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

   ### Search before asking
   
   - [x] I searched in the [issues](https://github.com/apache/fluss/issues) and 
found nothing similar.
   
   ### Fluss version
   
   main (development)
   
   Affected downstream build observed in 
`ververica-connector-fluss-1.20-vvr-11.6.0-jdk11`, and the same control flow 
exists in the current development branch.
   
   ### Please describe the bug 🐞
   
   When reading a primary-key table through the hybrid lake snapshot + Fluss 
log path, `LakeSnapshotAndLogSplitScanner` can create `SortMergeReader` with a 
null row comparator if the log scan is considered finished from the beginning.
   
   Triggering sequence:
   
   1. `LakeSnapshotAndLogSplitScanner` is constructed with `startingOffset >= 
stoppingOffset` or `stoppingOffset <= 0`.
   2. This sets `logScanFinished` to `true` before the first `pollBatch()`.
   3. The first `pollBatch()` enters the `logScanFinished` branch directly.
   4. That branch creates lake record iterators but does not initialize 
`rowComparator` from `SortedRecordReader.order()`.
   5. `SortMergeReader` receives `rowComparator == null`.
   6. If there are multiple lake snapshot iterators, `ConcatRecordIterator` 
adds them to a `PriorityQueue`, which triggers the comparator lambda and throws 
NPE.
   
   Observed stack trace excerpt:
   
   ```text
   Caused by: java.lang.NullPointerException
       at 
org.apache.fluss.client.table.scanner.SortMergeReader$ConcatRecordIterator.lambda$new$0(SortMergeReader.java:117)
       at java.util.PriorityQueue.siftUpUsingComparator(PriorityQueue.java:675)
       at java.util.PriorityQueue.offer(PriorityQueue.java:345)
       at java.util.PriorityQueue.add(PriorityQueue.java:326)
       at 
org.apache.fluss.client.table.scanner.SortMergeReader$ConcatRecordIterator.<init>(SortMergeReader.java:126)
       at 
org.apache.fluss.client.table.scanner.SortMergeReader$ConcatRecordIterator.wrap(SortMergeReader.java:136)
       at 
org.apache.fluss.client.table.scanner.SortMergeReader.<init>(SortMergeReader.java:79)
       at 
org.apache.fluss.client.table.scanner.batch.LakeSnapshotAndLogSplitScanner.pollBatch(LakeSnapshotAndLogSplitScanner.java:170)
   ```
   
   Expected behavior: the scanner should initialize the lake snapshot ordering 
comparator before constructing `SortMergeReader`, regardless of whether the log 
scan finishes before the first poll.
   
   Actual behavior: the initial `logScanFinished == true` branch bypasses 
`SortedRecordReader.order()` and can pass a null comparator to 
`SortMergeReader`.
   
   ### Solution
   
   Unify lake reader initialization in `LakeSnapshotAndLogSplitScanner` so both 
paths initialize `rowComparator` from `SortedRecordReader.order()` before 
creating `SortMergeReader`. Add a regression test for the `startingOffset == 
stoppingOffset` path with multiple lake splits.
   
   ### Are you willing to submit a PR?
   
   - [x] Im 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