[Feature] Support anchor-based merge in starting phase for chain table streaming read

[juntaozhang]

Search before asking

• I searched in the issues and found nothing similar.

Motivation

Chain table streaming read is designed around a lightweight starting phase for fast job startup. However, the starting snapshot can be produced in two ways today:

1. Lightweight mode (default): fast, but each partition is read independently.
2. After compact_chain_table: the snapshot branch already contains merged state, but there is an operational gap between the delta change and the next compaction. During that gap, the starting snapshot remains unmerged.

merge-snapshot in starting phase fills this gap: it gives users an explicit option to trade heavier startup cost for a starting snapshot that already reflects cross-branch deletes and updates, using the same merge logic as batch mode. The extra startup cost can be reduced by compact_chain_table. This is useful for workloads that require a fully reconciled initial state before processing subsequent incremental delta changes.

Another bugfix

Lightweight Phase 1 emits +I[100, …, 20250901] from the snapshot split.
The delta split's -D[100, …, 20250902] is dropped because there is no matching +I for the same key within that delta partition.
As a result, new downstream consumers see the stale +I and never observe the deletion;
the record only disappears after compact_chain_table merges the deletes into the snapshot branch and the streaming job is restarted.

Solution

ALTER TABLE t SET ('chain-table.streaming.merge-snapshot' = 'true');

• Default: false — keeps the existing lightweight Phase 1 behavior unchanged.
• When enabled: Phase 1 performs anchor-based chain merging. For each group, it merges the latest snapshot partition with delta partitions whose chain key is strictly greater than the snapshot chain key. This produces a correct starting snapshot that already reflects cross-branch deletes and updates.
• The trade-off is a heavier startup scan, which can be mitigated by periodically running compact_chain_table.

Anything else?

Are you willing to submit a PR?

• I'm willing to submit a PR!