This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 63d7ca5e3b [core] Fix file index pushdown for bitmap64 deletion
vectors (#9141)
63d7ca5e3b is described below
commit 63d7ca5e3bdc71a086777491772878af4fbb615d
Author: QuakeWang <[email protected]>
AuthorDate: Mon Aug 17 13:33:46 2026 +0800
[core] Fix file index pushdown for bitmap64 deletion vectors (#9141)
---
.../apache/paimon/reader/LimitRecordReader.java | 26 ++++++
.../paimon/reader/LimitRecordReaderTest.java | 93 ++++++++++++++++++++++
.../org/apache/paimon/io/FileIndexEvaluator.java | 7 ++
.../apache/paimon/operation/RawFileSplitRead.java | 8 +-
.../apache/paimon/io/FileIndexEvaluatorTest.java | 79 ++++++++++++++++++
.../paimon/operation/RawFileSplitReadTest.java | 57 ++++++++++++-
6 files changed, 266 insertions(+), 4 deletions(-)
diff --git
a/paimon-common/src/main/java/org/apache/paimon/reader/LimitRecordReader.java
b/paimon-common/src/main/java/org/apache/paimon/reader/LimitRecordReader.java
index 256453497f..2f0766e29a 100644
---
a/paimon-common/src/main/java/org/apache/paimon/reader/LimitRecordReader.java
+++
b/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 final class LimitRecordReader<T> implements
RecordReader<T> {
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 final class LimitRecordReader<T> implements
RecordReader<T> {
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();
+ }
+ }
}
diff --git
a/paimon-common/src/test/java/org/apache/paimon/reader/LimitRecordReaderTest.java
b/paimon-common/src/test/java/org/apache/paimon/reader/LimitRecordReaderTest.java
new file mode 100644
index 0000000000..6e5224d98d
--- /dev/null
+++
b/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();
+ }
+ }
+}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/io/FileIndexEvaluator.java
b/paimon-core/src/main/java/org/apache/paimon/io/FileIndexEvaluator.java
index 1a78622717..a82194cc76 100644
--- a/paimon-core/src/main/java/org/apache/paimon/io/FileIndexEvaluator.java
+++ b/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;
@@ -52,6 +53,12 @@ public class FileIndexEvaluator {
DataFileMeta file,
@Nullable DeletionVector dv)
throws IOException {
+ // File index selections use 32-bit positions. Fall back when they
cannot safely represent
+ // the file or its deletion vector.
+ if (file.rowCount() > RoaringBitmap32.MAX_VALUE || dv instanceof
Bitmap64DeletionVector) {
+ return FileIndexResult.REMAIN;
+ }
+
if (isNullOrEmpty(dataFilter) && topN == null) {
if (limit == null) {
return FileIndexResult.REMAIN;
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/RawFileSplitRead.java
b/paimon-core/src/main/java/org/apache/paimon/operation/RawFileSplitRead.java
index c66b0f2c2c..f5fe777b29 100644
---
a/paimon-core/src/main/java/org/apache/paimon/operation/RawFileSplitRead.java
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/RawFileSplitRead.java
@@ -41,6 +41,7 @@ import org.apache.paimon.predicate.Predicate;
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.ReadBatchSizer;
import org.apache.paimon.reader.ReaderSupplier;
import org.apache.paimon.reader.RecordReader;
@@ -221,7 +222,12 @@ public class RawFileSplitRead implements
SplitRead<InternalRow> {
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(
diff --git
a/paimon-core/src/test/java/org/apache/paimon/io/FileIndexEvaluatorTest.java
b/paimon-core/src/test/java/org/apache/paimon/io/FileIndexEvaluatorTest.java
index 5b91c64813..891ae8380a 100644
--- a/paimon-core/src/test/java/org/apache/paimon/io/FileIndexEvaluatorTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/io/FileIndexEvaluatorTest.java
@@ -18,12 +18,14 @@
package org.apache.paimon.io;
+import org.apache.paimon.deletionvectors.Bitmap64DeletionVector;
import org.apache.paimon.deletionvectors.BitmapDeletionVector;
import org.apache.paimon.fileindex.FileIndexFormat;
import org.apache.paimon.fileindex.FileIndexResult;
import org.apache.paimon.fileindex.FileIndexWriter;
import org.apache.paimon.fileindex.bitmap.BitmapFileIndex;
import org.apache.paimon.fileindex.bitmap.BitmapFileIndexFactory;
+import org.apache.paimon.fileindex.bitmap.BitmapIndexResult;
import org.apache.paimon.options.Options;
import org.apache.paimon.predicate.PredicateBuilder;
import org.apache.paimon.schema.TableSchema;
@@ -32,6 +34,7 @@ import org.apache.paimon.types.DataField;
import org.apache.paimon.types.DataType;
import org.apache.paimon.types.DataTypes;
import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.RoaringBitmap32;
import org.junit.jupiter.api.Test;
@@ -46,6 +49,64 @@ import static org.assertj.core.api.Assertions.assertThat;
/** Tests for {@link FileIndexEvaluator}. */
public class FileIndexEvaluatorTest {
+ @Test
+ public void testLimitIntersectsBitmapDeletionVector() throws Exception {
+ BitmapDeletionVector deletionVector = new BitmapDeletionVector();
+ for (int position = 0; position < 5; position++) {
+ deletionVector.delete(position);
+ }
+
+ FileIndexResult result =
+ FileIndexEvaluator.evaluate(
+ null,
+ null,
+ Collections.emptyList(),
+ null,
+ 10,
+ null,
+ DataFileTestUtils.newFile("data.avro", 0, 0, 19, 0L),
+ deletionVector);
+
+ assertThat(result).isInstanceOf(BitmapIndexResult.class);
+ assertThat(((BitmapIndexResult) result).get())
+ .isEqualTo(RoaringBitmap32.bitmapOfRange(5, 15));
+ }
+
+ @Test
+ public void testBitmap64DeletionVectorFallsBack() throws Exception {
+ Bitmap64DeletionVector deletionVector = new Bitmap64DeletionVector();
+ deletionVector.delete(0);
+
+ FileIndexResult result =
+ FileIndexEvaluator.evaluate(
+ null,
+ null,
+ Collections.emptyList(),
+ null,
+ 10,
+ null,
+ DataFileTestUtils.newFile("data.avro", 0, 0, 19, 0L),
+ deletionVector);
+
+ assertThat(result).isSameAs(FileIndexResult.REMAIN);
+ }
+
+ @Test
+ public void testLargeFileFallsBack() throws Exception {
+ FileIndexResult result =
+ FileIndexEvaluator.evaluate(
+ null,
+ null,
+ Collections.emptyList(),
+ null,
+ 10,
+ null,
+ fileWithRowCount(Integer.MAX_VALUE + 2L),
+ null);
+
+ assertThat(result).isSameAs(FileIndexResult.REMAIN);
+ }
+
@Test
public void testDataFilterIntersectsDeletionVector() throws Exception {
BitmapDeletionVector deletionVector = new BitmapDeletionVector();
@@ -95,6 +156,24 @@ public class FileIndexEvaluatorTest {
null);
}
+ private static DataFileMeta fileWithRowCount(long rowCount) {
+ return DataFileMeta.forAppend(
+ "data.avro",
+ 0,
+ rowCount,
+ SimpleStats.EMPTY_STATS,
+ 0,
+ 0,
+ 0,
+ Collections.emptyList(),
+ null,
+ null,
+ null,
+ null,
+ null,
+ null);
+ }
+
private static byte[] embeddedBitmapIndex() throws IOException {
BitmapFileIndex bitmapFileIndex = new BitmapFileIndex(DataTypes.INT(),
new Options());
FileIndexWriter indexWriter = bitmapFileIndex.createWriter();
diff --git
a/paimon-core/src/test/java/org/apache/paimon/operation/RawFileSplitReadTest.java
b/paimon-core/src/test/java/org/apache/paimon/operation/RawFileSplitReadTest.java
index cc67dd32ed..c1e12d61d3 100644
---
a/paimon-core/src/test/java/org/apache/paimon/operation/RawFileSplitReadTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/operation/RawFileSplitReadTest.java
@@ -18,10 +18,13 @@
package org.apache.paimon.operation;
+import org.apache.paimon.AppendOnlyFileStore;
import org.apache.paimon.CoreOptions;
import org.apache.paimon.data.BinaryString;
import org.apache.paimon.data.GenericRow;
import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.deletionvectors.Bitmap64DeletionVector;
+import org.apache.paimon.deletionvectors.DeletionVector;
import org.apache.paimon.format.FileFormat;
import org.apache.paimon.format.FlushingFileFormat;
import org.apache.paimon.format.FormatReaderFactory;
@@ -44,11 +47,15 @@ import org.apache.paimon.table.source.DataSplit;
import org.apache.paimon.table.source.InnerTableRead;
import org.apache.paimon.types.DataTypes;
import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.IOExceptionSupplier;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
+import java.util.ArrayList;
+import java.util.Collections;
import java.util.List;
+import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;
import static org.assertj.core.api.Assertions.assertThat;
@@ -134,6 +141,39 @@ class RawFileSplitReadTest {
}
}
+ @Test
+ void testLimitAfterBitmap64DeletionVector() throws Exception {
+ List<InternalRow> rows = new ArrayList<>();
+ for (int i = 0; i < 20; i++) {
+ rows.add(GenericRow.of(BinaryString.fromString("value-" + i), i));
+ }
+ FileStoreTable table = createTable("bitmap64-limit", rows);
+ DataSplit split = singleSplit(table);
+ assertThat(split.dataFiles()).hasSize(1);
+
+ DeletionVector deletionVector = new Bitmap64DeletionVector();
+ for (int position = 0; position < 5; position++) {
+ deletionVector.delete(position);
+ }
+ String fileName = split.dataFiles().get(0).fileName();
+ Map<String, IOExceptionSupplier<DeletionVector>>
deletionVectorFactories =
+ Collections.singletonMap(fileName, () -> deletionVector);
+
+ RawFileSplitRead read = ((AppendOnlyFileStore)
table.store()).newRead();
+ read.withLimit(10);
+ AtomicInteger count = new AtomicInteger();
+ try (RecordReader<InternalRow> reader =
+ read.createReader(
+ split.partition(),
+ split.bucket(),
+ split.dataFiles(),
+ deletionVectorFactories)) {
+ reader.forEachRemaining(ignored -> count.incrementAndGet());
+ }
+
+ assertThat(count).hasValue(10);
+ }
+
@Test
void testTableReadSharesBatchSizer() throws Exception {
FileStoreTable table = createTable("dynamic-batch-size", 20);
@@ -164,10 +204,21 @@ class RawFileSplitReadTest {
}
private FileStoreTable createTable(String directory) throws Exception {
- return createTable(directory, 1);
+ return createTable(
+ directory,
+
Collections.singletonList(GenericRow.of(BinaryString.fromString("value"), 42)));
}
private FileStoreTable createTable(String directory, int rowCount) throws
Exception {
+ List<InternalRow> rows = new ArrayList<>();
+ for (int i = 0; i < rowCount; i++) {
+ rows.add(GenericRow.of(BinaryString.fromString("value"), i + 42));
+ }
+ return createTable(directory, rows);
+ }
+
+ private FileStoreTable createTable(String directory, List<? extends
InternalRow> rows)
+ throws Exception {
Path tablePath = new Path(tempDir.resolve(directory).toUri());
Options options = new Options();
options.set(CoreOptions.PATH, tablePath.toString());
@@ -188,8 +239,8 @@ class RawFileSplitReadTest {
BatchWriteBuilder writeBuilder = table.newBatchWriteBuilder();
try (BatchTableWrite write = writeBuilder.newWrite();
BatchTableCommit commit = writeBuilder.newCommit()) {
- for (int i = 0; i < rowCount; i++) {
- write.write(GenericRow.of(BinaryString.fromString("value"), i
+ 42));
+ for (InternalRow row : rows) {
+ write.write(row);
}
commit.commit(write.prepareCommit());
}
