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