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(); }