以前我只是粗略地知道，Flink SQL 的回撤机制是为了保证数据正确而引入的；没想到它还会带来一些不易察觉的副作用。在实时盯盘这类对数据很敏感的场景里，需要花额外的功夫处理：变更日志乱序、中间错误结果暴露等。这篇文章讨论这些问题和解法。

回撤机制：为什么需要「作废」

批 SQL 查询一次，得到一个静态的结果。流 SQL 则是在不断变化的输入上持续维护一个结果：上游一变，下游已经发出去的旧结果就错了，必须有一种机制告诉下游「我之前说的那条作废」。这就是回撤的本质：它不是性能优化，而是流式 SQL 维护正确性的最小代价。

来看一个统计「词频的分布」的 SQL：先数每个词出现几次，再数「出现 N 次的词有几个」。

SELECT cnt, COUNT(cnt) AS freq
FROM (
    SELECT word, COUNT(*) AS cnt
    FROM words
    GROUP BY word
) t
GROUP BY cnt;

假设依次来了三个词：Hello、World、Hello。

关键在第三行：Hello 的次数从 1 变成 2，内层不能只发一条「Hello 现在是 2」，否则外层会以为又多了一个出现 2 次的词，却不知道「出现 1 次的词」要少一个。所以内层发出两条消息：先发 -U（update_before，撤回旧值），再发 +U（update_after，新值）。下游先减去旧值、再加上新值，最终结果才是正确的。

下面两张动图对比了有无回撤机制时的结果：

动图只能看，下面这个实验可以自己一步步输入：切换「发回撤」和「不发回撤」，看外层结果从哪一步开始出错；也留意每一步里外层结果怎样变——-U 和 +U 之间那一刻的结果，就是后面要说的中间结果。

回撤保证了最终结果的正确，但代价不小。表面上看只是一条消息变成了两条、数据量膨胀；更严重的是，Flink 是逐条事件流水线处理、单条事件原子的，本应原子完成的一次更新被拆成了两步，就会出现传统事务里「读未提交」的问题：下游可能读到两步之间那个错误的中间结果。同时，在分布式环境里这两条消息还可能乱序，必须另外保证最终一致。

Flink 为什么不用一条「复合的更新消息」？主要有两个原因：- 序列化简单：拆开后，无论哪种变更类型，消息结构都一样；- 分布式 Shuffle：即使用复合的 UPDATE 消息，在 Join、聚合等需要按 key 重新分发的环节，旧值和新值的 key 可能不同、要发往不同的节点，仍然得拆成 DELETE 和 INSERT，不如全程保持一致。

回撤带来的两类问题，解决的责任方不一样：

变更日志乱序：引擎来解决

看一个乱序的例子：

两个源表做外键关联后输出。由于关联键和 s2 的主键不一致，数据经过分布式 Shuffle 后，下游 Sink 算子可能收到三种顺序：

• 情况 1：和顺序处理时的消息序列相同，最终结果符合预期：(id=1, level=20, attr='b1')；

• 情况 2、情况 3：乱序了，如果不加处理，外部存储里 id=1 这一行会被错误地删除。

为此，Flink 引入了 SinkUpsertMaterializer 算子，在写出前对变更日志重新排序，保证最终结果正确。原理详见阿里云文档 Changelog 事件乱序处理原理。

不过，SinkUpsertMaterializer 的「回退」语义需要保留每个 key 的完整历史才能正确处理，有严重的性能瓶颈。社区在 FLIP-558 中把选择权交还给用户：提出 ON CONFLICT 语法，显式声明冲突时怎么处理，从而解决状态膨胀的问题。

中间错误结果：业务侧来承受

比起乱序，中间结果暴露给用户的体感更强：监控大盘上本该单调递增的指标，突然往下掉了一下。FLIP-558 也承认了这个语义缺陷，提出用水位线作为分布式屏障做压缩（永远不在 -U 和 +U 之间输出，永远不越过记录），用一定的延迟换取内部的一致性。但在这之前，只能靠业务侧的手段处理。

现象

• 聚合时指标下降：求和算子收到更新时，更新被拆成了 -U 和 +U。收到 -U 时，它先撤回之前的结果，再发出撤回之后的结果，于是本应直接从 3 变成 4 的指标，中间出现了一个错误的 2，收到 +U 之后才变成 4。

• 关联时关系为空：类似地，Join 算子收到 -U 后会撤回之前关联上的结果，并发出一条「没有关联上」的消息，下游查询时就会突然看不到数据。

在 Flink 中开启 MiniBatch 可以很大程度上缓解这类中间错误结果：攒一小批再处理，-U 和 +U 往往在同一批里相互抵消。但对数据非常敏感的场景（如实时监控），还需要从根本上杜绝。下面介绍两种方法。

方法一：写入时用版本列挡住

实时结果一般写入 Doris 这样的 OLAP 引擎供查询。Doris 的主键模型支持更新，Flink 写入时默认会忽略 -U 消息（由 sink.ignore.update-before 参数控制）。再把需要保证递增的字段设置为 Sequence 列，Doris 就只允许大的值覆盖小的值，中间那个变小的错误结果写不进去。

Doris 版本列的问题是不够灵活：只能按一个字段比较。实际场景中，经过复杂的关联、聚合后，往往既要保证多个指标递增、又要保证多个关系不丢，很难只选出一个版本字段。

Apache Paimon 的部分列更新可以定义多个 Sequence Group，每组字段各用自己的版本字段，能更灵活地处理多个字段都要保证递增的场景。

方法二：在输出前用自定义算子过滤

最简单的做法是在输出的最外层再套一层聚合：需要保持递增的指标取 MAX，其他指标取 LAST_VALUE。

更灵活的是 Flink 2.1 引入的 Process Table Function（PTF）：在 SQL 里自定义算子，能拿到每条消息的变更类型，也能使用自定义状态。

用它可以这样处理：

• 自定义聚合：收到 -U 时只更新内部的聚合状态、不输出结果，只在 +I / +U 时输出更新后的值，就不会出现指标突降；

• 自定义关联：直接忽略 -U，只输出最新的关联结果，就不会出现关联为空的消息。

具体实现可以参考 QFlink 项目中的示例。

三种做法对比

总结

这篇文章从回撤机制出发，分别从引擎和业务两个角度讨论了它带来的问题和解法。从业界流式计算的发展看，原子可见的单位要大于单条记录，才是流式 SQL 正确性的真正底线：

在 AI Coding 的时代，为什么还要用 SQL？SQL 是为了让人更容易上手开发，现在 AI 可以写代码了，直接生成原生的 DataStream 代码，没有回撤问题，不是更简单吗？我的回答是：SQL 生态不是一天建成的，软件都是分层的，上层要为下层屏蔽复杂性。Flink SQL 只是还需要变得更好而已。

参考资料

• 深入分析 Flink SQL 工作机制

• 阿里云 Flink 文档：Changelog 事件乱序处理原理

• Jamie Brandon，Internal consistency in streaming systems

• FLIP-558: Improvements to SinkUpsertMaterializer and changelog disorder

• Flink 文档：Process Table Functions