在使用Flink CDC时，遇到以下报错，是否意味着在Flink SQL中使用CDC只能原样读取呢？

在使用Flink CDC时，遇到以下报错，是否意味着在Flink SQL中使用CDC只能原样读取和写入数据，而不能对数据进行处理后再写入？
Flink SQL> insert into dc_ods_air_daily_city_air_data_airmaster_log select * from dc_ods_air_daily_city_air_data_airmaster;
[ERROR] Could not execute SQL statement. Reason:
java.lang.IllegalStateException: please declare primary key for sink table when query contains update/delete record.