This is an automated email from the ASF dual-hosted git repository.
luoyuxia pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/fluss.git
The following commit(s) were added to refs/heads/main by this push:
new 52c16789b [client] Fix data hole in primary key table batch read
(#3627)
52c16789b is described below
commit 52c16789b8b3676e4b467eebe8aee53ca2bd3094
Author: Junbo Wang <[email protected]>
AuthorDate: Sat Jul 11 10:50:17 2026 +0800
[client] Fix data hole in primary key table batch read (#3627)
---
.../client/table/scanner/SortMergeReader.java | 388 +++++++--------------
.../client/table/scanner/SortMergeReaderTest.java | 37 ++
.../fluss/spark/SparkPrimaryKeyTableReadTest.scala | 29 ++
3 files changed, 195 insertions(+), 259 deletions(-)
diff --git
a/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/SortMergeReader.java
b/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/SortMergeReader.java
index 088981090..f48a9c28d 100644
---
a/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/SortMergeReader.java
+++
b/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/SortMergeReader.java
@@ -26,31 +26,33 @@ import org.apache.fluss.utils.CloseableIterator;
import javax.annotation.Nullable;
-import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.Iterator;
import java.util.List;
import java.util.NoSuchElementException;
import java.util.PriorityQueue;
-import java.util.function.Function;
/**
- * A sort merge reader to merge snapshot record and change log. Both of them
should have the same
- * primary key encoding when comparing the keys.
+ * A sort merge reader that merges snapshot records and change log records
into a single stream
+ * sorted by primary key. Both inputs must share the same primary key encoding.
+ *
+ * <p>The merge is driven by a single {@link MergeIterator} (a two-way merge).
For the same key the
+ * change log wins over the snapshot: an update overrides the value, a delete
removes the row. Using
+ * one iterator keeps the result identical regardless of how many times {@link
#readBatch()} is
+ * called, so trailing change log records (keys greater than the max snapshot
key) are never
+ * dropped.
*/
public class SortMergeReader {
private final ProjectedRow snapshotProjectedPkRow;
private final CloseableIterator<LogRecord> snapshotRecordIterator;
private final Comparator<InternalRow> userKeyComparator;
- private CloseableIterator<KeyValueRow> changeLogIterator;
-
- private final SnapshotMergedRowIteratorWrapper
snapshotMergedRowIteratorWrapper;
-
- private final ChangeLogIteratorWrapper changeLogIteratorWrapper;
+ private final CloseableIterator<KeyValueRow> changeLogIterator;
private @Nullable final ProjectedRow projectedRow;
+ private @Nullable MergeIterator mergeIterator;
+
public SortMergeReader(
@Nullable int[] projectedFields,
int[] pkIndexes,
@@ -78,23 +80,129 @@ public class SortMergeReader {
this.snapshotRecordIterator =
ConcatRecordIterator.wrap(snapshotRecordIterators,
userKeyComparator, pkIndexes);
this.changeLogIterator = changeLogIterator;
- this.changeLogIteratorWrapper = new ChangeLogIteratorWrapper();
- this.snapshotMergedRowIteratorWrapper = new
SnapshotMergedRowIteratorWrapper();
- // to project to fields provided by user
this.projectedRow = projectedFields == null ? null :
ProjectedRow.from(projectedFields);
}
+ /**
+ * 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.
+ */
@Nullable
public CloseableIterator<InternalRow> readBatch() {
- if (!snapshotRecordIterator.hasNext()) {
- return changeLogIterator.hasNext()
- ? changeLogIteratorWrapper.replace(changeLogIterator)
- : null;
- } else {
- CloseableIterator<SortMergeRows> mergedRecordIterator =
- transform(snapshotRecordIterator,
this::sortMergeWithChangeLog);
-
- return
snapshotMergedRowIteratorWrapper.replace(mergedRecordIterator);
+ if (mergeIterator == null) {
+ mergeIterator = new MergeIterator();
+ }
+ return mergeIterator.hasNext() ? mergeIterator : null;
+ }
+
+ /** Two-way merge iterator over the snapshot and change log streams. */
+ private class MergeIterator implements CloseableIterator<InternalRow> {
+
+ // peeked head of each stream; null means not peeked yet or drained
+ private @Nullable LogRecord pendingSnapshot;
+ private @Nullable KeyValueRow pendingLog;
+
+ // next row to return (before projection) and whether it has been
computed
+ private @Nullable InternalRow nextRow;
+ private boolean nextComputed;
+
+ private @Nullable LogRecord peekSnapshot() {
+ if (pendingSnapshot == null && snapshotRecordIterator.hasNext()) {
+ pendingSnapshot = snapshotRecordIterator.next();
+ }
+ return pendingSnapshot;
+ }
+
+ private @Nullable KeyValueRow peekLog() {
+ if (pendingLog == null && changeLogIterator.hasNext()) {
+ pendingLog = changeLogIterator.next();
+ }
+ return pendingLog;
+ }
+
+ @Override
+ public boolean hasNext() {
+ if (!nextComputed) {
+ nextRow = advance();
+ nextComputed = true;
+ }
+ return nextRow != null;
+ }
+
+ @Override
+ public InternalRow next() {
+ if (!hasNext()) {
+ throw new NoSuchElementException();
+ }
+ InternalRow row = nextRow;
+ nextRow = null;
+ nextComputed = false;
+ return projectedRow == null ? row : projectedRow.replaceRow(row);
+ }
+
+ /** Advances to the next merged row, or {@code null} when both streams
are drained. */
+ private @Nullable InternalRow advance() {
+ while (true) {
+ LogRecord snapshot = peekSnapshot();
+ KeyValueRow log = peekLog();
+
+ if (snapshot == null && log == null) {
+ return null;
+ }
+
+ // only the change log remains
+ if (snapshot == null) {
+ InternalRow row = takeLog(log);
+ if (row != null) {
+ return row;
+ }
+ continue;
+ }
+
+ InternalRow snapshotRow = snapshot.getRow();
+
+ // only the snapshot remains
+ if (log == null) {
+ pendingSnapshot = null;
+ return snapshotRow;
+ }
+
+ int compareResult =
+ userKeyComparator.compare(
+
snapshotProjectedPkRow.replaceRow(snapshotRow), log.keyRow());
+ if (compareResult < 0) {
+ pendingSnapshot = null;
+ return snapshotRow;
+ } else if (compareResult > 0) {
+ InternalRow row = takeLog(log);
+ if (row != null) {
+ return row;
+ }
+ continue;
+ } else {
+ // same key: change log overrides snapshot, delete drops
both
+ pendingSnapshot = null;
+ InternalRow row = takeLog(log);
+ if (row != null) {
+ return row;
+ }
+ continue;
+ }
+ }
+ }
+
+ /**
+ * Consumes the change log head; returns its value row, or {@code
null} if it is a delete.
+ */
+ private @Nullable InternalRow takeLog(KeyValueRow log) {
+ pendingLog = null;
+ return log.isDelete() ? null : log.valueRow();
+ }
+
+ @Override
+ public void close() {
+ snapshotRecordIterator.close();
+ changeLogIterator.close();
}
}
@@ -168,242 +276,4 @@ public class SortMergeReader {
return priorityQueue.peek().next();
}
}
-
- private SortMergeRows sortMergeWithChangeLog(InternalRow lakeSnapshotRow) {
- // no log record, we return the snapshot record
- if (!changeLogIterator.hasNext()) {
- return new SortMergeRows(lakeSnapshotRow);
- }
- KeyValueRow logKeyValueRow = changeLogIterator.next();
- // now, let's compare with the snapshot row with log row
- int compareResult =
- userKeyComparator.compare(
- snapshotProjectedPkRow.replaceRow(lakeSnapshotRow),
- logKeyValueRow.keyRow());
- if (compareResult == 0) {
- // record of snapshot is equal to log, but the log record is
delete,
- // we shouldn't emit record
- if (logKeyValueRow.isDelete()) {
- return SortMergeRows.EMPTY;
- } else {
- // return the log record
- return new SortMergeRows(logKeyValueRow.valueRow());
- }
- }
-
- // the snapshot record is less than the log record, emit the
- // snapshot record
- if (compareResult < 0) {
- // need to put back the log record to log iterator to make the log
record
- // can be advanced again
- changeLogIterator =
- SingleElementHeadIterator.addElementToHead(logKeyValueRow,
changeLogIterator);
- return new SortMergeRows(lakeSnapshotRow);
- } else {
- // snapshot record > log record
- // we should emit the log record firsts; and still need to
iterator changelog to find
- // the first change log greater than the snapshot record
- List<InternalRow> emitRows = new ArrayList<>();
- // only emit the log record if it's not a delete operation
- if (!logKeyValueRow.isDelete()) {
- emitRows.add(logKeyValueRow.valueRow());
- }
- boolean shouldEmitSnapshotRecord = true;
- while (changeLogIterator.hasNext()) {
- // get the next log record
- logKeyValueRow = changeLogIterator.next();
- // compare with the snapshot row,
- compareResult =
- userKeyComparator.compare(
-
snapshotProjectedPkRow.replaceRow(lakeSnapshotRow),
- logKeyValueRow.keyRow());
- // if snapshot record < the log record
- if (compareResult < 0) {
- // we can break the loop
- changeLogIterator =
- SingleElementHeadIterator.addElementToHead(
- logKeyValueRow, changeLogIterator);
- break;
- } else if (compareResult > 0) {
- // snapshot record > the log record
- // the log record should be emitted
- if (!logKeyValueRow.isDelete()) {
- emitRows.add(logKeyValueRow.valueRow());
- }
- } else {
- // log record == snapshot record
- // the log record should be emitted if is not delete, but
the snapshot record
- // shouldn't be emitted
- if (!logKeyValueRow.isDelete()) {
- emitRows.add(logKeyValueRow.valueRow());
- }
- shouldEmitSnapshotRecord = false;
- }
- }
-
- if (shouldEmitSnapshotRecord) {
- emitRows.add(lakeSnapshotRow);
- }
- return new SortMergeRows(emitRows);
- }
- }
-
- private static class ChangeLogIteratorWrapper implements
CloseableIterator<InternalRow> {
- private CloseableIterator<KeyValueRow> changeLogRecordIterator;
- private KeyValueRow nextReturnRow;
-
- public ChangeLogIteratorWrapper() {}
-
- public ChangeLogIteratorWrapper replace(
- CloseableIterator<KeyValueRow> changeLogRecordIterator) {
- this.changeLogRecordIterator = changeLogRecordIterator;
- this.nextReturnRow = null;
- return this;
- }
-
- @Override
- public void close() {
- if (changeLogRecordIterator != null) {
- changeLogRecordIterator.close();
- }
- }
-
- @Override
- public boolean hasNext() {
- if (nextReturnRow != null) {
- return true;
- }
- while (changeLogRecordIterator != null &&
changeLogRecordIterator.hasNext()) {
- KeyValueRow row = changeLogRecordIterator.next();
- if (!row.isDelete()) {
- nextReturnRow = row;
- return true;
- }
- }
- return false;
- }
-
- @Override
- public InternalRow next() {
- if (nextReturnRow == null) {
- throw new NoSuchElementException();
- }
- KeyValueRow row = nextReturnRow;
- nextReturnRow = null; // Clear cache after consuming
- return row.valueRow();
- }
- }
-
- private class SnapshotMergedRowIteratorWrapper implements
CloseableIterator<InternalRow> {
- private CloseableIterator<SortMergeRows> currentLakeSnapshotRecords;
-
- private @Nullable Iterator<InternalRow> currentMergedRows;
-
- // the row to be returned
- private @Nullable InternalRow returnedRow;
-
- public SnapshotMergedRowIteratorWrapper replace(
- CloseableIterator<SortMergeRows> currentLakeSnapshotRecords) {
- this.currentLakeSnapshotRecords = currentLakeSnapshotRecords;
- this.returnedRow = null;
- this.currentMergedRows = null;
- return this;
- }
-
- @Override
- public void close() {
- currentLakeSnapshotRecords.close();
- }
-
- @Override
- public boolean hasNext() {
- if (returnedRow != null) {
- return true;
- }
- try {
- // Loop to skip empty merge results (e.g., when a snapshot
record is
- // deleted by changelog) and advance to the next non-empty
result.
- while (returnedRow == null) {
- if (currentMergedRows == null) {
- if (!currentLakeSnapshotRecords.hasNext()) {
- return false;
- }
-
- SortMergeRows sortMergeRows =
currentLakeSnapshotRecords.next();
- if (!sortMergeRows.mergedRows.isEmpty()) {
- currentMergedRows =
sortMergeRows.mergedRows.iterator();
- } else {
- // If mergedRows is empty (e.g., record was
deleted), continue
- // the loop to try the next snapshot record.
- continue;
- }
- }
-
- if (currentMergedRows.hasNext()) {
- returnedRow = currentMergedRows.next();
- } else {
- currentMergedRows = null;
- }
- }
- return true;
- } catch (Exception e) {
- throw new RuntimeException(e);
- }
- }
-
- @Override
- public InternalRow next() {
- InternalRow returnedRow =
- projectedRow == null
- ? this.returnedRow
- : projectedRow.replaceRow(this.returnedRow);
- // now, we can set the internalRow to null,
- // if no any row remain in current merged row, set the
currentMergedRows to null
- // to enable fetch next merged rows
- this.returnedRow = null;
- if (currentMergedRows != null && !currentMergedRows.hasNext()) {
- currentMergedRows = null;
- }
- return returnedRow;
- }
- }
-
- private static class SortMergeRows {
- private static final SortMergeRows EMPTY = new
SortMergeRows(Collections.emptyList());
-
- // the rows merge with change log, one snapshot row may advance
multiple change log
- private final List<InternalRow> mergedRows;
-
- public SortMergeRows(List<InternalRow> mergedRows) {
- this.mergedRows = mergedRows;
- }
-
- public SortMergeRows(InternalRow internalRow) {
- this.mergedRows = Collections.singletonList(internalRow);
- }
- }
-
- private <R> CloseableIterator<R> transform(
- CloseableIterator<LogRecord> originElementIterator,
- final Function<InternalRow, R> function) {
- return new CloseableIterator<R>() {
- private final CloseableIterator<LogRecord> inner =
originElementIterator;
-
- @Override
- public void close() {
- inner.close();
- }
-
- @Override
- public boolean hasNext() {
- return inner.hasNext();
- }
-
- @Override
- public R next() {
- LogRecord element = inner.next();
- return function.apply(element.getRow());
- }
- };
- }
}
diff --git
a/fluss-client/src/test/java/org/apache/fluss/client/table/scanner/SortMergeReaderTest.java
b/fluss-client/src/test/java/org/apache/fluss/client/table/scanner/SortMergeReaderTest.java
index 5ce97d5f6..4bffb250b 100644
---
a/fluss-client/src/test/java/org/apache/fluss/client/table/scanner/SortMergeReaderTest.java
+++
b/fluss-client/src/test/java/org/apache/fluss/client/table/scanner/SortMergeReaderTest.java
@@ -137,6 +137,43 @@ class SortMergeReaderTest {
.containsExactly(0, 1, 3, 4, 6, 7, 9);
}
+ @Test
+ void testReadBatchEmitsChangeLogBeyondSnapshotMaxKey() {
+ int keyIndex = 0;
+ int[] pkIndexes = new int[] {keyIndex};
+
+ // snapshot keys: 0, 1, 2
+ List<LogRecord> snapshotRecords = createRecords(0, 3, false);
+ // change log keys: 3, 4, 5, all inserts whose keys are greater than
the max
+ // snapshot key (2). This simulates a pk table with datalake tiering
where the
+ // newly-written rows (larger seq ids) are still in the Fluss log tail
while the
+ // older rows have been tiered into the lake snapshot.
+ List<KeyValueRow> changeLogRecords =
+ createRecords(3, 3, true).stream()
+ .map(logRecord -> new KeyValueRow(pkIndexes,
logRecord.getRow(), false))
+ .collect(Collectors.toList());
+
+ SortMergeReader sortMergeReader =
+ new SortMergeReader(
+ null,
+ pkIndexes,
+ CloseableIterator.wrap(snapshotRecords.iterator()),
+ new FlussRowComparator(keyIndex),
+ CloseableIterator.wrap(changeLogRecords.iterator()));
+
+ InternalRow.FieldGetter[] fieldGetters =
+ InternalRow.createFieldGetters(
+ RowType.of(new IntType(), new StringType(), new
StringType()));
+ List<InternalRow> actualRows;
+ try (CloseableIterator<InternalRow> iterator =
sortMergeReader.readBatch()) {
+ actualRows = materializeRows(iterator, fieldGetters);
+ }
+
+ // The change-log rows beyond the max snapshot key must not be dropped.
+ assertThat(actualRows.stream().map(row ->
row.getInt(0)).collect(Collectors.toList()))
+ .containsExactly(0, 1, 2, 3, 4, 5);
+ }
+
private CloseableIterator<InternalRow> projected(
CloseableIterator<LogRecord> originElementIterator, final
ProjectedRow projectedRow) {
return new CloseableIterator<InternalRow>() {
diff --git
a/fluss-spark/fluss-spark-ut/src/test/scala/org/apache/fluss/spark/SparkPrimaryKeyTableReadTest.scala
b/fluss-spark/fluss-spark-ut/src/test/scala/org/apache/fluss/spark/SparkPrimaryKeyTableReadTest.scala
index 4e30d552d..823521a18 100644
---
a/fluss-spark/fluss-spark-ut/src/test/scala/org/apache/fluss/spark/SparkPrimaryKeyTableReadTest.scala
+++
b/fluss-spark/fluss-spark-ut/src/test/scala/org/apache/fluss/spark/SparkPrimaryKeyTableReadTest.scala
@@ -519,6 +519,35 @@ class SparkPrimaryKeyTableReadTest extends
FlussSparkTestBase {
}
}
+ test("Spark Read: primary key table batch read has no data hole with
monotonic keys") {
+ withTable("fluss_fault_test_pk") {
+ val tablePath = createTablePath("fluss_fault_test_pk")
+ sql(s"""
+ |CREATE TABLE $DEFAULT_DATABASE.fluss_fault_test_pk (seq_id
BIGINT, payload STRING)
+ |TBLPROPERTIES("primary.key" = "seq_id", "bucket.num" = 1)
+ |""".stripMargin)
+
+ // First batch: seq_id 1..100, materialized into a kv snapshot.
+ val firstBatch = (1 to 100).map(i => s"($i, 'v$i')").mkString(", ")
+ sql(s"INSERT INTO $DEFAULT_DATABASE.fluss_fault_test_pk VALUES
$firstBatch")
+ flussServer.triggerAndWaitSnapshot(tablePath)
+
+ // Second batch: seq_id 101..200, only present in the log tail. All keys
are
+ // strictly greater than the max key in the snapshot, mimicking a
datagen job
+ // that keeps appending monotonically increasing primary keys.
+ val secondBatch = (101 to 200).map(i => s"($i, 'v$i')").mkString(", ")
+ sql(s"INSERT INTO $DEFAULT_DATABASE.fluss_fault_test_pk VALUES
$secondBatch")
+
+ // total_count must equal max_seq - min_seq + 1, i.e. no data hole.
+ checkAnswer(
+ sql(s"""
+ |SELECT COUNT(*) AS total_count, MAX(seq_id) AS max_seq,
MIN(seq_id) AS min_seq
+ |FROM $DEFAULT_DATABASE.fluss_fault_test_pk""".stripMargin),
+ Row(200L, 200L, 1L) :: Nil
+ )
+ }
+ }
+
private def partitionPredicate(df: DataFrame):
Option[org.apache.fluss.predicate.Predicate] = {
flussUpsertScan(df).flatMap(_.partitionPredicate)
}
