From 4d8ab1450dcf519cd7e0426850d17160ccf2de5f Mon Sep 17 00:00:00 2001 From: Gezi-lzq Date: Wed, 5 Aug 2026 16:04:20 +0000 Subject: [PATCH 1/4] [flink] Complete bounded log splits after filtered batches Use scanner progress to finish bounded log splits when filtering materializes no records. Add a batch regression test for statistics-filtered log tables. Fixes #3872 --- .../source/reader/FlinkSourceSplitReader.java | 5 ++++ .../source/FlinkTableSourceBatchITCase.java | 27 +++++++++++++++++ .../testutils/FlinkRowAssertionsUtils.java | 29 +++++++++++++++++++ 3 files changed, 61 insertions(+) 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..f106a8ad92b 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 @@ -468,6 +468,11 @@ private FlinkRecordsWithSplitIds forLogRecords(ScanRecords scanRecords) { splitIdByTableBucket.put(scanBucket, splitId); tableScanBuckets.add(scanBucket); List 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 diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/FlinkTableSourceBatchITCase.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/FlinkTableSourceBatchITCase.java index 3a80639ed63..5314f60f5fa 100644 --- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/FlinkTableSourceBatchITCase.java +++ b/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,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 collected = tEnv.executeSql(query).collect(); + assertThat(collectRowsUntilEndWithTimeout(collected)).isEmpty(); + } + @Test void testLakeTableQueryOnLakeDisabledTable() throws Exception { String tableName = prepareSourceTable(new String[] {"id", "name"}, null); diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/testutils/FlinkRowAssertionsUtils.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/testutils/FlinkRowAssertionsUtils.java index a489cc631b3..de3880f3bd9 100644 --- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/testutils/FlinkRowAssertionsUtils.java +++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/testutils/FlinkRowAssertionsUtils.java @@ -119,6 +119,35 @@ public static List collectBatchRows(CloseableIterator iterator) thr return actual; } + /** + * Collects all rows and fails if the Flink result iterator does not finish within one minute. + */ + public static List collectRowsUntilEndWithTimeout(CloseableIterator iterator) { + CompletableFuture> 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 collectRowsWithTimeout( CloseableIterator iterator, int expectedCount, From 68bf13ea32134881b26ce92919bd2b9a5a07b539 Mon Sep 17 00:00:00 2001 From: Gezi-lzq Date: Sun, 16 Aug 2026 10:58:35 +0000 Subject: [PATCH 2/4] [flink] Fix batch log scan with empty buckets --- .../source/reader/FlinkSourceSplitReader.java | 28 +++++++++++++++++-- .../source/FlinkTableSourceBatchITCase.java | 20 +++++++++++++ .../reader/FlinkSourceSplitReaderTest.java | 9 +++--- 3 files changed, 51 insertions(+), 6 deletions(-) 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 f106a8ad92b..1c0fa9b2097 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 @@ -178,7 +178,9 @@ public RecordsWithSplitIds 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 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; @@ -519,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 finishedSplitIds) { + Iterator> iterator = subscribedBuckets.entrySet().iterator(); + while (iterator.hasNext()) { + Map.Entry 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 toRecordAndPos( Iterator recordAndPosIterator) { return new CloseableIterator() { diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/FlinkTableSourceBatchITCase.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/FlinkTableSourceBatchITCase.java index 5314f60f5fa..96fa3df498d 100644 --- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/FlinkTableSourceBatchITCase.java +++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/FlinkTableSourceBatchITCase.java @@ -378,6 +378,26 @@ void testFilteredLogTableBatchScanCompletesWhenNoRecordsMatch() throws Exception 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 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); 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..8a94aa70f4f 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 @@ -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 subscribeSplits = Arrays.asList(split1, split2, split3); try (FlinkSourceSplitReader splitReader = @@ -352,9 +352,10 @@ void testSubscribeEmptySplits() throws Exception { // fetch records RecordsWithSplitIds 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()); } } From 2aa379944c3f5facc523ec37dd2a9775dddc7cb5 Mon Sep 17 00:00:00 2001 From: Gezi-lzq Date: Tue, 18 Aug 2026 09:26:10 +0000 Subject: [PATCH 3/4] [flink] Retire finished buckets in log record conversion --- .../fluss/flink/source/reader/FlinkSourceSplitReader.java | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) 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 1c0fa9b2097..a3157c792ac 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 @@ -178,9 +178,7 @@ public RecordsWithSplitIds fetch() throws IOException { return FlinkRecordsWithSplitIds.emptyRecords(); } ScanRecords scanRecords = logScanner.poll(POLL_TIMEOUT); - FlinkRecordsWithSplitIds records = forLogRecords(scanRecords); - removeFinishedSplits(records.finishedSplits()); - return records; + return forLogRecords(scanRecords); } } } @@ -518,6 +516,8 @@ public String next() { new FlinkRecordsWithSplitIds( splitRecords, splitIterator, tableScanBuckets.iterator(), finishedSplits); stoppingOffsets.forEach(recordsWithSplitIds::setTableBucketStoppingOffset); + // Build the current result before retiring finished buckets. + removeFinishedSplits(finishedSplits); return recordsWithSplitIds; } From 8595ec26a81d12ac809f627b82eabf81c1e79d28 Mon Sep 17 00:00:00 2001 From: Gezi-lzq Date: Tue, 18 Aug 2026 09:48:44 +0000 Subject: [PATCH 4/4] Revert "[flink] Retire finished buckets in log record conversion" This reverts commit 2aa379944c3f5facc523ec37dd2a9775dddc7cb5. --- .../fluss/flink/source/reader/FlinkSourceSplitReader.java | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) 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 a3157c792ac..1c0fa9b2097 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 @@ -178,7 +178,9 @@ public RecordsWithSplitIds fetch() throws IOException { return FlinkRecordsWithSplitIds.emptyRecords(); } ScanRecords scanRecords = logScanner.poll(POLL_TIMEOUT); - return forLogRecords(scanRecords); + FlinkRecordsWithSplitIds records = forLogRecords(scanRecords); + removeFinishedSplits(records.finishedSplits()); + return records; } } } @@ -516,8 +518,6 @@ public String next() { new FlinkRecordsWithSplitIds( splitRecords, splitIterator, tableScanBuckets.iterator(), finishedSplits); stoppingOffsets.forEach(recordsWithSplitIds::setTableBucketStoppingOffset); - // Build the current result before retiring finished buckets. - removeFinishedSplits(finishedSplits); return recordsWithSplitIds; }