[flink] Batch Log Table scan fix #3972

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<Long> 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).
+     *
+     * <p>{@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<TableBucket> removePartitions(Map<Long, String> 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<String> finishedSplits = new HashSet<>();
         Map<TableBucket, String> splitIdByTableBucket = new HashMap<>();
         List<TableBucket> tableScanBuckets = new ArrayList<>(scanRecords.buckets().size());
+        List<TableBucket> 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<TableBucket> buckets = tableScanBuckets.iterator();
         Iterator<String> splitIterator =
                 new Iterator<String>() {
@@ -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<String> finishedSplits =
                 (currentBoundedSplit instanceof HybridSnapshotLogSplit

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<RecordAndPos> 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<String> finishedSplits = new HashSet<>();
+            while (finishedSplits.isEmpty()) {
+                RecordsWithSplitIds<RecordAndPos> 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<RecordAndPos> records = splitReader.fetch();
+            assertThat(records.nextSplit()).isNull();
+            assertThat(records.finishedSplits()).isEmpty();
+        }
+    }
+
     // ------------------
 
     private void assignSplitsAndFetchUntilRetrieveRecords(