[client] Backport SortMergeReader changelog delete fix to 0.9

fluss-client/src/main/java/org/apache/fluss/client/table/scanner/SortMergeReader.java

@@ -321,23 +321,31 @@ public boolean hasNext() {
                 return true;
             }
             try {
-                // if currentMergedRows is null, we need to get the next mergedRows
-                if (currentMergedRows == null) {
-                    SortMergeRows sortMergeRows =
-                            currentLakeSnapshotRecords.hasNext()
-                                    ? currentLakeSnapshotRecords.next()
-                                    : null;
-                    //  next mergedRows is not null and is not empty, set the currentMergedRows
-                    if (sortMergeRows != null && !sortMergeRows.mergedRows.isEmpty()) {
-                        currentMergedRows = sortMergeRows.mergedRows.iterator();
+                // Loop to skip empty merge results (e.g., when a snapshot record is
+                // deleted by changelog) and advance to the next non-empty result.
+                while (returnedRow == null) {
+                    if (currentMergedRows == null) {
+                        if (!currentLakeSnapshotRecords.hasNext()) {
+                            return false;
+                        }
+
+                        SortMergeRows sortMergeRows = currentLakeSnapshotRecords.next();
+                        if (!sortMergeRows.mergedRows.isEmpty()) {
+                            currentMergedRows = sortMergeRows.mergedRows.iterator();
+                        } else {
+                            // If mergedRows is empty (e.g., record was deleted), continue
+                            // the loop to try the next snapshot record.
+                            continue;
+                        }
+                    }
+
+                    if (currentMergedRows.hasNext()) {
+                        returnedRow = currentMergedRows.next();
+                    } else {
+                        currentMergedRows = null;
                     }
                 }
-                // check if has next row, whether does, set the internalRow to returned in method
-                // next;
-                if (currentMergedRows != null && currentMergedRows.hasNext()) {
-                    returnedRow = currentMergedRows.next();
-                }
-                return returnedRow != null;
+                return true;
             } catch (Exception e) {
                 throw new RuntimeException(e);
             }

fluss-client/src/test/java/org/apache/fluss/client/table/scanner/SortMergeReaderTest.java

@@ -29,6 +29,7 @@
 import org.apache.fluss.types.StringType;
 import org.apache.fluss.utils.CloseableIterator;
 
+import org.junit.jupiter.api.Test;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.ValueSource;
 
@@ -107,6 +108,35 @@ void testReadBatch(boolean isProjected) {
         assertThat(actualRows).isEqualTo(materializeRows(expected, fieldGetters));
     }
 
+    @Test
+    void testReadBatchSkipsDeletedSnapshotRows() {
+        int keyIndex = 0;
+        int[] pkIndexes = new int[] {keyIndex};
+
+        List<LogRecord> snapshotRecords = createRecords(0, 10, false);
+        List<KeyValueRow> changeLogRecords = createDeleteRecords(pkIndexes, Arrays.asList(2, 5, 8));
+
+        SortMergeReader sortMergeReader =
+                new SortMergeReader(
+                        null,
+                        pkIndexes,
+                        CloseableIterator.wrap(snapshotRecords.iterator()),
+                        new FlussRowComparator(keyIndex),
+                        CloseableIterator.wrap(changeLogRecords.iterator()));
+
+        InternalRow.FieldGetter[] fieldGetters =
+                InternalRow.createFieldGetters(
+                        RowType.of(new IntType(), new StringType(), new StringType()));
+        List<InternalRow> actualRows;
+        try (CloseableIterator<InternalRow> iterator = sortMergeReader.readBatch()) {
+            actualRows = materializeRows(iterator, fieldGetters);
+        }
+
+        assertThat(actualRows).hasSize(7);
+        assertThat(actualRows.stream().map(row -> row.getInt(0)).collect(Collectors.toList()))
+                .containsExactly(0, 1, 3, 4, 6, 7, 9);
+    }
+
     private CloseableIterator<InternalRow> projected(
             CloseableIterator<LogRecord> originElementIterator, final ProjectedRow projectedRow) {
         return new CloseableIterator<InternalRow>() {
@@ -157,4 +187,16 @@ private List<LogRecord> createRecords(int startId, int count, boolean isLog) {
         }
         return logRecords;
     }
+
+    private List<KeyValueRow> createDeleteRecords(int[] pkIndexes, List<Integer> keys) {
+        List<KeyValueRow> deleteRecords = new ArrayList<>();
+        for (Integer key : keys) {
+            deleteRecords.add(
+                    new KeyValueRow(
+                            pkIndexes,
+                            row(key, BinaryString.fromString("a"), BinaryString.fromString("A")),
+                            true));
+        }
+        return deleteRecords;
+    }
 }