[Feature] Support anchor-based merge in starting phase for chain table streaming read
- Dominant language
- Java
- Stars
- 3.4k
- Forks
- 1.4k
- Avg merge
- 1d 11h
- Merged PRs (30d)
- 396
Description
### Search before asking
- [x] I searched in the [issues](https://github.com/apache/paimon/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?
- [x] I'm willing to submit a PR!
Contributor guide
No contributing guide indexed for this repository
Research direction
Start by tracing chain table streaming read's lightweight Phase 1 and the existing batch merge logic, then locate where the chain-table.streaming.merge-snapshot option should be applied. Verify that enabling it merges the latest snapshot with eligible delta partitions and preserves the default behavior when disabled; reproduce the stale deletion scenario described in the issue to validate the result.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- java
- Domain
- data-engineering, distributed-systems, stream-processing
- Issue type
- Feature
- Difficulty
- 5/5
- Estimated time
- Over a week
- Activity status
- Quiet
- Clarity
- Mostly clear
- Newbie friendliness
- 45/100