(fluss) branch main updated: [client] Initialize lake comparator before merge read (#3406)

[aborisov]

This is an automated email from the ASF dual-hosted git repository.

fresh-borzoni pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/fluss.git


The following commit(s) were added to refs/heads/main by this push:
     new bb0694c98 [client] Initialize lake comparator before merge read (#3406)
bb0694c98 is described below

commit bb0694c9829e0f11f916dd7051cbf5d9b9e61cd7
Author: yuxia Luo <luoyu...@alumni.sjtu.edu.cn>
AuthorDate: Sat May 30 00:14:17 2026 +0800

    [client] Initialize lake comparator before merge read (#3406)
---
 .../batch/LakeSnapshotAndLogSplitScanner.java      | 59 +++++++++++-----------
 1 file changed, 30 insertions(+), 29 deletions(-)

diff --git 
a/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/batch/LakeSnapshotAndLogSplitScanner.java
 
b/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/batch/LakeSnapshotAndLogSplitScanner.java
index f25b878d3..15352f63b 100644
--- 
a/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/batch/LakeSnapshotAndLogSplitScanner.java
+++ 
b/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/batch/LakeSnapshotAndLogSplitScanner.java
@@ -54,6 +54,7 @@ public class LakeSnapshotAndLogSplitScanner implements 
BatchScanner {
     private final @Nullable List<LakeSplit> lakeSplits;
     private Comparator<InternalRow> rowComparator;
     private List<CloseableIterator<LogRecord>> lakeRecordIterators = new 
ArrayList<>();
+    private boolean lakeRecordIteratorsInitialized;
     private final LakeSource<LakeSplit> lakeSource;
 
     private final int[] pkIndexes;
@@ -155,16 +156,7 @@ public class LakeSnapshotAndLogSplitScanner implements 
BatchScanner {
     @Override
     public CloseableIterator<InternalRow> pollBatch(Duration timeout) throws 
IOException {
         if (logScanFinished) {
-            if (lakeRecordIterators.isEmpty()) {
-                if (lakeSplits == null || lakeSplits.isEmpty()) {
-                    lakeRecordIterators = Collections.emptyList();
-                } else {
-                    for (LakeSplit lakeSplit : lakeSplits) {
-                        lakeRecordIterators.add(
-                                lakeSource.createRecordReader(() -> 
lakeSplit).read());
-                    }
-                }
-            }
+            initializeLakeRecordIterators();
             if (currentSortMergeReader == null) {
                 currentSortMergeReader =
                         new SortMergeReader(
@@ -179,25 +171,8 @@ public class LakeSnapshotAndLogSplitScanner implements 
BatchScanner {
             }
             return currentSortMergeReader.readBatch();
         } else {
-            if (lakeRecordIterators.isEmpty()) {
-                List<RecordReader> recordReaders = new ArrayList<>();
-                if (lakeSplits == null || lakeSplits.isEmpty()) {
-                    // pass null split to get rowComparator
-                    recordReaders.add(lakeSource.createRecordReader(() -> 
null));
-                } else {
-                    for (LakeSplit lakeSplit : lakeSplits) {
-                        recordReaders.add(lakeSource.createRecordReader(() -> 
lakeSplit));
-                    }
-                }
-                for (RecordReader reader : recordReaders) {
-                    if (reader instanceof SortedRecordReader) {
-                        rowComparator = ((SortedRecordReader) reader).order();
-                    } else {
-                        throw new UnsupportedOperationException(
-                                "lake records must instance of sorted view.");
-                    }
-                    lakeRecordIterators.add(reader.read());
-                }
+            initializeLakeRecordIterators();
+            if (logRows == null) {
                 logRows = new TreeMap<>(rowComparator);
             }
             pollLogRecords(timeout);
@@ -205,6 +180,32 @@ public class LakeSnapshotAndLogSplitScanner implements 
BatchScanner {
         }
     }
 
+    private void initializeLakeRecordIterators() throws IOException {
+        if (lakeRecordIteratorsInitialized) {
+            return;
+        }
+
+        List<RecordReader> recordReaders = new ArrayList<>();
+        if (lakeSplits == null || lakeSplits.isEmpty()) {
+            // pass null split to get rowComparator
+            recordReaders.add(lakeSource.createRecordReader(() -> null));
+        } else {
+            for (LakeSplit lakeSplit : lakeSplits) {
+                recordReaders.add(lakeSource.createRecordReader(() -> 
lakeSplit));
+            }
+        }
+        for (RecordReader reader : recordReaders) {
+            if (reader instanceof SortedRecordReader) {
+                rowComparator = ((SortedRecordReader) reader).order();
+            } else {
+                throw new UnsupportedOperationException(
+                        "lake records must instance of sorted view.");
+            }
+            lakeRecordIterators.add(reader.read());
+        }
+        lakeRecordIteratorsInitialized = true;
+    }
+
     private void pollLogRecords(Duration timeout) {
         ScanRecords scanRecords = logScanner.poll(timeout);
         for (ScanRecord scanRecord : scanRecords) {