[client] Fix data hole in primary key table batch read#3627
Open
beryllw wants to merge 1 commit into
Open
Conversation
beryllw
force-pushed
the
spark-kv-read-bugfix
branch
from
a0d4bdb to
cd0aaf5
Compare
SortMergeReader merged the snapshot and change log driven solely by the snapshot iterator, so once the snapshot was exhausted the trailing change log records (keys greater than the max snapshot key) were silently dropped. Spark's non-lake primary key batch read (FlussUpsertPartitionReader) calls readBatch() only once and hit this directly, producing a data hole where COUNT(*) < max_seq - min_seq + 1. Rewrite SortMergeReader as a single two-way merge iterator so the merged output is independent of how many times readBatch() is called (Spark calls it once, the lake scanner calls it repeatedly). Add a unit test covering change log keys beyond the max snapshot key and an end-to-end Spark ITCase.
beryllw
force-pushed
the
spark-kv-read-bugfix
branch
from
cd0aaf5 to
0497f35
Compare
Member
|
cc @Yohahaha could you help to review this? |
Yohahaha
approved these changes
Jul 9, 2026
Contributor
There was a problem hiding this comment.
Pull request overview
This PR fixes a correctness issue in Fluss primary-key table batch reads where trailing change-log records (with keys greater than the max snapshot key) could be silently dropped, causing “data holes” in Spark batch reads and inconsistent results depending on how the reader is consumed.
Changes:
- Rewrites
SortMergeReaderinto a single two-way merge iterator over snapshot + change log so trailing log records are emitted in-order and not dropped. - Adds a
SortMergeReaderunit test that validates change-log rows beyond the snapshot’s max key are emitted. - Adds a Spark UT reproducing/guarding against the monotonic-key “data hole” scenario.
Reviewed changes
Copilot reviewed 3 out of 3 changed files in this pull request and generated 2 comments.
| File | Description |
|---|---|
| fluss-spark/fluss-spark-ut/src/test/scala/org/apache/fluss/spark/SparkPrimaryKeyTableReadTest.scala | Adds Spark regression test asserting monotonic primary-key batch reads have no missing tail rows. |
| fluss-client/src/test/java/org/apache/fluss/client/table/scanner/SortMergeReaderTest.java | Adds unit test ensuring change-log records beyond snapshot max key are emitted. |
| fluss-client/src/main/java/org/apache/fluss/client/table/scanner/SortMergeReader.java | Replaces prior snapshot-driven merge with a two-way merge iterator over snapshot and change-log streams. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
Comment on lines
+202
to
206
| @Override | ||
| public void close() { | ||
| snapshotRecordIterator.close(); | ||
| changeLogIterator.close(); | ||
| } |
Comment on lines
+86
to
+89
| /** | ||
| * Returns the merged-row iterator, or {@code null} when nothing is left. The same instance is | ||
| * returned on every call, so repeated invocations keep draining the snapshot and change log. | ||
| */ |
wuchong
approved these changes
Jul 10, 2026
wuchong
left a comment
wuchong
left a comment
Member
There was a problem hiding this comment.
LGTM
I think the Copilots' review comment is not critical, because existing callers either fully consume the iterator before closing it or close the scanner when terminating early.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Purpose
Linked issue: close #3626
Brief change log
COUNT(*) < max_seq - min_seq + 1.SortMergeReader: the merge was driven by the snapshot iterator and stopped once the snapshot was exhausted, discarding trailing change log records. It surfaces on the non-lake Spark batch read path (FlussUpsertPartitionReader), which callsreadBatch()only once.Rewrite
SortMergeReaderas a single two-way merge iterator over the snapshot stream and the change log stream (peek both heads, advance the smaller key; for an equal key the change log overrides the snapshot; a delete removes the row).The merged output is now identical regardless of how many timesreadBatch()is called — Spark calls it once, the lake scanner calls it repeatedly — which removes the structural cause of the dropped tail and also unifies projection handling. The public API is unchanged, so no caller needs to be modified.Tests
API and Format
Documentation