[flink] Lake-batch fallback should use KV snapshot instead of reading log from earliest
- 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.
### Description
Follow-up to #3296.
In batch mode on a lake-enabled PK table with no lake snapshot yet, the fallback reads every bucket's log from EARLIEST, ignoring existing Fluss KV snapshots. This is OOM-prone on large never-tiered tables.
Spark fixed the equivalent in #3317 with per-bucket dispatch (snapshot+tail where KV snapshot exists, log-only otherwise).
Flink can't 1:1 port because `FlinkSourceSplitReader` has no sort-merge on the Fluss-only path. We may reuse `LakeSnapshotAndLogSplitScanner` (already used by Flink's lake reader, already sort-merges per PK) for the fallback or do smth similar
### Willingness to contribute
- [ ] I'm willing to submit a PR!
Contributor guide
No contributing guide indexed for this repository
Research direction
Start by inspecting FlinkSourceSplitReader and the existing LakeSnapshotAndLogSplitScanner used by the Flink lake reader. Compare the fallback described here with the per-bucket Spark fix in #3317 and the preceding context in #3296. Done means batch fallback uses available KV snapshots with only the remaining log tail, while buckets without snapshots retain log-only behavior.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- java
- Domain
- stream-processing
- Issue type
- Bug
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Quiet
- Clarity
- Mostly clear
- Newbie friendliness
- 50/100