Flink SQL 撤回失败排查记录：更新键丢失与动态字段

一、问题背景

Flink SQL 的统计结果会随着源数据变化而更新，不是只向数据库追加新记录。

例如，一笔 30 元订单从支付方式发生了变化，正确场景如下：

记录旧支付方式统计
撤回旧分类的统计
写入新分类的统计
最终只保留一笔、30 元

如果旧统计没有正确撤回，就可能变成：

支付分类	笔数	金额
旧分类，错误残留	1	30
新分类	1	30
合计	2	60

那为什么会撤回失败就是本文要探究的问题

二、基本概念

2.1 Flink SQL 传递的是变更流

Flink 用 RowKind 表示消息含义：

类型	标记	含义
INSERT	`+I`	新增结果
UPDATE_BEFORE	`-U`	更新前的旧结果
UPDATE_AFTER	`+U`	更新后的新结果
DELETE	`-D`	删除结果

例如，一组金额从 30 变成 50，可以通过撤回旧结果、写入新结果来表达。

具体输出哪些变更类型，取决于执行计划和下游能力，不是每次更新都必须产生四种消息。

2.2 inputUpsertKey（更新键）标识的是结果的身份字段

以聚合为例：

  
SELECT
    uuid,
    storeId,
    tradeDate,
    pmSign,
    SUM(amount) AS saleAmt
FROM orders
GROUP BY storeId, tradeDate, pmSign

同一个“门店 + 日期 + 支付分类”只有一条当前聚合结果，因此规划器能够推导：

更新键 = storeId + tradeDate + pmSign

金额可以从 30 变成 50，但仍是同一组统计，所以金额不属于这条结果的身份。

如果最终输出字段为：

0：uuid
1：storeId
2：tradeDate
3：pmSign
4：saleAmt

对应配置就是：

inputUpsertKey=[1,2,3]

这里记录的是字段位置，不是额外生成一个数据 ID。规划时决定使用哪些字段，运行时比较这些字段的值。

其他 SQL，例如 JOIN、窗口和去重，有各自的键推导规则，不是所有查询都直接取 GROUP BY。

2.3 Sink 主键与输入更新键的功能差异

键	本例	作用
Sink 主键	`uuid`	最终操作结果表的哪条记录
输入更新键	`storeId + tradeDate + pmSign`	上游哪条结果正在更新或撤回

SinkUpsertMaterializer 的状态可以理解为：

先按 Sink 主键找到状态桶
再在桶内匹配需要更新或撤回的输入记录

一般情况下，同一个 Sink 主键可能对应不同的上游输入结果，因此不能把每一条撤回都无条件翻译成“删除数据库这个主键”。算子需要先确定撤回的是哪条输入，以及撤回后是否还有有效结果。

本例希望 UUID 与完整分组一一对应，但规划器不会自动证明 MD5(CONCAT(...)) 能够唯一代表全部分组。

目标表声明 PRIMARY KEY (uuid) NOT ENFORCED，也不等于上游计算表达式的唯一性已经得到证明。NOT ENFORCED 表示 Flink 不主动检查该约束；数据库的物理主键是另一个层面。

三、产生原因之一：更新键在 SELECT 中没有成功传递

注意：“更新键在 SELECT 中没有成功传递”只是导致FlinkSQL从更新键判断退回到整行匹配

3.1 不是 SELECT 与 GROUP BY 字段不一样，就一定丢失

准确判断标准是：

经过最终 SELECT 后，规划器是否还能确认至少一组输出字段能够唯一识别上游结果。

例如：

  
SELECT
    storeId,
    pmSign,
    SUM(amount) AS saleAmt,
    COUNT(*) AS orderNum
FROM orders
GROUP BY storeId, pmSign

SELECT 比 GROUP BY 多了金额和笔数，但完整分组键仍然保留，因此可以继续识别 storeId + pmSign。

3.2 哪些情况可能丢失？

情况 A：没有输出完整分组键

  
SELECT storeId, SUM(amount) AS saleAmt
FROM orders
GROUP BY storeId, pmSign

上游按“门店 + 支付分类”分组，最终只输出门店。仅凭门店不能区分不同支付分类，因此原来的完整分组键不能继续传递。

如果没有其他可推导的更新键，就会没有可用的 inputUpsertKey。

情况 B：计算使不同分组可能产生相同输出

  
SELECT storeId, UPPER(pmSign) AS pmSign, SUM(amount) AS saleAmt
FROM orders
GROUP BY storeId, pmSign

原来的 abc 和 ABC 是不同分组，转换后都可能输出 ABC。计算后的字段不一定能唯一识别原分组。

情况 C：值没有变化，但规划器无法识别字段映射

本次嵌套 CAST 属于这一类：日期值没有改变，完整分组也没有改变，但规划器未能传递日期的更新键信息。

因此，不能简单归结为 SELECT 与 GROUP BY 的业务语义不一致。

四、产生原因之二：整行匹配遇到动态字段

4.1 更新键丢失后如何匹配？

在本次实际存在的 SinkUpsertMaterializer 中：

有 inputUpsertKey：比较指定的更新键字段
没有 inputUpsertKey：比较整行，忽略 RowKind

忽略 RowKind，是因为保存的旧结果可能是 INSERT，到达的撤回可能是 DELETE，它们的变更类型不同，但可能对应同一条业务结果。

不是每个作业都需要这个算子；本文讨论的是规划器实际生成了该算子的链路。

4.2 撤回不一定是最终旧行的原样副本

举例如下

最终 SELECT 包含：

  
CURRENT_TIMESTAMP AS modifyDate

处理链路：

聚合产生业务结果或撤回
    ↓
最终 SELECT 计算 UUID、modifyDate 等
    ↓
materializer
    ↓
Sink

第一次输出：

聚合结果：金额30
SELECT 在 10:00:01 处理
最终旧行：金额30，modifyDate=10:00:01

随后撤回：

聚合产生旧业务结果的撤回：金额30
SELECT 在 10:00:02 再处理
最终撤回行（重新计算日期）：金额30，modifyDate=10:00:02

同一份旧业务内容再次经过 SELECT，动态字段会在这次处理时重新计算，导致撤回的消息和原始消息不匹配，最终导致撤销失败

4.3 匹配失败为什么会留下旧统计？

Flink 1.17.2 的 retractRow 中，找不到匹配行时直接返回：

  
final int index = findFirst(values, retract);
if (index == -1) {
    LOG.info(STATE_CLEARED_WARN_MSG);
    return;
}

后续移除状态、发送删除的逻辑没有执行。

撤回已到达 materializer
    ↓
没有匹配到旧行
    ↓
没有向 Sink 发送删除
    ↓
旧分类残留

这不是数据库主键删除失败，而是删除没有继续发出去。

该分支的日志会提示状态 TTL，但本例保留 48 小时 TTL、几秒内复现，不能仅凭提示认定状态真的过期。

五、本地复现与验证

代码就不贴了，直接贴流程和最终执行的日志
验证环境：Flink 1.17.2、JDK 8、本地 Flink 作业、测试内存主键 Sink

输入过程：

插入旧支付类型 01，金额 30。
等第一条结果实际写入 Sink。
等待 1200 毫秒。
撤回旧分类，插入新支付类型 03，金额仍为 30。
检查原比较器、真实状态及 Sink 是否实际收到 DELETE。

实际验证结果：

组别	SELECT与GROUP不一致	是否使用动态字段	更新键	原比较器匹配	最终记录数
A	不一致	使用	丢失	失败	2，旧数据残留
B	一致	使用	完整保留	成功	1
C	不一致	不使用	丢失	成功	1

动态字段就是指CURRENT_TIMESTAMP这类函数，每次就算都会是不同的值

完整输出日志如下

A-no-key-current-time

  
字段顺序：[uuid, storeId, tradeDate, pmSign, saleAmt, orderNum, modifyDate]
inputUpsertKey=[]
对应字段=[]

进入真实 materializer：+I[0ad542e355eb4b1a67047d801af5d091, 10, 2026-10-09, 01, 30.0000, 1, 2026-10-12 15:45:23.618]
处理前已保存的行：[]
处理后已保存的行：[+I[0ad542e355eb4b1a67047d801af5d091, 10, 2026-10-09, 01, 30.0000, 1, 2026-10-12 15:45:23.618]]

进入真实 materializer：-D[0ad542e355eb4b1a67047d801af5d091, 10, 2026-10-09, 01, 30.0000, 1, 2026-10-12 15:45:24.902]
处理前已保存的行：[+I[0ad542e355eb4b1a67047d801af5d091, 10, 2026-10-09, 01, 30.0000, 1, 2026-10-12 15:45:23.618]]
原 equaliser 是否匹配：false
处理后已保存的行：[+I[0ad542e355eb4b1a67047d801af5d091, 10, 2026-10-09, 01, 30.0000, 1, 2026-10-12 15:45:23.618]]

进入真实 materializer：+I[3b5a8b978863b58d9901ca07028a3e92, 10, 2026-10-09, 03, 30.0000, 1, 2026-10-12 15:45:24.902]
处理前已保存的行：[]
处理后已保存的行：[+I[3b5a8b978863b58d9901ca07028a3e92, 10, 2026-10-09, 03, 30.0000, 1, 2026-10-12 15:45:24.902]]

实际到达 Sink 的消息：[+I[0ad542e355eb4b1a67047d801af5d091, 10, 2026-10-09, 01, 30.0000, 1, 2026-10-12 15:45:23.618], +I[3b5a8b978863b58d9901ca07028a3e92, 10, 2026-10-09, 03, 30.0000, 1, 2026-10-12 15:45:24.902]]
最终 Sink 记录：[+I[0ad542e355eb4b1a67047d801af5d091, 10, 2026-10-09, 01, 30.0000, 1, 2026-10-12 15:45:23.618], +I[3b5a8b978863b58d9901ca07028a3e92, 10, 2026-10-09, 03, 30.0000, 1, 2026-10-12 15:45:24.902]]
最终记录数：2

B-with-key-current-time

  
字段顺序：[uuid, storeId, tradeDate, pmSign, saleAmt, orderNum, modifyDate]
inputUpsertKey=[1, 2, 3]
对应字段=[storeId, tradeDate, pmSign]

进入真实 materializer：+I[0ad542e355eb4b1a67047d801af5d091, 10, 2026-10-09, 01, 30.0000, 1, 2026-10-12 15:45:26.294]
处理前已保存的行：[]
处理后已保存的行：[+I[0ad542e355eb4b1a67047d801af5d091, 10, 2026-10-09, 01, 30.0000, 1, 2026-10-12 15:45:26.294]]

进入真实 materializer：-D[0ad542e355eb4b1a67047d801af5d091, 10, 2026-10-09, 01, 30.0000, 1, 2026-10-12 15:45:27.601]
处理前已保存的行：[+I[0ad542e355eb4b1a67047d801af5d091, 10, 2026-10-09, 01, 30.0000, 1, 2026-10-12 15:45:26.294]]
原 equaliser 是否匹配：true
处理后已保存的行：[]

进入真实 materializer：+I[3b5a8b978863b58d9901ca07028a3e92, 10, 2026-10-09, 03, 30.0000, 1, 2026-10-12 15:45:27.601]
处理前已保存的行：[]
处理后已保存的行：[+I[3b5a8b978863b58d9901ca07028a3e92, 10, 2026-10-09, 03, 30.0000, 1, 2026-10-12 15:45:27.601]]

实际到达 Sink 的消息：[+I[0ad542e355eb4b1a67047d801af5d091, 10, 2026-10-09, 01, 30.0000, 1, 2026-10-12 15:45:26.294], -D[0ad542e355eb4b1a67047d801af5d091, 10, 2026-10-09, 01, 30.0000, 1, 2026-10-12 15:45:27.601], +I[3b5a8b978863b58d9901ca07028a3e92, 10, 2026-10-09, 03, 30.0000, 1, 2026-10-12 15:45:27.601]]
最终 Sink 记录：[+I[3b5a8b978863b58d9901ca07028a3e92, 10, 2026-10-09, 03, 30.0000, 1, 2026-10-12 15:45:27.601]]
最终记录数：1

C-no-key-fixed-time

  
字段顺序：[uuid, storeId, tradeDate, pmSign, saleAmt, orderNum, modifyDate]
inputUpsertKey=[]
对应字段=[]

进入真实 materializer：+I[0ad542e355eb4b1a67047d801af5d091, 10, 2026-10-09, 01, 30.0000, 1, fixed-time]
处理前已保存的行：[]
处理后已保存的行：[+I[0ad542e355eb4b1a67047d801af5d091, 10, 2026-10-09, 01, 30.0000, 1, fixed-time]]

进入真实 materializer：-D[0ad542e355eb4b1a67047d801af5d091, 10, 2026-10-09, 01, 30.0000, 1, fixed-time]
处理前已保存的行：[+I[0ad542e355eb4b1a67047d801af5d091, 10, 2026-10-09, 01, 30.0000, 1, fixed-time]]
原 equaliser 是否匹配：true
处理后已保存的行：[]

进入真实 materializer：+I[3b5a8b978863b58d9901ca07028a3e92, 10, 2026-10-09, 03, 30.0000, 1, fixed-time]
处理前已保存的行：[]
处理后已保存的行：[+I[3b5a8b978863b58d9901ca07028a3e92, 10, 2026-10-09, 03, 30.0000, 1, fixed-time]]

实际到达 Sink 的消息：[+I[0ad542e355eb4b1a67047d801af5d091, 10, 2026-10-09, 01, 30.0000, 1, fixed-time], -D[0ad542e355eb4b1a67047d801af5d091, 10, 2026-10-09, 01, 30.0000, 1, fixed-time], +I[3b5a8b978863b58d9901ca07028a3e92, 10, 2026-10-09, 03, 30.0000, 1, fixed-time]]
最终 Sink 记录：[+I[3b5a8b978863b58d9901ca07028a3e92, 10, 2026-10-09, 03, 30.0000, 1, fixed-time]]
最终记录数：1

三组分别证明：

A：更新键丢失与动态时间变化叠加，撤回失败。
B：动态时间本身不必然导致失败。
C：整行匹配本身也不必然失败。

六、参考资料

Flink 动态表与更新流：流式结果如何需要新增、更新和撤回