[Bug] Spark batch read on primary key table returns a data hole (missing newest rows)

[beryllw]

Search before asking

• I searched in the issues and found nothing similar.

Fluss version

0.9.0 (latest release)

Please describe the bug 🐞

Batch-reading a primary key table from Spark can silently drop the newest rows, producing a "data hole". For a table continuously written with monotonically increasing primary keys, this query:

SELECT COUNT(*) AS total_count,
       MAX(seq_id) AS max_seq,
       MIN(seq_id) AS min_seq
FROM fluss_catalog.fluss.fluss_fault_test_pk;

returns a total_count smaller than max_seq - min_seq + 1 — a whole batch of the latest records is missing.

This is a plain primary key table; datalake tiering is NOT enabled. Reading the same data from Flink (with a filter + limit) returns the rows correctly, so the data does exist in Fluss — only the Spark batch read loses it.

How to reproduce

1. Create a primary key table (no datalake).
2. Write a batch of rows (e.g. seq_id 1..100) and let a KV snapshot be taken.
3. Append a new batch whose keys are all greater than the snapshot max key
   (e.g. seq_id 101..200); these rows stay in the log tail.
4. Run the aggregation query above from Spark.

Expected: [200, 200, 1]
Actual: [100, 100, 1] ← rows 101..200 are missing.

Root cause

Reading a primary key table merges the KV snapshot (sorted by pk) with the log tail (change log) via SortMergeReader. The merge was driven solely by the snapshot iterator: once all snapshot rows were consumed it stopped, dropping the trailing change log records whose keys are greater than the max snapshot key (they were put back during sort-merge to wait for a larger snapshot row that never comes).

Spark's non-lake batch reader FlussUpsertPartitionReader calls readBatch() exactly once, so it hits this directly. The lake scanner calls readBatch() repeatedly and happened to emit the tail on a later call, which is why datalake-enabled tables and Flink are not affected.

Solution

No response

Are you willing to submit a PR?

• I'm willing to submit a PR!