[core] Fix file index pushdown for bitmap64 deletion vectors

paimon-common/src/main/java/org/apache/paimon/reader/LimitRecordReader.java

@@ -18,6 +18,8 @@
 
 package org.apache.paimon.reader;
 
+import org.apache.paimon.fs.Path;
+
 import javax.annotation.Nullable;
 
 import java.io.IOException;
@@ -59,6 +61,9 @@ public RecordIterator<T> readBatch() throws IOException {
         if (iterator instanceof ScoreRecordIterator) {
             return new LimitScoreRecordIterator<>((ScoreRecordIterator<T>) iterator);
         }
+        if (iterator instanceof FileRecordIterator) {
+            return new LimitFileRecordIterator<>((FileRecordIterator<T>) iterator);
+        }
         return new LimitRecordIterator<>(iterator);
     }
 
@@ -114,4 +119,25 @@ public long returnedRowId() {
             return iterator.returnedRowId();
         }
     }
+
+    private class LimitFileRecordIterator<T> extends LimitRecordIterator<T>
+            implements FileRecordIterator<T> {
+
+        private final FileRecordIterator<T> iterator;
+
+        private LimitFileRecordIterator(FileRecordIterator<T> iterator) {
+            super(iterator);
+            this.iterator = iterator;
+        }
+
+        @Override
+        public long returnedPosition() {
+            return iterator.returnedPosition();
+        }
+
+        @Override
+        public Path filePath() {
+            return iterator.filePath();
+        }
+    }
 }

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

@@ -161,6 +161,22 @@ public void forEach(LongConsumer consumer) {
         }
     }
 
+    /** Returns a 32-bit copy containing positions lower than {@code maxExclusive}. */
+    public RoaringBitmap32 projectToBitmap32(long maxExclusive) {
+        long maximumExclusive = (long) RoaringBitmap32.MAX_VALUE + 1;
+        Preconditions.checkArgument(
+                maxExclusive >= 0 && maxExclusive <= maximumExclusive,
+                "Invalid 32-bit projection bound: %s",
+                maxExclusive);
+        if (bitmaps.length == 0) {
+            return new RoaringBitmap32();
+        }
+
+        RoaringBitmap projected = bitmaps[0].clone();
+        projected.remove(maxExclusive, 1L << Integer.SIZE);
+        return RoaringBitmap32.fromRoaringBitmap(projected);
+    }
+
     @VisibleForTesting
     int allocatedBitmapCount() {
         return bitmaps.length;

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

@@ -44,6 +44,10 @@ private RoaringBitmap32(RoaringBitmap roaringBitmap) {
         this.roaringBitmap = roaringBitmap;
     }
 
+    static RoaringBitmap32 fromRoaringBitmap(RoaringBitmap roaringBitmap) {
+        return new RoaringBitmap32(roaringBitmap);
+    }
+
     /**
      * Note: the result is read only, do not call any modify operation outside.
      *

paimon-common/src/test/java/org/apache/paimon/reader/LimitRecordReaderTest.java

@@ -0,0 +1,93 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.reader;
+
+import org.apache.paimon.fs.Path;
+
+import org.junit.jupiter.api.Test;
+
+import javax.annotation.Nullable;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for {@link LimitRecordReader}. */
+public class LimitRecordReaderTest {
+
+    @Test
+    public void testPreservesFileRecordIterator() throws Exception {
+        FileRecordIterator<Integer> fileIterator =
+                new FileRecordIterator<Integer>() {
+                    private int position = -1;
+
+                    @Override
+                    public long returnedPosition() {
+                        return position;
+                    }
+
+                    @Override
+                    public Path filePath() {
+                        return new Path("test-file.parquet");
+                    }
+
+                    @Nullable
+                    @Override
+                    public Integer next() {
+                        position++;
+                        return position < 3 ? position : null;
+                    }
+
+                    @Override
+                    public void releaseBatch() {}
+                };
+
+        FileRecordReader<Integer> fileReader =
+                new FileRecordReader<Integer>() {
+                    private boolean batchReturned;
+
+                    @Nullable
+                    @Override
+                    public FileRecordIterator<Integer> readBatch() {
+                        if (batchReturned) {
+                            return null;
+                        }
+                        batchReturned = true;
+                        return fileIterator;
+                    }
+
+                    @Override
+                    public void close() {}
+                };
+
+        try (RecordReader<Integer> reader = LimitRecordReader.limit(fileReader, 2)) {
+            RecordReader.RecordIterator<Integer> batch = reader.readBatch();
+            assertThat(batch).isInstanceOf(FileRecordIterator.class);
+
+            FileRecordIterator<?> limited = (FileRecordIterator<?>) batch;
+            assertThat(limited.filePath()).isEqualTo(new Path("test-file.parquet"));
+            assertThat(limited.next()).isEqualTo(0);
+            assertThat(limited.returnedPosition()).isEqualTo(0);
+            assertThat(limited.next()).isEqualTo(1);
+            assertThat(limited.returnedPosition()).isEqualTo(1);
+            assertThat(limited.next()).isNull();
+            limited.releaseBatch();
+
+            assertThat(reader.readBatch()).isNull();
+        }
+    }
+}

paimon-core/src/main/java/org/apache/paimon/deletionvectors/Bitmap64DeletionVector.java

@@ -95,6 +95,11 @@ public void forEachDeletedPosition(LongConsumer consumer) {
         roaringBitmap.forEach(consumer);
     }
 
+    /** Returns a 32-bit copy containing deleted positions lower than {@code maxExclusive}. */
+    public RoaringBitmap32 projectToBitmap32(long maxExclusive) {
+        return roaringBitmap.projectToBitmap32(maxExclusive);
+    }
+
     @Override
     public int serializeTo(DataOutputStream out) throws IOException {
         roaringBitmap.runLengthEncode(); // run-length encode the bitmap before serializing

paimon-core/src/main/java/org/apache/paimon/io/FileIndexEvaluator.java

@@ -18,6 +18,7 @@
 
 package org.apache.paimon.io;
 
+import org.apache.paimon.deletionvectors.Bitmap64DeletionVector;
 import org.apache.paimon.deletionvectors.BitmapDeletionVector;
 import org.apache.paimon.deletionvectors.DeletionVector;
 import org.apache.paimon.fileindex.FileIndexPredicate;
@@ -33,6 +34,7 @@
 import javax.annotation.Nullable;
 
 import java.io.IOException;
+import java.util.Iterator;
 import java.util.List;
 import java.util.Objects;
 import java.util.stream.Collectors;
@@ -57,35 +59,57 @@ public static FileIndexResult evaluate(
                 return FileIndexResult.REMAIN;
             } else {
                 // limit can not work with other predicates.
-                return createBaseSelection(file, dv).limit(limit);
+                return createLimitSelection(file, dv, limit);
             }
         }
 
+        if (isNullOrEmpty(dataFilter)
+                && topN != null
+                && (file.rowCount() > RoaringBitmap32.MAX_VALUE || !supportsBitmapSelection(dv))) {
+            return FileIndexResult.REMAIN;
+        }
+
         try (FileIndexPredicate predicate =
                 createFileIndexPredicate(fileIO, dataSchema, dataFilePathFactory, file)) {
             if (predicate == null) {
                 return FileIndexResult.REMAIN;
             }
 
-            BitmapIndexResult selection = createBaseSelection(file, dv);
+            BitmapIndexResult selection = null;
             FileIndexResult result;
             if (!isNullOrEmpty(dataFilter)) {
                 Predicate filter = PredicateBuilder.and(dataFilter.toArray(new Predicate[0]));
                 result = predicate.evaluate(filter);
-                result = result.and(selection);
+                if (result instanceof BitmapIndexResult) {
+                    // Bitmap file indexes cannot represent positions beyond RoaringBitmap32.
+                    if (file.rowCount() > RoaringBitmap32.MAX_VALUE) {
+                        return FileIndexResult.REMAIN;
+                    }
+                    BitmapIndexResult bitmapResult = (BitmapIndexResult) result;
+                    if (bitmapResult.get().getCardinality() == file.rowCount()) {
+                        return FileIndexResult.REMAIN;
+                    }
+                    if (dv instanceof Bitmap64DeletionVector) {
+                        result = excludeDeletedPositions(bitmapResult, dv);
+                    } else if (supportsBitmapSelection(dv)) {
+                        selection = createBaseSelection(file, dv);
+                        result = result.and(selection);
+                    }
+                }
             } else if (topN != null) {
                 // 1. TopN cannot work with filter, because a filter may not completely filter out
                 // all records, any unfiltered records can affect the calculation results of TopN
                 // 2. evaluateTopN with selection, because we must filter out the data based on
                 // deletion vector before selecting TopN records.
+                selection = createBaseSelection(file, dv);
                 result = predicate.evaluateTopN(topN, selection);
             } else {
                 return FileIndexResult.REMAIN;
             }
 
             // if all position selected, or if only and not the deletion
             // the effect will not obvious, just return REMAIN.
-            if (Objects.equals(result, selection)) {
+            if (selection != null && Objects.equals(result, selection)) {
                 return FileIndexResult.REMAIN;
             }
 
@@ -97,15 +121,75 @@ public static FileIndexResult evaluate(
         }
     }
 
+    private static FileIndexResult createLimitSelection(
+            DataFileMeta file, @Nullable DeletionVector dv, int limit) {
+        if (dv == null) {
+            return new BitmapIndexResult(
+                    () -> RoaringBitmap32.bitmapOfRange(0, Math.min(file.rowCount(), limit)));
+        }
+
+        if (dv instanceof BitmapDeletionVector && file.rowCount() <= RoaringBitmap32.MAX_VALUE) {
+            return createBaseSelection(file, dv).limit(limit);
+        }
+
+        RoaringBitmap32 selection = new RoaringBitmap32();
+        long position = 0;
+        int remaining = limit;
+        while (remaining > 0 && position < file.rowCount()) {
+            if (position > RoaringBitmap32.MAX_VALUE) {
+                return FileIndexResult.REMAIN;
+            }
+            if (!dv.isDeleted(position)) {
+                selection.add((int) position);
+                remaining--;
+            }
+            position++;
+        }
+        return new BitmapIndexResult(() -> selection);
+    }
+
     private static BitmapIndexResult createBaseSelection(
             DataFileMeta file, @Nullable DeletionVector dv) {
-        BitmapIndexResult selection =
-                new BitmapIndexResult(() -> RoaringBitmap32.bitmapOfRange(0, file.rowCount()));
-        if (dv instanceof BitmapDeletionVector) {
-            RoaringBitmap32 deletion = ((BitmapDeletionVector) dv).get();
-            selection = selection.andNot(deletion);
-        }
-        return selection;
+        return new BitmapIndexResult(
+                () -> {
+                    RoaringBitmap32 selection = RoaringBitmap32.bitmapOfRange(0, file.rowCount());
+                    if (dv == null) {
+                        return selection;
+                    }
+
+                    RoaringBitmap32 deletion;
+                    if (dv instanceof BitmapDeletionVector) {
+                        deletion = ((BitmapDeletionVector) dv).get();
+                    } else if (dv instanceof Bitmap64DeletionVector) {
+                        deletion = ((Bitmap64DeletionVector) dv).projectToBitmap32(file.rowCount());
+                    } else {
+                        return selection;
+                    }
+                    selection.andNot(deletion);
+                    return selection;
+                });
+    }
+
+    private static boolean supportsBitmapSelection(@Nullable DeletionVector dv) {
+        return dv == null
+                || dv instanceof BitmapDeletionVector
+                || dv instanceof Bitmap64DeletionVector;
+    }
+
+    private static BitmapIndexResult excludeDeletedPositions(
+            BitmapIndexResult candidates, DeletionVector dv) {
+        return new BitmapIndexResult(
+                () -> {
+                    RoaringBitmap32 result = new RoaringBitmap32();
+                    Iterator<Integer> iterator = candidates.get().iterator();
+                    while (iterator.hasNext()) {
+                        int position = iterator.next();
+                        if (!dv.isDeleted(position)) {
+                            result.add(position);
+                        }
+                    }
+                    return result;
+                });
     }
 
     @Nullable

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

@@ -41,6 +41,7 @@
 import org.apache.paimon.predicate.TopN;
 import org.apache.paimon.reader.EmptyFileRecordReader;
 import org.apache.paimon.reader.FileRecordReader;
+import org.apache.paimon.reader.LimitRecordReader;
 import org.apache.paimon.reader.ReaderSupplier;
 import org.apache.paimon.reader.RecordReader;
 import org.apache.paimon.schema.SchemaManager;
@@ -213,7 +214,12 @@ public RecordReader<InternalRow> createReader(
                             null));
         }
 
-        return ConcatRecordReader.create(suppliers);
+        RecordReader<InternalRow> reader = ConcatRecordReader.create(suppliers);
+        // Apply the final limit after deletion vectors when no later predicate can drop rows.
+        if (topN == null && (filters == null || filters.isEmpty())) {
+            return LimitRecordReader.limit(reader, limit);
+        }
+        return reader;
     }
 
     FileRecordReader<InternalRow> createFileReader(

paimon-core/src/test/java/org/apache/paimon/deletionvectors/DeletionVectorTest.java

@@ -23,6 +23,7 @@
 import org.apache.paimon.fs.Path;
 import org.apache.paimon.reader.FileRecordIterator;
 import org.apache.paimon.reader.FileRecordReader;
+import org.apache.paimon.utils.RoaringBitmap32;
 
 import org.junit.jupiter.api.Test;
 
@@ -160,6 +161,18 @@ public void testBitmap64DeletionVector() {
         }
     }
 
+    @Test
+    public void testBitmap64DeletionVectorProjection() {
+        Bitmap64DeletionVector deletionVector = new Bitmap64DeletionVector();
+        deletionVector.delete(1);
+        deletionVector.delete(9);
+        deletionVector.delete(10);
+        deletionVector.delete(Integer.MAX_VALUE + 1L);
+        deletionVector.delete((1L << Integer.SIZE) + 1);
+
+        assertThat(deletionVector.projectToBitmap32(10)).isEqualTo(RoaringBitmap32.bitmapOf(1, 9));
+    }
+
     @Test
     public void testBitmapDeletionVectorTo64() {
         HashSet<Integer> toDelete = new HashSet<>();