apache / apache/paimon

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

Open
#8,720 0 comments 0 reactions 0 assignees View on GitHub
enhancement
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

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.