diff --git a/docs/docs/multimodal-table/global-index/full-text.mdx b/docs/docs/multimodal-table/global-index/full-text.mdx index af6ad4916bdb..52d3b1ba7cb6 100644 --- a/docs/docs/multimodal-table/global-index/full-text.mdx +++ b/docs/docs/multimodal-table/global-index/full-text.mdx @@ -300,8 +300,13 @@ BM25 statistics are always those of the full corpus: Bitmap, Multivalue, FM) are decided by the index, the same way [vector search pre-filters](./vector#vector-search) rows. When the index can only produce candidates (a conjunct no index could evaluate, or `contains`, - ends-with and `LIKE` on a BTree index), the candidates are verified by reading - their filter columns. + ends-with and `LIKE` on a BTree index), those candidates are never ranked as + if they matched: with `global-index.filter.refine-from-data=true` they are + verified by reading their filter columns, otherwise (the default) they are + excluded and a warning is logged, so the result may hold fewer than `limit` + rows. The read runs on the caller and can cover every candidate row, which is + why it is opt-in; prefer an index that answers the predicate exactly, such as + Bitmap or FM for `contains`. - Rows whose filter columns are not covered by a scalar index follow `scalar-index.search-mode`: diff --git a/docs/docs/multimodal-table/global-index/vector.mdx b/docs/docs/multimodal-table/global-index/vector.mdx index 98ea0a5372a2..c17170e9f9b3 100644 --- a/docs/docs/multimodal-table/global-index/vector.mdx +++ b/docs/docs/multimodal-table/global-index/vector.mdx @@ -285,6 +285,13 @@ The scalar filter is evaluated with matching scalar global indexes before vector BTree index for frequently used metadata filters, such as `category`, `tenant_id`, or `event_time`, so vector search can restrict the candidate row ids before running ANN search. +An index answer that is only a candidate set — `contains`, ends-with or `LIKE` on a BTree index, +or a conjunction with a member no index can evaluate — is never ranked as if it matched, because +a non-matching but closer row would take a top-k slot from a matching one. By default such +candidates are excluded and a warning is logged, so the result may hold fewer than `limit` rows; +set `global-index.filter.refine-from-data=true` to verify them by reading their filter columns +instead. That read runs on the caller and can cover every candidate row. + diff --git a/docs/generated/core_configuration.html b/docs/generated/core_configuration.html index 2fb5acc403d8..3fd3783defa8 100644 --- a/docs/generated/core_configuration.html +++ b/docs/generated/core_configuration.html @@ -866,6 +866,12 @@ String Global index root directory, if not set, the global index files will be stored under the <table-root-directory>/index. + +
global-index.filter.refine-from-data
+ false + Boolean + Whether a vector, hybrid or full-text search may read the filter columns of candidate rows to verify a row filter that the scalar global index can only answer with a superset, such as contains, ends-with or like on a BTree index or a conjunction with a member no index can evaluate. When false, such candidates are excluded from the search, which never returns a non-matching row but may return fewer than the requested top-k. When true, the read runs on the caller and may cover every candidate row. +
global-index.row-count-per-shard
100000 diff --git a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java index 60605bc74c7e..a7de6e1b4c17 100644 --- a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java +++ b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java @@ -3000,6 +3000,21 @@ public String toString() { .defaultValue(GlobalIndexSearchMode.FAST) .withDescription("Search mode for full-text index queries."); + public static final ConfigOption GLOBAL_INDEX_FILTER_REFINE_FROM_DATA = + key("global-index.filter.refine-from-data") + .booleanType() + .defaultValue(false) + .withDescription( + "Whether a vector, hybrid or full-text search may read the filter " + + "columns of candidate rows to verify a row filter that the " + + "scalar global index can only answer with a superset, such " + + "as contains, ends-with or like on a BTree index or a " + + "conjunction with a member no index can evaluate. When " + + "false, such candidates are excluded from the search, which " + + "never returns a non-matching row but may return fewer than " + + "the requested top-k. When true, the read runs on the caller " + + "and may cover every candidate row."); + public static final ConfigOption GLOBAL_INDEX_THREAD_NUM = key("global-index.thread-num") .intType() @@ -4781,6 +4796,10 @@ public GlobalIndexSearchMode scalarIndexSearchMode() { return indexSearchMode(SCALAR_INDEX_SEARCH_MODE); } + public boolean globalIndexFilterRefineFromData() { + return options.get(GLOBAL_INDEX_FILTER_REFINE_FROM_DATA); + } + public GlobalIndexSearchMode vectorIndexSearchMode() { return indexSearchMode(VECTOR_INDEX_SEARCH_MODE); } 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 0f0b43810d5b..bd8f5444fe9a 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; @@ -54,6 +53,9 @@ import org.apache.paimon.utils.Range; import org.apache.paimon.utils.RoaringNavigableMap64; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + import javax.annotation.Nullable; import java.io.IOException; @@ -76,6 +78,9 @@ /** Base implementation for vector reads. */ public abstract class AbstractDataEvolutionVectorRead implements Serializable { + private static final Logger LOG = + LoggerFactory.getLogger(AbstractDataEvolutionVectorRead.class); + private static final long serialVersionUID = 1L; protected final FileStoreTable table; @@ -212,9 +217,14 @@ protected List preFilters(List sp } /** - * 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". + * Rows of the indexed splits that satisfy {@link #filter} according to the scalar global + * indexes, or {@code null} when no index can evaluate the predicate (no scalar index files, or + * a function the reader does not support); {@code null} means "cannot decide", never "no rows + * match". The set is exact: an index answer that may be a superset (see {@link + * FilteredRowIdReader#isExact}) is refined from the data when {@code + * global-index.filter.refine-from-data} allows it and excluded otherwise, because a superset + * ranked by the ANN would push matching rows out of the top-k where the engine-side filter + * cannot bring them back. */ @Nullable private RoaringNavigableMap64 scalarMatchedRows(List splits) { @@ -224,8 +234,10 @@ private RoaringNavigableMap64 scalarMatchedRows(List spl Set scalarIndexFiles = new TreeSet<>(Comparator.comparing(IndexFileMeta::fileName)); + RoaringNavigableMap64 splitRows = new RoaringNavigableMap64(); for (IndexVectorSearchSplit split : splits) { scalarIndexFiles.addAll(split.scalarIndexFiles()); + splitRows.addRange(new Range(split.rowRangeStart(), split.rowRangeEnd())); } Optional optionalScanner = @@ -236,11 +248,21 @@ private RoaringNavigableMap64 scalarMatchedRows(List spl } try (DataEvolutionGlobalIndexScanner scanner = optionalScanner.get()) { - Optional result = scanner.scan(filter); - if (!result.isPresent()) { + Optional evaluation = scanner.scanWithCoverage(filter); + if (!evaluation.isPresent()) { return null; } - return result.get().results(); + RoaringNavigableMap64 matched = evaluation.get().result().results(); + if (FilteredRowIdReader.isExact(table.rowType(), filter, evaluation.get())) { + return matched; + } + if (!table.coreOptions().globalIndexFilterRefineFromData()) { + FilteredRowIdReader.warnCandidatesExcluded(LOG, table, filter); + return new RoaringNavigableMap64(); + } + RoaringNavigableMap64 candidates = RoaringNavigableMap64.and(matched, splitRows); + return new FilteredRowIdReader(table, planSnapshot, partitionFilter, filter) + .matchingRowIds(candidates); } catch (IOException e) { throw new RuntimeException(e); } diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/DataEvolutionFullTextRead.java b/paimon-core/src/main/java/org/apache/paimon/table/source/DataEvolutionFullTextRead.java index 9a8227b1b9f6..34a5728e5739 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/source/DataEvolutionFullTextRead.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/source/DataEvolutionFullTextRead.java @@ -36,13 +36,7 @@ import org.apache.paimon.index.IndexFileMeta; import org.apache.paimon.index.IndexPathFactory; import org.apache.paimon.partition.PartitionPredicate; -import org.apache.paimon.predicate.CompoundPredicate; -import org.apache.paimon.predicate.Contains; -import org.apache.paimon.predicate.EndsWith; import org.apache.paimon.predicate.FullTextSearch; -import org.apache.paimon.predicate.LeafFunction; -import org.apache.paimon.predicate.LeafPredicate; -import org.apache.paimon.predicate.Like; import org.apache.paimon.predicate.Predicate; import org.apache.paimon.table.FileStoreTable; import org.apache.paimon.types.DataField; @@ -68,7 +62,6 @@ import java.util.concurrent.ExecutorService; import static org.apache.paimon.CoreOptions.GLOBAL_INDEX_THREAD_NUM; -import static org.apache.paimon.predicate.PredicateVisitor.collectFieldIds; import static org.apache.paimon.utils.Preconditions.checkNotNull; /** Implementation for {@link FullTextRead}. */ @@ -183,7 +176,8 @@ private GlobalIndexResult read( * index. When the index answer may be a superset (a conjunct it could not evaluate was * dropped, or a {@code contains} / {@code endsWith} / {@code like} leaf, which BTree * answers with every non-null row), the candidates are refined by reading their filter - * columns. + * columns if {@code global-index.filter.refine-from-data} allows it, and excluded + * otherwise. *
  • Rows whose filter columns are not covered follow {@code scalar-index.search-mode}: * excluded in {@code fast}, otherwise decided by reading their filter columns. * @@ -237,10 +231,17 @@ private RoaringNavigableMap64 matchedRows( RoaringNavigableMap64 fromIndex = RoaringNavigableMap64.and( evaluation.get().result().results(), decidedByIndex); - if (!isExact(evaluation.get()) && !fromIndex.isEmpty()) { - fromIndex = - new FilteredRowIdReader(table, planSnapshot, partitionFilter, filter) - .matchingRowIds(fromIndex); + if (!FilteredRowIdReader.isExact(table.rowType(), filter, evaluation.get()) + && !fromIndex.isEmpty()) { + if (table.coreOptions().globalIndexFilterRefineFromData()) { + fromIndex = + new FilteredRowIdReader( + table, planSnapshot, partitionFilter, filter) + .matchingRowIds(fromIndex); + } else { + FilteredRowIdReader.warnCandidatesExcluded(LOG, table, filter); + fromIndex = new RoaringNavigableMap64(); + } } matched.or(fromIndex); } else { @@ -270,37 +271,6 @@ private Optional evaluateWithIndexes( } } - /** - * Whether an index evaluation is an exact match set. The global index contract only promises - * candidates: a conjunct no index could evaluate is dropped, and BTree answers substring - * predicates with every non-null row. Both cases are refined from the data. - */ - private boolean isExact(GlobalIndexEvaluator.Evaluation evaluation) { - Set filterFieldIds = collectFieldIds(table.rowType(), filter); - if (!evaluation.contributingFieldIds().containsAll(filterFieldIds)) { - return false; - } - return !hasCandidateOnlyLeaf(filter); - } - - private static boolean hasCandidateOnlyLeaf(Predicate predicate) { - if (predicate instanceof CompoundPredicate) { - for (Predicate child : ((CompoundPredicate) predicate).children()) { - if (hasCandidateOnlyLeaf(child)) { - return true; - } - } - return false; - } - if (predicate instanceof LeafPredicate) { - LeafFunction function = ((LeafPredicate) predicate).function(); - return function instanceof Contains - || function instanceof EndsWith - || function instanceof Like; - } - return false; - } - private void warnUnindexedFilter() { if (table.coreOptions().scalarIndexSearchMode() == GlobalIndexSearchMode.FAST) { LOG.warn( diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/FilteredRowIdReader.java b/paimon-core/src/main/java/org/apache/paimon/table/source/FilteredRowIdReader.java index 8f88c511c2ef..da333e6ef583 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/source/FilteredRowIdReader.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/source/FilteredRowIdReader.java @@ -21,7 +21,14 @@ import org.apache.paimon.CoreOptions; import org.apache.paimon.Snapshot; import org.apache.paimon.data.InternalRow; +import org.apache.paimon.globalindex.GlobalIndexEvaluator; import org.apache.paimon.partition.PartitionPredicate; +import org.apache.paimon.predicate.CompoundPredicate; +import org.apache.paimon.predicate.Contains; +import org.apache.paimon.predicate.EndsWith; +import org.apache.paimon.predicate.LeafFunction; +import org.apache.paimon.predicate.LeafPredicate; +import org.apache.paimon.predicate.Like; import org.apache.paimon.predicate.Predicate; import org.apache.paimon.predicate.PredicateVisitor; import org.apache.paimon.reader.RecordReader; @@ -31,6 +38,8 @@ import org.apache.paimon.utils.Range; import org.apache.paimon.utils.RoaringNavigableMap64; +import org.slf4j.Logger; + import javax.annotation.Nullable; import java.io.IOException; @@ -40,6 +49,8 @@ import java.util.Map; import java.util.Set; +import static org.apache.paimon.predicate.PredicateVisitor.collectFieldIds; + /** * Evaluates a row predicate on the data for a given set of row ids and returns the ids that satisfy * it. Only the filter columns and the row id are read, so this is the exact counterpart of a scalar @@ -64,6 +75,53 @@ class FilteredRowIdReader { this.filter = filter; } + /** + * Whether a scalar global index evaluation of {@code filter} is an exact match set. The global + * index contract only promises candidates: {@link GlobalIndexEvaluator} drops a conjunct no + * index can evaluate, and BTree answers {@code contains} / {@code endsWith} / {@code like} with + * every non-null row. A search that ranks rows before the engine filters them must refine a + * non-exact answer through {@link #matchingRowIds} first. + */ + static boolean isExact( + RowType rowType, Predicate filter, GlobalIndexEvaluator.Evaluation evaluation) { + Set filterFieldIds = collectFieldIds(rowType, filter); + if (!evaluation.contributingFieldIds().containsAll(filterFieldIds)) { + return false; + } + return !hasCandidateOnlyLeaf(filter); + } + + private static boolean hasCandidateOnlyLeaf(Predicate predicate) { + if (predicate instanceof CompoundPredicate) { + for (Predicate child : ((CompoundPredicate) predicate).children()) { + if (hasCandidateOnlyLeaf(child)) { + return true; + } + } + return false; + } + if (predicate instanceof LeafPredicate) { + LeafFunction function = ((LeafPredicate) predicate).function(); + return function instanceof Contains + || function instanceof EndsWith + || function instanceof Like; + } + return false; + } + + /** Logs that a candidate-only index answer was dropped because refinement is disabled. */ + static void warnCandidatesExcluded(Logger log, FileStoreTable table, Predicate filter) { + log.warn( + "The scalar global index can only answer the row filter {} on table {} with " + + "candidates, and {} is false, so those rows are excluded from the " + + "search; the result may hold fewer rows than requested. Set the option " + + "to true to verify the candidates against the data, or build an index " + + "that answers the predicate exactly.", + filter, + table.name(), + CoreOptions.GLOBAL_INDEX_FILTER_REFINE_FROM_DATA.key()); + } + /** The subset of {@code rows} whose data satisfies the filter. */ RoaringNavigableMap64 matchingRowIds(RoaringNavigableMap64 rows) { RoaringNavigableMap64 matching = new RoaringNavigableMap64(); diff --git a/paimon-core/src/test/java/org/apache/paimon/table/source/FullTextSearchBuilderTest.java b/paimon-core/src/test/java/org/apache/paimon/table/source/FullTextSearchBuilderTest.java index 94f483ec6c7b..cb9a38adb250 100644 --- a/paimon-core/src/test/java/org/apache/paimon/table/source/FullTextSearchBuilderTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/table/source/FullTextSearchBuilderTest.java @@ -55,10 +55,11 @@ import org.apache.paimon.types.DataTypes; import org.apache.paimon.types.RowType; import org.apache.paimon.utils.Range; +import org.apache.paimon.utils.RoaringNavigableMap64; import org.junit.jupiter.api.Test; import org.junit.jupiter.params.ParameterizedTest; -import org.junit.jupiter.params.provider.ValueSource; +import org.junit.jupiter.params.provider.CsvSource; import javax.annotation.Nullable; @@ -800,14 +801,16 @@ public FullTextRead newFullTextRead() { .hasMessageContaining("does not support row filters"); } - @ParameterizedTest(name = "scalar-index.search-mode={0}") - @ValueSource(strings = {"fast", "full"}) - public void testFullTextSearchRefinesCandidateOnlyIndexResultsBeforeTopK(String scalarMode) - throws Exception { + @ParameterizedTest(name = "scalar-index.search-mode={0}, refine-from-data={1}") + @CsvSource({"fast, true", "full, true", "fast, false", "full, false"}) + public void testFullTextSearchCandidateOnlyIndexResultsAreNeverRanked( + String scalarMode, boolean refine) throws Exception { // BTree answers contains / endsWith / like with every non-null row (a candidate superset). // Ranking that superset lets a non-matching, higher-scoring row take the single slot and - // the matching row is lost before any engine-side filter can run. - FileStoreTable table = createTable("full_text_candidate_only_" + scalarMode, scalarMode); + // the matching row is lost before any engine-side filter can run. With refinement the + // candidates are verified from the data; without it they are excluded. + FileStoreTable table = + createTable("full_text_candidate_only_" + scalarMode + refine, scalarMode, refine); writeDocuments(table, RANKED_DOCUMENTS); buildAndCommitIndex(table, RANKED_DOCUMENTS); @@ -816,18 +819,26 @@ public void testFullTextSearchRefinesCandidateOnlyIndexResultsBeforeTopK(String Predicate endsWithZeta = builder.endsWith(1, BinaryString.fromString("zeta")); Predicate likeZeta = builder.like(1, BinaryString.fromString("%zeta")); - // Without a scalar index, full mode evaluates the predicate on the data: exact. + // Without a scalar index, full mode evaluates the predicate on the data regardless of + // the refine option: that read is governed by scalar-index.search-mode. if (scalarMode.equals("full")) { assertThat(searchWithFilter(table, containsZeta, 1).results()).containsExactly(5L); } buildAndCommitBTreeIndex(table, RANKED_DOCUMENTS); - assertThat(searchWithFilter(table, containsZeta, 1).results()).containsExactly(5L); - assertThat(searchWithFilter(table, endsWithZeta, 1).results()).containsExactly(5L); - assertThat(searchWithFilter(table, likeZeta, 1).results()).containsExactly(5L); - assertThat(searchWithFilter(table, containsZeta, 10).results()).containsExactly(5L); + for (Predicate candidateOnly : Arrays.asList(containsZeta, endsWithZeta, likeZeta)) { + for (int limit : new int[] {1, 10}) { + RoaringNavigableMap64 rows = + searchWithFilter(table, candidateOnly, limit).results(); + if (refine) { + assertThat(rows).as("%s limit %s", candidateOnly, limit).containsExactly(5L); + } else { + assertThat(rows).as("%s limit %s", candidateOnly, limit).isEmpty(); + } + } + } - // Exact operators on the same btree stay exact. + // Exact operators on the same btree stay exact either way. assertThat( searchWithFilter( table, @@ -835,16 +846,24 @@ public void testFullTextSearchRefinesCandidateOnlyIndexResultsBeforeTopK(String 1) .results()) .containsExactly(5L); + assertThat( + searchWithFilter( + table, + builder.startsWith(1, BinaryString.fromString("paimon z")), + 1) + .results()) + .containsExactly(5L); } - @ParameterizedTest(name = "scalar-index.search-mode={0}") - @ValueSource(strings = {"fast", "full"}) - public void testFullTextSearchPartiallyIndexedConjunctionIsExactBeforeTopK(String scalarMode) - throws Exception { + @ParameterizedTest(name = "scalar-index.search-mode={0}, refine-from-data={1}") + @CsvSource({"fast, true", "full, true", "fast, false", "full, false"}) + public void testFullTextSearchPartiallyIndexedConjunctionIsNeverRankedAsSuperset( + String scalarMode, boolean refine) throws Exception { // Only `id` is indexed. The evaluator drops the conjunct it cannot evaluate, which makes - // the index result a superset (every row with id >= 0); ranking that superset lets row 0 - // take the single slot although only row 5 satisfies the whole predicate. - FileStoreTable table = createTable("full_text_partial_and_" + scalarMode, scalarMode); + // the index result a superset (every row with id >= 0); ranking that superset would let + // row 0 take the single slot although only row 5 satisfies the whole predicate. + FileStoreTable table = + createTable("full_text_partial_and_" + scalarMode + refine, scalarMode, refine); writeDocuments(table, RANKED_DOCUMENTS); buildAndCommitIndex(table, RANKED_DOCUMENTS); buildAndCommitIdBTreeIndex(table, RANKED_DOCUMENTS.length); @@ -854,8 +873,15 @@ public void testFullTextSearchPartiallyIndexedConjunctionIsExactBeforeTopK(Strin PredicateBuilder.and( builder.greaterOrEqual(0, 0), builder.equal(1, BinaryString.fromString("paimon zeta"))); - assertThat(searchWithFilter(table, partiallyIndexed, 1).results()).containsExactly(5L); - assertThat(searchWithFilter(table, partiallyIndexed, 10).results()).containsExactly(5L); + for (int limit : new int[] {1, 10}) { + RoaringNavigableMap64 rows = searchWithFilter(table, partiallyIndexed, limit).results(); + if (refine || scalarMode.equals("full")) { + // full mode decides rows whose filter column is unindexed from the data anyway + assertThat(rows).containsExactly(5L); + } else { + assertThat(rows).isEmpty(); + } + } } private GlobalIndexResult searchWithFilter( @@ -871,6 +897,11 @@ private GlobalIndexResult searchWithFilter( } private FileStoreTable createTable(String name, String scalarSearchMode) throws Exception { + return createTable(name, scalarSearchMode, true); + } + + private FileStoreTable createTable(String name, String scalarSearchMode, boolean refine) + throws Exception { Identifier identifier = identifier(name); Schema schema = Schema.newBuilder() @@ -880,6 +911,9 @@ private FileStoreTable createTable(String name, String scalarSearchMode) throws .option(CoreOptions.ROW_TRACKING_ENABLED.key(), "true") .option(CoreOptions.DATA_EVOLUTION_ENABLED.key(), "true") .option(CoreOptions.SCALAR_INDEX_SEARCH_MODE.key(), scalarSearchMode) + .option( + CoreOptions.GLOBAL_INDEX_FILTER_REFINE_FROM_DATA.key(), + Boolean.toString(refine)) .build(); catalog.createTable(identifier, schema, false); return getTable(identifier); diff --git a/paimon-core/src/test/java/org/apache/paimon/table/source/VectorSearchRowFilterExactnessTest.java b/paimon-core/src/test/java/org/apache/paimon/table/source/VectorSearchRowFilterExactnessTest.java new file mode 100644 index 000000000000..e203229c30d9 --- /dev/null +++ b/paimon-core/src/test/java/org/apache/paimon/table/source/VectorSearchRowFilterExactnessTest.java @@ -0,0 +1,551 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.paimon.table.source; + +import org.apache.paimon.CoreOptions; +import org.apache.paimon.catalog.Identifier; +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.globalindex.GlobalIndexBuilderUtils; +import org.apache.paimon.globalindex.GlobalIndexResult; +import org.apache.paimon.globalindex.GlobalIndexSingleColumnWriter; +import org.apache.paimon.globalindex.btree.BTreeGlobalIndexerFactory; +import org.apache.paimon.globalindex.testvector.TestVectorGlobalIndexerFactory; +import org.apache.paimon.index.IndexFileMeta; +import org.apache.paimon.io.CompactIncrement; +import org.apache.paimon.io.DataIncrement; +import org.apache.paimon.options.Options; +import org.apache.paimon.predicate.Predicate; +import org.apache.paimon.predicate.PredicateBuilder; +import org.apache.paimon.schema.Schema; +import org.apache.paimon.table.FileStoreTable; +import org.apache.paimon.table.TableTestBase; +import org.apache.paimon.table.sink.BatchTableCommit; +import org.apache.paimon.table.sink.BatchTableWrite; +import org.apache.paimon.table.sink.BatchWriteBuilder; +import org.apache.paimon.table.sink.CommitMessage; +import org.apache.paimon.table.sink.CommitMessageImpl; +import org.apache.paimon.types.ArrayType; +import org.apache.paimon.types.DataField; +import org.apache.paimon.types.DataTypes; +import org.apache.paimon.utils.Range; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; + +import java.util.Arrays; +import java.util.Collections; +import java.util.List; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * Vector search with a row filter must rank an exact match set: candidate-only scalar index answers + * (BTree contains / partially indexed conjunctions) let a non-matching but closer row take a top-k + * slot, and the engine-side filter cannot recover the matching row afterwards. + */ +public class VectorSearchRowFilterExactnessTest extends TableTestBase { + + @Override + protected Schema schemaDefault() { + return schemaBuilder(true).build(); + } + + /** + * {@code refine} sets {@code global-index.filter.refine-from-data}, which defaults to false. + */ + private static Schema.Builder schemaBuilder(boolean refine) { + return Schema.newBuilder() + .column("id", DataTypes.INT()) + .column("name", DataTypes.STRING()) + .column("vec", new ArrayType(DataTypes.FLOAT())) + .option(CoreOptions.BUCKET.key(), "-1") + .option(CoreOptions.ROW_TRACKING_ENABLED.key(), "true") + .option(CoreOptions.DATA_EVOLUTION_ENABLED.key(), "true") + .option( + CoreOptions.GLOBAL_INDEX_FILTER_REFINE_FROM_DATA.key(), + Boolean.toString(refine)) + .option("test.vector.dimension", "2") + .option("test.vector.metric", "l2"); + } + + private FileStoreTable createTable(String name, Schema.Builder schema) throws Exception { + Identifier identifier = identifier(name); + catalog.createTable(identifier, schema.build(), false); + return getTable(identifier); + } + + private static org.apache.paimon.utils.RoaringNavigableMap64 search( + FileStoreTable table, Predicate filter, int limit) { + return table.newVectorSearchBuilder() + .withVector(new float[] {1.0f, 0.0f}) + .withVectorColumn("vec") + .withLimit(limit) + .withFilter(filter) + .executeLocal() + .results(); + } + + @Test + public void testContainsOnBTreeColumnRanksOnlyMatchingRows() throws Exception { + createTableDefault(); + FileStoreTable table = getTableDefault(); + // Row 0 is the nearest neighbour of the query but does not contain "zeta". + String[] names = {"alpha", "beta zeta", "gamma"}; + float[][] vectors = {{1.0f, 0.0f}, {0.6f, 0.8f}, {0.0f, 1.0f}}; + write(table, names, vectors); + buildAndCommitVectorIndex(table, vectors); + buildAndCommitNameBTreeIndex(table, names); + + Predicate containsZeta = + new PredicateBuilder(table.rowType()).contains(1, BinaryString.fromString("zeta")); + GlobalIndexResult result = + table.newVectorSearchBuilder() + .withVector(new float[] {1.0f, 0.0f}) + .withVectorColumn("vec") + .withLimit(1) + .withFilter(containsZeta) + .executeLocal(); + + assertThat(result.results()).containsExactly(1L); + } + + @Test + public void testPartiallyIndexedConjunctionRanksOnlyMatchingRows() throws Exception { + createTableDefault(); + FileStoreTable table = getTableDefault(); + String[] names = {"alpha", "beta zeta", "gamma"}; + float[][] vectors = {{1.0f, 0.0f}, {0.6f, 0.8f}, {0.0f, 1.0f}}; + write(table, names, vectors); + buildAndCommitVectorIndex(table, vectors); + buildAndCommitIdBTreeIndex(table, names.length); + + // `id` is indexed, `name` is not: the evaluator drops the second conjunct. + PredicateBuilder builder = new PredicateBuilder(table.rowType()); + Predicate partiallyIndexed = + PredicateBuilder.and( + builder.greaterOrEqual(0, 0), + builder.equal(1, BinaryString.fromString("beta zeta"))); + GlobalIndexResult result = + table.newVectorSearchBuilder() + .withVector(new float[] {1.0f, 0.0f}) + .withVectorColumn("vec") + .withLimit(1) + .withFilter(partiallyIndexed) + .executeLocal(); + + assertThat(result.results()).containsExactly(1L); + } + + @Test + public void testOtherCandidateOnlyOperatorsAndEmptyRefinement() throws Exception { + createTableDefault(); + FileStoreTable table = getTableDefault(); + String[] names = {"alpha", "beta zeta", "gamma"}; + float[][] vectors = {{1.0f, 0.0f}, {0.6f, 0.8f}, {0.0f, 1.0f}}; + write(table, names, vectors); + buildAndCommitVectorIndex(table, vectors); + buildAndCommitNameBTreeIndex(table, names); + + PredicateBuilder builder = new PredicateBuilder(table.rowType()); + assertThat(search(table, builder.endsWith(1, BinaryString.fromString("zeta")), 1)) + .containsExactly(1L); + assertThat(search(table, builder.like(1, BinaryString.fromString("%zeta")), 1)) + .containsExactly(1L); + // The candidate set is every non-null row; refining it leaves nothing. + assertThat(search(table, builder.contains(1, BinaryString.fromString("omega")), 3)) + .isEmpty(); + // An OR with a branch no index can evaluate is not narrowed by the index at all. + assertThat( + search( + table, + PredicateBuilder.or( + builder.equal(0, 1), + builder.equal(1, BinaryString.fromString("x"))), + 1)) + .doesNotContain(0L, 2L); + } + + @Test + public void testCandidatesAreRefinedPerIndexRange() throws Exception { + createTableDefault(); + FileStoreTable table = getTableDefault(); + // Two vector index ranges; the matching row of each range is not its nearest neighbour. + String[] names = {"alpha", "beta zeta", "gamma", "delta zeta", "epsilon", "eta"}; + float[][] vectors = { + {1.0f, 0.0f}, {0.6f, 0.8f}, {0.0f, 1.0f}, {0.6f, -0.8f}, {1.0f, 0.1f}, {0.0f, -1.0f} + }; + write(table, names, vectors); + buildAndCommitVectorIndex(table, vectors, new Range(0, 2)); + buildAndCommitVectorIndex(table, vectors, new Range(3, 5)); + buildAndCommitNameBTreeIndex(table, names); + + Predicate containsZeta = + new PredicateBuilder(table.rowType()).contains(1, BinaryString.fromString("zeta")); + assertThat(search(table, containsZeta, 2)).containsExactlyInAnyOrder(1L, 3L); + assertThat(search(table, containsZeta, 10)).containsExactlyInAnyOrder(1L, 3L); + } + + @Test + public void testRefinementCombinedWithDeletionVectors() throws Exception { + Identifier identifier = identifier("vector_refine_dv"); + catalog.createTable( + identifier, + schemaBuilder(true) + .option(CoreOptions.DELETION_VECTORS_ENABLED.key(), "true") + .build(), + false); + FileStoreTable table = getTable(identifier); + String[] names = {"alpha", "beta zeta", "gamma zeta"}; + float[][] vectors = {{1.0f, 0.0f}, {0.6f, 0.8f}, {0.0f, 1.0f}}; + write(table, names, vectors); + buildAndCommitVectorIndex(table, vectors); + buildAndCommitNameBTreeIndex(table, names); + DeletionVectorTestUtils.commitDeletionVectors(table, 1L); + + Predicate containsZeta = + new PredicateBuilder(table.rowType()).contains(1, BinaryString.fromString("zeta")); + assertThat(search(table, containsZeta, 1)).containsExactly(2L); + } + + @Test + public void testBatchAndHybridVectorSearchRefineCandidates() throws Exception { + createTableDefault(); + FileStoreTable table = getTableDefault(); + String[] names = {"alpha", "beta zeta", "gamma"}; + float[][] vectors = {{1.0f, 0.0f}, {0.6f, 0.8f}, {0.0f, 1.0f}}; + write(table, names, vectors); + buildAndCommitVectorIndex(table, vectors); + buildAndCommitNameBTreeIndex(table, names); + Predicate containsZeta = + new PredicateBuilder(table.rowType()).contains(1, BinaryString.fromString("zeta")); + + List batch = + table.newBatchVectorSearchBuilder() + .withVectors(new float[][] {{1.0f, 0.0f}, {0.0f, 1.0f}}) + .withVectorColumn("vec") + .withLimit(1) + .withFilter(containsZeta) + .executeBatchLocal(); + assertThat(batch).hasSize(2); + assertThat(batch.get(0).results()).containsExactly(1L); + assertThat(batch.get(1).results()).containsExactly(1L); + + GlobalIndexResult hybrid = + table.newHybridSearchBuilder() + .addVectorRoute("vec", new float[] {1.0f, 0.0f}, 1) + .withFilter(containsZeta) + .withLimit(1) + .executeLocal(); + assertThat(hybrid.results()).containsExactly(1L); + } + + // --------------------------------------------------------------------------------------- + // global-index.filter.refine-from-data = false (the default) + // --------------------------------------------------------------------------------------- + + @Test + public void testRefineFromDataIsOffByDefault() throws Exception { + FileStoreTable table = + createTable( + "vector_refine_default", + Schema.newBuilder() + .column("id", DataTypes.INT()) + .column("name", DataTypes.STRING()) + .column("vec", new ArrayType(DataTypes.FLOAT())) + .option(CoreOptions.BUCKET.key(), "-1") + .option(CoreOptions.ROW_TRACKING_ENABLED.key(), "true") + .option(CoreOptions.DATA_EVOLUTION_ENABLED.key(), "true") + .option("test.vector.dimension", "2") + .option("test.vector.metric", "l2")); + assertThat(table.coreOptions().globalIndexFilterRefineFromData()).isFalse(); + + String[] names = {"alpha", "beta zeta", "gamma"}; + float[][] vectors = {{1.0f, 0.0f}, {0.6f, 0.8f}, {0.0f, 1.0f}}; + write(table, names, vectors); + buildAndCommitVectorIndex(table, vectors); + buildAndCommitNameBTreeIndex(table, names); + + // Candidates are excluded: never the non-matching row 0, and no data is read. + Predicate containsZeta = + new PredicateBuilder(table.rowType()).contains(1, BinaryString.fromString("zeta")); + assertThat(search(table, containsZeta, 1)).isEmpty(); + } + + @Test + public void testRefineDisabledExcludesEveryCandidateOnlyAnswer() throws Exception { + FileStoreTable table = createTable("vector_refine_off", schemaBuilder(false)); + String[] names = {"alpha", "beta zeta", "gamma"}; + float[][] vectors = {{1.0f, 0.0f}, {0.6f, 0.8f}, {0.0f, 1.0f}}; + write(table, names, vectors); + buildAndCommitVectorIndex(table, vectors); + buildAndCommitNameBTreeIndex(table, names); + + PredicateBuilder builder = new PredicateBuilder(table.rowType()); + for (Predicate candidateOnly : + Arrays.asList( + builder.contains(1, BinaryString.fromString("zeta")), + builder.endsWith(1, BinaryString.fromString("zeta")), + builder.like(1, BinaryString.fromString("%zeta")))) { + assertThat(search(table, candidateOnly, 3)).as(candidateOnly.toString()).isEmpty(); + } + + // Exact answers on the same index are unaffected by the option. + assertThat(search(table, builder.equal(1, BinaryString.fromString("beta zeta")), 1)) + .containsExactly(1L); + assertThat(search(table, builder.startsWith(1, BinaryString.fromString("beta")), 1)) + .containsExactly(1L); + assertThat( + search( + table, + builder.in( + 1, + Arrays.asList( + BinaryString.fromString("beta zeta"), + BinaryString.fromString("gamma"))), + 2)) + .containsExactlyInAnyOrder(1L, 2L); + + // A conjunction with a member no index can evaluate is a superset as well: excluded. + FileStoreTable partial = createTable("vector_refine_off_partial", schemaBuilder(false)); + write(partial, names, vectors); + buildAndCommitVectorIndex(partial, vectors); + buildAndCommitIdBTreeIndex(partial, names.length); + PredicateBuilder partialBuilder = new PredicateBuilder(partial.rowType()); + assertThat( + search( + partial, + PredicateBuilder.and( + partialBuilder.greaterOrEqual(0, 0), + partialBuilder.equal( + 1, BinaryString.fromString("beta zeta"))), + 3)) + .isEmpty(); + assertThat(search(partial, partialBuilder.greaterOrEqual(0, 1), 1)).containsExactly(1L); + } + + @Test + public void testRefineDisabledAppliesToBatchAndHybridSearch() throws Exception { + FileStoreTable table = createTable("vector_refine_off_batch", schemaBuilder(false)); + String[] names = {"alpha", "beta zeta", "gamma"}; + float[][] vectors = {{1.0f, 0.0f}, {0.6f, 0.8f}, {0.0f, 1.0f}}; + write(table, names, vectors); + buildAndCommitVectorIndex(table, vectors); + buildAndCommitNameBTreeIndex(table, names); + Predicate containsZeta = + new PredicateBuilder(table.rowType()).contains(1, BinaryString.fromString("zeta")); + + List batch = + table.newBatchVectorSearchBuilder() + .withVectors(new float[][] {{1.0f, 0.0f}, {0.0f, 1.0f}}) + .withVectorColumn("vec") + .withLimit(1) + .withFilter(containsZeta) + .executeBatchLocal(); + assertThat(batch).hasSize(2); + assertThat(batch.get(0).results().isEmpty()).isTrue(); + assertThat(batch.get(1).results().isEmpty()).isTrue(); + + GlobalIndexResult hybrid = + table.newHybridSearchBuilder() + .addVectorRoute("vec", new float[] {1.0f, 0.0f}, 1) + .withFilter(containsZeta) + .withLimit(1) + .executeLocal(); + assertThat(hybrid.results().isEmpty()).isTrue(); + } + + @Test + public void testRefineDisabledKeepsUnindexedColumnsOnTheDataPathInFullMode() throws Exception { + // The option gates only candidate refinement; rows whose filter column has no index at + // all still follow scalar-index.search-mode. + FileStoreTable table = + createTable( + "vector_refine_off_full", + schemaBuilder(false) + .option(CoreOptions.SCALAR_INDEX_SEARCH_MODE.key(), "full")); + String[] names = {"alpha", "beta zeta", "gamma"}; + float[][] vectors = {{1.0f, 0.0f}, {0.6f, 0.8f}, {0.0f, 1.0f}}; + write(table, names, vectors); + buildAndCommitVectorIndex(table, vectors); + + Predicate nameFilter = + new PredicateBuilder(table.rowType()) + .equal(1, BinaryString.fromString("beta zeta")); + assertThat(search(table, nameFilter, 1)).containsExactly(1L); + } + + @ParameterizedTest(name = "scalar-index.search-mode={0}") + @ValueSource(strings = {"fast", "full"}) + public void testFilterOnUnindexedColumn(String scalarMode) throws Exception { + // No scalar index at all: the filter cannot be answered by any index. + Identifier identifier = identifier("vector_unindexed_" + scalarMode); + catalog.createTable( + identifier, + Schema.newBuilder() + .column("id", DataTypes.INT()) + .column("name", DataTypes.STRING()) + .column("vec", new ArrayType(DataTypes.FLOAT())) + .option(CoreOptions.BUCKET.key(), "-1") + .option(CoreOptions.ROW_TRACKING_ENABLED.key(), "true") + .option(CoreOptions.DATA_EVOLUTION_ENABLED.key(), "true") + .option(CoreOptions.SCALAR_INDEX_SEARCH_MODE.key(), scalarMode) + .option("test.vector.dimension", "2") + .option("test.vector.metric", "l2") + .build(), + false); + FileStoreTable table = getTable(identifier); + String[] names = {"alpha", "beta zeta", "gamma"}; + float[][] vectors = {{1.0f, 0.0f}, {0.6f, 0.8f}, {0.0f, 1.0f}}; + write(table, names, vectors); + buildAndCommitVectorIndex(table, vectors); + + Predicate nameFilter = + new PredicateBuilder(table.rowType()) + .equal(1, BinaryString.fromString("beta zeta")); + GlobalIndexResult result = + table.newVectorSearchBuilder() + .withVector(new float[] {1.0f, 0.0f}) + .withVectorColumn("vec") + .withLimit(1) + .withFilter(nameFilter) + .executeLocal(); + + // Never a non-matching row; full mode must find the matching one from the data. + assertThat(result.results()).doesNotContain(0L, 2L); + if (scalarMode.equals("full")) { + assertThat(result.results()).containsExactly(1L); + } + } + + private void write(FileStoreTable table, String[] names, float[][] vectors) throws Exception { + BatchWriteBuilder writeBuilder = table.newBatchWriteBuilder(); + try (BatchTableWrite write = writeBuilder.newWrite(); + BatchTableCommit commit = writeBuilder.newCommit()) { + for (int i = 0; i < names.length; i++) { + write.write( + GenericRow.of( + i, + BinaryString.fromString(names[i]), + new GenericArray(vectors[i]))); + } + commit.commit(write.prepareCommit()); + } + } + + private void buildAndCommitVectorIndex(FileStoreTable table, float[][] vectors) + throws Exception { + buildAndCommitVectorIndex(table, vectors, new Range(0, vectors.length - 1)); + } + + /** Indexes {@code vectors[rowRange]} as one index file; local ids start at 0. */ + private void buildAndCommitVectorIndex(FileStoreTable table, float[][] vectors, Range rowRange) + throws Exception { + Options options = table.coreOptions().toConfiguration(); + DataField vectorField = table.rowType().getField("vec"); + GlobalIndexSingleColumnWriter writer = + (GlobalIndexSingleColumnWriter) + GlobalIndexBuilderUtils.createIndexWriter( + table, + TestVectorGlobalIndexerFactory.IDENTIFIER, + vectorField, + options); + for (long rowId = rowRange.from; rowId <= rowRange.to; rowId++) { + writer.write(vectors[(int) rowId], rowId - rowRange.from); + } + commitIndex( + table, + GlobalIndexBuilderUtils.toIndexFileMetas( + table.fileIO(), + table.store().pathFactory().globalIndexFileFactory(), + table.coreOptions(), + rowRange, + vectorField.id(), + TestVectorGlobalIndexerFactory.IDENTIFIER, + writer.finish())); + } + + private void buildAndCommitNameBTreeIndex(FileStoreTable table, String[] names) + throws Exception { + Options options = table.coreOptions().toConfiguration(); + DataField nameField = table.rowType().getField("name"); + GlobalIndexSingleColumnWriter writer = + (GlobalIndexSingleColumnWriter) + GlobalIndexBuilderUtils.createIndexWriter( + table, BTreeGlobalIndexerFactory.IDENTIFIER, nameField, options); + // The btree writer needs sorted keys. + Integer[] order = new Integer[names.length]; + for (int i = 0; i < names.length; i++) { + order[i] = i; + } + Arrays.sort(order, (a, b) -> names[a].compareTo(names[b])); + for (int rowId : order) { + writer.write(BinaryString.fromString(names[rowId]), rowId); + } + commitIndex( + table, + GlobalIndexBuilderUtils.toIndexFileMetas( + table.fileIO(), + table.store().pathFactory().globalIndexFileFactory(), + table.coreOptions(), + new Range(0, names.length - 1), + nameField.id(), + BTreeGlobalIndexerFactory.IDENTIFIER, + writer.finish())); + } + + private void buildAndCommitIdBTreeIndex(FileStoreTable table, int rowCount) throws Exception { + Options options = table.coreOptions().toConfiguration(); + DataField idField = table.rowType().getField("id"); + GlobalIndexSingleColumnWriter writer = + (GlobalIndexSingleColumnWriter) + GlobalIndexBuilderUtils.createIndexWriter( + table, BTreeGlobalIndexerFactory.IDENTIFIER, idField, options); + for (int i = 0; i < rowCount; i++) { + writer.write(i, i); + } + commitIndex( + table, + GlobalIndexBuilderUtils.toIndexFileMetas( + table.fileIO(), + table.store().pathFactory().globalIndexFileFactory(), + table.coreOptions(), + new Range(0, rowCount - 1), + idField.id(), + BTreeGlobalIndexerFactory.IDENTIFIER, + writer.finish())); + } + + private static void commitIndex(FileStoreTable table, List indexFiles) + throws Exception { + CommitMessage message = + new CommitMessageImpl( + BinaryRow.EMPTY_ROW, + 0, + null, + DataIncrement.indexIncrement(indexFiles), + CompactIncrement.emptyIncrement()); + try (BatchTableCommit commit = table.newBatchWriteBuilder().newCommit()) { + commit.commit(Collections.singletonList(message)); + } + } +} diff --git a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/FullTextSearchTest.scala b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/FullTextSearchTest.scala index dfdd82087797..b99170eef594 100644 --- a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/FullTextSearchTest.scala +++ b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/FullTextSearchTest.scala @@ -553,4 +553,49 @@ class FullTextSearchTest extends PaimonSparkTestBase { assert(mixed.subsetOf(Set(3, 5))) } } + + test("full-text search - candidate-only index answers are excluded unless refine-from-data") { + withTable("T") { + createRankedTable() + // A BTree answers LIKE '%zeta%' with every non-null row; that superset must not be ranked. + spark + .sql("CALL sys.create_global_index(table => 'test.T', index_column => 'content', index_type => 'btree')") + .collect() + + val excluded = spark + .sql(s""" + |SELECT id + |FROM full_text_search('T', 'content', '$rankedQuery', 1) + |WHERE content LIKE '%zeta%' + |""".stripMargin) + .collect() + assert(excluded.isEmpty) + + spark.sql("ALTER TABLE T SET TBLPROPERTIES ('global-index.filter.refine-from-data' = 'true')") + val refined = spark + .sql(s""" + |SELECT id + |FROM full_text_search('T', 'content', '$rankedQuery', 1) + |WHERE content LIKE '%zeta%' + |""".stripMargin) + .collect() + .map(_.getInt(0)) + .toSeq + assert(refined == Seq(5)) + + // Exact operators on the same index work either way. + spark.sql( + "ALTER TABLE T SET TBLPROPERTIES ('global-index.filter.refine-from-data' = 'false')") + val exact = spark + .sql(s""" + |SELECT id + |FROM full_text_search('T', 'content', '$rankedQuery', 1) + |WHERE content = 'paimon zeta' + |""".stripMargin) + .collect() + .map(_.getInt(0)) + .toSeq + assert(exact == Seq(5)) + } + } }