This is an automated email from the ASF dual-hosted git repository. XiaoHongbo-Hope pushed a commit to branch release-2.1 in repository https://gitbox.apache.org/repos/asf/paimon.git
commit 2af7567e5fdad5b40cdd760e1120588d9a604850 Author: sanshi <[email protected]> AuthorDate: Fri Sep 18 12:26:40 2026 +0800 [core] Apply deletion vectors during data evolution file index pushdown (#9922) --- .../org/apache/paimon/io/FileIndexEvaluator.java | 47 ++++++++++-- .../paimon/operation/DataEvolutionSplitRead.java | 25 +++++-- .../apache/paimon/io/FileIndexEvaluatorTest.java | 83 ++++++++++++++-------- .../paimon/table/DataEvolutionFileIndexTest.java | 68 +++++++++++++++++- 4 files changed, 177 insertions(+), 46 deletions(-) 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 a82194cc76..42e62d7b99 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 @@ -34,6 +34,7 @@ import org.apache.paimon.utils.RoaringBitmap32; 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; @@ -53,6 +54,27 @@ public class FileIndexEvaluator { DataFileMeta file, @Nullable DeletionVector dv) throws IOException { + return evaluate( + fileIO, dataSchema, dataFilter, topN, limit, dataFilePathFactory, file, dv, 0L); + } + + /** + * Evaluates a file index with a deletion vector whose positions may be relative to a containing + * file range. + * + * @param fileOffset offset from this file's local position to the deletion vector position + */ + public static FileIndexResult evaluate( + FileIO fileIO, + TableSchema dataSchema, + List<Predicate> dataFilter, + @Nullable TopN topN, + @Nullable Integer limit, + DataFilePathFactory dataFilePathFactory, + DataFileMeta file, + @Nullable DeletionVector dv, + long fileOffset) + 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) { @@ -64,7 +86,7 @@ public class FileIndexEvaluator { return FileIndexResult.REMAIN; } else { // limit can not work with other predicates. - return createBaseSelection(file, dv).limit(limit); + return createBaseSelection(file, dv, fileOffset).limit(limit); } } @@ -74,7 +96,7 @@ public class FileIndexEvaluator { return FileIndexResult.REMAIN; } - BitmapIndexResult selection = createBaseSelection(file, dv); + BitmapIndexResult selection = createBaseSelection(file, dv, fileOffset); FileIndexResult result; if (!isNullOrEmpty(dataFilter)) { Predicate filter = PredicateBuilder.and(dataFilter.toArray(new Predicate[0])); @@ -90,26 +112,37 @@ public class FileIndexEvaluator { return FileIndexResult.REMAIN; } + if (!result.remain()) { + return FileIndexResult.SKIP; + } + // if all position selected, or if only and not the deletion // the effect will not obvious, just return REMAIN. if (Objects.equals(result, selection)) { return FileIndexResult.REMAIN; } - if (!result.remain()) { - return FileIndexResult.SKIP; - } - return result; } } private static BitmapIndexResult createBaseSelection( - DataFileMeta file, @Nullable DeletionVector dv) { + DataFileMeta file, @Nullable DeletionVector dv, long fileOffset) { BitmapIndexResult selection = new BitmapIndexResult(() -> RoaringBitmap32.bitmapOfRange(0, file.rowCount())); if (dv instanceof BitmapDeletionVector) { RoaringBitmap32 deletion = ((BitmapDeletionVector) dv).get(); + if (fileOffset != 0) { + RoaringBitmap32 localDeletion = new RoaringBitmap32(); + Iterator<Integer> iterator = deletion.iterator(); + while (iterator.hasNext()) { + long localPosition = (long) iterator.next() - fileOffset; + if (localPosition >= 0 && localPosition < file.rowCount()) { + localDeletion.add((int) localPosition); + } + } + deletion = localDeletion; + } selection = selection.andNot(deletion); } return selection; diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/DataEvolutionSplitRead.java b/paimon-core/src/main/java/org/apache/paimon/operation/DataEvolutionSplitRead.java index de0f8fd5ee..0b05ee835e 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/DataEvolutionSplitRead.java +++ b/paimon-core/src/main/java/org/apache/paimon/operation/DataEvolutionSplitRead.java @@ -670,6 +670,12 @@ public class DataEvolutionSplitRead implements SplitRead<InternalRow> { FileIndexResult fileIndexResult = null; if (fileIndexReadEnabled) { + DeletionVector dv = deletionVector == null ? null : deletionVector.deletionVector; + long fileOffset = + dv == null || dv.isEmpty() + ? 0L + : deletionVectorOffset( + file.nonNullRowIdRange(), rowRanges, deletionVector); fileIndexResult = FileIndexEvaluator.evaluate( fileIO, @@ -679,7 +685,8 @@ public class DataEvolutionSplitRead implements SplitRead<InternalRow> { null, dataFilePathFactory, file, - null); + dv, + fileOffset); if (!fileIndexResult.remain()) { return new EmptyFileRecordReader<>(); } @@ -778,6 +785,15 @@ public class DataEvolutionSplitRead implements SplitRead<InternalRow> { return reader; } + return new ApplyDeletionVectorReader( + reader, + deletionVector.deletionVector, + // Convert anchor-range DV positions to this reader's local returned positions. + deletionVectorOffset(readerRange, rowRanges, deletionVector)); + } + + private long deletionVectorOffset( + Range readerRange, List<Range> rowRanges, DeletionVectorWithRange deletionVector) { checkArgument( selectedRangesContainedByDeletionVector( readerRange, rowRanges, deletionVector.range), @@ -785,12 +801,7 @@ public class DataEvolutionSplitRead implements SplitRead<InternalRow> { deletionVector.range, rowRanges, readerRange); - - return new ApplyDeletionVectorReader( - reader, - deletionVector.deletionVector, - // Convert anchor-range DV positions to this reader's local returned positions. - readerRange.from - deletionVector.range.from); + return readerRange.from - deletionVector.range.from; } private boolean selectedRangesContainedByDeletionVector( 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 891ae8380a..bd7f781855 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 @@ -112,37 +112,60 @@ public class FileIndexEvaluatorTest { BitmapDeletionVector deletionVector = new BitmapDeletionVector(); deletionVector.delete(0); - FileIndexResult result = - FileIndexEvaluator.evaluate( - null, - tableSchema(), - Collections.singletonList( - new PredicateBuilder( - RowType.of( - new DataType[] {DataTypes.INT()}, - new String[] {"a"})) - .equal(0, 1)), - null, - null, - null, - DataFileMeta.forAppend( - "file", - 0, - 2, - SimpleStats.EMPTY_STATS, - 0, - 0, - 0, - Collections.emptyList(), - embeddedBitmapIndex(), - null, - null, - null, - null, - null), - deletionVector); + assertThat(evaluateEmbeddedIndex(deletionVector, 0L)).isSameAs(FileIndexResult.SKIP); + } + + @Test + public void testFullyDeletedFileIsSkipped() throws Exception { + BitmapDeletionVector deletionVector = new BitmapDeletionVector(); + deletionVector.delete(0); + deletionVector.delete(1); - assertThat(result).isSameAs(FileIndexResult.SKIP); + assertThat(evaluateEmbeddedIndex(deletionVector, 0L)).isSameAs(FileIndexResult.SKIP); + } + + @Test + public void testDataFilterIntersectsOffsetDeletionVector() throws Exception { + BitmapDeletionVector hit = new BitmapDeletionVector(); + hit.delete(10); + assertThat(evaluateEmbeddedIndex(hit, 10L)).isSameAs(FileIndexResult.SKIP); + + BitmapDeletionVector neighbour = new BitmapDeletionVector(); + neighbour.delete(11); + assertThat(evaluateEmbeddedIndex(neighbour, 10L).remain()).isTrue(); + } + + private static FileIndexResult evaluateEmbeddedIndex( + BitmapDeletionVector deletionVector, long fileOffset) throws IOException { + return FileIndexEvaluator.evaluate( + null, + tableSchema(), + Collections.singletonList( + new PredicateBuilder( + RowType.of( + new DataType[] {DataTypes.INT()}, + new String[] {"a"})) + .equal(0, 1)), + null, + null, + null, + DataFileMeta.forAppend( + "file", + 0, + 2, + SimpleStats.EMPTY_STATS, + 0, + 0, + 0, + Collections.emptyList(), + embeddedBitmapIndex(), + null, + null, + null, + null, + null), + deletionVector, + fileOffset); } private static TableSchema tableSchema() { diff --git a/paimon-core/src/test/java/org/apache/paimon/table/DataEvolutionFileIndexTest.java b/paimon-core/src/test/java/org/apache/paimon/table/DataEvolutionFileIndexTest.java index caa3e7d42b..9b0b7da89f 100644 --- a/paimon-core/src/test/java/org/apache/paimon/table/DataEvolutionFileIndexTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/table/DataEvolutionFileIndexTest.java @@ -33,6 +33,7 @@ import org.apache.paimon.fileindex.FileIndexOptions; import org.apache.paimon.fileindex.bitmap.BitmapFileIndexFactory; import org.apache.paimon.fileindex.bloomfilter.BloomFilterFileIndexFactory; import org.apache.paimon.fileindex.bsi.BitSliceIndexBitmapFileIndexFactory; +import org.apache.paimon.fs.Path; import org.apache.paimon.index.IndexFileMeta; import org.apache.paimon.io.CompactIncrement; import org.apache.paimon.io.DataFileMeta; @@ -477,6 +478,48 @@ public class DataEvolutionFileIndexTest extends DataEvolutionTestBase { assertRow(assertSingleRow(rows), 50); } + @Test + public void testBitmapSelectionComposesWithDeletionVectorForNonZeroRowId() throws Exception { + Map<String, String> options = bitmapOptions("f1"); + options.put(CoreOptions.DELETION_VECTORS_ENABLED.key(), "true"); + FileStoreTable table = createTable("bitmap_dv_non_zero_row_id", options); + writeAllColumns(table, ROW_COUNT); + writeAllColumns(table, ROW_COUNT); + + deleteRowsFrom(table, ROW_COUNT, 50); + + FileStoreTable latest = getTable(identifier(table.name())); + DataSplit targetSplit = + latest.newReadBuilder().newScan().plan().splits().stream() + .map(split -> (DataSplit) split) + .filter( + split -> + split.dataFiles().stream() + .anyMatch( + file -> + file.nonNullFirstRowId() + == ROW_COUNT)) + .findFirst() + .orElseThrow(IllegalStateException::new); + DataFileMeta targetFile = + targetSplit.dataFiles().stream() + .filter(file -> file.nonNullFirstRowId() == ROW_COUNT) + .findFirst() + .orElseThrow(IllegalStateException::new); + Path targetPath = + latest.store() + .pathFactory() + .createDataFilePathFactory(targetSplit.partition(), targetSplit.bucket()) + .toPath(targetFile); + assertThat(latest.fileIO().delete(targetPath, false)).isTrue(); + + // The second group can only succeed now if the bitmap hit is intersected with its DV + // before opening the missing data file. The first group remains visible. + RowType readType = rowTypeWithRowId(rowType()).project(SpecialFields.ROW_ID.name(), "f1"); + List<InternalRow> rows = readWithFilter(table, equalF1(f1(50)), readType); + assertThat(rowIds(rows)).containsExactly(50L); + } + @Test public void testMergedGroupKeptWhenFilterColumnOverwritten() throws Exception { FileStoreTable table = createTable("overwritten", Collections.emptyMap()); @@ -503,7 +546,28 @@ public class DataEvolutionFileIndexTest extends DataEvolutionTestBase { FileStoreTable latest = getTable(identifier(table.name())); List<DataFileMeta> dataFiles = ((DataSplit) latest.newReadBuilder().newScan().plan().splits().get(0)).dataFiles(); - String anchor = retrieveAnchorFile(dataFiles, file -> file).fileName(); + deleteRows(latest, retrieveAnchorFile(dataFiles, file -> file), positions); + } + + private void deleteRowsFrom(FileStoreTable table, long firstRowId, long... positions) + throws Exception { + FileStoreTable latest = getTable(identifier(table.name())); + DataFileMeta anchor = + latest.newReadBuilder().newScan().plan().splits().stream() + .map(split -> (DataSplit) split) + .flatMap(split -> split.dataFiles().stream()) + .filter(file -> file.nonNullFirstRowId() == firstRowId) + .findFirst() + .orElseThrow( + () -> + new IllegalArgumentException( + "Cannot find data file with first row id " + + firstRowId)); + deleteRows(latest, anchor, positions); + } + + private void deleteRows(FileStoreTable latest, DataFileMeta anchor, long... positions) + throws Exception { BaseAppendDeleteFileMaintainer maintainer = BaseAppendDeleteFileMaintainer.forUnawareAppend( @@ -514,7 +578,7 @@ public class DataEvolutionFileIndexTest extends DataEvolutionTestBase { for (long position : positions) { deletionVector.delete(position); } - maintainer.notifyNewDeletionVector(anchor, deletionVector); + maintainer.notifyNewDeletionVector(anchor.fileName(), deletionVector); List<IndexFileMeta> newIndexFiles = new ArrayList<>(); for (IndexManifestEntry entry : maintainer.persist()) {
