Streaming hybrid lake reads may skip splits and rescan after recovery
- 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