[core] Introduce mergedRowCount in Split (apache#7090)

Author: JingsongLi

paimon-core/src/main/java/org/apache/paimon/globalindex/IndexedSplit.java

@@ -35,6 +35,7 @@
 import java.util.Arrays;
 import java.util.List;
 import java.util.Objects;
+import java.util.OptionalLong;
 
 /** Indexed split for global index. */
 public class IndexedSplit implements Split {
@@ -71,6 +72,11 @@ public long rowCount() {
         return rowRanges.stream().mapToLong(r -> r.to - r.from + 1).sum();
     }
 
+    @Override
+    public OptionalLong mergedRowCount() {
+        return OptionalLong.of(rowCount());
+    }
+
     @Override
     public boolean equals(Object o) {
         if (this == o) {

paimon-core/src/main/java/org/apache/paimon/table/FallbackReadFileStoreTable.java

@@ -63,6 +63,7 @@
 import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;
+import java.util.OptionalLong;
 import java.util.Set;
 import java.util.stream.Collectors;
 
@@ -249,6 +250,7 @@ private boolean sameRowTypeIgnoreNullable(RowType mainRowType, RowType fallbackR
 
     /** Split for fallback read. */
     public interface FallbackSplit extends Split {
+
         boolean isFallback();
 
         Split wrapped();
@@ -281,6 +283,11 @@ public Split wrapped() {
         public long rowCount() {
             return split.rowCount();
         }
+
+        @Override
+        public OptionalLong mergedRowCount() {
+            return split.mergedRowCount();
+        }
     }
 
     /** DataSplit fallback implementation. */

paimon-core/src/main/java/org/apache/paimon/table/format/FormatDataSplit.java

@@ -25,6 +25,7 @@
 import javax.annotation.Nullable;
 
 import java.util.Objects;
+import java.util.OptionalLong;
 
 /** {@link FormatDataSplit} for format table. */
 public class FormatDataSplit implements Split {
@@ -85,6 +86,11 @@ public long rowCount() {
         return -1;
     }
 
+    @Override
+    public OptionalLong mergedRowCount() {
+        return OptionalLong.empty();
+    }
+
     @Override
     public boolean equals(Object o) {
         if (this == o) {

paimon-core/src/main/java/org/apache/paimon/table/object/ObjectTableImpl.java

@@ -51,6 +51,7 @@
 import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;
+import java.util.OptionalLong;
 
 import static org.apache.paimon.data.BinaryString.fromString;
 
@@ -189,6 +190,11 @@ public boolean equals(Object o) {
         public int hashCode() {
             return Objects.hash(location);
         }
+
+        @Override
+        public OptionalLong mergedRowCount() {
+            return OptionalLong.empty();
+        }
     }
 
     private static class ObjectRead implements InnerTableRead {

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

@@ -35,6 +35,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
+import java.util.OptionalLong;
 
 /**
  * A split describes chain table read scope. It follows DataSplit's custom serialization pattern and
@@ -87,6 +88,11 @@ public long rowCount() {
         return sum;
     }
 
+    @Override
+    public OptionalLong mergedRowCount() {
+        return OptionalLong.empty();
+    }
+
     @Override
     public boolean equals(Object o) {
         if (this == o) {

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

@@ -55,7 +55,6 @@
 
 import static org.apache.paimon.io.DataFilePathFactory.INDEX_PATH_SUFFIX;
 import static org.apache.paimon.utils.Preconditions.checkArgument;
-import static org.apache.paimon.utils.Preconditions.checkState;
 
 /** Input splits. Needed by most batch computation engines. */
 public class DataSplit implements Split {
@@ -143,17 +142,31 @@ public long rowCount() {
         return rowCount;
     }
 
-    /** Whether it is possible to calculate the merged row count. */
-    public boolean mergedRowCountAvailable() {
-        return rawConvertible
-                && (dataDeletionFiles == null
-                        || dataDeletionFiles.stream()
-                                .allMatch(f -> f == null || f.cardinality() != null));
-    }
+    @Override
+    public OptionalLong mergedRowCount() {
+        if (!rawConvertible
+                || (dataDeletionFiles != null
+                        && !dataDeletionFiles.stream()
+                                .allMatch(f -> f == null || f.cardinality() != null))) {
+            return OptionalLong.empty();
+        }
 
-    public long mergedRowCount() {
-        checkState(mergedRowCountAvailable());
-        return partialMergedRowCount();
+        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;
+                }
+            }
+        }
+        return OptionalLong.of(sum);
     }
 
     public Object minValue(int fieldIndex, DataField dataField, SimpleStatsEvolutions evolutions) {

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

@@ -37,6 +37,7 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Optional;
+import java.util.OptionalLong;
 
 import static org.apache.paimon.table.source.PushDownUtils.minmaxAvailable;
 
@@ -144,10 +145,10 @@ private Optional<StartingScanner.Result> applyPushDownLimit() {
 
         List<Split> limitedSplits = new ArrayList<>();
         for (DataSplit dataSplit : splits) {
-            if (dataSplit.rawConvertible()) {
-                long partialMergedRowCount = dataSplit.partialMergedRowCount();
+            OptionalLong mergedRowCount = dataSplit.mergedRowCount();
+            if (mergedRowCount.isPresent()) {
                 limitedSplits.add(dataSplit);
-                scannedRowCount += partialMergedRowCount;
+                scannedRowCount += mergedRowCount.getAsLong();
                 if (scannedRowCount >= pushDownLimit) {
                     SnapshotReader.Plan newPlan =
                             new PlanImpl(plan.watermark(), plan.snapshotId(), limitedSplits);

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

@@ -22,6 +22,8 @@
 
 import javax.annotation.Nullable;
 
+import java.util.OptionalLong;
+
 /** A wrapper class for {@link Split} that adds query authorization information. */
 public class QueryAuthSplit implements Split {
 
@@ -48,4 +50,9 @@ public TableQueryAuthResult authResult() {
     public long rowCount() {
         return split.rowCount();
     }
+
+    @Override
+    public OptionalLong mergedRowCount() {
+        return split.mergedRowCount();
+    }
 }

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

@@ -23,6 +23,7 @@
 import java.io.Serializable;
 import java.util.List;
 import java.util.Optional;
+import java.util.OptionalLong;
 
 /**
  * An input split for reading.
@@ -32,8 +33,20 @@
 @Public
 public interface Split extends Serializable {
 
+    /**
+     * The row count in files, may be duplicated, such as in the primary key table and the
+     * Data-Evolution Append table.
+     */
     long rowCount();
 
+    /**
+     * Return the merged row count of data files. For example, when the delete vector is enabled in
+     * the primary key table, the number of rows that have been deleted will be subtracted from the
+     * returned result. In the Data Evolution mode of the Append table, the actual number of rows
+     * will be returned.
+     */
+    OptionalLong mergedRowCount();
+
     /**
      * If all files in this split can be read without merging, returns an {@link Optional} wrapping
      * a list of {@link RawFile}s to be read without merging. Otherwise, returns {@link

paimon-core/src/main/java/org/apache/paimon/table/system/AggregationFieldsTable.java

@@ -53,6 +53,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
+import java.util.OptionalLong;
 import java.util.function.Function;
 
 import static org.apache.paimon.catalog.Identifier.SYSTEM_TABLE_SPLITTER;
@@ -160,6 +161,11 @@ public boolean equals(Object o) {
         public int hashCode() {
             return Objects.hash(location);
         }
+
+        @Override
+        public OptionalLong mergedRowCount() {
+            return OptionalLong.empty();
+        }
     }
 
     /** {@link TableRead} implementation for {@link AggregationFieldsTable}. */