Flink CDC在A表物理删除，有没有办法在W里面插入个逻辑删除？

在Flink CDC SQL Client中，INSERT INTO W SELECT * FROM A 的场景下，如果表 A 中发生了物理删除操作（即直接从数据库中删除某一行），Flink CDC可以通过捕获 RowKind.DELETE 事件感知到这一变更。然而，默认情况下，这种物理删除会直接反映为下游表 W 中对应行的移除。如果您希望将物理删除改写为逻辑删除（例如通过设置一个标志位字段来标记删除状态），可以通过以下方式实现。

解决方案：使用逻辑删除字段

为了在目标表 W 中插入逻辑删除记录，您需要对 Flink SQL 进行扩展处理，具体步骤如下：

1. 修改目标表结构

在目标表 W 中添加一个逻辑删除标志字段（如 is_deleted），用于标识某条记录是否被逻辑删除。例如：

CREATE TABLE W (
    id INT,
    name STRING,
    is_deleted BOOLEAN, -- 逻辑删除标志字段
    PRIMARY KEY (id) NOT ENFORCED
) WITH (
    'connector' = '...', -- 根据实际使用的连接器配置
    ...
);

2. 使用 Flink 的 ROW 和 CASE 表达式改写逻辑

在 INSERT INTO W SELECT * FROM A 的查询中，通过 Flink SQL 的 ROW 和 CASE 表达式，将捕获到的 RowKind.DELETE 事件改写为逻辑删除记录。示例如下：

INSERT INTO W
SELECT 
    id,
    name,
    CASE 
        WHEN ROW_KIND = 'DELETE' THEN TRUE -- 如果是删除事件，设置 is_deleted 为 TRUE
        ELSE FALSE                         -- 否则设置为 FALSE
    END AS is_deleted
FROM A;

3. 配置 Flink CDC 捕获删除事件

确保 Flink CDC 能够正确捕获 RowKind.DELETE 事件。这通常需要在源表 A 的配置中启用变更数据捕获（CDC）功能。例如，对于 MySQL CDC 源表，可以使用以下配置：

CREATE TABLE A (
    id INT,
    name STRING,
    PRIMARY KEY (id) NOT ENFORCED
) WITH (
    'connector' = 'mysql-cdc',
    'hostname' = 'localhost',
    'port' = '3306',
    'username' = 'root',
    'password' = 'password',
    'database-name' = 'test_db',
    'table-name' = 'A'
);

4. 验证逻辑删除效果

在目标表 W 中，逻辑删除的效果表现为新增一条记录，其中 is_deleted 字段被设置为 TRUE。您可以通过以下查询验证结果：

SELECT * FROM W WHERE is_deleted = TRUE;

注意事项

• 主键要求：目标表 W 必须定义主键，以确保逻辑删除能够正确更新对应的记录。
• 性能影响：逻辑删除会增加目标表的数据量，可能对查询性能产生一定影响，请根据实际需求权衡。
• 兼容性：某些目标存储系统（如 ClickHouse）对 UPDATE 和 DELETE 操作支持有限，可能导致逻辑删除性能下降。建议在设计时充分考虑目标存储的特性。

总结

通过在目标表中添加逻辑删除标志字段，并结合 Flink SQL 的 CASE 表达式改写删除事件，您可以实现将物理删除转换为逻辑删除的功能。这种方法不仅保留了删除记录的历史信息，还避免了直接移除数据带来的数据丢失风险。