Flink CDC在flink-sql中不知道应该怎样对应了？

在 Flink SQL 中处理 JSON 格式的数据时，你可以使用 FLINK_TABLE_SCHEMA 或者 ROW 结构来描述 JSON 数据，并使用 SELECT * FROM ... 来读取数据。
如果你需要在 Flink SQL 中添加一个名为 __op 的虚拟列并将其值设置为某个固定值，你可以这样做：

INSERT INTO result_table SELECT *, '__op' AS __op FROM source_table;

这样，新的结果表 result_table 就包含了源表 source_table 所有的列，再加上一个新的虚拟列 __op，其值为字符串 '__op'。
请注意，由于 Flink SQL 目前尚不支持直接从 Kafka 读取数据，你需要先将 Kafka 的数据加载到一个中间表，然后再进行查询。例如：

CREATE TABLE source_table (data ROW <field1 STRING, field2 INT>) WITH (
  'connector.type' = 'kafka',
  ...
);

INSERT INTO result_table SELECT *, '__op' AS __op FROM source_table;

在Flink SQL中，您可以使用JSON_OBJECT函数来创建一个包含虚拟列的JSON对象。以下是一个示例：

SELECT JSON_OBJECT(
    'id', id,
    'name', name,
    'value', value,
    '__op', 'insert' -- 这里添加虚拟列__op，值为'insert'
) AS json_data
FROM kafka_source;

在这个示例中，我们从Kafka源表中选择id、name和value列，并使用JSON_OBJECT函数创建一个包含虚拟列__op的JSON对象。您可以根据需要修改这个示例，以适应您的具体需求。