[core] Refactor row id pushdown to DataEvolutionFileStoreScan

Author: JingsongLi

docs/layouts/shortcodes/generated/core_configuration.html

@@ -1037,12 +1037,6 @@
             <td>String</td>
             <td>Time field for record level expire. It supports the following types: `timestamps in seconds with INT`,`timestamps in seconds with BIGINT`, `timestamps in milliseconds with BIGINT` or `timestamp`.</td>
         </tr>
-        <tr>
-            <td><h5>row-id-push-down.enabled</h5></td>
-            <td style="word-wrap: break-word;">false</td>
-            <td>Boolean</td>
-            <td>Whether to enable row id push down for scan. Currently, only the data evolution table supports row id push down.</td>
-        </tr>
         <tr>
             <td><h5>row-tracking.enabled</h5></td>
             <td style="word-wrap: break-word;">false</td>

paimon-api/src/main/java/org/apache/paimon/CoreOptions.java

@@ -2146,14 +2146,6 @@ public InlineElement getDescription() {
                     .withDescription(
                             "Whether to try upgrading the data files after overwriting a primary key table.");
 
-    public static final ConfigOption<Boolean> ROW_ID_PUSH_DOWN_ENABLED =
-            key("row-id-push-down.enabled")
-                    .booleanType()
-                    .defaultValue(false)
-                    .withDescription(
-                            "Whether to enable row id push down for scan."
-                                    + " Currently, only the data evolution table supports row id push down.");
-
     private final Options options;
 
     public CoreOptions(Map<String, String> options) {
@@ -3338,10 +3330,6 @@ public boolean overwriteUpgrade() {
         return options.get(OVERWRITE_UPGRADE);
     }
 
-    public boolean rowIdPushDownEnabled() {
-        return options.get(ROW_ID_PUSH_DOWN_ENABLED);
-    }
-
     /** Specifies the merge engine for table with primary key. */
     public enum MergeEngine implements DescribedEnum {
         DEDUPLICATE("deduplicate", "De-duplicate and keep the last row."),

paimon-common/src/main/java/org/apache/paimon/predicate/RowIdPredicateVisitor.java

@@ -22,6 +22,7 @@
 
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Optional;
 
 import static org.apache.paimon.table.SpecialFields.ROW_ID;
 
@@ -40,10 +41,10 @@
  *       AND _ROW_ID IN (1, 2)}).
  * </ul>
  */
-public class RowIdPredicateVisitor implements PredicateVisitor<List<Range>> {
+public class RowIdPredicateVisitor implements PredicateVisitor<Optional<List<Range>>> {
 
     @Override
-    public List<Range> visit(LeafPredicate predicate) {
+    public Optional<List<Range>> visit(LeafPredicate predicate) {
         if (ROW_ID.name().equals(predicate.fieldName())) {
             LeafFunction function = predicate.function();
             if (function instanceof Equal || function instanceof In) {
@@ -53,57 +54,55 @@ public List<Range> visit(LeafPredicate predicate) {
                 }
                 // The list output by getRangesFromList is already sorted,
                 // and has no overlap
-                return Range.getRangesFromList(rowIds);
+                return Optional.of(Range.toRanges(rowIds));
             }
         }
-        return null;
+        return Optional.empty();
     }
 
     @Override
-    public List<Range> visit(CompoundPredicate predicate) {
+    public Optional<List<Range>> visit(CompoundPredicate predicate) {
         CompoundPredicate.Function function = predicate.function();
-        List<Range> rowIds = null;
+        Optional<List<Range>> rowIds = Optional.empty();
         // `And` means we should get the intersection of all children.
         if (function instanceof And) {
             for (Predicate child : predicate.children()) {
-                List<Range> childList = child.visit(this);
-                if (childList == null) {
+                Optional<List<Range>> childList = child.visit(this);
+                if (!childList.isPresent()) {
                     continue;
                 }
 
-                if (rowIds == null) {
-                    rowIds = childList;
-                } else {
-                    rowIds = Range.and(rowIds, childList);
-                }
+                rowIds =
+                        rowIds.map(ranges -> Optional.of(Range.and(ranges, childList.get())))
+                                .orElse(childList);
 
                 // shortcut for intersection
-                if (rowIds.isEmpty()) {
+                if (rowIds.get().isEmpty()) {
                     return rowIds;
                 }
             }
         } else if (function instanceof Or) {
             // `Or` means we should get the union of all children
-            rowIds = new ArrayList<>();
+            rowIds = Optional.of(new ArrayList<>());
             for (Predicate child : predicate.children()) {
-                List<Range> childList = child.visit(this);
-                if (childList == null) {
-                    return null;
+                Optional<List<Range>> childList = child.visit(this);
+                if (!childList.isPresent()) {
+                    return Optional.empty();
                 }
 
-                rowIds.addAll(childList);
-                rowIds = Range.sortAndMergeOverlap(rowIds, true);
+                rowIds.get().addAll(childList.get());
+                rowIds = Optional.of(Range.sortAndMergeOverlap(rowIds.get(), true));
             }
         } else {
-            // unexpected function type, just return null
-            return null;
+            // unexpected function type, just return empty
+            return Optional.empty();
         }
         return rowIds;
     }
 
     @Override
-    public List<Range> visit(TransformPredicate predicate) {
+    public Optional<List<Range>> visit(TransformPredicate predicate) {
         // do not support transform predicate now.
-        return null;
+        return Optional.empty();
     }
 }

paimon-common/src/main/java/org/apache/paimon/utils/Range.java

@@ -22,9 +22,9 @@
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.Comparator;
+import java.util.Iterator;
 import java.util.List;
 import java.util.Objects;
-import java.util.stream.Collectors;
 
 /** Range represents from (inclusive) and to (inclusive). */
 public class Range implements Serializable {
@@ -178,31 +178,29 @@ public static List<Range> mergeSortedAsPossible(List<Range> ranges) {
         return result;
     }
 
-    public static List<Range> getRangesFromList(List<Long> origLongs) {
-        if (origLongs == null || origLongs.isEmpty()) {
-            return Collections.emptyList();
+    public static List<Range> toRanges(Iterable<Long> ids) {
+        List<Range> ranges = new ArrayList<>();
+        Iterator<Long> iterator = ids.iterator();
+
+        if (!iterator.hasNext()) {
+            return ranges;
         }
 
-        List<Long> longs = origLongs.stream().distinct().sorted().collect(Collectors.toList());
-
-        ArrayList<Range> ranges = new ArrayList<>();
-        Long rangeStart = null;
-        Long rangeEnd = null;
-        for (Long cur : longs) {
-            if (rangeStart == null) {
-                rangeStart = cur;
-                rangeEnd = cur;
-            } else if (rangeEnd == cur - 1) {
-                rangeEnd = cur;
-            } else {
+        long rangeStart = iterator.next();
+        long rangeEnd = rangeStart;
+
+        while (iterator.hasNext()) {
+            long current = iterator.next();
+            if (current != rangeEnd + 1) {
+                // Save the current range and start a new one
                 ranges.add(new Range(rangeStart, rangeEnd));
-                rangeStart = cur;
-                rangeEnd = cur;
+                rangeStart = current;
             }
+            rangeEnd = current;
         }
-        if (rangeStart != null) {
-            ranges.add(new Range(rangeStart, rangeEnd));
-        }
+        // Add the last range
+        ranges.add(new Range(rangeStart, rangeEnd));
+
         return ranges;
     }
 

paimon-common/src/main/java/org/apache/paimon/utils/RoaringNavigableMap64.java

@@ -18,16 +18,13 @@
 
 package org.apache.paimon.utils;
 
-import org.apache.paimon.annotation.VisibleForTesting;
-
 import org.roaringbitmap.longlong.Roaring64NavigableMap;
 
 import java.io.ByteArrayInputStream;
 import java.io.ByteArrayOutputStream;
 import java.io.DataInputStream;
 import java.io.DataOutputStream;
 import java.io.IOException;
-import java.util.ArrayList;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Objects;
@@ -135,42 +132,16 @@ public void deserialize(byte[] rbmBytes) throws IOException {
         }
     }
 
-    @VisibleForTesting
     /**
      * Converts this bitmap to a list of contiguous ranges.
      *
      * <p>This is useful for interoperability with APIs that expect List&lt;Range&gt;.
      */
     public List<Range> toRangeList() {
-        List<Range> ranges = new ArrayList<>();
-        Iterator<Long> iterator = roaring64NavigableMap.iterator();
-
-        if (!iterator.hasNext()) {
-            return ranges;
-        }
-
-        long rangeStart = iterator.next();
-        long rangeEnd = rangeStart;
-
-        while (iterator.hasNext()) {
-            long current = iterator.next();
-            if (current == rangeEnd + 1) {
-                // Extend the current range
-                rangeEnd = current;
-            } else {
-                // Save the current range and start a new one
-                ranges.add(new Range(rangeStart, rangeEnd));
-                rangeStart = current;
-                rangeEnd = current;
-            }
-        }
-        // Add the last range
-        ranges.add(new Range(rangeStart, rangeEnd));
-
-        return ranges;
+        // TODO Optimize this to avoid iterator all ids
+        return Range.toRanges(roaring64NavigableMap::iterator);
     }
 
-    @VisibleForTesting
     public static RoaringNavigableMap64 bitmapOf(long... dat) {
         RoaringNavigableMap64 roaringBitmap64 = new RoaringNavigableMap64();
         for (long ele : dat) {

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

@@ -24,7 +24,10 @@
 import org.apache.paimon.manifest.PartitionEntry;
 import org.apache.paimon.metrics.MetricRegistry;
 import org.apache.paimon.partition.PartitionPredicate;
+import org.apache.paimon.predicate.CompoundPredicate;
+import org.apache.paimon.predicate.LeafPredicate;
 import org.apache.paimon.predicate.Predicate;
+import org.apache.paimon.predicate.RowIdPredicateVisitor;
 import org.apache.paimon.predicate.TopN;
 import org.apache.paimon.predicate.VectorSearch;
 import org.apache.paimon.table.FileStoreTable;
@@ -47,6 +50,7 @@
 import java.util.Optional;
 
 import static org.apache.paimon.globalindex.GlobalIndexScanBuilder.parallelScan;
+import static org.apache.paimon.table.SpecialFields.ROW_ID;
 
 /** Scan for data evolution table. */
 public class DataEvolutionBatchScan implements DataTableScan {
@@ -71,11 +75,43 @@ public DataTableScan withShard(int indexOfThisSubtask, int numberOfParallelSubta
 
     @Override
     public InnerTableScan withFilter(Predicate predicate) {
+        if (predicate == null) {
+            return this;
+        }
+
+        predicate.visit(new RowIdPredicateVisitor()).ifPresent(this::withRowRanges);
+        predicate = removeRowIdFilter(predicate);
         this.filter = predicate;
         batchScan.withFilter(predicate);
         return this;
     }
 
+    private Predicate removeRowIdFilter(Predicate filter) {
+        if (filter instanceof LeafPredicate
+                && ROW_ID.name().equals(((LeafPredicate) filter).fieldName())) {
+            return null;
+        } else if (filter instanceof CompoundPredicate) {
+            CompoundPredicate compoundPredicate = (CompoundPredicate) filter;
+
+            List<Predicate> newChildren = new ArrayList<>();
+            for (Predicate child : compoundPredicate.children()) {
+                Predicate newChild = removeRowIdFilter(child);
+                if (newChild != null) {
+                    newChildren.add(newChild);
+                }
+            }
+
+            if (newChildren.isEmpty()) {
+                return null;
+            } else if (newChildren.size() == 1) {
+                return newChildren.get(0);
+            } else {
+                return new CompoundPredicate(compoundPredicate.function(), newChildren);
+            }
+        }
+        return filter;
+    }
+
     @Override
     public InnerTableScan withVectorSearch(VectorSearch vectorSearch) {
         this.vectorSearch = vectorSearch;
@@ -157,9 +193,13 @@ public InnerTableScan withLevelFilter(Filter<Integer> levelFilter) {
 
     @Override
     public InnerTableScan withRowRanges(List<Range> rowRanges) {
+        if (rowRanges == null) {
+            return this;
+        }
+
         this.pushedRowRanges = rowRanges;
         if (globalIndexResult != null) {
-            throw new IllegalStateException("");
+            throw new IllegalStateException("Cannot push row ranges after global index eval.");
         }
         return this;
     }

paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreScan.java

@@ -243,6 +243,7 @@ public FileStoreScan keepStats() {
     @Override
     public FileStoreScan withRowRanges(List<Range> rowRanges) {
         this.rowRanges = rowRanges;
+        manifestsReader.withRowRanges(rowRanges);
         return this;
     }
 
@@ -274,7 +275,6 @@ public Plan plan() {
         ManifestsReader.Result manifestsResult = readManifests();
         Snapshot snapshot = manifestsResult.snapshot;
         List<ManifestFileMeta> manifests = manifestsResult.filteredManifests;
-        manifests = postFilterManifests(manifests);
 
         Iterator<ManifestEntry> iterator = readManifestEntries(manifests, false);
         if (supportsLimitPushManifestEntries()) {
@@ -434,10 +434,6 @@ protected TableSchema scanTableSchema(long id) {
     /** Note: Keep this thread-safe. */
     protected abstract boolean filterByStats(ManifestEntry entry);
 
-    protected List<ManifestFileMeta> postFilterManifests(List<ManifestFileMeta> manifests) {
-        return manifests;
-    }
-
     protected boolean postFilterManifestEntriesEnabled() {
         return false;
     }