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]
