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 1b47fba765 [core] Push down deletion vectors for merged data evolution
groups (#9987)
1b47fba765 is described below
commit 1b47fba765d11ad244d355dbad89041f751fef3d
Author: sanshi <[email protected]>
AuthorDate: Thu Sep 24 17:41:21 2026 +0800
[core] Push down deletion vectors for merged data evolution groups (#9987)
---
.../org/apache/paimon/io/FileIndexEvaluator.java | 36 +++++++++
.../paimon/operation/DataEvolutionSplitRead.java | 72 +++++++++++++++---
.../paimon/table/DataEvolutionFileIndexTest.java | 86 ++++++++++++++++++++--
3 files changed, 175 insertions(+), 19 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 42e62d7b99..9d88923f2f 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
@@ -126,6 +126,42 @@ public class FileIndexEvaluator {
}
}
+ /**
+ * Intersects a previously evaluated file-index result with the live rows
from a deletion
+ * vector.
+ *
+ * <p>The file index uses positions local to {@code file}, while a
deletion vector may use
+ * positions relative to the merged group's anchor range. {@code
fileOffset} converts the latter
+ * to the former. Keeping this operation separate from {@link #evaluate}
allows callers to reuse
+ * an index result instead of reading the file-index sidecar a second time.
+ *
+ * <p>Only bitmap index results can be narrowed to a subset of live rows.
A non-bitmap result
+ * remains conservative, except that an empty live-row selection always
proves that the file can
+ * be skipped.
+ */
+ public static FileIndexResult intersectDeletionVector(
+ FileIndexResult result,
+ DataFileMeta file,
+ @Nullable DeletionVector dv,
+ long fileOffset) {
+ if (file.rowCount() > RoaringBitmap32.MAX_VALUE
+ || dv == null
+ || dv.isEmpty()
+ || dv instanceof Bitmap64DeletionVector) {
+ return result;
+ }
+ BitmapIndexResult liveRows = createBaseSelection(file, dv, fileOffset);
+ if (!liveRows.remain()) {
+ return FileIndexResult.SKIP;
+ }
+ if (result instanceof BitmapIndexResult) {
+ FileIndexResult intersected = result.and(liveRows);
+ return intersected.remain() ? intersected : FileIndexResult.SKIP;
+ }
+ return result;
+ }
+
+ /** Returns all local file positions minus the positions deleted by a
bitmap DV. */
private static BitmapIndexResult createBaseSelection(
DataFileMeta file, @Nullable DeletionVector dv, long fileOffset) {
BitmapIndexResult 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 0b05ee835e..696831fbf6 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
@@ -250,11 +250,23 @@ public class DataEvolutionSplitRead implements
SplitRead<InternalRow> {
} else {
suppliers.add(
() -> {
- if (skipByFileIndex(filters, needMergeFiles,
dataFilePathFactory)) {
+ // Reject the group using only file indexes before
opening its DV.
+ // This keeps the common selective path free of a
DV sidecar read.
+ List<FileIndexResultEntry> fileIndexResults =
+ evaluateFileIndexes(
+ filters, needMergeFiles,
dataFilePathFactory);
+ if (fileIndexResults.stream()
+ .anyMatch(entry ->
!entry.result.remain())) {
return new EmptyFileRecordReader<>();
}
DeletionVectorWithRange deletionVector =
readDeletionVector(needMergeFiles,
deletionVectorFactory);
+ if (deletionVector != null
+ && !deletionVector.deletionVector.isEmpty()
+ && skipByFileIndex(
+ fileIndexResults, rowRanges,
deletionVector)) {
+ return new EmptyFileRecordReader<>();
+ }
return createUnionReader(
needMergeFiles,
partition,
@@ -822,26 +834,24 @@ public class DataEvolutionSplitRead implements
SplitRead<InternalRow> {
}
/**
- * Whether the file index proves that no row of a merged group can match
the filters. Only plain
- * data files are considered: {@link #mergeRangesAndSort} guarantees they
all span the row id
- * range of the whole group, while a blob or vector-store file only covers
a sub range and can
- * not prove anything for the other rows.
+ * Evaluates each applicable file index once for the winning fields of a
merged row-id group.
*
- * <p>A column can be written by several files of the group; the column
merge takes each field
- * from the newest file that wrote it and older copies are dead. Files
arrive newest first
- * ({@link #mergeRangesAndSort}), so each file's index is evaluated only
over the columns it is
- * the newest writer of. A stale value in an older file must not veto a
group whose winning file
- * matches, mirroring the winner selection in {@link
DataEvolutionFileStoreScan#evolutionStats}.
+ * <p>Only normal data files are considered. Blob and vector-store files
can cover only a subset
+ * of the group range, so their indexes cannot prove that the whole group
has no match. Files
+ * are visited newest first; {@code claimedFieldIds} prevents an older
copy of an overwritten
+ * field from vetoing the group. The returned results are retained so a
later deletion-vector
+ * pass can intersect them without reopening file-index sidecars.
*/
- private boolean skipByFileIndex(
+ private List<FileIndexResultEntry> evaluateFileIndexes(
@Nullable List<Predicate> filters,
List<DataFileMeta> files,
DataFilePathFactory pathFactory)
throws IOException {
if (!fileIndexReadEnabled || isNullOrEmpty(filters)) {
- return false;
+ return Collections.emptyList();
}
+ List<FileIndexResultEntry> results = new ArrayList<>();
Set<Integer> claimedFieldIds = new HashSet<>();
for (DataFileMeta file : files) {
if (isBlobFile(file.fileName()) ||
isVectorStoreFile(file.fileName())) {
@@ -868,6 +878,33 @@ public class DataEvolutionSplitRead implements
SplitRead<InternalRow> {
FileIndexResult result =
FileIndexEvaluator.evaluate(
fileIO, dataSchema, dataFilters, null, null,
pathFactory, file, null);
+ results.add(new FileIndexResultEntry(file, result));
+ if (!result.remain()) {
+ return results;
+ }
+ }
+ return results;
+ }
+
+ /**
+ * Applies a group deletion vector to already evaluated file-index results.
+ *
+ * <p>A merged group can only be skipped when one of the files responsible
for the final
+ * predicate values has no live candidate rows. This method only
intersects the saved results;
+ * it does not reopen or reevaluate any file index. The offset passed to
the evaluator maps
+ * positions in the group's DV anchor range to positions local to the
current file.
+ */
+ private boolean skipByFileIndex(
+ List<FileIndexResultEntry> fileIndexResults,
+ List<Range> rowRanges,
+ DeletionVectorWithRange deletionVector) {
+ DeletionVector dv = deletionVector.deletionVector;
+ for (FileIndexResultEntry entry : fileIndexResults) {
+ long fileOffset =
+ deletionVectorOffset(entry.file.nonNullRowIdRange(),
rowRanges, deletionVector);
+ FileIndexResult result =
+ FileIndexEvaluator.intersectDeletionVector(
+ entry.result, entry.file, dv, fileOffset);
if (!result.remain()) {
return true;
}
@@ -1139,6 +1176,17 @@ public class DataEvolutionSplitRead implements
SplitRead<InternalRow> {
}
}
+ private static class FileIndexResultEntry {
+
+ private final DataFileMeta file;
+ private final FileIndexResult result;
+
+ private FileIndexResultEntry(DataFileMeta file, FileIndexResult
result) {
+ this.file = file;
+ this.result = result;
+ }
+ }
+
private static class DeletionVectorWithRange {
private final Range range;
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 9b0b7da89f..30f4be2dca 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
@@ -71,6 +71,7 @@ import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.Objects;
import java.util.stream.Collectors;
import static org.apache.paimon.table.SpecialFields.rowTypeWithRowId;
@@ -541,6 +542,77 @@ public class DataEvolutionFileIndexTest extends
DataEvolutionTestBase {
assertThat(query(table, equalF1(f1(50)))).isEmpty();
}
+ @Test
+ public void testMergedGroupFileIndexComposesWithDeletionVector() throws
Exception {
+ Map<String, String> options = new HashMap<>();
+ options.put(CoreOptions.DELETION_VECTORS_ENABLED.key(), "true");
+ FileStoreTable table = createTable("merged_bitmap_dv", options);
+ writeSplitColumns(table, ROW_COUNT, bitmapOptions("f1"),
Collections.emptyMap());
+ writeSplitColumns(table, ROW_COUNT, bitmapOptions("f1"),
Collections.emptyMap());
+
+ 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 anchor =
+ retrieveAnchorFile(
+ targetSplit.dataFiles().stream()
+ .filter(file -> file.nonNullFirstRowId() ==
ROW_COUNT)
+ .collect(Collectors.toList()),
+ file -> file);
+ Path anchorPath =
+ latest.store()
+ .pathFactory()
+ .createDataFilePathFactory(targetSplit.partition(),
targetSplit.bucket())
+ .toPath(anchor);
+ assertThat(latest.fileIO().delete(anchorPath, false)).isTrue();
+
+ // The deleted row is the only bitmap hit in the second merged group.
The missing anchor
+ // file therefore proves that the group was skipped before any union
reader opened it.
+ RowType readType =
+
rowTypeWithRowId(rowType()).project(SpecialFields.ROW_ID.name(), "f1", "f2");
+ List<InternalRow> rows = readWithFilter(table, equalF1(f1(50)),
readType);
+ assertThat(rowIds(rows)).containsExactlyElementsOf(rowIds(0,
ROW_COUNT));
+
+ FileStoreTable neighbour = createTable("merged_bitmap_dv_neighbour",
options);
+ writeSplitColumns(neighbour, ROW_COUNT, bitmapOptions("f1"),
Collections.emptyMap());
+ deleteRows(neighbour, 51);
+ assertRow(assertSingleRow(query(neighbour, equalF1(f1(50)))), 50);
+ }
+
+ @Test
+ public void testMergedGroupFileIndexSkipsBeforeReadingDeletionVector()
throws Exception {
+ Map<String, String> options = new HashMap<>();
+ options.put(CoreOptions.DELETION_VECTORS_ENABLED.key(), "true");
+ FileStoreTable table = createTable("merged_bitmap_before_dv", options);
+ writeSplitColumns(table, ROW_COUNT, bitmapOptions("f1"),
Collections.emptyMap());
+ deleteRows(table, 50);
+
+ FileStoreTable latest = getTable(identifier(table.name()));
+ DataSplit split = (DataSplit)
latest.newReadBuilder().newScan().plan().splits().get(0);
+ Path deletionVectorPath =
+ split.deletionFiles().get().stream()
+ .filter(Objects::nonNull)
+ .map(file -> new Path(file.path()))
+ .findFirst()
+ .orElseThrow(IllegalStateException::new);
+ assertThat(latest.fileIO().delete(deletionVectorPath, false)).isTrue();
+
+ // The bitmap index already rejects this value, so the missing DV file
must not be read.
+ assertThat(readWithFilter(table, equalF1(MISSING_F1))).isEmpty();
+ }
+
/** Commits a deletion vector for the anchor file of the only row id group
of {@code table}. */
private void deleteRows(FileStoreTable table, long... positions) throws
Exception {
FileStoreTable latest = getTable(identifier(table.name()));
@@ -552,17 +624,17 @@ public class DataEvolutionFileIndexTest extends
DataEvolutionTestBase {
private void deleteRowsFrom(FileStoreTable table, long firstRowId, long...
positions)
throws Exception {
FileStoreTable latest = getTable(identifier(table.name()));
- DataFileMeta anchor =
+ List<DataFileMeta> group =
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));
+ .collect(Collectors.toList());
+ if (group.isEmpty()) {
+ throw new IllegalArgumentException(
+ "Cannot find data file with first row id " + firstRowId);
+ }
+ DataFileMeta anchor = retrieveAnchorFile(group, file -> file);
deleteRows(latest, anchor, positions);
}
