beryllw opened a new issue, #3626:
URL: https://github.com/apache/fluss/issues/3626

   ### Search before asking
   
   - [x] I searched in the [issues](https://github.com/apache/fluss/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?
   
   - [x] I'm willing to submit a PR!


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: [email protected]

For queries about this service, please contact Infrastructure at:
[email protected]

Reply via email to