From f3e488162700d622a3296f430fd539b45e63ec36 Mon Sep 17 00:00:00 2001 From: QuakeWang Date: Mon, 10 Aug 2026 15:22:33 +0800 Subject: [PATCH] [core] Fix file index pushdown for bitmap64 deletion vectors File index limit and TopN selections only excluded 32-bit deletion vectors, so bitmap64 deletions could reduce pushed-down results after reading. Large files could also force unsupported 32-bit position selections. Project bounded bitmap64 positions for file indexes, fall back when positions cannot be represented, and enforce limits after deletion-vector filtering. Signed-off-by: QuakeWang --- .../utils/OptimizedRoaringBitmap64.java | 16 + .../apache/paimon/utils/RoaringBitmap32.java | 4 + .../Bitmap64DeletionVector.java | 5 + .../apache/paimon/io/FileIndexEvaluator.java | 106 +++++- .../paimon/operation/RawFileSplitRead.java | 8 +- .../deletionvectors/DeletionVectorTest.java | 13 + .../paimon/io/FileIndexEvaluatorTest.java | 353 ++++++++++++++++++ .../operation/RawFileSplitReadTest.java | 58 ++- .../table/PrimaryKeySimpleTableTest.java | 17 +- 9 files changed, 561 insertions(+), 19 deletions(-) create mode 100644 paimon-core/src/test/java/org/apache/paimon/io/FileIndexEvaluatorTest.java diff --git a/paimon-common/src/main/java/org/apache/paimon/utils/OptimizedRoaringBitmap64.java b/paimon-common/src/main/java/org/apache/paimon/utils/OptimizedRoaringBitmap64.java index 7171771e9e65..02f92eaaa7ea 100644 --- a/paimon-common/src/main/java/org/apache/paimon/utils/OptimizedRoaringBitmap64.java +++ b/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; diff --git a/paimon-common/src/main/java/org/apache/paimon/utils/RoaringBitmap32.java b/paimon-common/src/main/java/org/apache/paimon/utils/RoaringBitmap32.java index adb7b0d6112d..375300869f85 100644 --- a/paimon-common/src/main/java/org/apache/paimon/utils/RoaringBitmap32.java +++ b/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. * diff --git a/paimon-core/src/main/java/org/apache/paimon/deletionvectors/Bitmap64DeletionVector.java b/paimon-core/src/main/java/org/apache/paimon/deletionvectors/Bitmap64DeletionVector.java index ee06d7232b38..0ca1784f18be 100644 --- a/paimon-core/src/main/java/org/apache/paimon/deletionvectors/Bitmap64DeletionVector.java +++ b/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 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 025887cefccd..39d9d609d933 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; @@ -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,27 +59,49 @@ 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.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; @@ -85,7 +109,7 @@ public static FileIndexResult evaluate( // 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 iterator = candidates.get().iterator(); + while (iterator.hasNext()) { + int position = iterator.next(); + if (!dv.isDeleted(position)) { + result.add(position); + } + } + return result; + }); } @Nullable 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 87f7063a81c0..63d0ab884f9f 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.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 createReader( null)); } - return ConcatRecordReader.create(suppliers); + RecordReader 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 createFileReader( diff --git a/paimon-core/src/test/java/org/apache/paimon/deletionvectors/DeletionVectorTest.java b/paimon-core/src/test/java/org/apache/paimon/deletionvectors/DeletionVectorTest.java index 1800f4c3328e..465fe9ed4897 100644 --- a/paimon-core/src/test/java/org/apache/paimon/deletionvectors/DeletionVectorTest.java +++ b/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 toDelete = new HashSet<>(); 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 new file mode 100644 index 000000000000..1f1d2a39167a --- /dev/null +++ b/paimon-core/src/test/java/org/apache/paimon/io/FileIndexEvaluatorTest.java @@ -0,0 +1,353 @@ +/* + * 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.io; + +import org.apache.paimon.deletionvectors.Bitmap64DeletionVector; +import org.apache.paimon.deletionvectors.BitmapDeletionVector; +import org.apache.paimon.deletionvectors.DeletionVector; +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.fileindex.bloomfilter.BloomFilterFileIndex; +import org.apache.paimon.fileindex.bloomfilter.BloomFilterFileIndexFactory; +import org.apache.paimon.fileindex.rangebitmap.RangeBitmapFileIndex; +import org.apache.paimon.fileindex.rangebitmap.RangeBitmapFileIndexFactory; +import org.apache.paimon.options.Options; +import org.apache.paimon.predicate.FieldRef; +import org.apache.paimon.predicate.Predicate; +import org.apache.paimon.predicate.PredicateBuilder; +import org.apache.paimon.predicate.TopN; +import org.apache.paimon.schema.TableSchema; +import org.apache.paimon.stats.SimpleStats; +import org.apache.paimon.types.DataField; +import org.apache.paimon.types.DataTypes; +import org.apache.paimon.utils.RoaringBitmap32; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; + +import java.io.ByteArrayOutputStream; +import java.util.Collections; +import java.util.Map; +import java.util.function.LongConsumer; + +import static org.apache.paimon.predicate.SortValue.NullOrdering.NULLS_LAST; +import static org.apache.paimon.predicate.SortValue.SortDirection.ASCENDING; +import static org.assertj.core.api.Assertions.assertThat; + +/** Tests for {@link FileIndexEvaluator}. */ +class FileIndexEvaluatorTest { + + private static final String FIELD_NAME = "value"; + + @ParameterizedTest + @ValueSource(booleans = {false, true}) + void testLimitSkipsDeletedPositions(boolean bitmap64) throws Exception { + DataFileMeta file = DataFileTestUtils.newFile("data.avro", 0, 0, 19, 0L); + DeletionVector deletionVector = + bitmap64 ? new Bitmap64DeletionVector() : new BitmapDeletionVector(); + for (int position = 0; position < 5; position++) { + deletionVector.delete(position); + } + + FileIndexResult result = + FileIndexEvaluator.evaluate( + null, null, Collections.emptyList(), null, 10, null, file, deletionVector); + + assertThat(result).isInstanceOf(BitmapIndexResult.class); + assertThat(((BitmapIndexResult) result).get()) + .isEqualTo(RoaringBitmap32.bitmapOfRange(5, 15)); + } + + @Test + void testLimitWithHighBitmap64DeletionPosition() throws Exception { + DataFileMeta file = fileWithRowCount(Integer.MAX_VALUE + 2L); + DeletionVector deletionVector = new Bitmap64DeletionVector(); + deletionVector.delete(0); + deletionVector.delete(Integer.MAX_VALUE + 1L); + + FileIndexResult result = + FileIndexEvaluator.evaluate( + null, null, Collections.emptyList(), null, 10, null, file, deletionVector); + + assertThat(result).isInstanceOf(BitmapIndexResult.class); + assertThat(((BitmapIndexResult) result).get()) + .isEqualTo(RoaringBitmap32.bitmapOfRange(1, 11)); + } + + @Test + void testLimitDoesNotIterateBitmap64DeletionVector() throws Exception { + DataFileMeta file = DataFileTestUtils.newFile("data.avro", 0, 0, 19, 0L); + DeletionVector deletionVector = + new Bitmap64DeletionVector() { + @Override + public void forEachDeletedPosition(LongConsumer consumer) { + throw new AssertionError( + "Limit evaluation must not expand the deletion vector."); + } + }; + deletionVector.delete(0); + + FileIndexResult result = + FileIndexEvaluator.evaluate( + null, null, Collections.emptyList(), null, 1, null, file, deletionVector); + + assertThat(result).isInstanceOf(BitmapIndexResult.class); + assertThat(((BitmapIndexResult) result).get()).isEqualTo(RoaringBitmap32.bitmapOf(1)); + } + + @Test + void testFilterAbandonsBitmapPushdownForLargeFile() throws Exception { + TableSchema schema = tableSchema(); + FileIndexWriter indexWriter = + new BitmapFileIndex(DataTypes.INT(), new Options()).createWriter(); + indexWriter.writeRecord(0); + DataFileMeta file = + fileWithRowCount(Integer.MAX_VALUE + 2L) + .copy(embeddedIndex(BitmapFileIndexFactory.BITMAP_INDEX, indexWriter)); + DeletionVector deletionVector = new Bitmap64DeletionVector(); + deletionVector.delete(Integer.MAX_VALUE + 1L); + Predicate filter = new PredicateBuilder(schema.logicalRowType()).equal(0, 0); + + FileIndexResult result = + FileIndexEvaluator.evaluate( + null, + schema, + Collections.singletonList(filter), + null, + null, + null, + file, + deletionVector); + + assertThat(result).isSameAs(FileIndexResult.REMAIN); + } + + @ParameterizedTest + @ValueSource(booleans = {false, true}) + void testFilterExcludesDeletedPositions(boolean bitmap64) throws Exception { + TableSchema schema = tableSchema(); + FileIndexWriter indexWriter = + new BitmapFileIndex(DataTypes.INT(), new Options()).createWriter(); + for (int position = 0; position < 10; position++) { + indexWriter.writeRecord(position % 2); + } + DataFileMeta file = + DataFileTestUtils.newFile("data.avro", 0, 0, 9, 0L) + .copy(embeddedIndex(BitmapFileIndexFactory.BITMAP_INDEX, indexWriter)); + DeletionVector deletionVector = + bitmap64 + ? new Bitmap64DeletionVector() { + @Override + public void forEachDeletedPosition(LongConsumer consumer) { + throw new AssertionError( + "Bitmap filter evaluation must not iterate the deletion vector."); + } + + @Override + public RoaringBitmap32 projectToBitmap32(long maxExclusive) { + throw new AssertionError( + "Bitmap filter evaluation must not project the deletion vector."); + } + } + : new BitmapDeletionVector(); + deletionVector.delete(0); + deletionVector.delete(2); + Predicate filter = new PredicateBuilder(schema.logicalRowType()).equal(0, 0); + + FileIndexResult result = + FileIndexEvaluator.evaluate( + null, + schema, + Collections.singletonList(filter), + null, + null, + null, + file, + deletionVector); + + assertThat(result).isInstanceOf(BitmapIndexResult.class); + assertThat(((BitmapIndexResult) result).get()).isEqualTo(RoaringBitmap32.bitmapOf(4, 6, 8)); + } + + @Test + void testTopNDoesNotIterateBitmap64DeletionVector() throws Exception { + TableSchema schema = tableSchema(); + FileIndexWriter indexWriter = + new RangeBitmapFileIndex(DataTypes.INT(), new Options()).createWriter(); + for (int value = 0; value < 5; value++) { + indexWriter.writeRecord(value); + } + DataFileMeta file = + fileWithRowCount(5) + .copy(embeddedIndex(RangeBitmapFileIndexFactory.RANGE_BITMAP, indexWriter)); + DeletionVector deletionVector = + new Bitmap64DeletionVector() { + @Override + public void forEachDeletedPosition(LongConsumer consumer) { + throw new AssertionError( + "TopN evaluation must not iterate the deletion vector."); + } + }; + deletionVector.delete(0); + deletionVector.delete(4); + FieldRef field = new FieldRef(0, FIELD_NAME, DataTypes.INT()); + TopN topN = new TopN(field, ASCENDING, NULLS_LAST, 2); + + FileIndexResult result = + FileIndexEvaluator.evaluate( + null, + schema, + Collections.emptyList(), + topN, + null, + null, + file, + deletionVector); + + assertThat(result).isInstanceOf(BitmapIndexResult.class); + assertThat(((BitmapIndexResult) result).get()).isEqualTo(RoaringBitmap32.bitmapOf(1, 2)); + } + + @Test + void testBloomRemainDoesNotIterateBitmap64DeletionVector() throws Exception { + FileIndexWriter indexWriter = + new BloomFilterFileIndex(DataTypes.INT(), new Options()).createWriter(); + indexWriter.writeRecord(1); + + FileIndexResult result = evaluateBloomFilter(indexWriter, 1); + + assertThat(result).isSameAs(FileIndexResult.REMAIN); + } + + @Test + void testBloomSkipDoesNotIterateBitmap64DeletionVector() throws Exception { + FileIndexWriter indexWriter = + new BloomFilterFileIndex(DataTypes.INT(), new Options()).createWriter(); + indexWriter.writeRecord(null); + + FileIndexResult result = evaluateBloomFilter(indexWriter, 1); + + assertThat(result).isSameAs(FileIndexResult.SKIP); + } + + @Test + void testBloomSkipForLargeFile() throws Exception { + TableSchema schema = tableSchema(); + FileIndexWriter indexWriter = + new BloomFilterFileIndex(DataTypes.INT(), new Options()).createWriter(); + indexWriter.writeRecord(null); + DataFileMeta file = + fileWithRowCount(Integer.MAX_VALUE + 2L) + .copy(embeddedIndex(BloomFilterFileIndexFactory.BLOOM_FILTER, indexWriter)); + Predicate filter = new PredicateBuilder(schema.logicalRowType()).equal(0, 1); + + FileIndexResult result = + FileIndexEvaluator.evaluate( + null, + schema, + Collections.singletonList(filter), + null, + null, + null, + file, + null); + + assertThat(result).isSameAs(FileIndexResult.SKIP); + } + + private static FileIndexResult evaluateBloomFilter(FileIndexWriter indexWriter, int value) + throws Exception { + TableSchema schema = tableSchema(); + DataFileMeta file = + DataFileTestUtils.newFile("data.avro", 0, 0, 0, 0L) + .copy(embeddedIndex(BloomFilterFileIndexFactory.BLOOM_FILTER, indexWriter)); + DeletionVector deletionVector = + new Bitmap64DeletionVector() { + @Override + public void forEachDeletedPosition(LongConsumer consumer) { + throw new AssertionError( + "Bloom filter evaluation must not expand the deletion vector."); + } + + @Override + public RoaringBitmap32 projectToBitmap32(long maxExclusive) { + throw new AssertionError( + "Bloom filter evaluation must not project the deletion vector."); + } + }; + deletionVector.delete(0); + Predicate filter = new PredicateBuilder(schema.logicalRowType()).equal(0, value); + + return FileIndexEvaluator.evaluate( + null, + schema, + Collections.singletonList(filter), + null, + null, + null, + file, + deletionVector); + } + + private static TableSchema tableSchema() { + DataField field = new DataField(0, FIELD_NAME, DataTypes.INT()); + return new TableSchema( + 0, + Collections.singletonList(field), + field.id(), + Collections.emptyList(), + Collections.emptyList(), + Collections.emptyMap(), + 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[] embeddedIndex(String indexType, FileIndexWriter indexWriter) + throws Exception { + Map indexes = + Collections.singletonMap(indexType, indexWriter.serializedBytes()); + ByteArrayOutputStream out = new ByteArrayOutputStream(); + try (FileIndexFormat.Writer writer = FileIndexFormat.createWriter(out)) { + writer.writeColumnIndexes(Collections.singletonMap(FIELD_NAME, indexes)); + } + return out.toByteArray(); + } +} 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 fd4fcb2d4853..1ece67555efe 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; @@ -43,11 +46,15 @@ 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; @@ -133,12 +140,59 @@ public FormatReaderFactory createReaderFactory( } } + @Test + void testLimitAfterBitmap64DeletionVectorWithoutFileIndex() throws Exception { + List 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, false); + 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> deletionVectorFactories = + Collections.singletonMap(fileName, () -> deletionVector); + + RawFileSplitRead read = ((AppendOnlyFileStore) table.store()).newRead(); + read.withLimit(10); + AtomicInteger count = new AtomicInteger(); + try (RecordReader reader = + read.createReader( + split.partition(), + split.bucket(), + split.dataFiles(), + deletionVectorFactories)) { + reader.forEachRemaining(ignored -> count.incrementAndGet()); + } + + assertThat(count).hasValue(10); + } + private FileStoreTable createTable(String directory) throws Exception { + return createTable( + directory, + Collections.singletonList(GenericRow.of(BinaryString.fromString("value"), 42))); + } + + private FileStoreTable createTable(String directory, List rows) + throws Exception { + return createTable(directory, rows, true); + } + + private FileStoreTable createTable( + String directory, List rows, boolean fileIndexReadEnabled) + throws Exception { Path tablePath = new Path(tempDir.resolve(directory).toUri()); Options options = new Options(); options.set(CoreOptions.PATH, tablePath.toString()); options.set(CoreOptions.BUCKET, 1); options.set(CoreOptions.BUCKET_KEY, "first"); + options.set(CoreOptions.FILE_INDEX_READ_ENABLED, fileIndexReadEnabled); Schema schema = Schema.newBuilder() .column("first", DataTypes.STRING()) @@ -153,7 +207,9 @@ private FileStoreTable createTable(String directory) throws Exception { BatchWriteBuilder writeBuilder = table.newBatchWriteBuilder(); try (BatchTableWrite write = writeBuilder.newWrite(); BatchTableCommit commit = writeBuilder.newCommit()) { - write.write(GenericRow.of(BinaryString.fromString("value"), 42)); + for (InternalRow row : rows) { + write.write(row); + } commit.commit(write.prepareCommit()); } return table; diff --git a/paimon-core/src/test/java/org/apache/paimon/table/PrimaryKeySimpleTableTest.java b/paimon-core/src/test/java/org/apache/paimon/table/PrimaryKeySimpleTableTest.java index 0fa383dd4d94..a219e64302ec 100644 --- a/paimon-core/src/test/java/org/apache/paimon/table/PrimaryKeySimpleTableTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/table/PrimaryKeySimpleTableTest.java @@ -126,6 +126,7 @@ import static org.apache.paimon.CoreOptions.CHANGELOG_PRODUCER; import static org.apache.paimon.CoreOptions.ChangelogProducer.LOOKUP; import static org.apache.paimon.CoreOptions.DELETION_VECTORS_ENABLED; +import static org.apache.paimon.CoreOptions.DELETION_VECTOR_BITMAP64; import static org.apache.paimon.CoreOptions.FILE_FORMAT; import static org.apache.paimon.CoreOptions.FILE_FORMAT_PARQUET; import static org.apache.paimon.CoreOptions.FILE_FORMAT_PER_LEVEL; @@ -1369,8 +1370,9 @@ public void testDeletionVectorsCombineWithFileIndexPushDownParquet() throws Exce } } - @Test - public void testTopNPushDownInDeletionVectorMode() throws Exception { + @ParameterizedTest + @ValueSource(booleans = {false, true}) + public void testTopNPushDownInDeletionVectorMode(boolean bitmap64) throws Exception { String indexColumnName = "b"; FileStoreTable table = createFileStoreTable( @@ -1378,6 +1380,7 @@ public void testTopNPushDownInDeletionVectorMode() throws Exception { conf.set(BUCKET, 1); conf.set(FILE_FORMAT, FILE_FORMAT_PARQUET); conf.set(DELETION_VECTORS_ENABLED, true); + conf.set(DELETION_VECTOR_BITMAP64, bitmap64); conf.set("parquet.block.size", "524288"); conf.set("parquet.page.size.row.check.min", "100"); conf.set("parquet.page.row.count.limit", "300"); @@ -1413,7 +1416,7 @@ public void testTopNPushDownInDeletionVectorMode() throws Exception { // test bottom k { - int k = new Random().nextInt(100); + int k = 50; RoaringBitmap32 bitmap = RoaringBitmap32.bitmapOfRange(min, min + k); DataField field = table.schema().nameToFieldMap().get(indexColumnName); FieldRef ref = new FieldRef(field.id(), field.name(), field.type()); @@ -1435,7 +1438,7 @@ public void testTopNPushDownInDeletionVectorMode() throws Exception { // test top k { - int k = new Random().nextInt(100); + int k = 50; RoaringBitmap32 bitmap = RoaringBitmap32.bitmapOfRange(max - k, max); DataField field = table.schema().nameToFieldMap().get(indexColumnName); FieldRef ref = new FieldRef(field.id(), field.name(), field.type()); @@ -1456,14 +1459,16 @@ public void testTopNPushDownInDeletionVectorMode() throws Exception { } } - @Test - public void testLimitPushDownInDeletionVectorMode() throws Exception { + @ParameterizedTest + @ValueSource(booleans = {false, true}) + public void testLimitPushDownInDeletionVectorMode(boolean bitmap64) throws Exception { FileStoreTable table = createFileStoreTable( conf -> { conf.set(BUCKET, 2); conf.set(FILE_FORMAT, FILE_FORMAT_PARQUET); conf.set(DELETION_VECTORS_ENABLED, true); + conf.set(DELETION_VECTOR_BITMAP64, bitmap64); conf.set(SOURCE_SPLIT_TARGET_SIZE, MemorySize.ofBytes(1)); conf.set("parquet.block.size", "524288"); conf.set("parquet.page.size.row.check.min", "100");