如何在Flink SQL中对Upsert（非仅追加）表执行区间连接？

核心问题说明 #

Flink原生区间连接仅支持**仅追加（Append-Only）**表，因为Upsert流（如CDC同步的带主键表）的更新会导致区间连接的状态维护逻辑冲突，因此规划器会直接拒绝你的原查询。结合你的场景（会话1小时后不再更新、无新订单），可以通过以下两种方案实现需求，同时确保状态自动清理。

方案一：Temporal Join + 状态TTL（推荐） #

Temporal Join原生支持Upsert流的关联，配合状态TTL可自动清理过期会话状态，完全匹配你的业务规则：

1. 配置状态TTL #

先通过SQL设置会话状态的过期时间，确保1小时后自动清理：

SET table.exec.state.ttl = '1 h';

2. 编写Temporal Join查询 #

使用FOR SYSTEM_TIME AS OF语法关联订单下单时刻的会话最新版本，同时加上时间区间过滤：

SELECT 
    o.order_id,
    o.session_id,
    s.user_id,
    o.order_time,
    s.start_time
FROM orders o
JOIN sessions FOR SYSTEM_TIME AS OF o.order_time s
  ON o.session_id = s.session_id
WHERE 
    o.order_time BETWEEN s.start_time AND s.start_time + INTERVAL '1' HOUR;

关键逻辑说明

• FOR SYSTEM_TIME AS OF o.order_time：确保订单关联的是下单时刻的会话最新版本，适配会话可能的更新操作
• 时间过滤条件：严格限制订单仅关联生命周期内的会话
• 状态TTL：1小时后，未更新的会话状态会被自动清理，避免状态膨胀

方案二：转Append-Only流后使用区间连接 #

如果一定要用区间连接语法，可先将会话表转为仅追加流（去重保留最新版本），再执行区间连接：

1. 生成去重后的会话仅追加流 #

CREATE VIEW sessions_append_only AS
SELECT 
    session_id,
    user_id,
    start_time
    -- 其他需要的字段
FROM (
    SELECT 
        *,
        ROW_NUMBER() OVER (PARTITION BY session_id ORDER BY start_time DESC) AS rn
    FROM sessions
) t
WHERE rn = 1;

2. 配置TTL并执行区间连接 #

SET table.exec.state.ttl = '1 h';

SELECT 
    o.order_id,
    o.session_id,
    s.user_id,
    o.order_time,
    s.start_time
FROM orders o
JOIN sessions_append_only s
  ON o.session_id = s.session_id
  AND o.order_time BETWEEN s.start_time AND s.start_time + INTERVAL '1' HOUR;

注意事项

• 该方案需额外做去重处理，不如Temporal Join高效
• 会话更新时会输出新的版本，旧版本状态会通过TTL自动清理，避免无效关联

关键配置验证 #

• 确保两张表的水印配置正确：WATERMARK FOR xxx AS xxx - INTERVAL '5' SECOND，保证时间窗口的乱序数据处理能力
• 状态TTL仅对会话表的状态生效，订单表的状态会因关联逻辑自动关联清理（无匹配会话的订单状态会被及时清理）

内容的提问来源于stack exchange，提问作者EgorBr