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 93fa0a5b4f66d0017cc957bb98c4f99c58eb9030 Author: YangJie <[email protected]> AuthorDate: Thu Sep 17 04:56:39 2026 -0400 [core] Fall back to raw vector search when scalar index cannot evaluate filter (#9909) --- .../source/AbstractDataEvolutionVectorRead.java | 57 +++++++++++- .../table/source/DataEvolutionBatchVectorRead.java | 2 +- .../table/source/DataEvolutionVectorRead.java | 2 +- .../table/source/VectorSearchBuilderTest.java | 101 +++++++++++++++++++++ .../vectorsearch/FlinkDataEvolutionVectorRead.java | 2 +- .../spark/read/SparkDataEvolutionVectorRead.java | 2 +- 6 files changed, 158 insertions(+), 8 deletions(-) diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractDataEvolutionVectorRead.java b/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractDataEvolutionVectorRead.java index 83ea140270..0f0b43810d 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractDataEvolutionVectorRead.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractDataEvolutionVectorRead.java @@ -88,6 +88,13 @@ public abstract class AbstractDataEvolutionVectorRead implements Serializable { /** Snapshot the plan was built against; pins filters and raw reads to it. */ @Nullable protected Snapshot planSnapshot; + /** + * Scalar pre-filter computed while splitting index/raw splits for the current read; lets {@link + * #preFilters} reuse it instead of scanning the scalar index twice. Only written and read on + * the single thread performing the read. + */ + @Nullable private transient RoaringNavigableMap64 scalarPreFilter; + private static final Comparator<long[]> WEAKEST_SCORE_FIRST = Comparator.<long[]>comparingDouble(a -> Float.intBitsToFloat((int) a[1])) .thenComparing((a, b) -> Long.compare(b[0], a[0])); @@ -132,6 +139,42 @@ public abstract class AbstractDataEvolutionVectorRead implements Serializable { table.coreOptions().toConfiguration()); } + /** + * Splits the search splits into index and raw splits, then moves the index splits aside when + * the scalar index cannot evaluate {@link #filter}: the index path answers with all-or-nothing + * bitmaps, so with an unevaluable filter it can neither honor the filter (an empty bitmap + * silently drops every covered row) nor ignore it (the top-K would be polluted by non-matching + * rows). Instead, suppress the index splits and route their ranges through the raw search, + * where the exact final-read filter decides; the raw read merges ranges, so overlap with an + * existing raw split is deduplicated. Splits routed this way reuse the vector index type so raw + * scoring keeps its metric. The computed scalar pre-filter is cached for {@link #preFilters} to + * reuse. + */ + protected void prepareSplits( + List<? extends VectorSearchSplit> splits, + List<IndexVectorSearchSplit> indexSplits, + List<RawVectorSearchSplit> rawSplits) { + splitSearchSplits(splits, indexSplits, rawSplits); + scalarPreFilter = null; + if (filter == null || indexSplits.isEmpty()) { + return; + } + RoaringNavigableMap64 matchedRows = scalarMatchedRows(indexSplits); + if (matchedRows != null) { + scalarPreFilter = matchedRows; + return; + } + List<Range> ranges = new ArrayList<>(); + List<IndexFileMeta> scalarIndexFiles = new ArrayList<>(); + for (IndexVectorSearchSplit split : indexSplits) { + ranges.add(new Range(split.rowRangeStart(), split.rowRangeEnd())); + scalarIndexFiles.addAll(split.scalarIndexFiles()); + } + rawSplits.add( + new RawVectorSearchSplit(ranges, scalarIndexFiles, vectorIndexType(indexSplits))); + indexSplits.clear(); + } + protected List<RoaringNavigableMap64> preFilters(List<IndexVectorSearchSplit> splits) { List<Range> indexedRowRanges = new ArrayList<>(splits.size()); for (IndexVectorSearchSplit split : splits) { @@ -141,7 +184,8 @@ public abstract class AbstractDataEvolutionVectorRead implements Serializable { RoaringNavigableMap64 liveRows = GlobalIndexLiveRowFilter.liveRows( table, planSnapshot, partitionFilter, indexedRowRanges); - RoaringNavigableMap64 matchedRows = scalarMatchedRows(splits); + RoaringNavigableMap64 matchedRows = + scalarPreFilter != null ? scalarPreFilter : scalarMatchedRows(splits); List<RoaringNavigableMap64> includeRowIds = new ArrayList<>(splits.size()); boolean hasFilter = false; @@ -167,6 +211,11 @@ public abstract class AbstractDataEvolutionVectorRead implements Serializable { return hasFilter ? includeRowIds : Collections.emptyList(); } + /** + * Row ids the scalar index reports as matching {@link #filter}, or {@code null} when the index + * cannot evaluate the predicate (no scalar index files, or a function the reader does not + * support). {@code null} means "cannot decide", never "no rows match". + */ @Nullable private RoaringNavigableMap64 scalarMatchedRows(List<IndexVectorSearchSplit> splits) { if (filter == null) { @@ -183,13 +232,13 @@ public abstract class AbstractDataEvolutionVectorRead implements Serializable { DataEvolutionGlobalIndexScanner.create( table, planSnapshot, partitionFilter, scalarIndexFiles); if (!optionalScanner.isPresent()) { - return new RoaringNavigableMap64(); + return null; } try (DataEvolutionGlobalIndexScanner scanner = optionalScanner.get()) { Optional<GlobalIndexResult> result = scanner.scan(filter); if (!result.isPresent()) { - return new RoaringNavigableMap64(); + return null; } return result.get().results(); } catch (IOException e) { @@ -632,7 +681,7 @@ public abstract class AbstractDataEvolutionVectorRead implements Serializable { return table.copyWithoutTimeTravel(pinOptions); } - protected static void splitSearchSplits( + private static void splitSearchSplits( List<? extends VectorSearchSplit> splits, List<IndexVectorSearchSplit> indexSplits, List<RawVectorSearchSplit> rawSplits) { diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/DataEvolutionBatchVectorRead.java b/paimon-core/src/main/java/org/apache/paimon/table/source/DataEvolutionBatchVectorRead.java index 6c83a62b23..a94f0d6dba 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/source/DataEvolutionBatchVectorRead.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/source/DataEvolutionBatchVectorRead.java @@ -70,7 +70,7 @@ public class DataEvolutionBatchVectorRead extends AbstractDataEvolutionVectorRea int n = vectors.length; List<IndexVectorSearchSplit> indexSplits = new ArrayList<>(); List<RawVectorSearchSplit> rawSplits = new ArrayList<>(); - splitSearchSplits(splits, indexSplits, rawSplits); + prepareSplits(splits, indexSplits, rawSplits); if (indexSplits.isEmpty() && rawSplits.isEmpty()) { List<GlobalIndexResult> empty = new ArrayList<>(n); for (int i = 0; i < n; i++) { diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/DataEvolutionVectorRead.java b/paimon-core/src/main/java/org/apache/paimon/table/source/DataEvolutionVectorRead.java index 9179dd7512..fbac5ef7b7 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/source/DataEvolutionVectorRead.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/source/DataEvolutionVectorRead.java @@ -68,7 +68,7 @@ public class DataEvolutionVectorRead extends AbstractDataEvolutionVectorRead imp protected GlobalIndexResult readSplits(List<? extends VectorSearchSplit> splits) { List<IndexVectorSearchSplit> indexSplits = new ArrayList<>(); List<RawVectorSearchSplit> rawSplits = new ArrayList<>(); - splitSearchSplits(splits, indexSplits, rawSplits); + prepareSplits(splits, indexSplits, rawSplits); if (indexSplits.isEmpty() && rawSplits.isEmpty()) { return GlobalIndexResult.createEmpty(); } diff --git a/paimon-core/src/test/java/org/apache/paimon/table/source/VectorSearchBuilderTest.java b/paimon-core/src/test/java/org/apache/paimon/table/source/VectorSearchBuilderTest.java index efec98c1f3..e9c972d978 100644 --- a/paimon-core/src/test/java/org/apache/paimon/table/source/VectorSearchBuilderTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/table/source/VectorSearchBuilderTest.java @@ -20,6 +20,7 @@ package org.apache.paimon.table.source; import org.apache.paimon.CoreOptions; import org.apache.paimon.data.BinaryRow; +import org.apache.paimon.data.BinaryString; import org.apache.paimon.data.GenericArray; import org.apache.paimon.data.GenericRow; import org.apache.paimon.data.InternalRow; @@ -31,6 +32,7 @@ import org.apache.paimon.globalindex.GlobalIndexResult; import org.apache.paimon.globalindex.GlobalIndexSingleColumnWriter; import org.apache.paimon.globalindex.ResultEntry; import org.apache.paimon.globalindex.ScoredGlobalIndexResult; +import org.apache.paimon.globalindex.bitmap.MultiValueGlobalIndexerFactory; import org.apache.paimon.globalindex.btree.BTreeGlobalIndexerFactory; import org.apache.paimon.globalindex.testvector.TestVectorGlobalIndexer; import org.apache.paimon.globalindex.testvector.TestVectorGlobalIndexerFactory; @@ -587,6 +589,105 @@ public class VectorSearchBuilderTest extends TableTestBase { assertThat(result.results()).containsExactlyInAnyOrder(2L, 3L); } + @Test + public void testVectorSearchFilterIndexCannotEvaluate() throws Exception { + catalog.createTable( + identifier("vector_search_unsupported_scalar_predicate"), + withVectorSchemaOptions( + Schema.newBuilder() + .column("id", DataTypes.INT()) + .column("tags", new ArrayType(DataTypes.STRING())) + .column( + VECTOR_FIELD_NAME, + new ArrayType(DataTypes.FLOAT()))) + .build(), + false); + FileStoreTable table = getTable(identifier("vector_search_unsupported_scalar_predicate")); + + // row 0 has NULL tags and is the closest to the query: dropping the index + // pre-filter alone would pollute the top-K with it, and keeping the old empty + // bitmap would drop every row — only routing the search through the exact + // final-read filter returns the closest *matching* rows + float[][] vectors = {{0.0f, 0.0f}, {1.0f, 0.0f}, {2.0f, 0.0f}, {3.0f, 0.0f}}; + writeVectorsWithTags(table, vectors, 0); + buildAndCommitIndex(table, vectors); + buildAndCommitMultiValueIndex(table, vectors.length); + + // IS NOT NULL on the multivalue-indexed array column is a function the reader + // cannot evaluate; the scalar pre-filter must step aside and let the exact + // final-read filter decide instead of answering "no rows match" + Predicate filter = + new PredicateBuilder(table.rowType()) + .isNotNull(table.rowType().getFieldIndex("tags")); + GlobalIndexResult result = + table.newVectorSearchBuilder() + .withVector(new float[] {0.0f, 0.0f}) + .withLimit(2) + .withVectorColumn(VECTOR_FIELD_NAME) + .withFilter(filter) + .executeLocal(); + + assertThat(result.results()).containsExactly(1L, 2L); + assertThat(readIds(table, result)).containsExactly(1, 2); + } + + private void writeVectorsWithTags(FileStoreTable table, float[][] vectors, int nullTagsRow) + throws Exception { + BatchWriteBuilder writeBuilder = table.newBatchWriteBuilder(); + try (BatchTableWrite write = writeBuilder.newWrite(); + BatchTableCommit commit = writeBuilder.newCommit()) { + for (int i = 0; i < vectors.length; i++) { + GenericArray tags = + i == nullTagsRow + ? null + : new GenericArray( + new BinaryString[] {BinaryString.fromString("t" + i)}); + write.write(GenericRow.of(i, tags, new GenericArray(vectors[i]))); + } + commit.commit(write.prepareCommit()); + } + } + + private void buildAndCommitMultiValueIndex(FileStoreTable table, int rowCount) + throws Exception { + Options options = table.coreOptions().toConfiguration(); + DataField tagsField = table.rowType().getField("tags"); + + GlobalIndexSingleColumnWriter writer = + (GlobalIndexSingleColumnWriter) + GlobalIndexBuilderUtils.createIndexWriter( + table, + MultiValueGlobalIndexerFactory.IDENTIFIER, + tagsField, + options); + for (int row = 0; row < rowCount; row++) { + writer.write(BinaryString.fromString("t" + row), row); + } + List<ResultEntry> entries = writer.finish(rowCount); + + List<IndexFileMeta> indexFiles = + GlobalIndexBuilderUtils.toIndexFileMetas( + table.fileIO(), + table.store().pathFactory().globalIndexFileFactory(), + table.coreOptions(), + new Range(0, rowCount - 1), + tagsField.id(), + MultiValueGlobalIndexerFactory.IDENTIFIER, + entries); + + DataIncrement dataIncrement = DataIncrement.indexIncrement(indexFiles); + CommitMessage message = + new CommitMessageImpl( + BinaryRow.EMPTY_ROW, + 0, + null, + dataIncrement, + CompactIncrement.emptyIncrement()); + try (BatchTableCommit commit = table.newBatchWriteBuilder().newCommit()) { + commit.commit(Collections.singletonList(message)); + } + } + @Test public void testVectorSearchRawSearchUsesScalarPreFilter() throws Exception { catalog.createTable( diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/vectorsearch/FlinkDataEvolutionVectorRead.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/vectorsearch/FlinkDataEvolutionVectorRead.java index 0e2b9a9df2..192bd54002 100644 --- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/vectorsearch/FlinkDataEvolutionVectorRead.java +++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/vectorsearch/FlinkDataEvolutionVectorRead.java @@ -91,7 +91,7 @@ public class FlinkDataEvolutionVectorRead extends DataEvolutionVectorRead { this.planSnapshot = plan.snapshot(); List<IndexVectorSearchSplit> indexSplits = new ArrayList<>(); List<RawVectorSearchSplit> rawSplits = new ArrayList<>(); - splitSearchSplits(plan.splits(), indexSplits, rawSplits); + prepareSplits(plan.splits(), indexSplits, rawSplits); if (indexSplits.isEmpty() && rawSplits.isEmpty()) { return GlobalIndexResult.createEmpty(); } diff --git a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/read/SparkDataEvolutionVectorRead.java b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/read/SparkDataEvolutionVectorRead.java index a65f7dcb82..47e2a1114b 100644 --- a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/read/SparkDataEvolutionVectorRead.java +++ b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/read/SparkDataEvolutionVectorRead.java @@ -74,7 +74,7 @@ public class SparkDataEvolutionVectorRead extends DataEvolutionVectorRead { this.planSnapshot = plan.snapshot(); List<IndexVectorSearchSplit> indexSplits = new ArrayList<>(); List<RawVectorSearchSplit> rawSplits = new ArrayList<>(); - splitSearchSplits(plan.splits(), indexSplits, rawSplits); + prepareSplits(plan.splits(), indexSplits, rawSplits); if (indexSplits.isEmpty() && rawSplits.isEmpty()) { return GlobalIndexResult.createEmpty(); }
