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 9731d53979 [core] Push down DV-aware bitmap selection for merged
groups (#10387)
9731d53979 is described below
commit 9731d5397952fd6fe556dcba1d36852cd8300ddf
Author: sanshi <[email protected]>
AuthorDate: Mon Oct 5 20:40:19 2026 +0800
[core] Push down DV-aware bitmap selection for merged groups (#10387)
---
.../paimon/operation/DataEvolutionSplitRead.java | 29 +++++++----
.../paimon/table/DataEvolutionFileIndexTest.java | 58 ++++++++++++++++++++++
2 files changed, 76 insertions(+), 11 deletions(-)
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 470a15f4a5..b053834633 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
@@ -286,10 +286,13 @@ public class DataEvolutionSplitRead implements
SplitRead<InternalRow> {
DeletionVectorWithRange deletionVector =
readDeletionVector(needMergeFiles,
deletionVectorFactory);
if (deletionVector != null
- && !deletionVector.deletionVector.isEmpty()
- && skipByFileIndex(
- fileIndexResults, rowRanges,
deletionVector)) {
- return new EmptyFileRecordReader<>();
+ &&
!deletionVector.deletionVector.isEmpty()) {
+ fileIndexResults =
+ intersectFileIndexResults(
+ fileIndexResults, rowRanges,
deletionVector);
+ if (fileIndexResults == null) {
+ return new EmptyFileRecordReader<>();
+ }
}
BitmapIndexResult groupSelection =
buildGroupSelection(needMergeFiles,
fileIndexResults);
@@ -994,16 +997,19 @@ public class DataEvolutionSplitRead implements
SplitRead<InternalRow> {
/**
* 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.
+ * <p>A merged group can be skipped when one of the files responsible for
the final predicate
+ * values has no live candidate rows. Otherwise, the intersected results
are retained for the
+ * group-level bitmap selection. 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(
+ @Nullable
+ private List<FileIndexResultEntry> intersectFileIndexResults(
List<FileIndexResultEntry> fileIndexResults,
List<Range> rowRanges,
DeletionVectorWithRange deletionVector) {
DeletionVector dv = deletionVector.deletionVector;
+ List<FileIndexResultEntry> intersectedResults = new
ArrayList<>(fileIndexResults.size());
for (FileIndexResultEntry entry : fileIndexResults) {
long fileOffset =
deletionVectorOffset(entry.file.nonNullRowIdRange(),
rowRanges, deletionVector);
@@ -1011,10 +1017,11 @@ public class DataEvolutionSplitRead implements
SplitRead<InternalRow> {
FileIndexEvaluator.intersectDeletionVector(
entry.result, entry.file, dv, fileOffset);
if (!result.remain()) {
- return true;
+ return null;
}
+ intersectedResults.add(new FileIndexResultEntry(entry.file,
result));
}
- return false;
+ return intersectedResults;
}
/**
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 e323817166..c2629f4f98 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
@@ -19,6 +19,7 @@
package org.apache.paimon.table;
import org.apache.paimon.CoreOptions;
+import org.apache.paimon.append.ForceSingleBatchReader;
import org.apache.paimon.catalog.Identifier;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.data.BinaryString;
@@ -26,10 +27,12 @@ import org.apache.paimon.data.GenericRow;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.data.Timestamp;
import org.apache.paimon.data.serializer.InternalRowSerializer;
+import org.apache.paimon.deletionvectors.ApplyDeletionVectorReader;
import org.apache.paimon.deletionvectors.BitmapDeletionVector;
import org.apache.paimon.deletionvectors.DeletionVector;
import org.apache.paimon.deletionvectors.append.BaseAppendDeleteFileMaintainer;
import org.apache.paimon.fileindex.FileIndexOptions;
+import org.apache.paimon.fileindex.bitmap.ApplyBitmapIndexRecordReader;
import org.apache.paimon.fileindex.bitmap.BitmapFileIndexFactory;
import org.apache.paimon.fileindex.bloomfilter.BloomFilterFileIndexFactory;
import org.apache.paimon.fileindex.bsi.BitSliceIndexBitmapFileIndexFactory;
@@ -42,6 +45,7 @@ import org.apache.paimon.manifest.FileKind;
import org.apache.paimon.manifest.IndexManifestEntry;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.predicate.PredicateBuilder;
+import org.apache.paimon.reader.DataEvolutionFileReader;
import org.apache.paimon.reader.RecordReader;
import org.apache.paimon.schema.Schema;
import org.apache.paimon.schema.SchemaChange;
@@ -65,6 +69,7 @@ import org.junit.jupiter.params.provider.ValueSource;
import javax.annotation.Nullable;
+import java.lang.reflect.Field;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
@@ -649,6 +654,31 @@ public class DataEvolutionFileIndexTest extends
DataEvolutionTestBase {
assertRow(assertSingleRow(query(neighbour, equalF1(f1(50)))), 50);
}
+ @Test
+ public void testMergedGroupDvAwareBitmapSelectionIsPushedDown() throws
Exception {
+ Map<String, String> options = new HashMap<>();
+ options.put(CoreOptions.DELETION_VECTORS_ENABLED.key(), "true");
+ FileStoreTable table = createTable("merged_bitmap_dv_partial",
options);
+ writeSplitColumns(table, ROW_COUNT, bitmapOptions("f1"),
Collections.emptyMap());
+
+ deleteRows(table, 51);
+
+ Predicate filter = PredicateBuilder.or(equalF1(f1(50)),
equalF1(f1(51)));
+ FileStoreTable latest = getTable(identifier(table.name()));
+ ReadBuilder readBuilder = latest.newReadBuilder().withFilter(filter);
+ DataSplit split = (DataSplit)
readBuilder.newScan().plan().splits().get(0);
+ InternalRowSerializer serializer = new
InternalRowSerializer(latest.rowType());
+ List<InternalRow> rows = new ArrayList<>();
+ try (RecordReader<InternalRow> reader =
readBuilder.newRead().createReader(split)) {
+ // Bypass the final DV filter. The result is still correct only
when the DV-aware bitmap
+ // {50} (rather than the original index bitmap {50, 51}) reached
every field reader.
+ removeFinalDeletionVectorReaders(reader);
+ reader.forEachRemaining(row -> rows.add(serializer.copy(row)));
+ }
+
+ assertRow(assertSingleRow(rows), 50);
+ }
+
@Test
public void testMergedGroupFileIndexSkipsBeforeReadingDeletionVector()
throws Exception {
Map<String, String> options = new HashMap<>();
@@ -1094,6 +1124,34 @@ public class DataEvolutionFileIndexTest extends
DataEvolutionTestBase {
return rows;
}
+ @SuppressWarnings("unchecked")
+ private static void
removeFinalDeletionVectorReaders(RecordReader<InternalRow> reader)
+ throws Exception {
+ assertThat(reader).isInstanceOf(DataEvolutionFileReader.class);
+ RecordReader<InternalRow>[] fieldReaders =
+ (RecordReader<InternalRow>[]) readField(reader, "readers");
+ for (int i = 0; i < fieldReaders.length; i++) {
+ if (fieldReaders[i] == null) {
+ continue;
+ }
+
+
assertThat(fieldReaders[i]).isInstanceOf(ForceSingleBatchReader.class);
+ RecordReader<InternalRow> fieldReader =
+ (RecordReader<InternalRow>) readField(fieldReaders[i],
"multiBatchReader");
+
assertThat(fieldReader).isInstanceOf(ApplyDeletionVectorReader.class);
+ RecordReader<InternalRow> bitmapReader =
+ ((ApplyDeletionVectorReader) fieldReader).reader();
+
assertThat(bitmapReader).isInstanceOf(ApplyBitmapIndexRecordReader.class);
+ fieldReaders[i] = new ForceSingleBatchReader(bitmapReader);
+ }
+ }
+
+ private static Object readField(Object target, String name) throws
Exception {
+ Field field = target.getClass().getDeclaredField(name);
+ field.setAccessible(true);
+ return field.get(target);
+ }
+
private void assertMergedGroup(FileStoreTable table) throws Exception {
FileStoreTable latest = getTable(identifier(table.name()));
List<Split> splits = latest.newReadBuilder().newScan().plan().splits();
