[flink] Complete bounded log splits after filtered batches

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

@@ -468,6 +468,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

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

@@ -50,6 +50,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;
@@ -349,6 +350,32 @@ 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 testLakeTableQueryOnLakeDisabledTable() throws Exception {
         String tableName = prepareSourceTable(new String[] {"id", "name"}, null);

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,