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 759041653725c4aa0b927d9fd586b5e37f41891d Author: Xiangyi Zhu <[email protected]> AuthorDate: Fri Sep 18 11:46:06 2026 +0800 [core][spark][docs] Support row filters in full-text search (#9855) --- .../multimodal-table/global-index/full-text.mdx | 62 +++ .../global-index/hybrid-search.mdx | 12 + docs/docs/primary-key-table/global-index.mdx | 4 +- .../table/source/DataEvolutionFullTextRead.java | 208 ++++++- .../table/source/DataEvolutionFullTextScan.java | 88 ++- .../paimon/table/source/FilteredRowIdReader.java | 132 +++++ .../paimon/table/source/FullTextSearchBuilder.java | 11 + .../table/source/FullTextSearchBuilderImpl.java | 42 +- .../table/source/HybridSearchBuilderImpl.java | 13 +- .../table/source/IndexFullTextSearchSplit.java | 46 +- .../paimon/table/source/RawFullTextReadImpl.java | 33 +- .../table/source/FullTextSearchBuilderTest.java | 617 ++++++++++++++++++++- .../table/source/PrimaryKeyFullTextSearchTest.java | 32 ++ .../index/NativeFullTextRowFilterTest.java | 517 +++++++++++++++++ .../org/apache/paimon/spark/PaimonBaseScan.scala | 4 +- .../paimon/spark/sql/FullTextSearchTest.scala | 228 ++++++++ .../apache/paimon/spark/sql/HybridSearchTest.scala | 57 +- 17 files changed, 2026 insertions(+), 80 deletions(-) diff --git a/docs/docs/multimodal-table/global-index/full-text.mdx b/docs/docs/multimodal-table/global-index/full-text.mdx index 8edef0e54d..af6ad4916b 100644 --- a/docs/docs/multimodal-table/global-index/full-text.mdx +++ b/docs/docs/multimodal-table/global-index/full-text.mdx @@ -206,6 +206,17 @@ FROM full_text_search( 10 ) ORDER BY __paimon_search_score DESC, id; + +-- Filter rows before ranking: the top 10 is taken among rows with category = 'lake'. +SELECT id, content, __paimon_search_score +FROM full_text_search( + 'my_table', + 'content', + '{"match":{"query":"paimon lake format"}}', + 10 +) +WHERE category = 'lake' AND dt >= '2026-09-01' +ORDER BY __paimon_search_score DESC, id; ``` Spark exposes the score as the `__paimon_search_score` metadata column. @@ -230,6 +241,19 @@ try (RecordReader<InternalRow> reader = readBuilder.newRead().createReader(plan) } ``` +Add a row filter with `withFilter` to rank only matching rows: + +```java +Predicate filter = new PredicateBuilder(table.rowType()).equal(1, BinaryString.fromString("lake")); + +GlobalIndexResult result = + table.newFullTextSearchBuilder() + .withQuery("content", "{\"match\":{\"query\":\"paimon lake\"}}") + .withFilter(filter) + .withLimit(10) + .executeLocal(); +``` + </TabItem> <TabItem value="python-sdk" label="Python SDK"> @@ -259,6 +283,44 @@ When `full-text-index.search-mode` is `full` or `detail`, Paimon also covers unindexed row ranges by reading raw rows and building a temporary native full-text index for the searched column. +### Row Filters + +A full-text search can carry a row filter: a Spark `WHERE` clause on the +`full_text_search` result, `withFilter` on the Java builder, or `withFilter` on +a hybrid search. Partition predicates in the filter prune partitions; the +remaining predicates are evaluated **before** top-k ranking, so the result is +the top-k among matching rows rather than a filtered subset of the unfiltered +top-k. + +Non-partition predicates are resolved into an exact set of matching rows +**before** ranking, and that set is handed to the existing full-text index, so +BM25 statistics are always those of the full corpus: + +- Rows whose filter columns are covered by a scalar global index (BTree, + 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. +- Rows whose filter columns are not covered by a scalar index follow + `scalar-index.search-mode`: + +| `scalar-index.search-mode` | Rows without an index on the filter columns | +|---|---| +| `fast` (default) | Excluded from the search; a warning is logged | +| `full` / `detail` | Decided by reading their filter columns | + +Build a scalar index on the columns you filter by to avoid the data read. +With `full-text-index.search-mode=fast`, only rows covered by the full-text +index are considered in either case; rows outside the coverage are searched +through a temporary index in `full` mode, where the filter is applied the same +way. Predicates the engine cannot push down (for example, UDFs) are applied by +the engine after the search, so such queries may return fewer than `limit` +rows. + +Primary-key full-text indexes do not support row filters yet; see +[Primary-Key Indexes](../../primary-key-table/global-index#limitations). + ## Query DSL The query DSL is a JSON object with one top-level query type. Use the examples diff --git a/docs/docs/multimodal-table/global-index/hybrid-search.mdx b/docs/docs/multimodal-table/global-index/hybrid-search.mdx index 09ef65e11b..81bfae13e0 100644 --- a/docs/docs/multimodal-table/global-index/hybrid-search.mdx +++ b/docs/docs/multimodal-table/global-index/hybrid-search.mdx @@ -397,3 +397,15 @@ increase route limits when a small candidate set restricts final recall. For Java and Python, the subsequent table scan does not preserve ranking order. See [Read Scored Results](./manage-indexes#read-scored-results) to access scores by row ID. + +## Row Filters + +A row filter given to a hybrid search (`WHERE` on the Spark `hybrid_search` +result, or `withFilter` on the Java builder) is applied to every route before +its own top-k, so vector and full-text candidates are drawn from the same +filtered row set before the ranker merges them. Full-text routes follow the same +rules as a standalone full-text search; see +[Full-Text Row Filters](./full-text#row-filters). In particular, under +`scalar-index.search-mode=fast` a full-text route whose filter columns have no +scalar index contributes no candidates, so the fused result comes from the +other routes only. diff --git a/docs/docs/primary-key-table/global-index.mdx b/docs/docs/primary-key-table/global-index.mdx index 82ca5f097b..a5c2a8d986 100644 --- a/docs/docs/primary-key-table/global-index.mdx +++ b/docs/docs/primary-key-table/global-index.mdx @@ -620,6 +620,8 @@ create a table with the desired definition and migrate the data. - FAST is the default search mode and excludes uncovered files from primary-key Vector and Full Text results. Vector supports exact fallback for uncovered files in FULL and DETAIL modes; primary-key Full Text supports only FAST until compaction creates persistent archives. -- Full Text routes support partition pruning but not arbitrary row predicates before Top-K. +- Primary-key Full Text routes support partition pruning but not arbitrary row predicates + before Top-K. Row filters on full-text search are supported for Data Evolution tables only; see + [Row Filters](../multimodal-table/global-index/full-text#row-filters). - Hybrid search cannot mix source-backed physical routes with global row-ID routes. - Online replacement between two definitions on the same column is not supported. 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 40e492968c..9a8227b1b9 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 @@ -18,7 +18,11 @@ package org.apache.paimon.table.source; +import org.apache.paimon.CoreOptions.GlobalIndexSearchMode; import org.apache.paimon.Snapshot; +import org.apache.paimon.globalindex.DataEvolutionGlobalIndexCoverage; +import org.apache.paimon.globalindex.DataEvolutionGlobalIndexScanner; +import org.apache.paimon.globalindex.GlobalIndexEvaluator; import org.apache.paimon.globalindex.GlobalIndexIOMeta; import org.apache.paimon.globalindex.GlobalIndexReadThreadPool; import org.apache.paimon.globalindex.GlobalIndexReader; @@ -32,31 +36,49 @@ import org.apache.paimon.index.GlobalIndexMeta; 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; import org.apache.paimon.utils.IOUtils; 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; import java.util.ArrayList; +import java.util.Comparator; import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.Optional; +import java.util.Set; +import java.util.TreeSet; import java.util.concurrent.CompletableFuture; 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}. */ public class DataEvolutionFullTextRead implements FullTextRead { + private static final Logger LOG = LoggerFactory.getLogger(DataEvolutionFullTextRead.class); + private final FileStoreTable table; @Nullable private final PartitionPredicate partitionFilter; + @Nullable private final Predicate filter; private final int limit; private final DataField textColumn; private final String query; @@ -67,8 +89,19 @@ public class DataEvolutionFullTextRead implements FullTextRead { int limit, List<DataField> textColumns, String query) { + this(table, partitionFilter, null, limit, textColumns, query); + } + + public DataEvolutionFullTextRead( + FileStoreTable table, + @Nullable PartitionPredicate partitionFilter, + @Nullable Predicate filter, + int limit, + List<DataField> textColumns, + String query) { this.table = table; this.partitionFilter = partitionFilter; + this.filter = filter; this.limit = limit; if (textColumns.size() != 1) { throw new IllegalArgumentException( @@ -100,10 +133,12 @@ public class DataEvolutionFullTextRead implements FullTextRead { ExecutorService executor = GlobalIndexReadThreadPool.getExecutorService(parallelism); Map<String, List<IndexFullTextSearchSplit>> splitsByColumn = new HashMap<>(); + List<IndexFullTextSearchSplit> indexSplits = new ArrayList<>(); List<Range> rawRowRanges = new ArrayList<>(); for (FullTextSearchSplit split : splits) { if (split instanceof IndexFullTextSearchSplit) { IndexFullTextSearchSplit indexSplit = (IndexFullTextSearchSplit) split; + indexSplits.add(indexSplit); splitsByColumn .computeIfAbsent(indexSplit.columnName(), k -> new ArrayList<>()) .add(indexSplit); @@ -115,14 +150,22 @@ public class DataEvolutionFullTextRead implements FullTextRead { GlobalIndexFileReader indexFileReader = m -> table.fileIO().newInputStream(m.filePath()); RoaringNavigableMap64 liveRows = GlobalIndexLiveRowFilter.liveRows(table, planSnapshot, partitionFilter, null); + RoaringNavigableMap64 matchedRows = matchedRows(indexSplits, planSnapshot, liveRows); ScoredGlobalIndexResult result = - evalQuery(splitsByColumn, indexPathFactory, indexFileReader, executor, liveRows); + evalQuery( + splitsByColumn, + indexPathFactory, + indexFileReader, + executor, + liveRows, + matchedRows); if (!rawRowRanges.isEmpty()) { result = new RawFullTextReadImpl( table, planSnapshot, partitionFilter, + filter, limit, textColumn, this::evalQuery) @@ -131,12 +174,153 @@ public class DataEvolutionFullTextRead implements FullTextRead { return result.topK(limit); } + /** + * Rows of the indexed splits that satisfy {@link #filter}, or {@code null} when there is no + * filter. The set is exact: rows enter top-k ranking only when the whole predicate holds. + * + * <ul> + * <li>Rows whose filter columns are covered by a scalar global index are decided by the + * 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. + * <li>Rows whose filter columns are not covered follow {@code scalar-index.search-mode}: + * excluded in {@code fast}, otherwise decided by reading their filter columns. + * </ul> + * + * <p>Either way the rows stay in the existing full-text index and are handed to it as an + * include bitmap, so BM25 statistics are those of the full corpus. + */ + @Nullable + private RoaringNavigableMap64 matchedRows( + List<IndexFullTextSearchSplit> indexSplits, + @Nullable Snapshot planSnapshot, + @Nullable RoaringNavigableMap64 liveRows) { + if (filter == null || indexSplits.isEmpty()) { + return null; + } + + RoaringNavigableMap64 covered = new RoaringNavigableMap64(); + Set<IndexFileMeta> scalarIndexFiles = + new TreeSet<>(Comparator.comparing(IndexFileMeta::fileName)); + for (IndexFullTextSearchSplit split : indexSplits) { + for (Range range : split.searchRowRanges()) { + covered.addRange(range); + } + scalarIndexFiles.addAll(split.scalarIndexFiles()); + } + if (liveRows != null) { + covered.and(liveRows); + } + + RoaringNavigableMap64 unindexed = new RoaringNavigableMap64(); + for (Range range : + new DataEvolutionGlobalIndexCoverage( + table, + planSnapshot, + partitionFilter, + scalarIndexFiles, + table.coreOptions().scalarIndexSearchMode()) + .unindexedRanges(table.rowType(), filter)) { + unindexed.addRange(range); + } + unindexed.and(covered); + RoaringNavigableMap64 decidedByIndex = + RoaringNavigableMap64.or(new RoaringNavigableMap64(), covered); + decidedByIndex.andNot(unindexed); + + RoaringNavigableMap64 matched = new RoaringNavigableMap64(); + if (!decidedByIndex.isEmpty()) { + Optional<GlobalIndexEvaluator.Evaluation> evaluation = + evaluateWithIndexes(scalarIndexFiles, planSnapshot); + if (evaluation.isPresent()) { + RoaringNavigableMap64 fromIndex = + RoaringNavigableMap64.and( + evaluation.get().result().results(), decidedByIndex); + if (!isExact(evaluation.get()) && !fromIndex.isEmpty()) { + fromIndex = + new FilteredRowIdReader(table, planSnapshot, partitionFilter, filter) + .matchingRowIds(fromIndex); + } + matched.or(fromIndex); + } else { + warnUnindexedFilter(); + } + } + if (!unindexed.isEmpty()) { + matched.or( + new FilteredRowIdReader(table, planSnapshot, partitionFilter, filter) + .matchingRowIds(unindexed)); + } + return matched; + } + + private Optional<GlobalIndexEvaluator.Evaluation> evaluateWithIndexes( + Set<IndexFileMeta> scalarIndexFiles, @Nullable Snapshot planSnapshot) { + Optional<DataEvolutionGlobalIndexScanner> optionalScanner = + DataEvolutionGlobalIndexScanner.create( + table, planSnapshot, partitionFilter, scalarIndexFiles); + if (!optionalScanner.isPresent()) { + return Optional.empty(); + } + try (DataEvolutionGlobalIndexScanner scanner = optionalScanner.get()) { + return scanner.scanWithCoverage(filter); + } catch (IOException e) { + throw new RuntimeException(e); + } + } + + /** + * 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<Integer> 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( + "Full-text search on table {} has a row filter {} that no scalar global index " + + "can evaluate; indexed rows are excluded from the result because " + + "scalar-index.search-mode is fast. Build a scalar index on the " + + "filtered columns or use search mode full.", + table.name(), + filter); + } + } + ScoredGlobalIndexResult evalQuery( Map<String, List<IndexFullTextSearchSplit>> splitsByColumn, IndexPathFactory indexPathFactory, GlobalIndexFileReader indexFileReader, - ExecutorService executor) { - return evalQuery(splitsByColumn, indexPathFactory, indexFileReader, executor, null); + ExecutorService executor, + @Nullable RoaringNavigableMap64 includeRowIds) { + return evalQuery( + splitsByColumn, indexPathFactory, indexFileReader, executor, null, includeRowIds); } private ScoredGlobalIndexResult evalQuery( @@ -144,14 +328,16 @@ public class DataEvolutionFullTextRead implements FullTextRead { IndexPathFactory indexPathFactory, GlobalIndexFileReader indexFileReader, ExecutorService executor, - @Nullable RoaringNavigableMap64 liveRows) { + @Nullable RoaringNavigableMap64 liveRows, + @Nullable RoaringNavigableMap64 matchedRows) { return evalColumnQuery( textColumn.name(), splitsByColumn, indexPathFactory, indexFileReader, executor, - liveRows); + liveRows, + matchedRows); } private ScoredGlobalIndexResult evalColumnQuery( @@ -160,7 +346,8 @@ public class DataEvolutionFullTextRead implements FullTextRead { IndexPathFactory indexPathFactory, GlobalIndexFileReader indexFileReader, ExecutorService executor, - @Nullable RoaringNavigableMap64 liveRows) { + @Nullable RoaringNavigableMap64 liveRows, + @Nullable RoaringNavigableMap64 matchedRows) { List<IndexFullTextSearchSplit> columnSplits = splitsByColumn.get(column); if (columnSplits == null || columnSplits.isEmpty()) { return ScoredGlobalIndexResult.createEmpty(); @@ -183,7 +370,7 @@ public class DataEvolutionFullTextRead implements FullTextRead { split.fullTextIndexFiles(), indexFileReader, executor, - includeRowIds(split, liveRows))); + includeRowIds(split, liveRows, matchedRows))); } CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join(); @@ -201,7 +388,9 @@ public class DataEvolutionFullTextRead implements FullTextRead { @Nullable private static RoaringNavigableMap64 includeRowIds( - IndexFullTextSearchSplit split, @Nullable RoaringNavigableMap64 liveRows) { + IndexFullTextSearchSplit split, + @Nullable RoaringNavigableMap64 liveRows, + @Nullable RoaringNavigableMap64 matchedRows) { RoaringNavigableMap64 include = new RoaringNavigableMap64(); for (Range range : split.searchRowRanges()) { include.addRange(range); @@ -209,6 +398,9 @@ public class DataEvolutionFullTextRead implements FullTextRead { if (liveRows != null) { include.and(liveRows); } + if (matchedRows != null) { + include.and(matchedRows); + } long physicalRowCount = split.rowRangeEnd() - split.rowRangeStart() + 1; return include.getLongCardinality() == physicalRowCount ? null : include; } diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/DataEvolutionFullTextScan.java b/paimon-core/src/main/java/org/apache/paimon/table/source/DataEvolutionFullTextScan.java index fbd1bd83d1..b909a87007 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/source/DataEvolutionFullTextScan.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/source/DataEvolutionFullTextScan.java @@ -28,6 +28,7 @@ import org.apache.paimon.index.IndexFileHandler; import org.apache.paimon.index.IndexFileMeta; import org.apache.paimon.manifest.IndexManifestEntry; import org.apache.paimon.partition.PartitionPredicate; +import org.apache.paimon.predicate.Predicate; import org.apache.paimon.table.FileStoreTable; import org.apache.paimon.table.source.snapshot.TimeTravelUtil; import org.apache.paimon.types.DataField; @@ -49,6 +50,7 @@ import java.util.Set; import java.util.TreeSet; import java.util.stream.Collectors; +import static org.apache.paimon.predicate.PredicateVisitor.collectFieldIds; import static org.apache.paimon.utils.Preconditions.checkNotNull; /** Implementation for {@link FullTextScan}. */ @@ -56,6 +58,7 @@ public class DataEvolutionFullTextScan implements FullTextScan { private final FileStoreTable table; @Nullable private final PartitionPredicate partitionFilter; + @Nullable private final Predicate filter; private final List<DataField> textColumns; @Nullable private final Snapshot pinnedSnapshot; @@ -78,8 +81,18 @@ public class DataEvolutionFullTextScan implements FullTextScan { @Nullable PartitionPredicate partitionFilter, List<DataField> textColumns, @Nullable Snapshot pinnedSnapshot) { + this(table, partitionFilter, null, textColumns, pinnedSnapshot); + } + + public DataEvolutionFullTextScan( + FileStoreTable table, + @Nullable PartitionPredicate partitionFilter, + @Nullable Predicate filter, + List<DataField> textColumns, + @Nullable Snapshot pinnedSnapshot) { this.table = table; this.partitionFilter = partitionFilter; + this.filter = filter; this.textColumns = textColumns; this.pinnedSnapshot = pinnedSnapshot; } @@ -98,6 +111,7 @@ public class DataEvolutionFullTextScan implements FullTextScan { idToColumn.put(textColumn.id(), textColumn.name()); } + Set<Integer> filterFieldIds = collectFieldIds(table.rowType(), filter); @Nullable Snapshot snapshot = pinnedSnapshot != null ? pinnedSnapshot : TimeTravelUtil.tryTravelOrLatest(table); @@ -111,34 +125,54 @@ public class DataEvolutionFullTextScan implements FullTextScan { if (globalIndex == null) { return false; } - return !matchedTextColumnIds(globalIndex, textColumnIds).isEmpty() - && supportsFullTextSearch(entry.indexFile().indexType()); + return isFullTextIndexFile(entry.indexFile(), textColumnIds) + || containsAnyField(globalIndex, filterFieldIds); }; List<IndexFileMeta> allIndexFiles = indexFileHandler.scan(snapshot, indexFileFilter).stream() .map(IndexManifestEntry::indexFile) .collect(Collectors.toList()); + List<IndexFileMeta> fullTextIndexFiles = new ArrayList<>(); + List<IndexFileMeta> scalarIndexFiles = new ArrayList<>(); + for (IndexFileMeta indexFile : allIndexFiles) { + boolean fullText = isFullTextIndexFile(indexFile, textColumnIds); + if (fullText) { + fullTextIndexFiles.add(indexFile); + } + // A dedicated full-text index cannot evaluate scalar predicates; a multi-field index + // (text carried next to scalar extra fields) can serve both roles. + GlobalIndexMeta meta = checkNotNull(indexFile.globalIndexMeta()); + if (containsAnyField(meta, filterFieldIds) && (!fullText || hasExtraFields(meta))) { + scalarIndexFiles.add(indexFile); + } + } + // Build splits: for each chosen full-text range, attach the scalar index files that can + // pre-filter its rows. Rows the full-text index does not cover go to the raw split; rows + // it covers but no scalar index does are resolved from the data at read time. List<FullTextSearchSplit> splits = new ArrayList<>(); for (IndexRangeSelection selection : - chooseIndexRanges(allIndexFiles, textColumnIds, idToColumn)) { + chooseIndexRanges(fullTextIndexFiles, textColumnIds, idToColumn)) { splits.add( new IndexFullTextSearchSplit( selection.columnName, selection.fileRange.from, selection.fileRange.to, selection.files, - selection.searchRanges)); + selection.searchRanges, + scalarIndexFiles( + scalarIndexFiles, + Collections.singletonList(selection.fileRange)))); } - if (!allIndexFiles.isEmpty()) { + if (!fullTextIndexFiles.isEmpty()) { List<Range> rawRowRanges = new DataEvolutionGlobalIndexCoverage( table, snapshot, partitionFilter, - allIndexFiles, + fullTextIndexFiles, table.coreOptions().fullTextIndexSearchMode()) .unindexedRanges(textColumnIds); if (!rawRowRanges.isEmpty()) { @@ -408,6 +442,48 @@ public class DataEvolutionFullTextScan implements FullTextScan { } } + private static boolean isFullTextIndexFile( + IndexFileMeta indexFile, Set<Integer> textColumnIds) { + GlobalIndexMeta globalIndex = checkNotNull(indexFile.globalIndexMeta()); + return !matchedTextColumnIds(globalIndex, textColumnIds).isEmpty() + && supportsFullTextSearch(indexFile.indexType()); + } + + private static boolean containsAnyField(GlobalIndexMeta meta, Set<Integer> fieldIds) { + if (fieldIds.contains(meta.indexFieldId())) { + return true; + } + int[] extraFieldIds = meta.extraFieldIds(); + if (extraFieldIds != null) { + for (int extraFieldId : extraFieldIds) { + if (fieldIds.contains(extraFieldId)) { + return true; + } + } + } + return false; + } + + private static boolean hasExtraFields(GlobalIndexMeta meta) { + int[] extraFieldIds = meta.extraFieldIds(); + return extraFieldIds != null && extraFieldIds.length > 0; + } + + private static List<IndexFileMeta> scalarIndexFiles( + List<IndexFileMeta> scalarIndexFiles, List<Range> rowRanges) { + List<IndexFileMeta> result = new ArrayList<>(); + for (IndexFileMeta indexFile : scalarIndexFiles) { + Range indexRange = checkNotNull(indexFile.globalIndexMeta()).rowRange(); + for (Range rowRange : rowRanges) { + if (rowRange.hasIntersection(indexRange)) { + result.add(indexFile); + break; + } + } + } + return result; + } + private static boolean supportsFullTextSearch(String indexType) { GlobalIndexerFactory factory; try { 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 new file mode 100644 index 0000000000..8f88c511c2 --- /dev/null +++ b/paimon-core/src/main/java/org/apache/paimon/table/source/FilteredRowIdReader.java @@ -0,0 +1,132 @@ +/* + * 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.Snapshot; +import org.apache.paimon.data.InternalRow; +import org.apache.paimon.partition.PartitionPredicate; +import org.apache.paimon.predicate.Predicate; +import org.apache.paimon.predicate.PredicateVisitor; +import org.apache.paimon.reader.RecordReader; +import org.apache.paimon.table.FileStoreTable; +import org.apache.paimon.table.SpecialFields; +import org.apache.paimon.types.RowType; +import org.apache.paimon.utils.Range; +import org.apache.paimon.utils.RoaringNavigableMap64; + +import javax.annotation.Nullable; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; + +/** + * 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 + * global index lookup: a search can hand the result to its index as an include bitmap without + * touching the index corpus. + */ +class FilteredRowIdReader { + + private final FileStoreTable table; + @Nullable private final Snapshot planSnapshot; + @Nullable private final PartitionPredicate partitionFilter; + private final Predicate filter; + + FilteredRowIdReader( + FileStoreTable table, + @Nullable Snapshot planSnapshot, + @Nullable PartitionPredicate partitionFilter, + Predicate filter) { + this.table = table; + this.planSnapshot = planSnapshot; + this.partitionFilter = partitionFilter; + this.filter = filter; + } + + /** The subset of {@code rows} whose data satisfies the filter. */ + RoaringNavigableMap64 matchingRowIds(RoaringNavigableMap64 rows) { + RoaringNavigableMap64 matching = new RoaringNavigableMap64(); + if (rows.isEmpty()) { + return matching; + } + List<Range> rowRanges = Range.sortAndMergeOverlap(rows.toRangeList(), true); + RowType readType = readType(); + int rowIdIndex = readType.getFieldIndex(SpecialFields.ROW_ID.name()); + + TableScan.Plan plan = + readBuilder(readType, false).withRowRanges(rowRanges).newScan().plan(); + try (RecordReader<InternalRow> reader = + readBuilder(readType, true).newRead().executeFilter().createReader(plan)) { + reader.forEachRemaining( + row -> { + long rowId = row.getLong(rowIdIndex); + if (rows.contains(rowId)) { + matching.add(rowId); + } + }); + } catch (IOException e) { + throw new RuntimeException("Failed to evaluate the row filter on table data.", e); + } + return matching; + } + + private RowType readType() { + RowType tableRowType = table.rowType(); + Set<String> filterFields = PredicateVisitor.collectFieldNames(filter); + List<String> readFields = new ArrayList<>(); + for (String field : tableRowType.getFieldNames()) { + if (filterFields.contains(field)) { + readFields.add(field); + } + } + return SpecialFields.rowTypeWithRowId(tableRowType.project(readFields)); + } + + private ReadBuilder readBuilder(RowType readType, boolean withFilter) { + ReadBuilder readBuilder = readTable().newReadBuilder().withReadType(readType); + if (partitionFilter != null) { + readBuilder.withPartitionFilter(partitionFilter); + } + if (withFilter) { + readBuilder.withFilter(filter); + } + return readBuilder; + } + + private FileStoreTable readTable() { + if (planSnapshot == null) { + return table; + } + Map<String, String> pinOptions = new HashMap<>(); + pinOptions.put( + CoreOptions.SCAN_MODE.key(), CoreOptions.StartupMode.FROM_SNAPSHOT.toString()); + pinOptions.put(CoreOptions.SCAN_SNAPSHOT_ID.key(), String.valueOf(planSnapshot.id())); + pinOptions.put(CoreOptions.SCAN_VERSION.key(), null); + pinOptions.put(CoreOptions.SCAN_TAG_NAME.key(), null); + pinOptions.put(CoreOptions.SCAN_WATERMARK.key(), null); + pinOptions.put(CoreOptions.SCAN_TIMESTAMP.key(), null); + pinOptions.put(CoreOptions.SCAN_TIMESTAMP_MILLIS.key(), null); + return table.copyWithoutTimeTravel(pinOptions); + } +} diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/FullTextSearchBuilder.java b/paimon-core/src/main/java/org/apache/paimon/table/source/FullTextSearchBuilder.java index 5b091670da..9e07fcdbd9 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/source/FullTextSearchBuilder.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/source/FullTextSearchBuilder.java @@ -20,6 +20,7 @@ package org.apache.paimon.table.source; import org.apache.paimon.globalindex.GlobalIndexResult; import org.apache.paimon.partition.PartitionPredicate; +import org.apache.paimon.predicate.Predicate; import java.io.Serializable; @@ -32,6 +33,16 @@ public interface FullTextSearchBuilder extends Serializable { "This full-text search builder does not support partition filters."); } + /** + * Push a row filter. Rows that do not satisfy the predicate are excluded before top-k ranking, + * so the returned top-k is the top-k among matching rows. Partition predicates contained in the + * filter are extracted and applied as partition filters. + */ + default FullTextSearchBuilder withFilter(Predicate predicate) { + throw new UnsupportedOperationException( + "This full-text search builder does not support row filters."); + } + /** The top k results to return. */ FullTextSearchBuilder withLimit(int limit); diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/FullTextSearchBuilderImpl.java b/paimon-core/src/main/java/org/apache/paimon/table/source/FullTextSearchBuilderImpl.java index 77d74c9a5f..baa6dab5f9 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/source/FullTextSearchBuilderImpl.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/source/FullTextSearchBuilderImpl.java @@ -23,15 +23,21 @@ import org.apache.paimon.catalog.TableQueryAuthResult; import org.apache.paimon.index.pk.PrimaryKeyIndexDefinition; import org.apache.paimon.index.pk.PrimaryKeyIndexDefinitions; import org.apache.paimon.partition.PartitionPredicate; +import org.apache.paimon.predicate.Predicate; +import org.apache.paimon.predicate.PredicateBuilder; import org.apache.paimon.table.FileStoreTable; import org.apache.paimon.table.InnerTable; import org.apache.paimon.types.DataField; +import org.apache.paimon.utils.Pair; import javax.annotation.Nullable; +import java.util.Arrays; import java.util.Collections; +import java.util.List; import java.util.Optional; +import static org.apache.paimon.partition.PartitionPredicate.splitPartitionPredicatesAndDataPredicates; import static org.apache.paimon.utils.Preconditions.checkArgument; import static org.apache.paimon.utils.Preconditions.checkNotNull; @@ -46,6 +52,7 @@ public class FullTextSearchBuilderImpl implements FullTextSearchBuilder { private String fieldName; private String query; private PartitionPredicate partitionFilter; + @Nullable private Predicate filter; @Nullable private Snapshot pinnedSnapshot; public FullTextSearchBuilderImpl(InnerTable table) { @@ -54,10 +61,37 @@ public class FullTextSearchBuilderImpl implements FullTextSearchBuilder { @Override public FullTextSearchBuilder withPartitionFilter(PartitionPredicate partitionFilter) { - this.partitionFilter = partitionFilter; + addPartitionFilter(partitionFilter); return this; } + @Override + public FullTextSearchBuilder withFilter(Predicate predicate) { + Pair<Optional<PartitionPredicate>, List<Predicate>> pair = + splitPartitionPredicatesAndDataPredicates( + predicate, table.rowType(), table.partitionKeys()); + if (pair.getLeft().isPresent()) { + addPartitionFilter(pair.getLeft().get()); + } + if (!pair.getRight().isEmpty()) { + Predicate dataFilter = PredicateBuilder.and(pair.getRight()); + this.filter = + this.filter == null ? dataFilter : PredicateBuilder.and(filter, dataFilter); + } + return this; + } + + private void addPartitionFilter(@Nullable PartitionPredicate partitionFilter) { + if (partitionFilter == null) { + return; + } + this.partitionFilter = + this.partitionFilter == null + ? partitionFilter + : PartitionPredicate.and( + Arrays.asList(this.partitionFilter, partitionFilter)); + } + @Override public FullTextSearchBuilder withLimit(int limit) { this.limit = limit; @@ -82,6 +116,7 @@ public class FullTextSearchBuilderImpl implements FullTextSearchBuilder { : new DataEvolutionFullTextScan( table, partitionFilter, + filter, Collections.singletonList(textColumn), pinnedSnapshot); } @@ -97,6 +132,7 @@ public class FullTextSearchBuilderImpl implements FullTextSearchBuilder { : new DataEvolutionFullTextRead( table, partitionFilter, + filter, limit, Collections.singletonList(textColumn), query); @@ -121,6 +157,10 @@ public class FullTextSearchBuilderImpl implements FullTextSearchBuilder { PrimaryKeyIndexDefinitions.create(table.schema()).definitions()) { if (definition.family() == PrimaryKeyIndexDefinition.Family.FULL_TEXT && definition.fieldId() == textColumn.id()) { + if (filter != null) { + throw new UnsupportedOperationException( + "Primary-key full-text search does not support non-partition filters yet."); + } return Optional.of(definition); } } diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/HybridSearchBuilderImpl.java b/paimon-core/src/main/java/org/apache/paimon/table/source/HybridSearchBuilderImpl.java index 35460de7b4..4183af92bb 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/source/HybridSearchBuilderImpl.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/source/HybridSearchBuilderImpl.java @@ -165,16 +165,6 @@ public class HybridSearchBuilderImpl implements HybridSearchBuilder { if (limit <= 0) { throw new IllegalArgumentException("Limit must be positive, got: " + limit); } - if (filter != null) { - for (HybridSearchRoute route : routes) { - if (!route.isVector()) { - throw new UnsupportedOperationException( - "Hybrid search with full-text routes does not support non-partition " - + "filters because full-text indexes cannot apply row-id " - + "pre-filters before top-k ranking."); - } - } - } } @Override @@ -377,6 +367,9 @@ public class HybridSearchBuilderImpl implements HybridSearchBuilder { if (partitionFilter != null) { fullTextSearchBuilder.withPartitionFilter(partitionFilter); } + if (filter != null) { + fullTextSearchBuilder.withFilter(filter); + } return fullTextSearchBuilder; } } diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/IndexFullTextSearchSplit.java b/paimon-core/src/main/java/org/apache/paimon/table/source/IndexFullTextSearchSplit.java index 54a55536f1..a7e93bfefa 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/source/IndexFullTextSearchSplit.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/source/IndexFullTextSearchSplit.java @@ -37,7 +37,8 @@ public class IndexFullTextSearchSplit extends FullTextSearchSplit { private static final long serialVersionUID = 1L; - private static final int VERSION = 1; + /** Version 2 appends the scalar index files usable as a row pre-filter. */ + private static final int VERSION = 2; private static final ThreadLocal<IndexFileMetaSerializer> INDEX_SERIALIZER = ThreadLocal.withInitial(IndexFileMetaSerializer::new); @@ -47,6 +48,7 @@ public class IndexFullTextSearchSplit extends FullTextSearchSplit { private long rowRangeEnd; private List<Range> searchRowRanges; private transient List<IndexFileMeta> fullTextIndexFiles; + private transient List<IndexFileMeta> scalarIndexFiles; public IndexFullTextSearchSplit( long rowRangeStart, long rowRangeEnd, List<IndexFileMeta> fullTextIndexFiles) { @@ -72,6 +74,22 @@ public class IndexFullTextSearchSplit extends FullTextSearchSplit { long rowRangeEnd, List<IndexFileMeta> fullTextIndexFiles, List<Range> searchRowRanges) { + this( + columnName, + rowRangeStart, + rowRangeEnd, + fullTextIndexFiles, + searchRowRanges, + Collections.emptyList()); + } + + public IndexFullTextSearchSplit( + String columnName, + long rowRangeStart, + long rowRangeEnd, + List<IndexFileMeta> fullTextIndexFiles, + List<Range> searchRowRanges, + List<IndexFileMeta> scalarIndexFiles) { this.columnName = columnName; this.rowRangeStart = rowRangeStart; this.rowRangeEnd = rowRangeEnd; @@ -86,6 +104,7 @@ public class IndexFullTextSearchSplit extends FullTextSearchSplit { } } this.searchRowRanges = Collections.unmodifiableList(ranges); + this.scalarIndexFiles = Collections.unmodifiableList(new ArrayList<>(scalarIndexFiles)); } public String columnName() { @@ -108,24 +127,35 @@ public class IndexFullTextSearchSplit extends FullTextSearchSplit { return fullTextIndexFiles; } + /** Scalar global index files intersecting this split, used to pre-filter rows. */ + public List<IndexFileMeta> scalarIndexFiles() { + return scalarIndexFiles; + } + private void writeObject(ObjectOutputStream out) throws IOException { out.defaultWriteObject(); out.writeInt(VERSION); IndexFileMetaSerializer serializer = INDEX_SERIALIZER.get(); DataOutputViewStreamWrapper view = new DataOutputViewStreamWrapper(out); serializer.serializeList(fullTextIndexFiles, view); + serializer.serializeList(scalarIndexFiles, view); } private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundException { in.defaultReadObject(); int version = in.readInt(); - if (version != VERSION) { + if (version < 1 || version > VERSION) { throw new IOException("Unsupported IndexFullTextSearchSplit version: " + version); } IndexFileMetaSerializer serializer = INDEX_SERIALIZER.get(); DataInputViewStreamWrapper view = new DataInputViewStreamWrapper(in); this.fullTextIndexFiles = Collections.unmodifiableList(new ArrayList<>(serializer.deserializeList(view))); + this.scalarIndexFiles = + version >= 2 + ? Collections.unmodifiableList( + new ArrayList<>(serializer.deserializeList(view))) + : Collections.emptyList(); if (searchRowRanges == null) { searchRowRanges = Collections.singletonList(new Range(rowRangeStart, rowRangeEnd)); } else { @@ -143,13 +173,19 @@ public class IndexFullTextSearchSplit extends FullTextSearchSplit { && rowRangeEnd == that.rowRangeEnd && Objects.equals(columnName, that.columnName) && Objects.equals(searchRowRanges, that.searchRowRanges) - && Objects.equals(fullTextIndexFiles, that.fullTextIndexFiles); + && Objects.equals(fullTextIndexFiles, that.fullTextIndexFiles) + && Objects.equals(scalarIndexFiles, that.scalarIndexFiles); } @Override public int hashCode() { return Objects.hash( - columnName, rowRangeStart, rowRangeEnd, searchRowRanges, fullTextIndexFiles); + columnName, + rowRangeStart, + rowRangeEnd, + searchRowRanges, + fullTextIndexFiles, + scalarIndexFiles); } @Override @@ -166,6 +202,8 @@ public class IndexFullTextSearchSplit extends FullTextSearchSplit { + searchRowRanges + ", fullTextIndexFiles=" + fullTextIndexFiles + + ", scalarIndexFiles=" + + scalarIndexFiles + '}'; } } diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/RawFullTextReadImpl.java b/paimon-core/src/main/java/org/apache/paimon/table/source/RawFullTextReadImpl.java index 86c6d2a1d2..44a7e63a36 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/source/RawFullTextReadImpl.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/source/RawFullTextReadImpl.java @@ -37,6 +37,7 @@ import org.apache.paimon.index.IndexFileMeta; import org.apache.paimon.index.IndexPathFactory; import org.apache.paimon.options.Options; import org.apache.paimon.partition.PartitionPredicate; +import org.apache.paimon.predicate.Predicate; import org.apache.paimon.reader.RecordReader; import org.apache.paimon.table.FileStoreTable; import org.apache.paimon.table.SpecialFields; @@ -68,6 +69,7 @@ class RawFullTextReadImpl { private final FileStoreTable table; @Nullable private final Snapshot planSnapshot; @Nullable private final PartitionPredicate partitionFilter; + @Nullable private final Predicate filter; private final int limit; private final DataField textColumn; private final IndexSearch indexSearch; @@ -79,9 +81,21 @@ class RawFullTextReadImpl { int limit, DataField textColumn, IndexSearch indexSearch) { + this(table, planSnapshot, partitionFilter, null, limit, textColumn, indexSearch); + } + + RawFullTextReadImpl( + FileStoreTable table, + @Nullable Snapshot planSnapshot, + @Nullable PartitionPredicate partitionFilter, + @Nullable Predicate filter, + int limit, + DataField textColumn, + IndexSearch indexSearch) { this.table = table; this.planSnapshot = planSnapshot; this.partitionFilter = partitionFilter; + this.filter = filter; this.limit = limit; this.textColumn = textColumn; this.indexSearch = indexSearch; @@ -111,22 +125,33 @@ class RawFullTextReadImpl { int rowIdIndex = readType.getFieldIndex(SpecialFields.ROW_ID.name()); Map<String, RawFullTextIndex> rawIndexes = createRawFullTextIndexes(splitsByColumn, readType, rawRowRanges); + // Every raw row is indexed so the temporary index scores against the full raw corpus; + // the row filter only decides which rows the search may return. + RoaringNavigableMap64 matchingRows = filter == null ? null : new RoaringNavigableMap64(); try { try (RecordReader<InternalRow> reader = readBuilder.newRead().createReader(plan); CloseableIterator<InternalRow> iterator = reader.toCloseableIterator()) { while (iterator.hasNext()) { InternalRow row = iterator.next(); + long rowId = row.getLong(rowIdIndex); for (RawFullTextIndex rawIndex : rawIndexes.values()) { - long rowId = row.getLong(rowIdIndex); rawIndex.write(row, rowId); } + if (matchingRows != null && filter.test(row)) { + matchingRows.add(rowId); + } } } } catch (Exception e) { throw new RuntimeException("Failed to read raw rows for full-text search.", e); } + if (matchingRows != null && matchingRows.isEmpty()) { + IOUtils.closeAllQuietly(rawIndexes.values()); + return ScoredGlobalIndexResult.createEmpty(); + } + try { Map<String, List<IndexFullTextSearchSplit>> rawSplitsByColumn = new HashMap<>(); for (Map.Entry<String, RawFullTextIndex> entry : rawIndexes.entrySet()) { @@ -143,7 +168,8 @@ class RawFullTextReadImpl { m -> new MemorySeekableInputStream( rawFileBytes(rawIndexes, m.filePath().getName())), - executor) + executor, + matchingRows) .topK(limit); } finally { IOUtils.closeAllQuietly(rawIndexes.values()); @@ -284,7 +310,8 @@ class RawFullTextReadImpl { Map<String, List<IndexFullTextSearchSplit>> splitsByColumn, IndexPathFactory indexPathFactory, GlobalIndexFileReader indexFileReader, - ExecutorService executor); + ExecutorService executor, + @Nullable RoaringNavigableMap64 includeRowIds); } private static class RawFullTextIndex implements Closeable { 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 0c4f2b5122..94f483ec6c 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 @@ -57,6 +57,10 @@ import org.apache.paimon.types.RowType; 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 javax.annotation.Nullable; import java.io.ByteArrayInputStream; import java.io.ByteArrayOutputStream; @@ -351,22 +355,554 @@ public class FullTextSearchBuilderTest extends TableTestBase { } @Test - public void testHybridSearchRejectsDataFilterWithFullTextRoute() throws Exception { + public void testHybridSearchAppliesDataFilterToFullTextRoute() throws Exception { + createTableDefault(); + FileStoreTable table = getTableDefault(); + + writeDocuments(table, RANKED_DOCUMENTS); + buildAndCommitIndex(table, RANKED_DOCUMENTS); + buildAndCommitIdBTreeIndex(table, RANKED_DOCUMENTS.length); + + Predicate idFilter = new PredicateBuilder(table.rowType()).greaterOrEqual(0, 3); + ScoredGlobalIndexResult result = + table.newHybridSearchBuilder() + .addFullTextRoute(TEXT_FIELD_NAME, matchQuery("paimon lake"), 2, 1.0f) + .withFilter(idFilter) + .withLimit(2) + .executeLocal(); + + // Rows 0-2 score higher, but the filter removes them before the route's top-k. + assertThat(result.results()).hasSize(2); + assertThat(result.results()).isSubsetOf(3L, 4L, 5L); + } + + // ====================== Row filter tests ====================== + + /** + * Rows 0-2 match both query terms of {@code "paimon lake"} (score 1.0) while rows 3-5 match + * only {@code "paimon"} (score 0.5), so an unfiltered top-k is always taken from rows 0-2. + */ + private static final String[] RANKED_DOCUMENTS = { + "paimon lake alpha", + "paimon lake beta", + "paimon lake gamma", + "paimon delta", + "paimon epsilon", + "paimon zeta" + }; + + @Test + public void testFullTextSearchWithFilterRanksOnlyMatchingRows() throws Exception { + createTableDefault(); + FileStoreTable table = getTableDefault(); + writeDocuments(table, RANKED_DOCUMENTS); + buildAndCommitIndex(table, RANKED_DOCUMENTS); + buildAndCommitIdBTreeIndex(table, RANKED_DOCUMENTS.length); + + GlobalIndexResult unfiltered = searchWithFilter(table, null, 2); + assertThat(unfiltered.results()).isSubsetOf(0L, 1L, 2L); + + Predicate idFilter = new PredicateBuilder(table.rowType()).greaterOrEqual(0, 3); + GlobalIndexResult filtered = searchWithFilter(table, idFilter, 2); + assertThat(filtered.results()).hasSize(2); + assertThat(filtered.results()).isSubsetOf(3L, 4L, 5L); + assertThat(readIds(table, filtered)).allMatch(id -> id >= 3); + + // The scores are the full-text scores of the surviving rows, not rescored. + ScoredGlobalIndexResult scored = (ScoredGlobalIndexResult) filtered; + for (long rowId : scored.results()) { + assertThat(scored.scoreGetter().score(rowId)).isEqualTo(0.5f); + } + } + + @Test + public void testFullTextSearchWithCompoundFilters() throws Exception { + createTableDefault(); + FileStoreTable table = getTableDefault(); + writeDocuments(table, RANKED_DOCUMENTS); + buildAndCommitIndex(table, RANKED_DOCUMENTS); + buildAndCommitIdBTreeIndex(table, RANKED_DOCUMENTS.length); + + PredicateBuilder builder = new PredicateBuilder(table.rowType()); + Predicate between = + PredicateBuilder.and(builder.greaterOrEqual(0, 2), builder.lessOrEqual(0, 4)); + assertThat(searchWithFilter(table, between, 10).results()) + .containsExactlyInAnyOrder(2L, 3L, 4L); + + Predicate either = PredicateBuilder.or(builder.equal(0, 0), builder.equal(0, 5)); + assertThat(searchWithFilter(table, either, 10).results()).containsExactlyInAnyOrder(0L, 5L); + + Predicate in = builder.in(0, Arrays.asList(1, 4)); + assertThat(searchWithFilter(table, in, 10).results()).containsExactlyInAnyOrder(1L, 4L); + + // A filter that matches rows the query does not still yields only query matches. + Predicate all = builder.greaterOrEqual(0, 0); + assertThat(searchWithFilter(table, all, 10).results()) + .containsExactlyInAnyOrder(0L, 1L, 2L, 3L, 4L, 5L); + } + + @Test + public void testFullTextSearchWithFilterMatchingNoRows() throws Exception { + createTableDefault(); + FileStoreTable table = getTableDefault(); + writeDocuments(table, RANKED_DOCUMENTS); + buildAndCommitIndex(table, RANKED_DOCUMENTS); + buildAndCommitIdBTreeIndex(table, RANKED_DOCUMENTS.length); + + Predicate impossible = new PredicateBuilder(table.rowType()).greaterThan(0, 100); + GlobalIndexResult result = searchWithFilter(table, impossible, 10); + assertThat(result.results().isEmpty()).isTrue(); + assertThat(readIds(table, result)).isEmpty(); + } + + @Test + public void testFullTextSearchFastModeExcludesRowsWithoutScalarIndex() throws Exception { + createTableDefault(); + FileStoreTable table = getTableDefault(); + writeDocuments(table, RANKED_DOCUMENTS); + buildAndCommitIndex(table, RANKED_DOCUMENTS); + // No scalar index on "id": in fast mode nothing can be verified, so nothing is returned. + + Predicate idFilter = new PredicateBuilder(table.rowType()).greaterOrEqual(0, 3); + FullTextSearchBuilder builder = + table.newFullTextSearchBuilder() + .withQuery(TEXT_FIELD_NAME, matchQuery("paimon")) + .withLimit(10) + .withFilter(idFilter); + FullTextScan.Plan plan = builder.newFullTextScan().scan(); + assertThat(plan.splits()).noneMatch(RawFullTextSearchSplit.class::isInstance); + assertThat(builder.newFullTextRead().read(plan).results().isEmpty()).isTrue(); + } + + @Test + public void testFullTextSearchFullModeScansRowsWithoutScalarIndex() throws Exception { + FileStoreTable table = createTable("full_text_filter_full_mode", "full"); + writeDocuments(table, RANKED_DOCUMENTS); + buildAndCommitIndex(table, RANKED_DOCUMENTS); + // No scalar index on "id": scalar-index.search-mode=full routes every row to a raw + // scan where the predicate is evaluated on the data. + + Predicate idFilter = new PredicateBuilder(table.rowType()).greaterOrEqual(0, 3); + FullTextSearchBuilder builder = + table.newFullTextSearchBuilder() + .withQuery(TEXT_FIELD_NAME, matchQuery("paimon lake")) + .withLimit(2) + .withFilter(idFilter); + FullTextScan.Plan plan = builder.newFullTextScan().scan(); + // The rows stay in the full-text index; the filter is resolved by reading `id`. + assertThat(plan.splits()).noneMatch(RawFullTextSearchSplit.class::isInstance); + + GlobalIndexResult result = builder.newFullTextRead().read(plan); + assertThat(result.results()).hasSize(2); + assertThat(result.results()).isSubsetOf(3L, 4L, 5L); + } + + @Test + public void testFullTextSearchFullModeMixesIndexAndDataFilterEvaluation() throws Exception { + FileStoreTable table = createTable("full_text_filter_partial_scalar", "full"); + writeDocuments(table, RANKED_DOCUMENTS); + buildAndCommitIndex(table, RANKED_DOCUMENTS); + // The scalar index covers rows 0-2 only; rows 3-5 are decided by reading `id`. + buildAndCommitIdBTreeIndexRange(table, new Range(0, 2)); + + Predicate idFilter = new PredicateBuilder(table.rowType()).greaterOrEqual(0, 1); + FullTextSearchBuilder builder = + table.newFullTextSearchBuilder() + .withQuery(TEXT_FIELD_NAME, matchQuery("paimon")) + .withLimit(10) + .withFilter(idFilter); + FullTextScan.Plan plan = builder.newFullTextScan().scan(); + assertThat(rawSplits(plan)).isEmpty(); + assertThat(indexSplits(plan)).allMatch(split -> !split.scalarIndexFiles().isEmpty()); + + GlobalIndexResult result = builder.newFullTextRead().read(plan); + assertThat(result.results()).containsExactlyInAnyOrder(1L, 2L, 3L, 4L, 5L); + } + + @Test + public void testFullTextSearchFastModeKeepsDataFilterInsideIndexedRanges() throws Exception { + // full-text mode fast, scalar mode full: rows outside the full-text coverage stay out + // even though their filter columns are unindexed; covered rows are filtered from data. + Identifier identifier = identifier("full_text_fast_scalar_full"); + Schema schema = + Schema.newBuilder() + .column("id", DataTypes.INT()) + .column(TEXT_FIELD_NAME, DataTypes.STRING()) + .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(), "full") + .build(); + catalog.createTable(identifier, schema, false); + FileStoreTable table = getTable(identifier); + writeDocuments(table, RANKED_DOCUMENTS); + buildAndCommitIndexRange( + table, + Arrays.copyOfRange(RANKED_DOCUMENTS, 0, 4), + Collections.singletonList(table.rowType().getField(TEXT_FIELD_NAME)), + 0); + + Predicate idFilter = new PredicateBuilder(table.rowType()).greaterOrEqual(0, 2); + FullTextSearchBuilder builder = + table.newFullTextSearchBuilder() + .withQuery(TEXT_FIELD_NAME, matchQuery("paimon")) + .withLimit(10) + .withFilter(idFilter); + FullTextScan.Plan plan = builder.newFullTextScan().scan(); + assertThat(rawSplits(plan)).isEmpty(); + + assertThat(builder.newFullTextRead().read(plan).results()) + .containsExactlyInAnyOrder(2L, 3L); + } + + @Test + public void testFullTextSearchFilterCombinedWithDeletionVectors() throws Exception { + Identifier identifier = identifier("full_text_filter_dv"); + Schema schema = + Schema.newBuilder() + .column("id", DataTypes.INT()) + .column(TEXT_FIELD_NAME, DataTypes.STRING()) + .option(CoreOptions.BUCKET.key(), "-1") + .option(CoreOptions.ROW_TRACKING_ENABLED.key(), "true") + .option(CoreOptions.DATA_EVOLUTION_ENABLED.key(), "true") + .option(CoreOptions.DELETION_VECTORS_ENABLED.key(), "true") + .build(); + catalog.createTable(identifier, schema, false); + FileStoreTable table = getTable(identifier); + writeDocuments(table, RANKED_DOCUMENTS); + buildAndCommitIndex(table, RANKED_DOCUMENTS); + buildAndCommitIdBTreeIndex(table, RANKED_DOCUMENTS.length); + commitDeletionVectors(table, 3L); + + Predicate idFilter = new PredicateBuilder(table.rowType()).greaterOrEqual(0, 3); + GlobalIndexResult result = searchWithFilter(table, idFilter, 10); + assertThat(result.results()).containsExactlyInAnyOrder(4L, 5L); + } + + @Test + public void testFullTextSearchFilterAcrossMultipleIndexRanges() throws Exception { + createTableDefault(); + FileStoreTable table = getTableDefault(); + writeDocuments(table, RANKED_DOCUMENTS); + List<DataField> textFields = + Collections.singletonList(table.rowType().getField(TEXT_FIELD_NAME)); + buildAndCommitIndexRange(table, Arrays.copyOfRange(RANKED_DOCUMENTS, 0, 3), textFields, 0); + buildAndCommitIndexRange(table, Arrays.copyOfRange(RANKED_DOCUMENTS, 3, 6), textFields, 3); + buildAndCommitIdBTreeIndexRange(table, new Range(0, 2)); + buildAndCommitIdBTreeIndexRange(table, new Range(3, 5)); + + Predicate in = new PredicateBuilder(table.rowType()).in(0, Arrays.asList(1, 4)); + FullTextSearchBuilder builder = + table.newFullTextSearchBuilder() + .withQuery(TEXT_FIELD_NAME, matchQuery("paimon")) + .withLimit(10) + .withFilter(in); + FullTextScan.Plan plan = builder.newFullTextScan().scan(); + List<IndexFullTextSearchSplit> indexSplits = indexSplits(plan); + assertThat(indexSplits).hasSize(2); + // Each split only carries the scalar index file of its own range. + for (IndexFullTextSearchSplit split : indexSplits) { + assertThat(split.scalarIndexFiles()).hasSize(1); + GlobalIndexMeta scalarMeta = split.scalarIndexFiles().get(0).globalIndexMeta(); + assertThat(scalarMeta.rowRangeStart()).isEqualTo(split.rowRangeStart()); + assertThat(scalarMeta.rowRangeEnd()).isEqualTo(split.rowRangeEnd()); + } + + assertThat(builder.newFullTextRead().read(plan).results()) + .containsExactlyInAnyOrder(1L, 4L); + } + + @Test + public void testFullTextSearchFilterExtractsPartitionPredicate() throws Exception { + Identifier identifier = identifier("full_text_filter_partitioned"); + Schema schema = + Schema.newBuilder() + .column("pt", DataTypes.INT()) + .column("id", DataTypes.INT()) + .column(TEXT_FIELD_NAME, DataTypes.STRING()) + .partitionKeys("pt") + .option(CoreOptions.BUCKET.key(), "-1") + .option(CoreOptions.ROW_TRACKING_ENABLED.key(), "true") + .option(CoreOptions.DATA_EVOLUTION_ENABLED.key(), "true") + .build(); + catalog.createTable(identifier, schema, false); + FileStoreTable table = getTable(identifier); + + RowType partitionType = RowType.of(DataTypes.INT()); + InternalRowSerializer serializer = new InternalRowSerializer(partitionType); + BinaryRow partition1 = serializer.toBinaryRow(GenericRow.of(1)).copy(); + BinaryRow partition2 = serializer.toBinaryRow(GenericRow.of(2)).copy(); + + String[] first = {"paimon one", "paimon two"}; + String[] second = {"paimon three"}; + writePartitionedDocuments(table, 1, first); + buildAndCommitIndexForColumn(table, TEXT_FIELD_NAME, first, partition1); + writePartitionedDocuments(table, 2, second); + buildAndCommitIndexForColumn(table, TEXT_FIELD_NAME, second, partition2, first.length); + + // Only a partition predicate: no scalar index is needed, and no row is dropped. + Predicate partitionOnly = new PredicateBuilder(table.rowType()).equal(0, 2); + GlobalIndexResult result = searchWithFilter(table, partitionOnly, 10); + assertThat(result.results()).containsExactly(2L); + + FullTextScan.Plan plan = + table.newFullTextSearchBuilder() + .withQuery(TEXT_FIELD_NAME, matchQuery("paimon")) + .withLimit(10) + .withFilter(partitionOnly) + .newFullTextScan() + .scan(); + assertThat(indexSplits(plan)).hasSize(1); + assertThat(indexSplits(plan).get(0).scalarIndexFiles()).isEmpty(); + } + + @Test + public void testFullTextScanAttachesScalarIndexFilesOnlyForFilteredColumns() throws Exception { + createTableDefault(); + FileStoreTable table = getTableDefault(); + writeDocuments(table, RANKED_DOCUMENTS); + buildAndCommitIndex(table, RANKED_DOCUMENTS); + buildAndCommitIdBTreeIndex(table, RANKED_DOCUMENTS.length); + buildAndCommitBTreeIndex(table, RANKED_DOCUMENTS); + + FullTextSearchBuilder noFilter = + table.newFullTextSearchBuilder() + .withQuery(TEXT_FIELD_NAME, matchQuery("paimon")) + .withLimit(10); + for (IndexFullTextSearchSplit split : indexSplits(noFilter.newFullTextScan().scan())) { + assertThat(split.scalarIndexFiles()).isEmpty(); + } + + // A filter on "id" attaches the id btree, not the btree built on the text column. + Predicate idFilter = new PredicateBuilder(table.rowType()).greaterOrEqual(0, 3); + FullTextScan.Plan plan = + table.newFullTextSearchBuilder() + .withQuery(TEXT_FIELD_NAME, matchQuery("paimon")) + .withLimit(10) + .withFilter(idFilter) + .newFullTextScan() + .scan(); + List<IndexFullTextSearchSplit> indexSplits = indexSplits(plan); + assertThat(indexSplits).hasSize(1); + assertThat(indexSplits.get(0).scalarIndexFiles()).hasSize(1); + assertThat(indexSplits.get(0).scalarIndexFiles().get(0).globalIndexMeta().indexFieldId()) + .isEqualTo(table.rowType().getField("id").id()); + assertThat(indexSplits.get(0).fullTextIndexFiles()) + .allMatch(f -> f.indexType().equals(TestFullTextGlobalIndexerFactory.IDENTIFIER)); + } + + @Test + public void testFullTextSearchFilterOnTextColumnBTreeIsNotMistakenForFullText() + throws Exception { + createTableDefault(); + FileStoreTable table = getTableDefault(); + writeDocuments(table, RANKED_DOCUMENTS); + buildAndCommitIndex(table, RANKED_DOCUMENTS); + buildAndCommitBTreeIndex(table, RANKED_DOCUMENTS); + + // Equality on the text column is evaluated by its btree, the full-text query by the + // full-text index; both are on the same column. + Predicate exact = + new PredicateBuilder(table.rowType()) + .equal(1, BinaryString.fromString("paimon lake beta")); + assertThat(searchWithFilter(table, exact, 10).results()).containsExactly(1L); + } + + @Test + public void testFullTextSearchAccumulatesFilters() throws Exception { createTableDefault(); FileStoreTable table = getTableDefault(); + writeDocuments(table, RANKED_DOCUMENTS); + buildAndCommitIndex(table, RANKED_DOCUMENTS); + buildAndCommitIdBTreeIndex(table, RANKED_DOCUMENTS.length); + + PredicateBuilder builder = new PredicateBuilder(table.rowType()); + // Two withFilter calls are AND-ed, exactly like a single conjunction. + GlobalIndexResult result = + table.newFullTextSearchBuilder() + .withQuery(TEXT_FIELD_NAME, matchQuery("paimon")) + .withLimit(10) + .withFilter(builder.greaterOrEqual(0, 2)) + .withFilter(builder.lessOrEqual(0, 4)) + .executeLocal(); + assertThat(result.results()).containsExactlyInAnyOrder(2L, 3L, 4L); + } + + @Test + public void testFullTextSearchRawPathAppliesFilterWithoutShrinkingCorpus() throws Exception { + // full-text index covers rows 0-2 only, full-text mode full: rows 3-5 are searched raw. + // The temporary index is built over all raw rows and the filter is applied as a bitmap. + Identifier identifier = identifier("full_text_raw_prefilter"); + Schema schema = + Schema.newBuilder() + .column("id", DataTypes.INT()) + .column(TEXT_FIELD_NAME, DataTypes.STRING()) + .option(CoreOptions.BUCKET.key(), "-1") + .option(CoreOptions.ROW_TRACKING_ENABLED.key(), "true") + .option(CoreOptions.DATA_EVOLUTION_ENABLED.key(), "true") + .option(CoreOptions.FULL_TEXT_INDEX_SEARCH_MODE.key(), "full") + .build(); + catalog.createTable(identifier, schema, false); + FileStoreTable table = getTable(identifier); + writeDocuments(table, RANKED_DOCUMENTS); + buildAndCommitIndexRange( + table, + Arrays.copyOfRange(RANKED_DOCUMENTS, 0, 3), + Collections.singletonList(table.rowType().getField(TEXT_FIELD_NAME)), + 0); + buildAndCommitIdBTreeIndex(table, RANKED_DOCUMENTS.length); - Predicate idFilter = new PredicateBuilder(table.rowType()).equal(0, 1); + Predicate idFilter = new PredicateBuilder(table.rowType()).greaterOrEqual(0, 4); + FullTextSearchBuilder builder = + table.newFullTextSearchBuilder() + .withQuery(TEXT_FIELD_NAME, matchQuery("paimon")) + .withLimit(10) + .withFilter(idFilter); + FullTextScan.Plan plan = builder.newFullTextScan().scan(); + List<RawFullTextSearchSplit> rawSplits = rawSplits(plan); + assertThat(rawSplits).hasSize(1); + assertThat(rawSplits.get(0).rowRanges()).containsExactly(new Range(3, 5)); + + assertThat(builder.newFullTextRead().read(plan).results()) + .containsExactlyInAnyOrder(4L, 5L); + + // A filter matching no raw row leaves the raw side empty without failing. + Predicate none = new PredicateBuilder(table.rowType()).greaterOrEqual(0, 100); + assertThat(searchWithFilter(table, none, 10).results().isEmpty()).isTrue(); + } - assertThatThrownBy( - () -> - table.newHybridSearchBuilder() - .addFullTextRoute( - TEXT_FIELD_NAME, matchQuery("Paimon"), 3, 1.0f) - .withFilter(idFilter) - .withLimit(3) - .routeBuilders()) + @Test + public void testFullTextSearchBuilderWithFilterIsOptionalForImplementations() { + FullTextSearchBuilder minimal = + new FullTextSearchBuilder() { + @Override + public FullTextSearchBuilder withLimit(int limit) { + return this; + } + + @Override + public FullTextSearchBuilder withQuery(String fieldName, String query) { + return this; + } + + @Override + public FullTextScan newFullTextScan() { + throw new UnsupportedOperationException(); + } + + @Override + public FullTextRead newFullTextRead() { + throw new UnsupportedOperationException(); + } + }; + assertThatThrownBy(() -> minimal.withFilter(PredicateBuilder.alwaysTrue())) .isInstanceOf(UnsupportedOperationException.class) - .hasMessageContaining("does not support non-partition filters"); + .hasMessageContaining("does not support row filters"); + } + + @ParameterizedTest(name = "scalar-index.search-mode={0}") + @ValueSource(strings = {"fast", "full"}) + public void testFullTextSearchRefinesCandidateOnlyIndexResultsBeforeTopK(String scalarMode) + 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); + writeDocuments(table, RANKED_DOCUMENTS); + buildAndCommitIndex(table, RANKED_DOCUMENTS); + + PredicateBuilder builder = new PredicateBuilder(table.rowType()); + Predicate containsZeta = builder.contains(1, BinaryString.fromString("zeta")); + 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. + 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); + + // Exact operators on the same btree stay exact. + assertThat( + searchWithFilter( + table, + builder.equal(1, BinaryString.fromString("paimon zeta")), + 1) + .results()) + .containsExactly(5L); + } + + @ParameterizedTest(name = "scalar-index.search-mode={0}") + @ValueSource(strings = {"fast", "full"}) + public void testFullTextSearchPartiallyIndexedConjunctionIsExactBeforeTopK(String scalarMode) + 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); + writeDocuments(table, RANKED_DOCUMENTS); + buildAndCommitIndex(table, RANKED_DOCUMENTS); + buildAndCommitIdBTreeIndex(table, RANKED_DOCUMENTS.length); + + PredicateBuilder builder = new PredicateBuilder(table.rowType()); + Predicate partiallyIndexed = + 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); + } + + private GlobalIndexResult searchWithFilter( + FileStoreTable table, @Nullable Predicate filter, int limit) { + FullTextSearchBuilder builder = + table.newFullTextSearchBuilder() + .withQuery(TEXT_FIELD_NAME, matchQuery("paimon lake")) + .withLimit(limit); + if (filter != null) { + builder.withFilter(filter); + } + return builder.executeLocal(); + } + + private FileStoreTable createTable(String name, String scalarSearchMode) throws Exception { + Identifier identifier = identifier(name); + Schema schema = + Schema.newBuilder() + .column("id", DataTypes.INT()) + .column(TEXT_FIELD_NAME, DataTypes.STRING()) + .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(), scalarSearchMode) + .build(); + catalog.createTable(identifier, schema, false); + return getTable(identifier); + } + + private static List<IndexFullTextSearchSplit> indexSplits(FullTextScan.Plan plan) { + List<IndexFullTextSearchSplit> splits = new ArrayList<>(); + for (FullTextSearchSplit split : plan.splits()) { + if (split instanceof IndexFullTextSearchSplit) { + splits.add((IndexFullTextSearchSplit) split); + } + } + return splits; + } + + private static List<RawFullTextSearchSplit> rawSplits(FullTextScan.Plan plan) { + List<RawFullTextSearchSplit> splits = new ArrayList<>(); + for (FullTextSearchSplit split : plan.splits()) { + if (split instanceof RawFullTextSearchSplit) { + splits.add((RawFullTextSearchSplit) split); + } + } + return splits; } @Test @@ -848,11 +1384,13 @@ public class FullTextSearchBuilderTest extends TableTestBase { String[] documents = {"Apache Paimon", "full-text search"}; writeDocuments(table, documents); buildAndCommitIndex(table, documents); + buildAndCommitIdBTreeIndex(table, documents.length); FullTextScan.Plan plan = table.newFullTextSearchBuilder() .withQuery(TEXT_FIELD_NAME, matchQuery("Paimon")) .withLimit(2) + .withFilter(new PredicateBuilder(table.rowType()).greaterOrEqual(0, 0)) .newFullTextScan() .scan(); @@ -879,6 +1417,9 @@ public class FullTextSearchBuilderTest extends TableTestBase { assertThat(deserialized.fullTextIndexFiles().get(i).fileName()) .isEqualTo(original.fullTextIndexFiles().get(i).fileName()); } + assertThat(original.scalarIndexFiles()).hasSize(1); + assertThat(deserialized.scalarIndexFiles()).isEqualTo(original.scalarIndexFiles()); + assertThat(deserialized).isEqualTo(original); RawFullTextSearchSplit rawOriginal = new RawFullTextSearchSplit(Collections.singletonList(new Range(2, 3))); @@ -894,6 +1435,7 @@ public class FullTextSearchBuilderTest extends TableTestBase { } assertThat(rawDeserialized.rowRanges()).isEqualTo(rawOriginal.rowRanges()); + assertThat(rawDeserialized).isEqualTo(rawOriginal); } // ====================== Helper methods ====================== @@ -1152,6 +1694,16 @@ public class FullTextSearchBuilderTest extends TableTestBase { private void buildAndCommitIndexForColumn( FileStoreTable table, String columnName, String[] documents, BinaryRow partition) throws Exception { + buildAndCommitIndexForColumn(table, columnName, documents, partition, 0); + } + + private void buildAndCommitIndexForColumn( + FileStoreTable table, + String columnName, + String[] documents, + BinaryRow partition, + long rowStart) + throws Exception { Options options = table.coreOptions().toConfiguration(); DataField textField = table.rowType().getField(columnName); @@ -1167,7 +1719,7 @@ public class FullTextSearchBuilderTest extends TableTestBase { } List<ResultEntry> entries = writer.finish(); - Range rowRange = new Range(0, documents.length - 1); + Range rowRange = new Range(rowStart, rowStart + documents.length - 1); List<IndexFileMeta> indexFiles = GlobalIndexBuilderUtils.toIndexFileMetas( table.fileIO(), @@ -1187,6 +1739,47 @@ public class FullTextSearchBuilderTest extends TableTestBase { } } + private void buildAndCommitIdBTreeIndex(FileStoreTable table, int rowCount) throws Exception { + buildAndCommitIdBTreeIndexRange(table, new Range(0, rowCount - 1)); + } + + /** Builds a btree index on {@code id} for rows in {@code rowRange}; ids equal row ids. */ + private void buildAndCommitIdBTreeIndexRange(FileStoreTable table, Range rowRange) + throws Exception { + Options options = table.coreOptions().toConfiguration(); + DataField idField = table.rowType().getField("id"); + + GlobalIndexSingleColumnWriter writer = + (GlobalIndexSingleColumnWriter) + GlobalIndexBuilderUtils.createIndexWriter( + table, BTreeGlobalIndexerFactory.IDENTIFIER, idField, options); + for (long rowId = rowRange.from; rowId <= rowRange.to; rowId++) { + writer.write((int) rowId, rowId - rowRange.from); + } + List<ResultEntry> entries = writer.finish(); + + List<IndexFileMeta> indexFiles = + GlobalIndexBuilderUtils.toIndexFileMetas( + table.fileIO(), + table.store().pathFactory().globalIndexFileFactory(), + table.coreOptions(), + rowRange, + idField.id(), + BTreeGlobalIndexerFactory.IDENTIFIER, + entries); + + 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)); + } + } + private void buildAndCommitBTreeIndex(FileStoreTable table, String[] documents) throws Exception { Options options = table.coreOptions().toConfiguration(); diff --git a/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeyFullTextSearchTest.java b/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeyFullTextSearchTest.java index e3faa6a38d..400a609b38 100644 --- a/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeyFullTextSearchTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeyFullTextSearchTest.java @@ -19,6 +19,8 @@ package org.apache.paimon.table.source; import org.apache.paimon.CoreOptions; +import org.apache.paimon.predicate.Predicate; +import org.apache.paimon.predicate.PredicateBuilder; import org.apache.paimon.schema.TableSchema; import org.apache.paimon.table.FileStoreTable; import org.apache.paimon.types.DataField; @@ -32,6 +34,7 @@ import java.util.HashMap; import java.util.Map; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; @@ -77,6 +80,35 @@ class PrimaryKeyFullTextSearchTest { assertThat(scan).isInstanceOf(DataEvolutionFullTextScan.class); } + @Test + void testPrimaryKeyFullTextRejectsRowFilter() { + FileStoreTable table = table(false); + when(table.partitionKeys()).thenReturn(Collections.emptyList()); + Predicate rowFilter = new PredicateBuilder(table.rowType()).equal(0, 1); + + FullTextSearchBuilder builder = + new FullTextSearchBuilderImpl(table) + .withQuery("content", "hello") + .withLimit(10) + .withFilter(rowFilter); + + assertThatThrownBy(builder::newFullTextScan) + .isInstanceOf(UnsupportedOperationException.class) + .hasMessageContaining("Primary-key full-text search does not support"); + assertThatThrownBy(builder::newFullTextRead) + .isInstanceOf(UnsupportedOperationException.class) + .hasMessageContaining("Primary-key full-text search does not support"); + + // The same filter on a column served by the global full-text path is accepted. + assertThat( + new FullTextSearchBuilderImpl(table) + .withQuery("other", "hello") + .withLimit(10) + .withFilter(rowFilter) + .newFullTextScan()) + .isInstanceOf(DataEvolutionFullTextScan.class); + } + @Test void testHybridRouteUsesPrimaryKeyFullText() { FileStoreTable table = table(false); diff --git a/paimon-full-text/src/test/java/org/apache/paimon/fulltext/index/NativeFullTextRowFilterTest.java b/paimon-full-text/src/test/java/org/apache/paimon/fulltext/index/NativeFullTextRowFilterTest.java new file mode 100644 index 0000000000..e2898bdaee --- /dev/null +++ b/paimon-full-text/src/test/java/org/apache/paimon/fulltext/index/NativeFullTextRowFilterTest.java @@ -0,0 +1,517 @@ +/* + * 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.fulltext.index; + +import org.apache.paimon.CoreOptions; +import org.apache.paimon.data.BinaryRow; +import org.apache.paimon.data.BinaryString; +import org.apache.paimon.data.GenericRow; +import org.apache.paimon.fs.FileIOFinder; +import org.apache.paimon.fs.Path; +import org.apache.paimon.fs.local.LocalFileIO; +import org.apache.paimon.globalindex.GlobalIndexBuilderUtils; +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.btree.BTreeGlobalIndexerFactory; +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.FileSystemSchemaManager; +import org.apache.paimon.schema.Schema; +import org.apache.paimon.schema.SchemaUtils; +import org.apache.paimon.schema.TableSchema; +import org.apache.paimon.table.AppendOnlyFileStoreTable; +import org.apache.paimon.table.CatalogEnvironment; +import org.apache.paimon.table.FileStoreTable; +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.DataField; +import org.apache.paimon.types.DataType; +import org.apache.paimon.types.DataTypes; +import org.apache.paimon.types.RowType; +import org.apache.paimon.utils.Range; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.condition.EnabledIfSystemProperty; +import org.junit.jupiter.api.io.TempDir; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Random; +import java.util.function.Supplier; + +import static org.apache.paimon.CoreOptions.DATA_EVOLUTION_ENABLED; +import static org.apache.paimon.CoreOptions.GLOBAL_INDEX_ENABLED; +import static org.apache.paimon.CoreOptions.PATH; +import static org.apache.paimon.CoreOptions.ROW_TRACKING_ENABLED; +import static org.assertj.core.api.Assertions.assertThat; + +/** + * Row filters on full-text search against the native engine: correctness of filter-then-rank, and a + * manual benchmark comparing the index pre-filter with the over-fetch workaround. + */ +public class NativeFullTextRowFilterTest { + + private static final String[] WORDS = { + "paimon", "lake", "stream", "batch", "index", "vector", "search", "table", "commit", + "snapshot", "manifest", "bucket", "partition", "schema", "column", "row", "query", "scan", + "read", "write" + }; + + @TempDir java.nio.file.Path tempDir; + + @Test + public void testFilterThenRankMatchesFilteredUnfilteredRanking() throws Exception { + int rowCount = 2_000; + int categories = 8; + Dataset dataset = writeDataset("filter_rank", rowCount, categories, 42); + FileStoreTable table = dataset.table; + PredicateBuilder builder = new PredicateBuilder(table.rowType()); + + String query = matchQuery("paimon lake"); + int limit = 20; + + // Reference: rank every row without a filter, then keep the wanted category. BM25 scores + // do not depend on the filter, so filter-then-rank must equal rank-then-filter. + ScoredGlobalIndexResult all = + (ScoredGlobalIndexResult) + table.newFullTextSearchBuilder() + .withQuery("content", query) + .withLimit(rowCount) + .executeLocal(); + for (int category = 0; category < categories; category++) { + List<Long> expected = new ArrayList<>(); + for (long rowId : all.results()) { + if (dataset.categoryOf.get(rowId) == category) { + expected.add(rowId); + } + } + expected.sort( + (a, b) -> { + int byScore = + Float.compare( + all.scoreGetter().score(b), all.scoreGetter().score(a)); + return byScore != 0 ? byScore : Long.compare(a, b); + }); + List<Long> expectedTopK = expected.subList(0, Math.min(limit, expected.size())); + float cutoff = + expectedTopK.isEmpty() + ? 0f + : all.scoreGetter().score(expectedTopK.get(expectedTopK.size() - 1)); + + ScoredGlobalIndexResult filtered = + (ScoredGlobalIndexResult) + table.newFullTextSearchBuilder() + .withQuery("content", query) + .withLimit(limit) + .withFilter(builder.equal(1, category)) + .executeLocal(); + + assertThat(filtered.results().getLongCardinality()).isEqualTo(expectedTopK.size()); + for (long rowId : filtered.results()) { + assertThat(dataset.categoryOf.get(rowId)).isEqualTo(category); + assertThat(all.results().contains(rowId)).isTrue(); + // Every returned row scores at least as high as the k-th expected row; ties at the + // cutoff may be broken differently by the engine. + assertThat(filtered.scoreGetter().score(rowId)).isGreaterThanOrEqualTo(cutoff); + assertThat(filtered.scoreGetter().score(rowId)) + .isEqualTo(all.scoreGetter().score(rowId)); + } + } + + // A filter that matches nothing returns nothing. + GlobalIndexResult none = + table.newFullTextSearchBuilder() + .withQuery("content", query) + .withLimit(limit) + .withFilter(builder.equal(1, categories + 1)) + .executeLocal(); + assertThat(none.results().isEmpty()).isTrue(); + } + + @Test + public void testUnindexedFilterInFullModeKeepsOriginalRanking() throws Exception { + // `id` has no scalar index. In scalar-index.search-mode=full the filter must be resolved + // against the existing full-text index, not by rebuilding an index over the filtered + // rows: a smaller corpus changes BM25 document frequencies and average length, which + // changes which rows make the top-k, not just their reported scores. + int rowCount = 2_000; + Dataset dataset = writeDataset("unindexed_full_mode", rowCount, 8, 42); + FileStoreTable table = + (FileStoreTable) + dataset.table.copy( + Collections.singletonMap( + CoreOptions.SCALAR_INDEX_SEARCH_MODE.key(), "full")); + PredicateBuilder builder = new PredicateBuilder(table.rowType()); + String query = matchQuery("paimon lake"); + int limit = 20; + int idFrom = 1_800; + + ScoredGlobalIndexResult all = + (ScoredGlobalIndexResult) + table.newFullTextSearchBuilder() + .withQuery("content", query) + .withLimit(rowCount) + .executeLocal(); + List<Long> expected = new ArrayList<>(); + for (long rowId : all.results()) { + if (rowId >= idFrom) { + expected.add(rowId); + } + } + expected.sort( + (a, b) -> { + int byScore = + Float.compare(all.scoreGetter().score(b), all.scoreGetter().score(a)); + return byScore != 0 ? byScore : Long.compare(a, b); + }); + List<Long> expectedTopK = expected.subList(0, Math.min(limit, expected.size())); + float cutoff = all.scoreGetter().score(expectedTopK.get(expectedTopK.size() - 1)); + + ScoredGlobalIndexResult filtered = + (ScoredGlobalIndexResult) + table.newFullTextSearchBuilder() + .withQuery("content", query) + .withLimit(limit) + .withFilter(builder.greaterOrEqual(0, idFrom)) + .executeLocal(); + + List<Long> actual = new ArrayList<>(); + filtered.results().forEach(actual::add); + // Membership first: every returned row must reach the cutoff of the original ranking. + for (long rowId : actual) { + assertThat(rowId).isGreaterThanOrEqualTo(idFrom); + assertThat(all.scoreGetter().score(rowId)) + .as( + "row %s (original score %s) is below the filtered top-%s cutoff %s", + rowId, all.scoreGetter().score(rowId), limit, cutoff) + .isGreaterThanOrEqualTo(cutoff); + } + assertThat(actual).hasSize(expectedTopK.size()); + for (long rowId : actual) { + assertThat(filtered.scoreGetter().score(rowId)) + .as("row %s must keep its score from the original index", rowId) + .isEqualTo(all.scoreGetter().score(rowId)); + } + } + + @Test + public void testRawPathAppliesFilterOverTheWholeRawCorpus() throws Exception { + // Rows 0-999 are covered by the full-text index, rows 1000-1999 are not. In full-text + // mode full the uncovered rows are searched through a temporary index; a filter must not + // shrink that temporary corpus, so scores of the returned raw rows equal the scores of an + // unfiltered raw search. + int rowCount = 2_000; + int indexedRows = 1_000; + Dataset dataset = writeDataset("raw_corpus", rowCount, 8, 42, indexedRows); + Map<String, String> options = new HashMap<>(); + options.put(CoreOptions.FULL_TEXT_INDEX_SEARCH_MODE.key(), "full"); + options.put(CoreOptions.SCALAR_INDEX_SEARCH_MODE.key(), "full"); + FileStoreTable table = (FileStoreTable) dataset.table.copy(options); + PredicateBuilder builder = new PredicateBuilder(table.rowType()); + String query = matchQuery("paimon lake"); + int limit = 20; + int idFrom = 1_800; + + ScoredGlobalIndexResult unfiltered = + (ScoredGlobalIndexResult) + table.newFullTextSearchBuilder() + .withQuery("content", query) + .withLimit(rowCount) + .executeLocal(); + ScoredGlobalIndexResult filtered = + (ScoredGlobalIndexResult) + table.newFullTextSearchBuilder() + .withQuery("content", query) + .withLimit(limit) + .withFilter(builder.greaterOrEqual(0, idFrom)) + .executeLocal(); + + assertThat(filtered.results().getLongCardinality()).isEqualTo(limit); + for (long rowId : filtered.results()) { + assertThat(rowId).isGreaterThanOrEqualTo(idFrom); + assertThat(unfiltered.results().contains(rowId)).isTrue(); + assertThat(filtered.scoreGetter().score(rowId)) + .as("row %s must keep the score of the unfiltered raw corpus", rowId) + .isEqualTo(unfiltered.scoreGetter().score(rowId)); + } + } + + /** + * Run with {@code mvn -pl paimon-full-text test -Dtest=NativeFullTextRowFilterTest + * -DextraJavaTestArgs=-Dpaimon.benchmark=true}. Prints one line per strategy. + */ + @Test + @EnabledIfSystemProperty(named = "paimon.benchmark", matches = "true") + public void benchmarkRowFilterStrategies() throws Exception { + int rowCount = 200_000; + int categories = 100; // 1% selectivity per category + Dataset dataset = writeDataset("benchmark", rowCount, categories, 7); + FileStoreTable table = dataset.table; + FileStoreTable fullModeTable = + (FileStoreTable) + table.copy( + Collections.singletonMap( + CoreOptions.SCALAR_INDEX_SEARCH_MODE.key(), "full")); + PredicateBuilder builder = new PredicateBuilder(table.rowType()); + String query = matchQuery("paimon lake"); + int limit = 10; + int category = 17; + + System.out.printf( + "%nfull-text row filter benchmark: rows=%d, categories=%d, limit=%d%n", + rowCount, categories, limit); + long baseline = + time( + "no filter (baseline)", + () -> + table.newFullTextSearchBuilder() + .withQuery("content", query) + .withLimit(limit) + .executeLocal()); + long preFilter = + time( + "withFilter via btree index (1% selective)", + () -> + table.newFullTextSearchBuilder() + .withQuery("content", query) + .withLimit(limit) + .withFilter(builder.equal(1, category)) + .executeLocal()); + time( + "withFilter via btree index (dense, ~99% selective)", + () -> + table.newFullTextSearchBuilder() + .withQuery("content", query) + .withLimit(limit) + .withFilter(builder.notEqual(1, category)) + .executeLocal()); + time( + "over-fetch limit*100 then client-side filter (old workaround)", + () -> { + ScoredGlobalIndexResult result = + (ScoredGlobalIndexResult) + table.newFullTextSearchBuilder() + .withQuery("content", query) + .withLimit(limit * 100) + .executeLocal(); + List<Long> kept = new ArrayList<>(); + for (long rowId : result.results()) { + if (dataset.categoryOf.get(rowId) == category) { + kept.add(rowId); + } + } + return kept; + }); + Predicate unindexedFilter = builder.equal(2, 3); + time( + "withFilter on unindexed column, scalar mode full (raw scan)", + () -> + fullModeTable + .newFullTextSearchBuilder() + .withQuery("content", query) + .withLimit(limit) + .withFilter(unindexedFilter) + .executeLocal()); + System.out.printf( + "pre-filter overhead over baseline: %.1fx%n", (double) preFilter / baseline); + } + + private static long time(String name, Supplier<Object> action) { + for (int i = 0; i < 3; i++) { + action.get(); + } + int iterations = 10; + long best = Long.MAX_VALUE; + long total = 0; + for (int i = 0; i < iterations; i++) { + long start = System.nanoTime(); + action.get(); + long elapsed = System.nanoTime() - start; + best = Math.min(best, elapsed); + total += elapsed; + } + System.out.printf( + " %-64s best %7.1f ms avg %7.1f ms%n", + name, best / 1_000_000.0, total / iterations / 1_000_000.0); + return best; + } + + private static final class Dataset { + final FileStoreTable table; + final Map<Long, Integer> categoryOf; + + Dataset(FileStoreTable table, Map<Long, Integer> categoryOf) { + this.table = table; + this.categoryOf = categoryOf; + } + } + + /** + * Table (id INT, category INT, other INT, content STRING) with a native full-text index on + * {@code content} and a btree index on {@code category}; {@code other} stays unindexed. + */ + private Dataset writeDataset(String tableName, int rowCount, int categories, long seed) + throws Exception { + return writeDataset(tableName, rowCount, categories, seed, rowCount); + } + + private Dataset writeDataset( + String tableName, int rowCount, int categories, long seed, int fullTextIndexedRows) + throws Exception { + Path tablePath = new Path(tempDir.resolve(tableName).toUri()); + LocalFileIO fileIO = LocalFileIO.create(); + + RowType rowType = + RowType.of( + new DataType[] { + DataTypes.INT(), DataTypes.INT(), DataTypes.INT(), DataTypes.STRING() + }, + new String[] {"id", "category", "other", "content"}); + Options options = new Options(); + options.set(PATH, tablePath.toString()); + options.set(ROW_TRACKING_ENABLED, true); + options.set(DATA_EVOLUTION_ENABLED, true); + options.set(GLOBAL_INDEX_ENABLED, true); + TableSchema tableSchema = + SchemaUtils.forceCommit( + new FileSystemSchemaManager(fileIO, tablePath), + new Schema( + rowType.getFields(), + Collections.emptyList(), + Collections.emptyList(), + options.toMap(), + "")); + FileStoreTable table = + new AppendOnlyFileStoreTable( + FileIOFinder.find(tablePath), + tablePath, + tableSchema, + CatalogEnvironment.empty()); + + Random random = new Random(seed); + Map<Long, Integer> categoryOf = new HashMap<>(); + List<String> contents = new ArrayList<>(rowCount); + List<Integer> categoryValues = new ArrayList<>(rowCount); + BatchWriteBuilder writeBuilder = table.newBatchWriteBuilder(); + try (BatchTableWrite write = writeBuilder.newWrite(); + BatchTableCommit commit = writeBuilder.newCommit()) { + for (int i = 0; i < rowCount; i++) { + int category = random.nextInt(categories); + StringBuilder content = new StringBuilder(); + int words = 3 + random.nextInt(8); + for (int w = 0; w < words; w++) { + content.append(WORDS[random.nextInt(WORDS.length)]).append(' '); + } + contents.add(content.toString().trim()); + categoryValues.add(category); + categoryOf.put((long) i, category); + write.write( + GenericRow.of( + i, + category, + random.nextInt(10), + BinaryString.fromString(contents.get(i)))); + } + commit.commit(write.prepareCommit()); + } + + Range rowRange = new Range(0, rowCount - 1); + DataField contentField = table.rowType().getField("content"); + GlobalIndexSingleColumnWriter textWriter = + (GlobalIndexSingleColumnWriter) + GlobalIndexBuilderUtils.createIndexWriter( + table, + NativeFullTextGlobalIndexerFactory.IDENTIFIER, + contentField, + table.coreOptions().toConfiguration()); + for (int i = 0; i < fullTextIndexedRows; i++) { + textWriter.write(BinaryString.fromString(contents.get(i)), i); + } + List<IndexFileMeta> indexFiles = + new ArrayList<>( + GlobalIndexBuilderUtils.toIndexFileMetas( + table.fileIO(), + table.store().pathFactory().globalIndexFileFactory(), + table.coreOptions(), + new Range(0, fullTextIndexedRows - 1), + contentField.id(), + NativeFullTextGlobalIndexerFactory.IDENTIFIER, + textWriter.finish())); + + DataField categoryField = table.rowType().getField("category"); + GlobalIndexSingleColumnWriter categoryWriter = + (GlobalIndexSingleColumnWriter) + GlobalIndexBuilderUtils.createIndexWriter( + table, + BTreeGlobalIndexerFactory.IDENTIFIER, + categoryField, + table.coreOptions().toConfiguration()); + // The btree writer is an SST writer: keys must arrive in sorted order. + List<Integer> rowIdsByCategory = new ArrayList<>(rowCount); + for (int i = 0; i < rowCount; i++) { + rowIdsByCategory.add(i); + } + rowIdsByCategory.sort( + (a, b) -> { + int byCategory = Integer.compare(categoryValues.get(a), categoryValues.get(b)); + return byCategory != 0 ? byCategory : Integer.compare(a, b); + }); + for (int rowId : rowIdsByCategory) { + categoryWriter.write(categoryValues.get(rowId), rowId); + } + List<ResultEntry> categoryEntries = categoryWriter.finish(); + indexFiles.addAll( + GlobalIndexBuilderUtils.toIndexFileMetas( + table.fileIO(), + table.store().pathFactory().globalIndexFileFactory(), + table.coreOptions(), + rowRange, + categoryField.id(), + BTreeGlobalIndexerFactory.IDENTIFIER, + categoryEntries)); + + 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)); + } + return new Dataset(table, categoryOf); + } + + private static String matchQuery(String terms) { + return "{\"match\":{\"query\":\"" + terms + "\"}}"; + } +} diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonBaseScan.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonBaseScan.scala index dc71a9cfbf..ed3452d2c3 100644 --- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonBaseScan.scala +++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonBaseScan.scala @@ -131,9 +131,7 @@ abstract class PaimonBaseScan(table: InnerTable) ftBuilder.withPartitionFilter(PartitionPredicate.and(pushedPartitionFilters.asJava)) } if (pushedDataFilters.nonEmpty) { - throw new UnsupportedOperationException( - "Full-text search does not support non-partition filters because full-text indexes " + - "cannot apply row-id pre-filters before top-k ranking.") + ftBuilder.withFilter(PredicateBuilder.and(pushedDataFilters.asJava)) } ftBuilder.newFullTextRead().read(ftBuilder.newFullTextScan().scan()) } 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 4c131e1f03..dfdd820877 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 @@ -325,4 +325,232 @@ class FullTextSearchTest extends PaimonSparkTestBase { assert(searchResult.length == 10) } } + + // ========== Row filter tests ========== + + private def createRankedTable(extraProps: String = ""): Unit = { + spark.sql(s""" + |CREATE TABLE T (id INT, category STRING, content STRING) + |TBLPROPERTIES ( + | 'bucket' = '-1', + | 'global-index.row-count-per-shard' = '10000', + | 'row-tracking.enabled' = 'true', + | 'data-evolution.enabled' = 'true'$extraProps) + |""".stripMargin) + // Rows 0-2 match both terms of "paimon lake" (score 1.0); rows 3-5 only "paimon" (0.5). + spark.sql(""" + |INSERT INTO T VALUES + | (0, 'a', 'paimon lake alpha'), + | (1, 'a', 'paimon lake beta'), + | (2, 'b', 'paimon lake gamma'), + | (3, 'b', 'paimon delta'), + | (4, 'c', 'paimon epsilon'), + | (5, 'c', 'paimon zeta') + |""".stripMargin) + spark + .sql( + s"CALL sys.create_global_index(table => 'test.T', index_column => 'content', index_type => '$indexType')") + .collect() + } + + private val rankedQuery = """{"match":{"column":"content","terms":"paimon lake"}}""" + + test("full-text search - WHERE filter is applied before top-k") { + withTable("T") { + createRankedTable() + spark + .sql("CALL sys.create_global_index(table => 'test.T', index_column => 'id', index_type => 'btree')") + .collect() + + val unfiltered = spark + .sql(s"SELECT id FROM full_text_search('T', 'content', '$rankedQuery', 2)") + .collect() + .map(_.getInt(0)) + .toSet + assert(unfiltered.subsetOf(Set(0, 1, 2))) + + val filtered = spark + .sql(s""" + |SELECT id, __paimon_search_score + |FROM full_text_search('T', 'content', '$rankedQuery', 2) + |WHERE id >= 3 + |""".stripMargin) + .collect() + assert(filtered.length == 2) + assert(filtered.map(_.getInt(0)).toSet.subsetOf(Set(3, 4, 5))) + assert(filtered.forall(_.getFloat(1) == 0.5f)) + + // Equivalent to ranking the filtered subset without the index. + val expected = spark + .sql("SELECT id FROM T WHERE id >= 3 AND content LIKE '%paimon%' ORDER BY id LIMIT 3") + .collect() + .map(_.getInt(0)) + .toSet + assert(filtered.map(_.getInt(0)).toSet.subsetOf(expected)) + } + } + + test("full-text search - WHERE filter on string column with bitmap index") { + withTable("T") { + createRankedTable() + spark + .sql("CALL sys.create_global_index(table => 'test.T', index_column => 'category', index_type => 'bitmap')") + .collect() + + val result = spark + .sql(s""" + |SELECT id + |FROM full_text_search('T', 'content', '$rankedQuery', 10) + |WHERE category IN ('b', 'c') + |ORDER BY id + |""".stripMargin) + .collect() + .map(_.getInt(0)) + .toSeq + assert(result == Seq(2, 3, 4, 5)) + + val none = spark + .sql(s""" + |SELECT id + |FROM full_text_search('T', 'content', '$rankedQuery', 10) + |WHERE category = 'z' + |""".stripMargin) + .collect() + assert(none.isEmpty) + } + } + + test("full-text search - fast mode excludes rows whose filter column has no index") { + withTable("T") { + createRankedTable() + + val result = spark + .sql(s""" + |SELECT id + |FROM full_text_search('T', 'content', '$rankedQuery', 10) + |WHERE id >= 3 + |""".stripMargin) + .collect() + assert(result.isEmpty) + } + } + + test("full-text search - full mode scans rows whose filter column has no index") { + withTable("T") { + createRankedTable(",\n 'scalar-index.search-mode' = 'full'") + + val result = spark + .sql(s""" + |SELECT id + |FROM full_text_search('T', 'content', '$rankedQuery', 2) + |WHERE id >= 3 + |""".stripMargin) + .collect() + .map(_.getInt(0)) + .toSet + assert(result.size == 2) + assert(result.subsetOf(Set(3, 4, 5))) + } + } + + test("full-text search - WHERE filter combined with partition filter") { + withTable("T") { + spark.sql(""" + |CREATE TABLE T (id INT, content STRING, pt INT) + |PARTITIONED BY (pt) + |TBLPROPERTIES ( + | 'bucket' = '-1', + | 'global-index.row-count-per-shard' = '10000', + | 'row-tracking.enabled' = 'true', + | 'data-evolution.enabled' = 'true') + |""".stripMargin) + spark.sql(""" + |INSERT INTO T VALUES + | (0, 'paimon lake', 1), + | (1, 'paimon lake', 1), + | (2, 'paimon lake', 2), + | (3, 'paimon lake', 2) + |""".stripMargin) + spark + .sql( + s"CALL sys.create_global_index(table => 'test.T', index_column => 'content', index_type => '$indexType')") + .collect() + spark + .sql("CALL sys.create_global_index(table => 'test.T', index_column => 'id', index_type => 'btree')") + .collect() + + val result = spark + .sql(s""" + |SELECT id + |FROM full_text_search('T', 'content', '$rankedQuery', 10) + |WHERE pt = 2 AND id >= 3 + |""".stripMargin) + .collect() + .map(_.getInt(0)) + .toSeq + assert(result == Seq(3)) + } + } + + test("full-text search - WHERE filter with deletion vectors") { + withTable("T") { + createRankedTable(",\n 'deletion-vectors.enabled' = 'true'") + spark + .sql("CALL sys.create_global_index(table => 'test.T', index_column => 'id', index_type => 'btree')") + .collect() + spark.sql("DELETE FROM T WHERE id = 3") + + val result = spark + .sql(s""" + |SELECT id + |FROM full_text_search('T', 'content', '$rankedQuery', 10) + |WHERE id >= 3 + |ORDER BY id + |""".stripMargin) + .collect() + .map(_.getInt(0)) + .toSeq + assert(result == Seq(4, 5)) + } + } + + test("full-text search - predicate that cannot be pushed down is applied after the search") { + withTable("T") { + createRankedTable() + spark + .sql("CALL sys.create_global_index(table => 'test.T', index_column => 'id', index_type => 'btree')") + .collect() + + // `id % 2 = 1` is not a pushable predicate, so Spark evaluates it after the top-k: the + // result is a subset of the unfiltered top-k, never rows outside it, and may be short. + val unfiltered = spark + .sql(s"SELECT id FROM full_text_search('T', 'content', '$rankedQuery', 3)") + .collect() + .map(_.getInt(0)) + .toSet + val filtered = spark + .sql(s""" + |SELECT id + |FROM full_text_search('T', 'content', '$rankedQuery', 3) + |WHERE id % 2 = 1 + |""".stripMargin) + .collect() + .map(_.getInt(0)) + .toSet + assert(filtered.subsetOf(unfiltered)) + assert(filtered.forall(_ % 2 == 1)) + + // Mixed: the pushable half narrows the candidates before top-k, the rest post-filters. + val mixed = spark + .sql(s""" + |SELECT id + |FROM full_text_search('T', 'content', '$rankedQuery', 3) + |WHERE id >= 3 AND id % 2 = 1 + |""".stripMargin) + .collect() + .map(_.getInt(0)) + .toSet + assert(mixed.subsetOf(Set(3, 5))) + } + } } diff --git a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/HybridSearchTest.scala b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/HybridSearchTest.scala index 254bf50c77..a9ec19a326 100644 --- a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/HybridSearchTest.scala +++ b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/HybridSearchTest.scala @@ -315,7 +315,7 @@ class HybridSearchTest extends PaimonSparkTestBase { } } - test("hybrid full-text route rejects non-partition filters") { + test("hybrid full-text route applies non-partition filters before ranking") { withTable("T") { spark.sql(""" |CREATE TABLE T (id INT, content STRING) @@ -326,6 +326,8 @@ class HybridSearchTest extends PaimonSparkTestBase { | 'data-evolution.enabled' = 'true') |""".stripMargin) + // Row 0 scores higher for "paimon search"; the filter must remove it before the route's + // top-1 so that row 1 is returned instead of nothing. spark.sql(""" |INSERT INTO T VALUES | (0, 'paimon search'), @@ -336,39 +338,30 @@ class HybridSearchTest extends PaimonSparkTestBase { .sql("CALL sys.create_global_index(table => 'test.T', index_column => 'content', " + "index_type => 'test-fulltext')") .collect() + spark + .sql("CALL sys.create_global_index(table => 'test.T', index_column => 'id', " + + "index_type => 'btree')") + .collect() - val error = intercept[Exception] { - spark - .sql(""" - |SELECT id - |FROM hybrid_search( - | 'T', - | array(), - | array( - | named_struct( - | 'column', 'content', - | 'query', '{"match":{"column":"content","terms":"paimon"}}', - | 'limit', 1, - | 'weight', 1.0f, - | 'options', map())), - | 1) - |WHERE id = 1 - |""".stripMargin) - .collect() - } - - assert(containsMessage(error, "does not support non-partition filters")) - } - } + val result = spark + .sql(""" + |SELECT id + |FROM hybrid_search( + | 'T', + | array(), + | array( + | named_struct( + | 'column', 'content', + | 'query', '{"match":{"column":"content","terms":"paimon search"}}', + | 'limit', 1, + | 'weight', 1.0f, + | 'options', map())), + | 1) + |WHERE id = 1 + |""".stripMargin) + .collect() - private def containsMessage(error: Throwable, expected: String): Boolean = { - var current = error - while (current != null) { - if (current.getMessage != null && current.getMessage.contains(expected)) { - return true - } - current = current.getCause + assert(result.map(_.getInt(0)).toSeq == Seq(1)) } - false } }
