apache / apache/fluss

[flink] Union reads restart from earliest when all lake splits of a partition are pruned

Closed
#4,325 0 comments 0 reactions 0 assignees View on GitHub
Dominant language
Java
Stars
2.1k
Forks
625
Avg merge
3d 14h
Merged PRs (30d)
97

Description

### Search before asking

- [x] I searched existing issues and pull requests and found no report of this pruning case.

### Fluss version

main (`d920be6d059de2e97c3f41b641bcb8c5d1fe0dbd`), 1.0-SNAPSHOT; Flink 1.20.3 and Paimon 2.0.0.

### Description

When a predicate prunes all lake splits of an active Fluss partition, `LakeSplitGenerator` discards that partition's committed lake offsets. The generated Fluss splits start from `EARLIEST_OFFSET`, rereading retained log records already covered by the lake snapshot. This affects both primary-key and log tables.

In `generatePartitionTableSplit()`, partitions absent from the planned lake splits reach the remaining-Fluss-partitions loop. That loop passes `Collections.emptyMap()` to `generateSplit()`, even when the lake snapshot contains offsets for their buckets.

The partition handling introduced in #1527 covers partitions retained only in the lake after Fluss partition TTL. With predicate pushdown, an active partition can also have no planned lake splits because its files do not match the query. In that case, the snapshot offsets still define where to start reading the Fluss increment.

### Reproduction

A planner-level regression uses one active partition with two buckets and an empty result from the lake planner:

| Bucket | Snapshot log offset | Stopping offset | Expected start | Actual start |
| --- | --- | --- | --- | --- |
| 0 | 100 | 110 | 100 | EARLIEST_OFFSET |
| 1 | absent | 20 | EARLIEST_OFFSET | EARLIEST_OFFSET |

Both primary-key and log-table cases reproduce the mismatch. The observed impact is redundant historical reads; no data loss was observed in the reproduction.

### Solution

Pass the snapshot offset map when planning the remaining Fluss partitions. Keep the existing earliest-offset fallback for buckets without a committed offset, and preserve the stopping offsets.

### Are you willing to submit a PR?

Yes.

Contributor guide

No contributing guide indexed for this repository

Research direction

Start in generatePartitionTableSplit() and trace how the remaining-Fluss-partitions loop calls generateSplit() when the lake planner returns no splits. Use the described planner-level regression with one active partition and two buckets, covering both primary-key and log tables; done means committed snapshot offsets are used where present, EARLIEST_OFFSET remains the fallback, and stopping offsets are preserved.

Written by the indexing model from the issue text.

Assessment

Tech stack
java
Domain
stream-processing
Issue type
Bug
Difficulty
3/5
Estimated time
1-2 days
Activity status
Active
Clarity
Clearly specified
Newbie friendliness
76/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.