[flink] Fix hybrid lake split recovery

[luoyuxia]

Purpose

Linked issue: close #3931

Streaming hybrid reads of lake-enabled primary-key tables can stop the lake snapshot when an empty inner LakeSplit occurs before a later non-empty split. They also reset the emitted inner split index to 0, so checkpoint recovery rescans earlier lake splits.

This change makes the lake snapshot complete and records a resumable split-local position. The restore algorithm remains compatible with existing checkpoints that contain split index 0 and a global records-to-skip count.

Brief change log

• Continue scanning across empty inner lake splits until data is found or all splits are exhausted.
• Preserve the actual inner split index in each RecordAndPos and reset the read count when that index changes.
• Keep the restore skip count across inner split boundaries so both existing global positions and new split-local positions use the same recovery path.
• Add regression tests for empty inner splits, index propagation, split-local counts, and existing checkpoint recovery semantics.

Tests

• ./mvnw -nsu -pl fluss-flink/fluss-flink-common -Dtest=BoundedSplitReaderTest,SeekableLakeSnapshotSplitScannerTest test
• ./mvnw -nsu -pl fluss-flink/fluss-flink-common verify (560 unit tests and 56 integration tests passed)

API and Format

No API or storage format changes. The source split serializer version is unchanged; existing checkpoint positions remain readable and recover through the unified skip algorithm.

Documentation

No documentation changes are required for this bug fix.

[Copilot]

Pull request overview

Fixes recovery correctness for Flink hybrid reads of lake-enabled PK tables by (1) continuing lake snapshot scanning past empty inner LakeSplits and (2) persisting a resumable, split-local position (inner split index + per-split read count) so checkpoint restore doesn’t rescan earlier lake splits.

Changes:

• Make SeekableLakeSnapshotSplitScanner skip over empty inner lake splits instead of terminating the bounded scan early.
• Track and propagate the actual inner split index through RecordAndPos, resetting per-split read counters when the split index changes while keeping restore skip semantics compatible with legacy checkpoints.
• Add regression tests for empty-split scanning and both legacy/global and new split-local restore positions.

Reviewed changes

Copilot reviewed 4 out of 4 changed files in this pull request and generated no comments.

File	Description
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/reader/BoundedSplitReader.java	Tracks current inner split index, emits correct split-local positions, and keeps restore skip semantics compatible.
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/lake/reader/SeekableLakeSnapshotSplitScanner.java	Skips consecutive empty inner lake splits so later non-empty splits are still read.
fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/reader/BoundedSplitReaderTest.java	Adds tests for split index propagation and legacy vs split-local restore behavior across indexed batches.
fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/lake/reader/SeekableLakeSnapshotSplitScannerTest.java	Adds regression test ensuring empty inner lake splits don’t prematurely end scanning.

---

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.