用水位线列构建增量 ETL 加载

跟踪单调递增的水位线，只复制越过水位的行，并让水位推进与数据写入同事务从而可重跑。

目标

水位线把四小时的全量拷贝变成几秒钟的增量加载。三个不可妥协的点：先验证列的单调性；水位只推进到“实际已复制到目标库的最大值”而不是墙上时钟；用幂等 upsert 让崩溃后的重跑零成本。三者齐备，这个任务才配叫运维友好。

步骤要点

用 lag 窗口函数验证 updated_at 从不回退，计数为零才允许作为水位列，否则迟到更新会被永久漏掉。
建 etl_watermarks 表，每个管线一行存水位，加 FOR UPDATE 行锁串行化并发实例。
每批只复制 updated_at 大于水位且按时间排序的前五万行，用 ON CONFLICT 幂等 upsert 写入目标库。
批次结束后把水位推进到目标库中实际存在的最大 updated_at，绝不推进到 now()，崩溃时未复制的行依然高于水位。
定期做源与目标“截至水位行数”对比，两侧相等即证明增量加载无丢失。

排错提示

源端迟到回填旧时间戳的行会永远躺在水位之下，加每日按 id 哈希的全量比对或定期回填扫描；源端硬删除无法通过水位传播，用软删标志、CDC 日志或夜间反连接清理目标库缺失 id；两个实例并发时行锁让后者等待，引擎不支持就改用咨询锁或带心跳的租约列，批次保持小让等待短暂。

适用边界

水位正确性依赖源列单调性，没有数据库保证的顺序就必须有对账扫描兜底；相等时间戳跨批次的行需要大于等于语义或以唯一 id 决胜，否则会重复或漏行。

代码示例

-- nightly full copy (before):
--   4h, 40M rows, misses rows created mid-copy

-- incremental (after):
WITH wm AS (
  SELECT watermark FROM etl_watermarks
  WHERE pipeline = 'events' FOR UPDATE
)
INSERT INTO warehouse.events
SELECT * FROM source.events
WHERE updated_at > (SELECT watermark FROM wm)
ORDER BY updated_at LIMIT 50000
ON CONFLICT (id) DO UPDATE SET
  payload = EXCLUDED.payload;

-- then advance watermark to max(updated_at) IN DESTINATION
-- runtime: 8-40s, restart-safe, nothing lost