From b8274378a51bce5a61781f80b6ccebfe2e20fb67 Mon Sep 17 00:00:00 2001 From: Evan Date: Tue, 18 Aug 2026 10:46:56 +0200 Subject: [PATCH] fix for [flink] Batch Log Table scan fails when some buckets are empty #3972 Signed-off-by: Evan --- .../source/reader/FlinkSourceSplitReader.java | 42 +++++++++++- .../reader/FlinkSourceSplitReaderTest.java | 66 +++++++++++++++++++ 2 files changed, 107 insertions(+), 1 deletion(-) diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/reader/FlinkSourceSplitReader.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/reader/FlinkSourceSplitReader.java index 50e2143962c..5b223aa2b89 100644 --- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/reader/FlinkSourceSplitReader.java +++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/reader/FlinkSourceSplitReader.java @@ -258,7 +258,7 @@ private void subscribeLog(SourceSplitBase split, long startingOffset) { Optional stoppingOffsetOpt = logSplit.getStoppingOffset(); if (stoppingOffsetOpt.isPresent()) { Long stoppingOffset = stoppingOffsetOpt.get(); - if (startingOffset >= stoppingOffset) { + if (isEmptyLogSplit(startingOffset, stoppingOffset)) { // is empty log splits as no log record can be fetched emptyLogSplits.add(split.splitId()); isEmptyLogSplit = true; @@ -319,6 +319,19 @@ private void subscribeLog(SourceSplitBase split, long startingOffset) { } } + /** + * Whether no record can be read between {@code startingOffset} (inclusive) and {@code + * stoppingOffset} (exclusive). + * + *

{@code startingOffset} may still be the {@link LogScanner#EARLIEST_OFFSET} sentinel + * instead of a resolved offset, so it can only be compared when it is non-negative. A stopping + * offset of 0 means nothing has been written to the bucket yet, so the split is empty for any + * starting offset. + */ + private static boolean isEmptyLogSplit(long startingOffset, long stoppingOffset) { + return stoppingOffset == 0 || (startingOffset >= 0 && startingOffset >= stoppingOffset); + } + public Set removePartitions(Map removedPartitions) { // First, if the current active bounded split belongs to a removed partition and is not // LakeSnapshotSplit, finish it so it will not be restored. @@ -458,6 +471,7 @@ private FlinkRecordsWithSplitIds forLogRecords(ScanRecords scanRecords) { Set finishedSplits = new HashSet<>(); Map splitIdByTableBucket = new HashMap<>(); List tableScanBuckets = new ArrayList<>(scanRecords.buckets().size()); + List finishedBuckets = new ArrayList<>(); for (TableBucket scanBucket : scanRecords.buckets()) { long stoppingOffset = getStoppingOffset(scanBucket); String splitId = subscribedBuckets.get(scanBucket); @@ -480,10 +494,20 @@ private FlinkRecordsWithSplitIds forLogRecords(ScanRecords scanRecords) { if (lastRecord.logOffset() >= stoppingOffset - 1) { stoppingOffsets.put(scanBucket, stoppingOffset); finishedSplits.add(splitId); + finishedBuckets.add(scanBucket); } } splitRecords.put(splitId, toRecordAndPos(bucketScanRecords.iterator())); } + + // A finished split is unregistered by SourceReaderBase, while the log scanner would keep + // returning records appended to its bucket afterwards. Stop reading those buckets now, + // otherwise a later fetch reports records for an unregistered split, which makes + // SourceReaderBase fail with "Have records for a split that was not registered". + for (TableBucket finishedBucket : finishedBuckets) { + unsubscribeFinishedBucket(finishedBucket); + } + Iterator buckets = tableScanBuckets.iterator(); Iterator splitIterator = new Iterator() { @@ -546,6 +570,22 @@ private long getStoppingOffset(TableBucket tableBucket) { return stoppingOffsets.getOrDefault(tableBucket, Long.MAX_VALUE); } + /** + * Stops reading the log of {@code tableBucket} whose bounded split has reached its stopping + * offset, so that records appended afterwards are not fetched for the finished split anymore. + */ + private void unsubscribeFinishedBucket(TableBucket tableBucket) { + subscribedBuckets.remove(tableBucket); + stoppingOffsets.remove(tableBucket); + Long partitionId = tableBucket.getPartitionId(); + if (partitionId != null) { + logScanner.unsubscribe(partitionId, tableBucket.getBucket()); + } else { + logScanner.unsubscribe(tableBucket.getBucket()); + } + LOG.info("Unsubscribe to read log of bucket {} since its split is finished.", tableBucket); + } + private FlinkRecordsWithSplitIds finishCurrentBoundedSplit() throws IOException { Set finishedSplits = (currentBoundedSplit instanceof HybridSnapshotLogSplit diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/reader/FlinkSourceSplitReaderTest.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/reader/FlinkSourceSplitReaderTest.java index 5b0117a0f73..ab1b78881bf 100644 --- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/reader/FlinkSourceSplitReaderTest.java +++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/reader/FlinkSourceSplitReaderTest.java @@ -358,6 +358,72 @@ void testSubscribeEmptySplits() throws Exception { } } + @Test + void testSubscribeEmptySplitWithEarliestStartingOffset() throws Exception { + TablePath tablePath = TablePath.of(DEFAULT_DB, "test-subscribe-empty-split-earliest"); + Schema schema = + Schema.newBuilder() + .column("id", DataTypes.INT()) + .column("name", DataTypes.STRING()) + .build(); + long tableId = + createTable( + tablePath, + TableDescriptor.builder().schema(schema).distributedBy(1).build()); + + // the bucket is empty, so the stopping offset of the batch split is 0 while the starting + // offset is still the EARLIEST_OFFSET sentinel + LogSplit split = new LogSplit(new TableBucket(tableId, 0), null, EARLIEST_OFFSET, 0); + + try (FlinkSourceSplitReader splitReader = + createSplitReader(tablePath, schema.getRowType())) { + splitReader.handleSplitsChanges( + new SplitsAddition<>(Collections.singletonList((SourceSplitBase) split))); + + RecordsWithSplitIds records = splitReader.fetch(); + assertThat(records.finishedSplits()).containsExactly(split.splitId()); + assertThat(records.nextSplit()).isNull(); + } + } + + @Test + void testFinishedLogSplitIsNotFetchedAgain() throws Exception { + TablePath tablePath = TablePath.of(DEFAULT_DB, "test-finished-log-split"); + Schema schema = + Schema.newBuilder() + .column("id", DataTypes.INT()) + .column("name", DataTypes.STRING()) + .build(); + long tableId = + createTable( + tablePath, + TableDescriptor.builder().schema(schema).distributedBy(1).build()); + + appendRows(tablePath, 5); + LogSplit split = new LogSplit(new TableBucket(tableId, 0), null, 0, 5); + + try (FlinkSourceSplitReader splitReader = + createSplitReader(tablePath, schema.getRowType())) { + splitReader.handleSplitsChanges( + new SplitsAddition<>(Collections.singletonList((SourceSplitBase) split))); + + Set finishedSplits = new HashSet<>(); + while (finishedSplits.isEmpty()) { + RecordsWithSplitIds records = splitReader.fetch(); + finishedSplits.addAll(records.finishedSplits()); + records.recycle(); + } + assertThat(finishedSplits).containsExactly(split.splitId()); + + // records appended after the split finished must not be fetched for it anymore, + // SourceReaderBase has already unregistered the split at this point + appendRows(tablePath, 5); + RecordsWithSplitIds records = splitReader.fetch(); + assertThat(records.nextSplit()).isNull(); + assertThat(records.finishedSplits()).isEmpty(); + } + } + // ------------------ private void assignSplitsAndFetchUntilRetrieveRecords(