[flink] Fix batch log scan with empty buckets

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/reader/FlinkSourceSplitReader.java

@@ -178,7 +178,9 @@ public RecordsWithSplitIds<RecordAndPos> fetch() throws IOException {
                     return FlinkRecordsWithSplitIds.emptyRecords();
                 }
                 ScanRecords scanRecords = logScanner.poll(POLL_TIMEOUT);
-                return forLogRecords(scanRecords);
+                FlinkRecordsWithSplitIds records = forLogRecords(scanRecords);
+                removeFinishedSplits(records.finishedSplits());
+                return records;
             }
         }
     }
@@ -258,7 +260,7 @@ private void subscribeLog(SourceSplitBase split, long startingOffset) {
             Optional<Long> stoppingOffsetOpt = logSplit.getStoppingOffset();
             if (stoppingOffsetOpt.isPresent()) {
                 Long stoppingOffset = stoppingOffsetOpt.get();
-                if (startingOffset >= stoppingOffset) {
+                if (startingOffset >= stoppingOffset || stoppingOffset == 0) {
                     // is empty log splits as no log record can be fetched
                     emptyLogSplits.add(split.splitId());
                     isEmptyLogSplit = true;
@@ -468,6 +470,11 @@ private FlinkRecordsWithSplitIds forLogRecords(ScanRecords scanRecords) {
             splitIdByTableBucket.put(scanBucket, splitId);
             tableScanBuckets.add(scanBucket);
             List<ScanRecord> bucketScanRecords = scanRecords.records(scanBucket);
+            Long consumedUpToOffset = scanRecords.consumedUpToOffset(scanBucket);
+            if (consumedUpToOffset != null && consumedUpToOffset >= stoppingOffset) {
+                stoppingOffsets.put(scanBucket, stoppingOffset);
+                finishedSplits.add(splitId);
+            }
             if (!bucketScanRecords.isEmpty()) {
                 final ScanRecord lastRecord = bucketScanRecords.get(bucketScanRecords.size() - 1);
                 // We keep the maximum message timestamp in the fetch for calculating lags
@@ -514,6 +521,28 @@ public String next() {
         return recordsWithSplitIds;
     }
 
+    /**
+     * Retire log buckets after building the corresponding finished result.
+     *
+     * @param finishedSplitIds split IDs that were reported as finished
+     */
+    private void removeFinishedSplits(Set<String> finishedSplitIds) {
+        Iterator<Map.Entry<TableBucket, String>> iterator = subscribedBuckets.entrySet().iterator();
+        while (iterator.hasNext()) {
+            Map.Entry<TableBucket, String> entry = iterator.next();
+            if (finishedSplitIds.contains(entry.getValue())) {
+                TableBucket tableBucket = entry.getKey();
+                if (tableBucket.getPartitionId() == null) {
+                    logScanner.unsubscribe(tableBucket.getBucket());
+                } else {
+                    logScanner.unsubscribe(tableBucket.getPartitionId(), tableBucket.getBucket());
+                }
+                stoppingOffsets.remove(tableBucket);
+                iterator.remove();
+            }
+        }
+    }
+
     private CloseableIterator<RecordAndPos> toRecordAndPos(
             Iterator<ScanRecord> recordAndPosIterator) {
         return new CloseableIterator<RecordAndPos>() {

fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/FlinkTableSourceBatchITCase.java

@@ -51,6 +51,7 @@
 
 import static org.apache.fluss.flink.FlinkConnectorOptions.BOOTSTRAP_SERVERS;
 import static org.apache.fluss.flink.source.testutils.FlinkRowAssertionsUtils.assertResultsIgnoreOrder;
+import static org.apache.fluss.flink.source.testutils.FlinkRowAssertionsUtils.collectRowsUntilEndWithTimeout;
 import static org.apache.fluss.flink.source.testutils.FlinkRowAssertionsUtils.collectRowsWithTimeout;
 import static org.apache.fluss.server.testutils.FlussClusterExtension.BUILTIN_DATABASE;
 import static org.apache.fluss.testutils.DataTestUtils.row;
@@ -351,6 +352,52 @@ void testScanFullLogTable(boolean partitionTable) throws Exception {
         assertResultsIgnoreOrder(collected, expected, true);
     }
 
+    @Test
+    void testFilteredLogTableBatchScanCompletesWhenNoRecordsMatch() throws Exception {
+        String tableName = String.format("test_filtered_log_table_%s", RandomUtils.nextInt());
+        tEnv.executeSql(
+                String.format(
+                        "create table %s (id int, name varchar) with ("
+                                + "'bucket.num' = '1', "
+                                + "'table.statistics.columns' = 'id')",
+                        tableName));
+
+        TablePath tablePath = TablePath.of(databaseName, tableName);
+        try (Table table = conn.getTable(tablePath)) {
+            AppendWriter appendWriter = table.newAppend().createWriter();
+            for (int i = 1; i <= 5; i++) {
+                appendWriter.append(row(i, "name" + i));
+            }
+            appendWriter.flush();
+        }
+
+        String query = String.format("SELECT * FROM %s WHERE id > 100", tableName);
+        assertThat(tEnv.explainSql(query)).contains("filter=[>(id, 100)]");
+
+        CloseableIterator<Row> collected = tEnv.executeSql(query).collect();
+        assertThat(collectRowsUntilEndWithTimeout(collected)).isEmpty();
+    }
+
+    @Test
+    void testBatchLogTableScanWithEmptyBucket() throws Exception {
+        tEnv.getConfig().set(ExecutionConfigOptions.TABLE_EXEC_RESOURCE_DEFAULT_PARALLELISM, 1);
+        String tableName = String.format("test_empty_bucket_log_table_%s", RandomUtils.nextInt());
+        tEnv.executeSql(
+                String.format(
+                        "create table %s (id int, name varchar) with ('bucket.num' = '2')",
+                        tableName));
+
+        try (Table table = conn.getTable(TablePath.of(databaseName, tableName))) {
+            AppendWriter appendWriter = table.newAppend().createWriter();
+            appendWriter.append(row(1, "alpha"));
+            appendWriter.flush();
+        }
+
+        CloseableIterator<Row> collected =
+                tEnv.executeSql(String.format("SELECT * FROM %s", tableName)).collect();
+        assertThat(collectRowsUntilEndWithTimeout(collected)).containsExactly("+I[1, alpha]");
+    }
+
     @Test
     void testLakeTableQueryOnLakeDisabledTable() throws Exception {
         String tableName = prepareSourceTable(new String[] {"id", "name"}, null);

fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/reader/FlinkSourceSplitReaderTest.java

@@ -340,10 +340,10 @@ void testSubscribeEmptySplits() throws Exception {
                         tablePath,
                         TableDescriptor.builder().schema(schema).distributedBy(3).build());
 
-        // create two empty splits with log start offset equal to end offset
+        // create empty splits, including a split starting from EARLIEST_OFFSET
         LogSplit split1 = new LogSplit(new TableBucket(tableId, 0), null, 0, 0);
         LogSplit split2 = new LogSplit(new TableBucket(tableId, 1), null, 0, 0);
-        LogSplit split3 = new LogSplit(new TableBucket(tableId, 2), null, EARLIEST_OFFSET);
+        LogSplit split3 = new LogSplit(new TableBucket(tableId, 2), null, EARLIEST_OFFSET, 0);
         List<SourceSplitBase> subscribeSplits = Arrays.asList(split1, split2, split3);
 
         try (FlinkSourceSplitReader splitReader =
@@ -352,9 +352,10 @@ void testSubscribeEmptySplits() throws Exception {
 
             // fetch records
             RecordsWithSplitIds<RecordAndPos> records = splitReader.fetch();
-            // finished splits should be split1,split2
+            // all empty splits should finish without subscribing to the log
             assertThat(records.finishedSplits())
-                    .containsExactlyInAnyOrder(split1.splitId(), split2.splitId());
+                    .containsExactlyInAnyOrder(
+                            split1.splitId(), split2.splitId(), split3.splitId());
         }
     }
 

fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/testutils/FlinkRowAssertionsUtils.java

@@ -119,6 +119,35 @@ public static List<String> collectBatchRows(CloseableIterator<Row> iterator) thr
         return actual;
     }
 
+    /**
+     * Collects all rows and fails if the Flink result iterator does not finish within one minute.
+     */
+    public static List<String> collectRowsUntilEndWithTimeout(CloseableIterator<Row> iterator) {
+        CompletableFuture<List<String>> future =
+                CompletableFuture.supplyAsync(
+                        () -> {
+                            try {
+                                return collectBatchRows(iterator);
+                            } catch (Exception e) {
+                                throw new RuntimeException(e);
+                            }
+                        },
+                        EXECUTOR);
+        try {
+            return future.get(1, TimeUnit.MINUTES);
+        } catch (TimeoutException e) {
+            future.cancel(true);
+            try {
+                iterator.close();
+            } catch (Exception ignored) {
+                // The timeout is the test failure we need to report.
+            }
+            throw new AssertionError("Flink job did not finish within one minute.", e);
+        } catch (Exception e) {
+            throw new RuntimeException("Failed to collect Flink job results.", e.getCause());
+        }
+    }
+
     protected static List<String> collectRowsWithTimeout(
             CloseableIterator<Row> iterator,
             int expectedCount,