Delta Query 查询改写优化

引擎版本

• 仅实时计算引擎 VVR 8.0.11 及以上版本支持 Delta Join。

• VVR 8.0.11 及以上版本支持 INNER Join，VVR 11.3 及以上版本支持 LEFT/RIGHT/FULL OUTER Join，VVR 11.4 及以上版本支持级联 Delta Join。

查询结构

• 仅支持改写流作业内的双流 Join 算子，不支持改写 Window Join、Interval Join、Temporal Join 和 Lookup Join。

• Join 条件至少要有一组等值条件，且等值条件覆盖的列在两侧都要命中源表的单个索引。

• Join Key 对应的索引列，从 Source 到 Join、以及级联 Join 之间不能参与任何计算。

• 从 Source 到 Join、以及级联 Join 之间不能包含 Rand() 等非确定性函数，避免反查时无法复现相同结果。

• 当前对 Delta Join 作业内的节点进行了限制，暂不支持白名单外的其他节点。

  • Source 到 Join 间、级联 Join 间，仅允许 Project 和 Filter 节点。源表上暂不支持定义 Watermark。定义 Watermark 会在 Source 之后引入 WatermarkAssigner 节点，不在当前白名单内。

  • Join 到 Sink 间仅允许 Project、Filter、Lookup（VVR 11.6 及以上版本支持）和 Union（VVR 11.8 及以上版本支持）节点。

  • 使用 BEGIN STATEMENT SET 同时写入多个 Sink 时：

    • VVR 11.8 以下版本，要求每一路写入都包含 Join 节点。

    • VVR 11.8 及以上版本，允许其中某一路不包含 Join 节点，但该路只能出现 Project、Filter、Lookup 和 Union 节点。

业务逻辑

• 可以容忍结果表无法物理删除数据

Delta Join 只向下游发送 +I和 +U，不发送 -U 和 -D。以过滤条件 WHERE l.amount + r.amount >= 100 为例，当某条 Join 记录从 120 掉到 80 时：

+U(..., 120) → 满足条件，写入结果表
+U(..., 80)  → 不满足条件，被过滤

由于没有 -U 撤旧结果，结果表里会残留 amount=120 的数据。

默认情况下，包含非唯一键过滤条件的 Query 会无法转化为 Delta Join。如果业务上能容忍数据残留时，可以设置 'table.optimizer.delta-join.ignore-non-unique-key-filter' = 'true' 跳过校验。

对于需要删除数据的场景，推荐使用逻辑删除，把过滤条件改写为标记列。以下面的查询为例，上游的每次变更都会覆盖写入结果表，is_delete 始终反映最新状态，下游查询时加上 WHERE is_delete = 0 即可。

INSERT INTO order_item_wide
SELECT
  o.merchant_id, o.order_id, o.item_id, i.item_name, o.amount,
  CASE WHEN o.amount >= 100 THEN 0 ELSE 1 END AS is_delete
FROM orders AS o
JOIN items AS i
  ON o.merchant_id = i.merchant_id AND o.item_id = i.item_id;

源表中的记录被删除时同理，删除消息不会传导到结果表，此前写入的关联结果会保留。需要让下游感知删除时，请使用上述的逻辑删除方案，由上游通过标记列表达删除状态。

• 需要保证主键和 Join Key 的关系不可变，避免分布式乱序问题

例如某张源表，主键为user_id列，表上对应的 Join Key 为user_name，一旦user_id为1和user_name为Jim的数据组成一条数据，不论后续user_id为1对应的数据如何变化，user_name都不能更改为其他值，如Sam。

步骤一：创建源表和结果表

两侧源表都要让 Join Key 命中索引。本例的 Join Key 是 (merchant_id, item_id)：在 orders 表中它是主键前缀，需要声明为 Bucket Key，作为可用的前缀索引；在 items 表中它就是主键，不需要额外配置。

CREATE TABLE `my-catalog`.`my_db`.`orders` (
  merchant_id BIGINT,                    -- 商家 ID
  item_id BIGINT,                        -- 商品 ID
  order_id BIGINT,                       -- 订单 ID
  amount DECIMAL(18, 2),                 -- 订单当前金额
  PRIMARY KEY (merchant_id, item_id, order_id) NOT ENFORCED
) WITH (
  'bucket.key' = 'merchant_id,item_id'
);

CREATE TABLE `my-catalog`.`my_db`.`items` (
  merchant_id BIGINT,                    -- 商家 ID
  item_id BIGINT,                        -- 商品 ID
  item_name STRING,                      -- 商品名称
  PRIMARY KEY (merchant_id, item_id) NOT ENFORCED   -- Join Key 与主键一致
);

CREATE TABLE `my-catalog`.`my_db`.`order_item_wide` (
  merchant_id BIGINT,
  order_id BIGINT,
  item_id BIGINT,
  item_name STRING,
  amount DECIMAL(18, 2),
  PRIMARY KEY (merchant_id, order_id, item_id) NOT ENFORCED
);

步骤二：编写 Join 作业

USE CATALOG `my-catalog`;
USE `my_db`;

-- 启用 Delta Join 改写
SET 'table.optimizer.delta-join.strategy' = 'EVENTUAL';

-- Delta Join 语句本身不需要引入专用语法
INSERT INTO order_item_wide
SELECT
  o.merchant_id,
  o.order_id,
  o.item_id,
  i.item_name,
  o.amount
FROM orders AS o
JOIN items AS i
  ON o.merchant_id = i.merchant_id AND o.item_id = i.item_id;

步骤三：验证优化生效

作业部署上线、启动后，在“状态总览”界面的作业拓扑图中，如果看到如下的 Delta Join 节点，则表明双流 Join 已经成功被改写为 Delta Join。

Flink 作业参数

Fluss 表参数

以下参数在 Fluss 建表的 WITH 子句中设置，也可以通过 SQL Hint 针对单个作业调整。

通过 Hint 调整时，Hint 需要写在被反查的表之后：

INSERT INTO order_item_wide
SELECT o.merchant_id, o.order_id, o.item_id, i.item_name, o.amount
FROM orders /*+ OPTIONS('client.lookup.queue-size' = '51200') */ AS o
JOIN items /*+ OPTIONS('client.lookup.queue-size' = '51200') */ AS i
  ON o.merchant_id = i.merchant_id AND o.item_id = i.item_id;

引擎版本

• 仅实时计算引擎 VVR 11.8 及以上版本支持 Delta Aggregate。

查询结构

• 查询结构

  • 仅支持改写流作业中的单层非窗口分组聚合，不支持窗口聚合和级联聚合。

  • Group Key 必须包含源表的单个索引。

  • Group Key 对应的索引列，从 Source 到 Aggregate 之间不能参与任何计算。

  • 从 Source 到 Aggregate 不能包含 Rand() 等非确定性函数，避免反查时无法复现相同结果。

  • 当前对 Delta Aggregate 作业内的节点进行了限制，暂不支持白名单外的其他节点。

    • Source 到 Aggregate 间，仅允许 Project、Filter 和 MiniBatchAssigner 节点。源表上暂不支持定义 Watermark。

    • Aggregate 到 Sink 间，仅允许 Project 和 Filter 节点。

    • 使用 BEGIN STATEMENT SET 同时写入多个 Sink 时每一路写入都必须包含 Aggregate 节点。

• 聚合函数

  • 当前支持 SUM、COUNT、AVG、MIN、MAX、LISTAGG、FIRST_VALUE 和 LAST_VALUE。

  • 暂不支持 DISTINCT、带 FILTER 子句的聚合、UDAF、Python 聚合函数，以及其他未列出的聚合函数。

业务逻辑

• 可以容忍结果表无法物理删除数据

Delta Aggregate 只向下游发送+U和-D，不发送-U。以过滤条件HAVING SUM(amount) >= 100 为例，当某个分组的SUM(amount)从 120 更新为 80 时：

+U(..., 120) → 满足条件，写入结果表
+U(..., 80)  → 不满足条件，被过滤

由于没有-U撤旧结果，结果表中仍会保留SUM(amount)=120的记录。

默认情况下，包含非唯一键过滤条件的 Query 会无法转化为 Delta Aggregate。如果业务上能容忍数据残留时，可以设置 'table.optimizer.delta-agg.ignore-non-unique-key-filter' = 'true' 跳过校验。

需要删除数据的场景，同样推荐使用逻辑删除，写法参见 Delta Join 的对应说明。

与 Delta Join 不同的是，当源表中的记录被删除时，Delta Aggregate 不丢弃删除消息，会产生正确的聚合结果且发送给下游。当分组内明细被删空时会向下游发送-D。

• 需要保证主键和 Group Key 的关系不可变，避免分布式乱序问题

例如某张源表，主键为user_id列，表上对应的 Group Key 为user_name，一旦user_id为1和user_name为Jim的数据组成一条数据，不论后续user_id为1对应的数据如何变化，user_name都不能更改为其他值，如Sam。

步骤一：创建源表和结果表

本例的 GROUP BY merchant_id 没有覆盖源表主键 (merchant_id, order_id)，但它是主键的前缀，因此声明为 Bucket Key，作为可用的前缀索引。

CREATE TABLE `my-catalog`.`my_db`.`orders` (
  merchant_id BIGINT,                    -- 商家 ID
  order_id BIGINT,                       -- 订单 ID
  amount DECIMAL(18, 2),                 -- 订单当前金额
  PRIMARY KEY (merchant_id, order_id) NOT ENFORCED
) WITH (
  'bucket.key' = 'merchant_id'           -- 按 merchant_id 反查该商家的全部订单
);

CREATE TABLE `my-catalog`.`my_db`.`merchant_summary` (
  merchant_id BIGINT,
  order_count BIGINT,                    -- 该商家当前订单数
  total_amount DECIMAL(38, 2),           -- 该商家当前成交总额，精度需高于源表字段以避免累加溢出
  PRIMARY KEY (merchant_id) NOT ENFORCED
);

步骤二：编写 Agg 作业

USE CATALOG `my-catalog`;
USE `my_db`;

-- 启用 Delta Aggregate 改写
SET 'table.optimizer.delta-agg.strategy' = 'EVENTUAL';

-- Delta Aggregate 语句本身不需要引入专用语法
INSERT INTO merchant_summary
SELECT
  merchant_id,
  COUNT(*) AS order_count,
  SUM(amount) AS total_amount
FROM orders
GROUP BY merchant_id;

步骤三：验证优化生效

作业部署上线、启动后，在“状态总览”界面的作业拓扑图中，如果看到如下的 Delta Aggregate 节点，则表明 Aggregate 已经成功被改写为 Delta Aggregate。

Flink 作业参数

Fluss 表参数

Delta Aggregate 的点查同样走 Fluss 客户端，可调整的参数与取值建议和 Delta Join 完全一致，请参见 Delta Join 章节的 Fluss 表参数。