[flink] Batch Log Table query does not finish when column statistics filter out all records

[Gezi-lzq]

Thank you very much for your feedback!

Search before asking

• I searched in the issues and found nothing similar.

Fluss version

main (development)

Please describe the bug 🐞

What happened

When a Flink Batch job reads a Fluss Log Table with column statistics enabled, and the query predicate filters out all records in a bucket, the query returns no records but the Flink job remains in the RUNNING state.

The bounded batch read does not complete.

What I expected

The query should return an empty result, and the Flink Batch job should finish normally.

Steps to reproduce

Start Fluss and Flink, then run the following scenario:

1. Configure Flink to use batch mode with low parallelism:

   SET 'execution.runtime-mode' = 'batch';
   SET 'parallelism.default' = '1';

2. Create a Log Table with two buckets and statistics enabled for amount:

   CREATE TABLE stats_log (
       order_id BIGINT,
       amount INT
   ) WITH (
       'bucket.num' = '2',
       'table.statistics.columns' = 'amount'
   );

3. Write two records with positive values and ensure that the records are distributed across the two buckets, for example:

   amount = 10
   amount = 20
   

4. Run a query that matches no records:

   SELECT *
   FROM stats_log
   WHERE amount < 0;

5. Observe the result and the Flink job state.

6. Remove the table.statistics.columns configuration and repeat the same query with the same data as a control test.

Additional context

With table.statistics.columns = 'amount' enabled, the query returns no records, but the Flink job remains in the RUNNING state. Without this configuration, the same query returns Empty set and finishes normally.

Column statistics allow Fluss to skip batches that cannot match the predicate, returning no records while still advancing the consumed offset to the end. The Flink source reader checks split completion only from the last materialized record and ignores this progress-only result, so the split is never completed.

Solution

Use the consumed offset progress to determine whether a bounded split has reached its stopping offset, including when the scan returns no materialized records.

Are you willing to submit a PR?

• I'm willing to submit a PR!