From 12f57abbe33f067f019e0f1fb71b18ab11bdea30 Mon Sep 17 00:00:00 2001 From: yangjie01 Date: Tue, 15 Sep 2026 12:07:17 +0800 Subject: [PATCH] [core] Fall back to raw vector search when scalar index cannot fully evaluate filter Vector search with a WHERE filter delegates pre-filtering to the scalar global index. The index answers with all-or-nothing bitmaps and the index read path has no final-read filter, so the bitmap is trusted as the exact match set. Two cases broke that assumption: - The index could not evaluate the predicate at all (no scalar index files, or an unsupported function such as IS NOT NULL on a multivalue-indexed array column). scalarMatchedRows returned an empty bitmap, which was AND-ed into every index split, so all index-covered rows silently vanished from the result. - The index could evaluate only some of the filter's fields, e.g. "a = 1 AND tags IS NOT NULL" where the array conjunct is unsupported. The evaluator drops the unsupported conjunct and returns a superset matching "a = 1" alone, polluting the vector top-K with non-matching rows. Make scalarMatchedRows return null ("cannot decide exactly") in both cases: when scanWithCoverage is absent, and when the contributing fields do not cover every field the filter references. Then route the ranges the raw splits do not already cover through the raw search, where the exact final-read filter decides; the scalar pre-filter is cached for preFilters to reuse. Every reader entry (local, batch, Spark, Flink) demotes right after splitting its splits. --- .../AbstractDataEvolutionVectorRead.java | 86 ++++++++++- .../source/DataEvolutionBatchVectorRead.java | 1 + .../table/source/DataEvolutionVectorRead.java | 1 + .../table/source/VectorSearchBuilderTest.java | 146 ++++++++++++++++++ .../FlinkDataEvolutionVectorRead.java | 1 + .../read/SparkDataEvolutionVectorRead.java | 1 + 6 files changed, 230 insertions(+), 6 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 83ea140270e6..5bcd16f18150 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 @@ -28,7 +28,6 @@ import org.apache.paimon.globalindex.GlobalIndexEvaluator; import org.apache.paimon.globalindex.GlobalIndexIOMeta; import org.apache.paimon.globalindex.GlobalIndexReader; -import org.apache.paimon.globalindex.GlobalIndexResult; import org.apache.paimon.globalindex.GlobalIndexer; import org.apache.paimon.globalindex.GlobalIndexerFactoryUtils; import org.apache.paimon.globalindex.OffsetGlobalIndexReader; @@ -71,6 +70,7 @@ import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutorService; +import static org.apache.paimon.predicate.PredicateVisitor.collectFieldIds; import static org.apache.paimon.utils.Preconditions.checkNotNull; /** Base implementation for vector reads. */ @@ -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 WEAKEST_SCORE_FIRST = Comparator.comparingDouble(a -> Float.intBitsToFloat((int) a[1])) .thenComparing((a, b) -> Long.compare(b[0], a[0])); @@ -132,6 +139,56 @@ private GlobalIndexer createGlobalIndexer(String indexType, GlobalIndexMeta meta table.coreOptions().toConfiguration()); } + /** + * Moves index splits aside when the scalar index cannot EXACTLY evaluate {@link #filter} (it + * evaluated nothing, or only some of the filter's fields; see {@link #scalarMatchedRows}): the + * index path answers with all-or-nothing bitmaps and has no final-read filter, so an inexact + * bitmap can neither honor the filter (an empty bitmap silently drops every covered row, a + * superset lets non-matching rows into the top-K) nor be ignored. Instead, suppress the index + * splits and route the ranges the raw splits do not already cover through the raw search, where + * the exact final-read filter decides; already-covered ranges would duplicate that work. Splits + * routed this way reuse the vector index type so raw scoring keeps its metric. Must run after + * {@link #splitSearchSplits} and before the index splits are read, on the single thread + * performing the read; the computed pre-filter is cached for {@link #preFilters} to reuse. + */ + protected void demoteUncoverableIndexSplits( + List indexSplits, List rawSplits) { + scalarPreFilter = null; + if (filter == null || indexSplits.isEmpty()) { + return; + } + RoaringNavigableMap64 matchedRows = scalarMatchedRows(indexSplits); + if (matchedRows != null) { + scalarPreFilter = matchedRows; + return; + } + String indexType = vectorIndexType(indexSplits); + List uncovered = new ArrayList<>(); + List scalarIndexFiles = new ArrayList<>(); + for (IndexVectorSearchSplit split : indexSplits) { + if (!coveredByRawSplits(split.rowRangeStart(), split.rowRangeEnd(), rawSplits)) { + uncovered.add(new Range(split.rowRangeStart(), split.rowRangeEnd())); + scalarIndexFiles.addAll(split.scalarIndexFiles()); + } + } + if (!uncovered.isEmpty()) { + rawSplits.add(new RawVectorSearchSplit(uncovered, scalarIndexFiles, indexType)); + } + indexSplits.clear(); + } + + private static boolean coveredByRawSplits( + long start, long end, List rawSplits) { + for (RawVectorSearchSplit raw : rawSplits) { + for (Range range : raw.rowRanges()) { + if (range.from <= start && end <= range.to) { + return true; + } + } + } + return false; + } + protected List preFilters(List splits) { List indexedRowRanges = new ArrayList<>(splits.size()); for (IndexVectorSearchSplit split : splits) { @@ -141,7 +198,8 @@ protected List preFilters(List sp RoaringNavigableMap64 liveRows = GlobalIndexLiveRowFilter.liveRows( table, planSnapshot, partitionFilter, indexedRowRanges); - RoaringNavigableMap64 matchedRows = scalarMatchedRows(splits); + RoaringNavigableMap64 matchedRows = + scalarPreFilter != null ? scalarPreFilter : scalarMatchedRows(splits); List includeRowIds = new ArrayList<>(splits.size()); boolean hasFilter = false; @@ -167,6 +225,14 @@ protected List preFilters(List sp return hasFilter ? includeRowIds : Collections.emptyList(); } + /** + * Row ids the scalar index reports as matching {@link #filter}, or {@code null} when it cannot + * evaluate the predicate exactly: no scalar index files, an unsupported function (e.g. {@code + * IS NOT NULL} on a multivalue-indexed array), or only some of the filter's fields contributed + * (an unsupported conjunct in an AND yields a superset). {@code null} means "cannot decide", + * never "no rows match": the index path trusts this bitmap as the exact include set with no + * final-read filter, so the caller must route these rows to the raw search instead. + */ @Nullable private RoaringNavigableMap64 scalarMatchedRows(List splits) { if (filter == null) { @@ -183,15 +249,23 @@ private RoaringNavigableMap64 scalarMatchedRows(List spl DataEvolutionGlobalIndexScanner.create( table, planSnapshot, partitionFilter, scalarIndexFiles); if (!optionalScanner.isPresent()) { - return new RoaringNavigableMap64(); + return null; } try (DataEvolutionGlobalIndexScanner scanner = optionalScanner.get()) { - Optional result = scanner.scan(filter); + Optional result = scanner.scanWithCoverage(filter); if (!result.isPresent()) { - return new RoaringNavigableMap64(); + return null; + } + // Trust the bitmap as the exact match set only when every field the filter references + // actually contributed; a partially-evaluable predicate (an unsupported conjunct in an + // AND) yields a superset, which the index path cannot correct. + if (!result.get() + .contributingFieldIds() + .containsAll(collectFieldIds(table.rowType(), filter))) { + return null; } - return result.get().results(); + return result.get().result().results(); } catch (IOException e) { throw new RuntimeException(e); } 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 6c83a62b2368..88a02923db61 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 @@ -71,6 +71,7 @@ private List readBatch(List splits) { List indexSplits = new ArrayList<>(); List rawSplits = new ArrayList<>(); splitSearchSplits(splits, indexSplits, rawSplits); + demoteUncoverableIndexSplits(indexSplits, rawSplits); if (indexSplits.isEmpty() && rawSplits.isEmpty()) { List 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 9179dd7512c5..5d2c98b629b1 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 @@ -69,6 +69,7 @@ protected GlobalIndexResult readSplits(List splits) List indexSplits = new ArrayList<>(); List rawSplits = new ArrayList<>(); splitSearchSplits(splits, indexSplits, rawSplits); + demoteUncoverableIndexSplits(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 efec98c1f370..5ed50be03439 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 @@ 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.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,150 @@ public void testVectorSearchFullModeScansFilteredUnindexedData() throws Exceptio 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); + } + + @Test + public void testVectorSearchFilterPartiallyEvaluableAnd() throws Exception { + catalog.createTable( + identifier("vector_search_partial_and_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_partial_and_scalar_predicate")); + + // row 0 has NULL tags and is the closest to the query. The BTree index can evaluate + // `id >= 0` (all rows) but the multivalue index cannot evaluate `tags IS NOT NULL`, so + // the scalar index answers the AND with the `id`-only superset {0,1,2,3}. Trusting it + // would keep the index splits and pull the null-tags row 0 into the top-K; only demoting + // to the raw search, where the exact final-read filter drops row 0, 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); + buildAndCommitBTreeIndex(table, new int[] {0, 1, 2, 3}, new Range(0, 3)); + buildAndCommitMultiValueIndex(table, vectors.length); + + Predicate idFilter = new PredicateBuilder(table.rowType()).greaterOrEqual(0, 0); + Predicate tagsFilter = + new PredicateBuilder(table.rowType()) + .isNotNull(table.rowType().getFieldIndex("tags")); + Predicate filter = PredicateBuilder.and(idFilter, tagsFilter); + + 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 entries = writer.finish(rowCount); + + List 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 0e2b9a9df219..111bcf395c0a 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 @@ -92,6 +92,7 @@ public GlobalIndexResult read(VectorScan.Plan plan) { List indexSplits = new ArrayList<>(); List rawSplits = new ArrayList<>(); splitSearchSplits(plan.splits(), indexSplits, rawSplits); + demoteUncoverableIndexSplits(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 a65f7dcb8264..e9d14dceb950 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 @@ -75,6 +75,7 @@ public GlobalIndexResult read(VectorScan.Plan plan) { List indexSplits = new ArrayList<>(); List rawSplits = new ArrayList<>(); splitSearchSplits(plan.splits(), indexSplits, rawSplits); + demoteUncoverableIndexSplits(indexSplits, rawSplits); if (indexSplits.isEmpty() && rawSplits.isEmpty()) { return GlobalIndexResult.createEmpty(); }