[core] Correct merged row count for Data Evolution Split

Author: JingsongLi

paimon-core/src/main/java/org/apache/paimon/table/source/DataSplit.java

@@ -39,6 +39,7 @@
 import org.apache.paimon.types.DataTypes;
 import org.apache.paimon.utils.FunctionWithIOException;
 import org.apache.paimon.utils.InternalRowUtils;
+import org.apache.paimon.utils.RangeHelper;
 import org.apache.paimon.utils.SerializationUtils;
 
 import javax.annotation.Nullable;
@@ -144,29 +145,61 @@ public long rowCount() {
 
     @Override
     public OptionalLong mergedRowCount() {
-        if (!rawConvertible
-                || (dataDeletionFiles != null
-                        && !dataDeletionFiles.stream()
-                                .allMatch(f -> f == null || f.cardinality() != null))) {
-            return OptionalLong.empty();
+        if (rawMergedRowCountAvailable()) {
+            return OptionalLong.of(rawMergedRowCount());
         }
+        if (dataEvolutionRowCountAvailable()) {
+            return OptionalLong.of(dataEvolutionMergedRowCount());
+        }
+        return OptionalLong.empty();
+    }
 
+    private boolean rawMergedRowCountAvailable() {
+        return rawConvertible
+                && (dataDeletionFiles == null
+                        || dataDeletionFiles.stream()
+                                .allMatch(f -> f == null || f.cardinality() != null));
+    }
+
+    private long rawMergedRowCount() {
         long sum = 0L;
-        List<RawFile> rawFiles = convertToRawFiles().orElse(null);
-        if (rawFiles != null) {
-            for (int i = 0; i < rawFiles.size(); i++) {
-                RawFile rawFile = rawFiles.get(i);
-                DeletionFile deletionFile =
-                        dataDeletionFiles == null ? null : dataDeletionFiles.get(i);
-                Long cardinality = deletionFile == null ? null : deletionFile.cardinality();
-                if (deletionFile == null) {
-                    sum += rawFile.rowCount();
-                } else if (cardinality != null) {
-                    sum += rawFile.rowCount() - cardinality;
-                }
+        for (int i = 0; i < dataFiles.size(); i++) {
+            DataFileMeta file = dataFiles.get(i);
+            DeletionFile deletionFile = dataDeletionFiles == null ? null : dataDeletionFiles.get(i);
+            Long cardinality = deletionFile == null ? null : deletionFile.cardinality();
+            if (deletionFile == null) {
+                sum += file.rowCount();
+            } else if (cardinality != null) {
+                sum += file.rowCount() - cardinality;
             }
         }
-        return OptionalLong.of(sum);
+        return sum;
+    }
+
+    private boolean dataEvolutionRowCountAvailable() {
+        for (DataFileMeta file : dataFiles) {
+            if (file.firstRowId() == null) {
+                return false;
+            }
+        }
+        return true;
+    }
+
+    private long dataEvolutionMergedRowCount() {
+        long sum = 0L;
+        RangeHelper<DataFileMeta> rangeHelper =
+                new RangeHelper<>(
+                        DataFileMeta::nonNullFirstRowId,
+                        f -> f.nonNullFirstRowId() + f.rowCount() - 1);
+        List<List<DataFileMeta>> ranges = rangeHelper.mergeOverlappingRanges(dataFiles);
+        for (List<DataFileMeta> group : ranges) {
+            long maxCount = 0;
+            for (DataFileMeta file : group) {
+                maxCount = Math.max(maxCount, file.rowCount());
+            }
+            sum += maxCount;
+        }
+        return sum;
     }
 
     public Object minValue(int fieldIndex, DataField dataField, SimpleStatsEvolutions evolutions) {

paimon-core/src/test/java/org/apache/paimon/table/DataEvolutionTableTest.java

@@ -44,6 +44,7 @@
 import org.apache.paimon.table.source.EndOfScanException;
 import org.apache.paimon.table.source.ReadBuilder;
 import org.apache.paimon.table.source.Split;
+import org.apache.paimon.table.source.TableScan;
 import org.apache.paimon.types.DataTypes;
 import org.apache.paimon.types.RowType;
 import org.apache.paimon.utils.Range;
@@ -55,6 +56,7 @@
 import java.util.Collections;
 import java.util.Iterator;
 import java.util.List;
+import java.util.OptionalLong;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.stream.Collectors;
 
@@ -87,8 +89,20 @@ public void testBasic() throws Exception {
         }
 
         ReadBuilder readBuilder = getTableDefault().newReadBuilder();
-        RecordReader<InternalRow> reader =
-                readBuilder.newRead().createReader(readBuilder.newScan().plan());
+        TableScan.Plan plan = readBuilder.newScan().plan();
+
+        // assert merged row count
+        long noMergedRowCount = plan.splits().stream().mapToLong(Split::rowCount).sum();
+        long mergedRowCount =
+                plan.splits().stream()
+                        .map(Split::mergedRowCount)
+                        .filter(OptionalLong::isPresent)
+                        .mapToLong(OptionalLong::getAsLong)
+                        .sum();
+        assertThat(noMergedRowCount).isEqualTo(2);
+        assertThat(mergedRowCount).isEqualTo(1);
+
+        RecordReader<InternalRow> reader = readBuilder.newRead().createReader(plan);
         assertThat(reader).isInstanceOf(DataEvolutionFileReader.class);
         reader.forEachRemaining(
                 r -> {
@@ -105,10 +119,7 @@ public void testBasic() throws Exception {
                         .newRead()
                         .createReader(readBuilder.newScan().plan());
         AtomicInteger cnt = new AtomicInteger(0);
-        reader.forEachRemaining(
-                r -> {
-                    cnt.incrementAndGet();
-                });
+        reader.forEachRemaining(r -> cnt.incrementAndGet());
         assertThat(cnt.get()).isEqualTo(1);
 
         // projection with an empty read type
@@ -119,10 +130,7 @@ public void testBasic() throws Exception {
                         .newRead()
                         .createReader(readBuilder.newScan().plan());
         AtomicInteger cnt1 = new AtomicInteger(0);
-        reader.forEachRemaining(
-                r -> {
-                    cnt1.incrementAndGet();
-                });
+        reader.forEachRemaining(r -> cnt1.incrementAndGet());
         assertThat(cnt1.get()).isEqualTo(1);
     }