diff --git a/docs/docs/multimodal-table/global-index/full-text.mdx b/docs/docs/multimodal-table/global-index/full-text.mdx index 8edef0e54d42..ec385397de77 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 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(); +``` + @@ -259,6 +283,38 @@ 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 evaluated through the scalar global indexes +(BTree, Bitmap, Multivalue, FM) built on the filtered columns, the same way +[vector search pre-filters](./vector#vector-search) rows. Build a scalar index +on the columns you filter by. How rows whose filter columns are not covered by +a scalar index are handled follows `scalar-index.search-mode`: + +| `scalar-index.search-mode` | Rows with an index on the filter columns | Rows without one | +|---|---|---| +| `fast` (default) | Evaluated through the index | Excluded from the search; a warning is logged | +| `full` / `detail` | Evaluated through the index | Read from the data files and filtered row by row before a temporary full-text index ranks them | + +With `full-text-index.search-mode=fast`, only rows covered by the full-text +index are considered in either case. In `fast` mode a conjunction whose members +are only partly indexed (`indexed_col = 1 AND unindexed_col = 2`) is narrowed by +the indexed members alone, so the candidate set is a superset. Predicates the +engine cannot push down (for example, UDFs) are likewise applied by the engine +after the search. In both situations the rows returned are correct, but a query +may return fewer than `limit` rows; use `full` mode or index every filtered +column when the exact top-k matters. + +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 09ef65e11b9a..42fa37deb614 100644 --- a/docs/docs/multimodal-table/global-index/hybrid-search.mdx +++ b/docs/docs/multimodal-table/global-index/hybrid-search.mdx @@ -397,3 +397,12 @@ 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). diff --git a/docs/docs/primary-key-table/global-index.mdx b/docs/docs/primary-key-table/global-index.mdx index 82ca5f097b00..a5c2a8d986d3 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 40e492968ce2..1f69780bc2be 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,10 @@ package org.apache.paimon.table.source; +import org.apache.paimon.CoreOptions.GlobalIndexSearchMode; import org.apache.paimon.Snapshot; +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; @@ -33,19 +36,27 @@ import org.apache.paimon.index.IndexPathFactory; import org.apache.paimon.partition.PartitionPredicate; import org.apache.paimon.predicate.FullTextSearch; +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; @@ -55,8 +66,11 @@ /** 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 +81,19 @@ public DataEvolutionFullTextRead( int limit, List textColumns, String query) { + this(table, partitionFilter, null, limit, textColumns, query); + } + + public DataEvolutionFullTextRead( + FileStoreTable table, + @Nullable PartitionPredicate partitionFilter, + @Nullable Predicate filter, + int limit, + List textColumns, + String query) { this.table = table; this.partitionFilter = partitionFilter; + this.filter = filter; this.limit = limit; if (textColumns.size() != 1) { throw new IllegalArgumentException( @@ -100,43 +125,153 @@ private GlobalIndexResult read( ExecutorService executor = GlobalIndexReadThreadPool.getExecutorService(parallelism); Map> splitsByColumn = new HashMap<>(); + List indexSplits = new ArrayList<>(); List rawRowRanges = new ArrayList<>(); + List rawSplits = 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); } else if (split instanceof RawFullTextSearchSplit) { - rawRowRanges.addAll(((RawFullTextSearchSplit) split).rowRanges()); + RawFullTextSearchSplit rawSplit = (RawFullTextSearchSplit) split; + rawSplits.add(rawSplit); + rawRowRanges.addAll(rawSplit.rowRanges()); } } GlobalIndexFileReader indexFileReader = m -> table.fileIO().newInputStream(m.filePath()); RoaringNavigableMap64 liveRows = GlobalIndexLiveRowFilter.liveRows(table, planSnapshot, partitionFilter, null); + RoaringNavigableMap64 matchedRows = scalarMatchedRows(indexSplits, planSnapshot); 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) - .withRawSearch(result, rawRowRanges, splitsByColumn, executor); + .withRawSearch( + result, + rawRowRanges, + rawPreFilter(rawSplits, planSnapshot), + splitsByColumn, + executor); } return result.topK(limit); } + /** + * Rows of the indexed splits that satisfy {@link #filter} according to the scalar global + * indexes attached to them, or {@code null} when there is no filter. Rows whose filter columns + * are not indexed are absent from the result: in {@code fast} scalar search mode they are + * excluded, in other modes the scan already routed them to a raw split where the predicate is + * evaluated on the data. + */ + @Nullable + private RoaringNavigableMap64 scalarMatchedRows( + List indexSplits, @Nullable Snapshot planSnapshot) { + if (filter == null || indexSplits.isEmpty()) { + return null; + } + + Set scalarIndexFiles = + new TreeSet<>(Comparator.comparing(IndexFileMeta::fileName)); + for (IndexFullTextSearchSplit split : indexSplits) { + scalarIndexFiles.addAll(split.scalarIndexFiles()); + } + + Optional optionalScanner = + DataEvolutionGlobalIndexScanner.create( + table, planSnapshot, partitionFilter, scalarIndexFiles); + if (!optionalScanner.isPresent()) { + warnUnindexedFilter(); + return new RoaringNavigableMap64(); + } + + try (DataEvolutionGlobalIndexScanner scanner = optionalScanner.get()) { + Optional result = scanner.scan(filter); + if (!result.isPresent()) { + warnUnindexedFilter(); + return new RoaringNavigableMap64(); + } + return result.get().results(); + } catch (IOException e) { + throw new RuntimeException(e); + } + } + + /** + * Rows of the raw splits that may satisfy {@link #filter}: rows matched by the scalar indexes + * plus rows those indexes do not cover, or {@code null} when nothing can be pre-filtered. The + * raw read still evaluates the predicate row by row, so this only bounds the rows to scan. + */ + @Nullable + private RoaringNavigableMap64 rawPreFilter( + List rawSplits, @Nullable Snapshot planSnapshot) { + if (filter == null || rawSplits.isEmpty()) { + return null; + } + + Set scalarIndexFiles = + new TreeSet<>(Comparator.comparing(IndexFileMeta::fileName)); + for (RawFullTextSearchSplit split : rawSplits) { + scalarIndexFiles.addAll(split.scalarIndexFiles()); + } + Optional optionalScanner = + DataEvolutionGlobalIndexScanner.create( + table, planSnapshot, partitionFilter, scalarIndexFiles); + if (!optionalScanner.isPresent()) { + return null; + } + + RoaringNavigableMap64 include = new RoaringNavigableMap64(); + try (DataEvolutionGlobalIndexScanner scanner = optionalScanner.get()) { + Optional result = scanner.scanWithCoverage(filter); + if (!result.isPresent()) { + return null; + } + include.or(result.get().result().results()); + include.or( + scanner.unindexedRowsForContributingFields(result.get().contributingFieldIds()) + .results()); + } catch (IOException e) { + throw new RuntimeException(e); + } + return include; + } + + 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> splitsByColumn, IndexPathFactory indexPathFactory, GlobalIndexFileReader indexFileReader, ExecutorService executor) { - return evalQuery(splitsByColumn, indexPathFactory, indexFileReader, executor, null); + return evalQuery(splitsByColumn, indexPathFactory, indexFileReader, executor, null, null); } private ScoredGlobalIndexResult evalQuery( @@ -144,14 +279,16 @@ private ScoredGlobalIndexResult evalQuery( 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 +297,8 @@ private ScoredGlobalIndexResult evalColumnQuery( IndexPathFactory indexPathFactory, GlobalIndexFileReader indexFileReader, ExecutorService executor, - @Nullable RoaringNavigableMap64 liveRows) { + @Nullable RoaringNavigableMap64 liveRows, + @Nullable RoaringNavigableMap64 matchedRows) { List columnSplits = splitsByColumn.get(column); if (columnSplits == null || columnSplits.isEmpty()) { return ScoredGlobalIndexResult.createEmpty(); @@ -183,7 +321,7 @@ private ScoredGlobalIndexResult evalColumnQuery( split.fullTextIndexFiles(), indexFileReader, executor, - includeRowIds(split, liveRows))); + includeRowIds(split, liveRows, matchedRows))); } CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join(); @@ -201,7 +339,9 @@ private ScoredGlobalIndexResult evalColumnQuery( @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 +349,9 @@ private static RoaringNavigableMap64 includeRowIds( 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 fbd1bd83d133..8ab86ab2d8c5 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 @@ -18,6 +18,7 @@ package org.apache.paimon.table.source; +import org.apache.paimon.CoreOptions.GlobalIndexSearchMode; import org.apache.paimon.Snapshot; import org.apache.paimon.annotation.VisibleForTesting; import org.apache.paimon.globalindex.DataEvolutionGlobalIndexCoverage; @@ -28,6 +29,7 @@ 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 +51,7 @@ 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 +59,7 @@ public class DataEvolutionFullTextScan implements FullTextScan { private final FileStoreTable table; @Nullable private final PartitionPredicate partitionFilter; + @Nullable private final Predicate filter; private final List textColumns; @Nullable private final Snapshot pinnedSnapshot; @@ -78,8 +82,18 @@ public DataEvolutionFullTextScan( @Nullable PartitionPredicate partitionFilter, List textColumns, @Nullable Snapshot pinnedSnapshot) { + this(table, partitionFilter, null, textColumns, pinnedSnapshot); + } + + public DataEvolutionFullTextScan( + FileStoreTable table, + @Nullable PartitionPredicate partitionFilter, + @Nullable Predicate filter, + List textColumns, + @Nullable Snapshot pinnedSnapshot) { this.table = table; this.partitionFilter = partitionFilter; + this.filter = filter; this.textColumns = textColumns; this.pinnedSnapshot = pinnedSnapshot; } @@ -98,6 +112,7 @@ public Plan scan() { idToColumn.put(textColumn.id(), textColumn.name()); } + Set filterFieldIds = collectFieldIds(table.rowType(), filter); @Nullable Snapshot snapshot = pinnedSnapshot != null ? pinnedSnapshot : TimeTravelUtil.tryTravelOrLatest(table); @@ -111,38 +126,85 @@ public Plan scan() { if (globalIndex == null) { return false; } - return !matchedTextColumnIds(globalIndex, textColumnIds).isEmpty() - && supportsFullTextSearch(entry.indexFile().indexType()); + return isFullTextIndexFile(entry.indexFile(), textColumnIds) + || containsAnyField(globalIndex, filterFieldIds); }; List allIndexFiles = indexFileHandler.scan(snapshot, indexFileFilter).stream() .map(IndexManifestEntry::indexFile) .collect(Collectors.toList()); + List fullTextIndexFiles = new ArrayList<>(); + List 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. List splits = new ArrayList<>(); + List fullTextIndexedRanges = new ArrayList<>(); for (IndexRangeSelection selection : - chooseIndexRanges(allIndexFiles, textColumnIds, idToColumn)) { + chooseIndexRanges(fullTextIndexFiles, textColumnIds, idToColumn)) { + fullTextIndexedRanges.addAll(selection.searchRanges); 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()) { + GlobalIndexSearchMode fullTextSearchMode = + table.coreOptions().fullTextIndexSearchMode(); List rawRowRanges = new DataEvolutionGlobalIndexCoverage( table, snapshot, partitionFilter, - allIndexFiles, - table.coreOptions().fullTextIndexSearchMode()) + fullTextIndexFiles, + fullTextSearchMode) .unindexedRanges(textColumnIds); + if (filter != null) { + // Rows whose filter columns are not covered by a scalar index cannot be + // pre-filtered through the index; scan them raw so the predicate is evaluated + // on the data, following scalar-index.search-mode. + List scalarUnindexedRanges = + new DataEvolutionGlobalIndexCoverage( + table, + snapshot, + partitionFilter, + scalarIndexFiles, + table.coreOptions().scalarIndexSearchMode()) + .unindexedRanges(table.rowType(), filter); + if (fullTextSearchMode == GlobalIndexSearchMode.FAST) { + scalarUnindexedRanges = + Range.and( + scalarUnindexedRanges, + Range.sortAndMergeOverlap(fullTextIndexedRanges, true)); + } + rawRowRanges = + Range.sortAndMergeOverlap( + addAll(rawRowRanges, scalarUnindexedRanges), true); + } if (!rawRowRanges.isEmpty()) { - splits.add(new RawFullTextSearchSplit(rawRowRanges)); + splits.add( + new RawFullTextSearchSplit( + rawRowRanges, scalarIndexFiles(scalarIndexFiles, rawRowRanges))); } } @@ -408,6 +470,55 @@ private IndexRangeSelection( } } + private static boolean isFullTextIndexFile( + IndexFileMeta indexFile, Set textColumnIds) { + GlobalIndexMeta globalIndex = checkNotNull(indexFile.globalIndexMeta()); + return !matchedTextColumnIds(globalIndex, textColumnIds).isEmpty() + && supportsFullTextSearch(indexFile.indexType()); + } + + private static boolean containsAnyField(GlobalIndexMeta meta, Set 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 scalarIndexFiles( + List scalarIndexFiles, List rowRanges) { + List 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 List addAll(List left, List right) { + List result = new ArrayList<>(left.size() + right.size()); + result.addAll(left); + result.addAll(right); + return result; + } + private static boolean supportsFullTextSearch(String indexType) { GlobalIndexerFactory factory; try { 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 5b091670da18..9e07fcdbd972 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 @@ 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 @@ default FullTextSearchBuilder withPartitionFilter(PartitionPredicate partitionPr "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 77d74c9a5f6c..baa6dab5f980 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.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 FullTextSearchBuilderImpl(InnerTable table) { @Override public FullTextSearchBuilder withPartitionFilter(PartitionPredicate partitionFilter) { - this.partitionFilter = partitionFilter; + addPartitionFilter(partitionFilter); return this; } + @Override + public FullTextSearchBuilder withFilter(Predicate predicate) { + Pair, List> 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 FullTextScan newFullTextScan() { : new DataEvolutionFullTextScan( table, partitionFilter, + filter, Collections.singletonList(textColumn), pinnedSnapshot); } @@ -97,6 +132,7 @@ public FullTextRead newFullTextRead() { : new DataEvolutionFullTextRead( table, partitionFilter, + filter, limit, Collections.singletonList(textColumn), query); @@ -121,6 +157,10 @@ private Optional primaryKeyFullTextDefinition(DataFie 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 35460de7b472..4183af92bbff 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 @@ private void validateSearch() { 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 @@ protected FullTextSearchBuilder newFullTextSearchBuilder(HybridSearchRoute route 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 54a55536f179..a7e93bfefa50 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 INDEX_SERIALIZER = ThreadLocal.withInitial(IndexFileMetaSerializer::new); @@ -47,6 +48,7 @@ public class IndexFullTextSearchSplit extends FullTextSearchSplit { private long rowRangeEnd; private List searchRowRanges; private transient List fullTextIndexFiles; + private transient List scalarIndexFiles; public IndexFullTextSearchSplit( long rowRangeStart, long rowRangeEnd, List fullTextIndexFiles) { @@ -72,6 +74,22 @@ public IndexFullTextSearchSplit( long rowRangeEnd, List fullTextIndexFiles, List searchRowRanges) { + this( + columnName, + rowRangeStart, + rowRangeEnd, + fullTextIndexFiles, + searchRowRanges, + Collections.emptyList()); + } + + public IndexFullTextSearchSplit( + String columnName, + long rowRangeStart, + long rowRangeEnd, + List fullTextIndexFiles, + List searchRowRanges, + List scalarIndexFiles) { this.columnName = columnName; this.rowRangeStart = rowRangeStart; this.rowRangeEnd = rowRangeEnd; @@ -86,6 +104,7 @@ public IndexFullTextSearchSplit( } } this.searchRowRanges = Collections.unmodifiableList(ranges); + this.scalarIndexFiles = Collections.unmodifiableList(new ArrayList<>(scalarIndexFiles)); } public String columnName() { @@ -108,24 +127,35 @@ public List fullTextIndexFiles() { return fullTextIndexFiles; } + /** Scalar global index files intersecting this split, used to pre-filter rows. */ + public List 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 boolean equals(Object o) { && 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 String toString() { + 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 86c6d2a1d2fd..9f65fd7bbd63 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,8 @@ 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.predicate.PredicateVisitor; import org.apache.paimon.reader.RecordReader; import org.apache.paimon.table.FileStoreTable; import org.apache.paimon.table.SpecialFields; @@ -57,6 +59,7 @@ import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.Set; import java.util.UUID; import java.util.concurrent.ExecutorService; @@ -68,6 +71,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 +83,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; @@ -92,12 +108,35 @@ ScoredGlobalIndexResult withRawSearch( List rawRowRanges, Map> splitsByColumn, ExecutorService executor) { + return withRawSearch(indexedResult, rawRowRanges, null, splitsByColumn, executor); + } + + /** + * Searches the raw row ranges and overrides the indexed result on them. {@code preFilter} + * bounds the rows to scan (rows outside it cannot satisfy the filter); the filter itself is + * still evaluated on every scanned row. + */ + ScoredGlobalIndexResult withRawSearch( + ScoredGlobalIndexResult indexedResult, + List rawRowRanges, + @Nullable RoaringNavigableMap64 preFilter, + Map> splitsByColumn, + ExecutorService executor) { rawRowRanges = Range.sortAndMergeOverlap(rawRowRanges, true); if (rawRowRanges.isEmpty()) { return indexedResult; } - ScoredGlobalIndexResult rawResult = readRawSearch(rawRowRanges, splitsByColumn, executor); + List scanRowRanges = + preFilter == null + ? rawRowRanges + : Range.and( + rawRowRanges, + Range.sortAndMergeOverlap(preFilter.toRangeList(), true)); + ScoredGlobalIndexResult rawResult = + scanRowRanges.isEmpty() + ? ScoredGlobalIndexResult.createEmpty() + : readRawSearch(scanRowRanges, splitsByColumn, executor); return overrideWithRawSearch(indexedResult, rawRowRanges, rawResult); } @@ -105,15 +144,17 @@ private ScoredGlobalIndexResult readRawSearch( List rawRowRanges, Map> splitsByColumn, ExecutorService executor) { - RowType readType = SpecialFields.rowTypeWithRowId(table.rowType()); - TableScan.Plan plan = rawReadBuilder(readType).withRowRanges(rawRowRanges).newScan().plan(); - ReadBuilder readBuilder = rawReadBuilder(readType); + RowType readType = rawReadType(); + TableScan.Plan plan = + rawReadBuilder(readType, false).withRowRanges(rawRowRanges).newScan().plan(); + ReadBuilder readBuilder = rawReadBuilder(readType, true); int rowIdIndex = readType.getFieldIndex(SpecialFields.ROW_ID.name()); Map rawIndexes = createRawFullTextIndexes(splitsByColumn, readType, rawRowRanges); try { - try (RecordReader reader = readBuilder.newRead().createReader(plan); + try (RecordReader reader = + readBuilder.newRead().executeFilter().createReader(plan); CloseableIterator iterator = reader.toCloseableIterator()) { while (iterator.hasNext()) { InternalRow row = iterator.next(); @@ -247,11 +288,30 @@ private static byte[] rawFileBytes(Map rawIndexes, Str throw new IllegalArgumentException("Unknown raw full-text index file: " + fileName); } - private ReadBuilder rawReadBuilder(RowType readType) { + /** The text column, the row id, and the filter columns so the predicate can be evaluated. */ + private RowType rawReadType() { + RowType tableRowType = table.rowType(); + List readFields = new ArrayList<>(); + readFields.add(textColumn.name()); + if (filter != null) { + Set filterFields = PredicateVisitor.collectFieldNames(filter); + for (String field : tableRowType.getFieldNames()) { + if (filterFields.contains(field) && !readFields.contains(field)) { + readFields.add(field); + } + } + } + return SpecialFields.rowTypeWithRowId(tableRowType.project(readFields)); + } + + private ReadBuilder rawReadBuilder(RowType readType, boolean includeFilter) { ReadBuilder readBuilder = rawReadTable().newReadBuilder().withReadType(readType); if (partitionFilter != null) { readBuilder.withPartitionFilter(partitionFilter); } + if (includeFilter && filter != null) { + readBuilder.withFilter(filter); + } return readBuilder; } diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/RawFullTextSearchSplit.java b/paimon-core/src/main/java/org/apache/paimon/table/source/RawFullTextSearchSplit.java index a95ea76255e8..5c14ead53efb 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/source/RawFullTextSearchSplit.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/source/RawFullTextSearchSplit.java @@ -18,8 +18,15 @@ package org.apache.paimon.table.source; +import org.apache.paimon.index.IndexFileMeta; +import org.apache.paimon.index.IndexFileMetaSerializer; +import org.apache.paimon.io.DataInputViewStreamWrapper; +import org.apache.paimon.io.DataOutputViewStreamWrapper; import org.apache.paimon.utils.Range; +import java.io.IOException; +import java.io.ObjectInputStream; +import java.io.ObjectOutputStream; import java.util.ArrayList; import java.util.Collections; import java.util.List; @@ -30,32 +37,74 @@ public class RawFullTextSearchSplit extends FullTextSearchSplit { private static final long serialVersionUID = 1L; + private static final int VERSION = 1; + + private static final ThreadLocal INDEX_SERIALIZER = + ThreadLocal.withInitial(IndexFileMetaSerializer::new); + private final List rowRanges; + private transient List scalarIndexFiles; public RawFullTextSearchSplit(List rowRanges) { + this(rowRanges, Collections.emptyList()); + } + + public RawFullTextSearchSplit(List rowRanges, List scalarIndexFiles) { this.rowRanges = Collections.unmodifiableList(new ArrayList<>(rowRanges)); + this.scalarIndexFiles = Collections.unmodifiableList(new ArrayList<>(scalarIndexFiles)); } public List rowRanges() { return rowRanges; } + /** Scalar global index files intersecting the raw ranges, used to pre-filter rows. */ + public List 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(scalarIndexFiles, view); + } + + private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundException { + in.defaultReadObject(); + int version = in.readInt(); + if (version != VERSION) { + throw new IOException("Unsupported RawFullTextSearchSplit version: " + version); + } + IndexFileMetaSerializer serializer = INDEX_SERIALIZER.get(); + DataInputViewStreamWrapper view = new DataInputViewStreamWrapper(in); + this.scalarIndexFiles = + Collections.unmodifiableList(new ArrayList<>(serializer.deserializeList(view))); + } + @Override public boolean equals(Object o) { if (o == null || getClass() != o.getClass()) { return false; } RawFullTextSearchSplit that = (RawFullTextSearchSplit) o; - return Objects.equals(rowRanges, that.rowRanges); + return Objects.equals(rowRanges, that.rowRanges) + && Objects.equals(scalarIndexFiles, that.scalarIndexFiles); } @Override public int hashCode() { - return Objects.hash(rowRanges); + return Objects.hash(rowRanges, scalarIndexFiles); } @Override public String toString() { - return "RawFullTextSearchSplit{" + "rowRanges=" + rowRanges + '}'; + return "RawFullTextSearchSplit{" + + "rowRanges=" + + rowRanges + + ", scalarIndexFiles=" + + scalarIndexFiles + + '}'; } } 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 0c4f2b512257..042497abc6fa 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 @@ -58,6 +58,8 @@ import org.junit.jupiter.api.Test; +import javax.annotation.Nullable; + import java.io.ByteArrayInputStream; import java.io.ByteArrayOutputStream; import java.io.ObjectInputStream; @@ -351,22 +353,535 @@ public void testHybridSearchBuilderWithFullTextRoute() throws Exception { } @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(); + assertThat(plan.splits()).anyMatch(RawFullTextSearchSplit.class::isInstance); + + GlobalIndexResult result = builder.newFullTextRead().read(plan); + assertThat(result.results()).hasSize(2); + assertThat(result.results()).isSubsetOf(3L, 4L, 5L); + } + + @Test + public void testFullTextSearchFullModeMixesIndexedAndRawFilterEvaluation() 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 must be filtered on the raw path. + 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(); + List rawSplits = rawSplits(plan); + assertThat(rawSplits).hasSize(1); + assertThat(rawSplits.get(0).rowRanges()).containsExactly(new Range(3, 5)); + 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 testFullTextSearchFastModeKeepsRawScanInsideIndexedRanges() throws Exception { + // full-text mode fast, scalar mode full: rows outside the full-text coverage stay out + // even though their filter columns are unindexed. + 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(); + List rawSplits = rawSplits(plan); + assertThat(rawSplits).hasSize(1); + assertThat(rawSplits.get(0).rowRanges()).containsExactly(new Range(0, 3)); + + 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 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 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 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 testFullTextSearchPartiallyIndexedConjunctionIsSupersetInFastMode() + throws Exception { + // id is indexed, the text column is not (as a scalar): in fast mode the evaluator drops + // the conjunct it cannot evaluate, so the pre-filter is a superset and the caller's + // row-level filter (Spark's post-filter) still applies. This documents the contract + // shared with vector search rather than an ideal outcome. createTableDefault(); FileStoreTable table = getTableDefault(); + 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, 3), + builder.equal(1, BinaryString.fromString("paimon zeta"))); + GlobalIndexResult result = searchWithFilter(table, partiallyIndexed, 10); + assertThat(result.results()).containsExactlyInAnyOrder(3L, 4L, 5L); + + // An OR with an unevaluable branch cannot be narrowed at all and is treated as unindexed. + Predicate partiallyIndexedOr = + PredicateBuilder.or( + builder.equal(0, 0), builder.equal(1, BinaryString.fromString("x"))); + assertThat(searchWithFilter(table, partiallyIndexedOr, 10).results().isEmpty()).isTrue(); + + // In full scalar mode the same predicate is exact: unindexed columns go to the raw path. + FileStoreTable fullModeTable = + (FileStoreTable) + table.copy( + Collections.singletonMap( + CoreOptions.SCALAR_INDEX_SEARCH_MODE.key(), "full")); + assertThat(searchWithFilter(fullModeTable, partiallyIndexed, 10).results()) + .containsExactly(5L); + assertThat(searchWithFilter(fullModeTable, partiallyIndexedOr, 10).results()) + .containsExactly(0L); + } + + @Test + public void testFullTextSearchRawScanIsBoundedByScalarPreFilter() throws Exception { + // full-text index covers rows 0-2 only, the id btree covers every row, full-text mode + // full: rows 3-5 are searched raw, and the btree bounds the raw scan to the matching rows. + 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()).greaterOrEqual(0, 4); + FullTextSearchBuilder builder = + table.newFullTextSearchBuilder() + .withQuery(TEXT_FIELD_NAME, matchQuery("paimon")) + .withLimit(10) + .withFilter(idFilter); + FullTextScan.Plan plan = builder.newFullTextScan().scan(); + List rawSplits = rawSplits(plan); + assertThat(rawSplits).hasSize(1); + assertThat(rawSplits.get(0).rowRanges()).containsExactly(new Range(3, 5)); + assertThat(rawSplits.get(0).scalarIndexFiles()).hasSize(1); - Predicate idFilter = new PredicateBuilder(table.rowType()).equal(0, 1); + assertThat(builder.newFullTextRead().read(plan).results()) + .containsExactlyInAnyOrder(4L, 5L); + } - 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"); + } + + 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 indexSplits(FullTextScan.Plan plan) { + List splits = new ArrayList<>(); + for (FullTextSearchSplit split : plan.splits()) { + if (split instanceof IndexFullTextSearchSplit) { + splits.add((IndexFullTextSearchSplit) split); + } + } + return splits; + } + + private static List rawSplits(FullTextScan.Plan plan) { + List splits = new ArrayList<>(); + for (FullTextSearchSplit split : plan.splits()) { + if (split instanceof RawFullTextSearchSplit) { + splits.add((RawFullTextSearchSplit) split); + } + } + return splits; } @Test @@ -848,11 +1363,13 @@ public void testFullTextSearchSplitSerialization() throws Exception { 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,9 +1396,13 @@ public void testFullTextSearchSplitSerialization() throws Exception { 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))); + new RawFullTextSearchSplit( + Collections.singletonList(new Range(2, 3)), original.scalarIndexFiles()); bos = new ByteArrayOutputStream(); try (ObjectOutputStream out = new ObjectOutputStream(bos)) { out.writeObject(rawOriginal); @@ -894,6 +1415,8 @@ public void testFullTextSearchSplitSerialization() throws Exception { } assertThat(rawDeserialized.rowRanges()).isEqualTo(rawOriginal.rowRanges()); + assertThat(rawDeserialized.scalarIndexFiles()).isEqualTo(rawOriginal.scalarIndexFiles()); + assertThat(rawDeserialized).isEqualTo(rawOriginal); } // ====================== Helper methods ====================== @@ -1152,6 +1675,16 @@ private void buildAndCommitIndexForColumn( 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 +1700,7 @@ private void buildAndCommitIndexForColumn( } List entries = writer.finish(); - Range rowRange = new Range(0, documents.length - 1); + Range rowRange = new Range(rowStart, rowStart + documents.length - 1); List indexFiles = GlobalIndexBuilderUtils.toIndexFileMetas( table.fileIO(), @@ -1187,6 +1720,47 @@ private void buildAndCommitIndexForColumn( } } + 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 entries = writer.finish(); + + List 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 e3faa6a38d9e..400a609b3895 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.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 @@ void testDataEvolutionRetainsGlobalFullTextPath() { 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 000000000000..43e2dd08b557 --- /dev/null +++ b/paimon-full-text/src/test/java/org/apache/paimon/fulltext/index/NativeFullTextRowFilterTest.java @@ -0,0 +1,403 @@ +/* + * 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 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 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(); + } + + /** + * 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 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 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 categoryOf; + + Dataset(FileStoreTable table, Map 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 { + 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 categoryOf = new HashMap<>(); + List contents = new ArrayList<>(rowCount); + List 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 < rowCount; i++) { + textWriter.write(BinaryString.fromString(contents.get(i)), i); + } + List indexFiles = + new ArrayList<>( + GlobalIndexBuilderUtils.toIndexFileMetas( + table.fileIO(), + table.store().pathFactory().globalIndexFileFactory(), + table.coreOptions(), + rowRange, + 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 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 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 dc71a9cfbfbf..ed3452d2c39d 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 4c131e1f0312..dfdd82087797 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 254bf50c7797..a9ec19a326e7 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 } }