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 855de94e3b [core] Fix primary-key global-index scan planning (#8647)
855de94e3b is described below
commit 855de94e3bf4af8d04f1b1037d23496023699956
Author: Jingsong Lee <[email protected]>
AuthorDate: Wed Jul 15 14:14:18 2026 +0800
[core] Fix primary-key global-index scan planning (#8647)
Fix primary-key batch scan planning so incompatible global-index results
fail fast, split-backed results are not processed again, and sorted
indexes are evaluated only for applicable predicates.
The previous implementation silently ignored unsupported
`GlobalIndexResult` implementations and passed the full scan filter into
sorted-index evaluation, causing redundant planning and unnecessary
index-manifest reads.
---
.../paimon/table/source/PrimaryKeyBatchScan.java | 80 ++++++++++++++--------
.../source/PrimaryKeySortedIndexBatchScanTest.java | 40 +++++++++++
.../table/source/PrimaryKeyVectorScanTest.java | 5 ++
3 files changed, 95 insertions(+), 30 deletions(-)
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/source/PrimaryKeyBatchScan.java
b/paimon-core/src/main/java/org/apache/paimon/table/source/PrimaryKeyBatchScan.java
index f739c86f1b..7fc98c959e 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/source/PrimaryKeyBatchScan.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/source/PrimaryKeyBatchScan.java
@@ -27,14 +27,17 @@ import
org.apache.paimon.index.pk.PrimaryKeyIndexDefinitions;
import org.apache.paimon.manifest.FileKind;
import org.apache.paimon.manifest.IndexManifestEntry;
import org.apache.paimon.predicate.Predicate;
-import org.apache.paimon.schema.TableSchema;
-import org.apache.paimon.table.BucketMode;
+import org.apache.paimon.predicate.PredicateProjectionConverter;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.source.snapshot.SnapshotReader;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.RowType;
import javax.annotation.Nullable;
import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
@@ -42,7 +45,10 @@ import java.util.Set;
/** Batch scan for primary-key tables and indexes. */
public class PrimaryKeyBatchScan extends AbstractBatchTableScan {
- private final FileStoreTable table;
+ private final RowType rowType;
+ private final List<PrimaryKeyIndexDefinition> definitions;
+ private final Set<Integer> definitionFieldIds;
+ private final PredicateProjectionConverter indexPredicateExtractor;
private final @Nullable PrimaryKeySortedIndexScan.ReaderFactory
readerFactory;
@Nullable private Predicate filter;
@@ -59,7 +65,28 @@ public class PrimaryKeyBatchScan extends
AbstractBatchTableScan {
table.coreOptions(),
snapshotReader,
queryAuth);
- this.table = table;
+ this.rowType = table.schema().logicalRowType();
+ List<PrimaryKeyIndexDefinition> definitions = new ArrayList<>();
+ Set<Integer> definitionFieldIds = new HashSet<>();
+ for (PrimaryKeyIndexDefinition definition :
+
PrimaryKeyIndexDefinitions.create(table.schema()).definitions()) {
+ if (definition.family() == PrimaryKeyIndexDefinition.Family.BTREE
+ || definition.family() ==
PrimaryKeyIndexDefinition.Family.BITMAP) {
+ definitions.add(definition);
+ definitionFieldIds.add(definition.fieldId());
+ }
+ }
+ this.definitions = Collections.unmodifiableList(definitions);
+ this.definitionFieldIds =
Collections.unmodifiableSet(definitionFieldIds);
+ int[] indexFieldMapping = new int[rowType.getFieldCount()];
+ Arrays.fill(indexFieldMapping, -1);
+ for (int i = 0; i < rowType.getFieldCount(); i++) {
+ DataField field = rowType.getFields().get(i);
+ if (definitionFieldIds.contains(field.id())) {
+ indexFieldMapping[i] = i;
+ }
+ }
+ this.indexPredicateExtractor =
PredicateProjectionConverter.fromMapping(indexFieldMapping);
this.readerFactory = readerFactory;
}
@@ -72,9 +99,15 @@ public class PrimaryKeyBatchScan extends
AbstractBatchTableScan {
@Override
public PrimaryKeyBatchScan withGlobalIndexResult(GlobalIndexResult
globalIndexResult) {
- if (globalIndexResult instanceof GlobalIndexSplitResult) {
- this.globalIndexSplitResult = (GlobalIndexSplitResult)
globalIndexResult;
+ if (globalIndexResult == null) {
+ return this;
}
+ if (!(globalIndexResult instanceof GlobalIndexSplitResult)) {
+ throw new IllegalArgumentException(
+ "PrimaryKeyBatchScan requires a GlobalIndexSplitResult,
but found "
+ + globalIndexResult.getClass().getName());
+ }
+ this.globalIndexSplitResult = (GlobalIndexSplitResult)
globalIndexResult;
return this;
}
@@ -93,21 +126,21 @@ public class PrimaryKeyBatchScan extends
AbstractBatchTableScan {
@Override
protected Plan postProcessPlan(Plan dataPlan) {
- if (!(dataPlan instanceof SnapshotReader.Plan)) {
+ if (globalIndexSplitResult != null || !(dataPlan instanceof
SnapshotReader.Plan)) {
return dataPlan;
}
SnapshotReader.Plan snapshotPlan = (SnapshotReader.Plan) dataPlan;
- if (filter == null
- || !options().globalIndexEnabled()
- || !snapshotReader.hasNonPartitionFilter()
- || table.schema().primaryKeys().isEmpty()
- || !options().deletionVectorsEnabled()
- || options().deletionVectorsMergeOnRead()
- || (options().bucket() <= 0 && options().bucket() !=
BucketMode.POSTPONE_BUCKET)
+ if (!options().globalIndexEnabled()
+ || definitions.isEmpty()
|| snapshotPlan.snapshotId() == null
|| snapshotPlan.splits().isEmpty()) {
return dataPlan;
}
+ Predicate indexFilter =
+ filter == null ? null :
filter.visit(indexPredicateExtractor).orElse(null);
+ if (indexFilter == null) {
+ return dataPlan;
+ }
List<DataSplit> dataSplits = new ArrayList<>();
for (Split split : snapshotPlan.splits()) {
@@ -122,19 +155,6 @@ public class PrimaryKeyBatchScan extends
AbstractBatchTableScan {
if (snapshot == null) {
return dataPlan;
}
- TableSchema snapshotSchema =
table.schemaManager().schema(snapshot.schemaId());
- List<PrimaryKeyIndexDefinition> definitions =
-
PrimaryKeyIndexDefinitions.create(snapshotSchema).definitions();
- Set<Integer> scalarFields = new HashSet<>();
- for (PrimaryKeyIndexDefinition definition : definitions) {
- if (definition.family() == PrimaryKeyIndexDefinition.Family.BTREE
- || definition.family() ==
PrimaryKeyIndexDefinition.Family.BITMAP) {
- scalarFields.add(definition.fieldId());
- }
- }
- if (scalarFields.isEmpty()) {
- return dataPlan;
- }
IndexFileHandler indexFileHandler = snapshotReader.indexFileHandler();
if (indexFileHandler == null) {
@@ -148,7 +168,7 @@ public class PrimaryKeyBatchScan extends
AbstractBatchTableScan {
return entry.kind() == FileKind.ADD
&& meta != null
&& meta.sourceMeta() != null
- &&
scalarFields.contains(meta.indexFieldId());
+ &&
definitionFieldIds.contains(meta.indexFieldId());
});
PrimaryKeySortedIndexScan.Plan indexPlan =
PrimaryKeySortedIndexScan.plan(snapshotId, dataSplits,
definitions, indexEntries);
@@ -157,12 +177,12 @@ public class PrimaryKeyBatchScan extends
AbstractBatchTableScan {
? PrimaryKeySortedIndexScan.readerFactory(
snapshotReader.snapshotManager().fileIO(),
snapshotReader.pathFactory(),
- snapshotSchema.logicalRowType(),
+ rowType,
options().toConfiguration())
: readerFactory;
PrimaryKeySortedIndexScan.EvaluatedPlan evaluated =
PrimaryKeySortedIndexScan.evaluate(
- indexPlan, snapshotSchema.logicalRowType(), filter,
definitions, factory);
+ indexPlan, rowType, indexFilter, definitions, factory);
PrimaryKeySortedIndexResult result = new
PrimaryKeySortedIndexResult(evaluated);
return new PlanImpl(
snapshotPlan.watermark(),
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeySortedIndexBatchScanTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeySortedIndexBatchScanTest.java
index 0979f874db..1b6256576d 100644
---
a/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeySortedIndexBatchScanTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeySortedIndexBatchScanTest.java
@@ -64,6 +64,8 @@ import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.RETURNS_SELF;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
/** Tests automatic source-backed BTree/Bitmap evaluation in ordinary batch
planning. */
@@ -115,6 +117,44 @@ class PrimaryKeySortedIndexBatchScanTest {
assertThat(result.splits()).singleElement().isInstanceOf(DataSplit.class);
}
+ @Test
+ void testRejectsNonSplitGlobalIndexResult() {
+ ScanFixture fixture = fixture(reader(2));
+
+ assertThatThrownBy(
+ () ->
fixture.scan.withGlobalIndexResult(GlobalIndexResult.createEmpty()))
+ .isInstanceOf(IllegalArgumentException.class)
+
.hasMessageContaining(GlobalIndexSplitResult.class.getSimpleName());
+ }
+
+ @Test
+ void testPostProcessPlanSkipsIndexWhenGlobalIndexResultExists() {
+ ScanFixture fixture = fixture(reader(2));
+ fixture.scan.withGlobalIndexResult(mock(GlobalIndexSplitResult.class));
+ TableScan.Plan dataPlan =
+ new PlanImpl(
+ null,
+ 11L,
+ Collections.<Split>singletonList(dataSplit(11,
fixture.dataFile)));
+
+ TableScan.Plan result = fixture.scan.postProcessPlan(dataPlan);
+
+ assertThat(result).isSameAs(dataPlan);
+ }
+
+ @Test
+ void testUnindexedFilterSkipsSortedIndexPlanning() {
+ ScanFixture fixture = fixture(reader(2));
+ Predicate idFilter =
+ new PredicateBuilder(tableSchema(true,
2).logicalRowType()).equal(0, 42);
+ fixture.scan.withFilter(idFilter);
+
+ TableScan.Plan result = fixture.scan.plan();
+
+
assertThat(result.splits()).singleElement().isInstanceOf(DataSplit.class);
+ verify(fixture.scan.snapshotReader, never()).indexFileHandler();
+ }
+
private static TableSchema tableSchema(boolean globalIndexEnabled, int
bucket) {
Map<String, String> options = new HashMap<>();
options.put(CoreOptions.BUCKET.key(), Integer.toString(bucket));
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeyVectorScanTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeyVectorScanTest.java
index 369c90d711..5de2923507 100644
---
a/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeyVectorScanTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeyVectorScanTest.java
@@ -37,6 +37,9 @@ import org.apache.paimon.stats.SimpleStats;
import org.apache.paimon.table.BucketMode;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.source.snapshot.SnapshotReader;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.Filter;
import org.apache.paimon.utils.Range;
import org.apache.paimon.utils.SnapshotManager;
@@ -253,6 +256,8 @@ class PrimaryKeyVectorScanTest {
FileStoreTable table, SnapshotReader snapshotReader, Snapshot
snapshot) {
TableSchema schema = mock(TableSchema.class);
when(schema.primaryKeys()).thenReturn(Collections.singletonList("id"));
+ when(schema.logicalRowType())
+ .thenReturn(RowType.of(new DataField(1, "id",
DataTypes.INT().notNull())));
when(table.schema()).thenReturn(schema);
when(table.schemaManager()).thenReturn(mock(SchemaManager.class));
SnapshotManager snapshotManager = mock(SnapshotManager.class);
