apache / apache/fluss

Streaming hybrid lake reads may skip splits and rescan after recovery

Open
#3,931 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 in the [issues](https://github.com/apache/fluss/issues) and found nothing similar.

### Fluss version

main (development)

### Please describe the bug 🐞

Streaming union reads of a lake-enabled primary-key table use a
`LakeSnapshotAndFlussLogSplit` whose lake snapshot contains multiple inner
`LakeSplit`s. Two reader behaviors can break correct and efficient recovery:

1. `SeekableLakeSnapshotSplitScanner.pollBatch()` advances at most once when
the current inner split is exhausted. If the next inner split produces an
empty iterator, it returns `null`, which means end of input to the bounded
reader. Any later non-empty lake splits are skipped, and the reader switches
to the Fluss log at the snapshot's log-end offset, so the skipped snapshot
rows are not recovered.
2. `BoundedSplitReader.RecordAndPosBatch.replace()` initially stores the inner
lake split index, but `next()` calls the two-argument
`MutableRecordAndPos.setRecord()`, which resets the index to the default
value `0`. Consequently, checkpoints always restore the lake scan from
inner split 0 and skip a global record count, causing a complete physical
rescan of earlier inner splits.

A minimal first failure sequence is:

1. Plan three inner lake splits for one Fluss source split: A (non-empty), B
(empty after deletes or filtering), and C (non-empty).
2. Read A completely.
3. The scanner advances to B, observes no records, and returns `null`.
4. The bounded lake reader is considered finished, so C is never read.

Expected behavior:

- Empty inner lake splits are skipped until a non-empty split is found or all
inner splits are exhausted.
- Checkpointed positions contain the actual inner split index and a record
count relative to that split, allowing recovery to resume without rescanning
earlier inner splits.

Actual behavior:

- A non-empty split after an empty split can be omitted.
- The current inner split index is reset to 0 for every emitted record.

### Solution

- Make `SeekableLakeSnapshotSplitScanner` continue across consecutive empty
inner splits before reporting end of input.
- Preserve the batch's inner split index in every emitted `RecordAndPos` and
reset the read-record count when the inner split changes.
- Keep the restore skip count spanning split boundaries. This gives one restore
algorithm for both new positions `(actual index, split-local count)` and
existing checkpoints `(index 0, global count)`, without changing the split
state serializer version.
- Add unit tests for non-empty/empty/non-empty inner splits, split index
propagation, split-local counts, and recovery from the existing checkpoint
representation.

### 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 with SeekableLakeSnapshotSplitScanner.pollBatch() and BoundedSplitReader.RecordAndPosBatch.replace()/next(), including MutableRecordAndPos.setRecord(). Reproduce the non-empty/empty/non-empty sequence, then run or add the relevant unit tests for empty-split traversal, split-index propagation, split-local counts, and recovery from the existing checkpoint representation. Done means later non-empty splits are read and recovery avoids rescanning earlier splits while preserving compatibility.

Written by the indexing model from the issue text.

Assessment

Tech stack
java
Domain
data-engineering, stream-processing
Issue type
Bug
Difficulty
4/5
Estimated time
3-5 days
Activity status
Quiet
Clarity
Mostly clear
Newbie friendliness
65/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.