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 05a4375aa3 [core] Apply deletion vectors during data evolution file
index pushdown (#9922)
05a4375aa3 is described below
commit 05a4375aa3c638b5127228936e7d742f3a975391
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()) {
