apache · zhuxiangyi · Sep 15, 2026
diff --git 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,38 @@ When `full-text-index.search-mode` is `full` or `detail`, Paimon also covers
 unindexed row ranges by reading raw rows and building a temporary native
 full-text index for the searched column.
 
+### Row Filters
+
+A full-text search can carry a row filter: a Spark `WHERE` clause on the
+`full_text_search` result, `withFilter` on the Java builder, or `withFilter` on
+a hybrid search. Partition predicates in the filter prune partitions; the
+remaining predicates are evaluated **before** top-k ranking, so the result is
+the top-k among matching rows rather than a filtered subset of the unfiltered
+top-k.
+
+Non-partition predicates are evaluated through the scalar global indexes
+(BTree, Bitmap, Multivalue, FM) built on the filtered columns, the same way
+[vector search pre-filters](./vector#vector-search) rows. Build a scalar index
+on the columns you filter by. How rows whose filter columns are not covered by
+a scalar index are handled follows `scalar-index.search-mode`:
+
+| `scalar-index.search-mode` | Rows with an index on the filter columns | Rows without one |
+|---|---|---|
+| `fast` (default) | Evaluated through the index | Excluded from the search; a warning is logged |
+| `full` / `detail` | Evaluated through the index | Read from the data files and filtered row by row before a temporary full-text index ranks them |
+
+With `full-text-index.search-mode=fast`, only rows covered by the full-text
+index are considered in either case. In `fast` mode a conjunction whose members
+are only partly indexed (`indexed_col = 1 AND unindexed_col = 2`) is narrowed by
+the indexed members alone, so the candidate set is a superset. Predicates the
+engine cannot push down (for example, UDFs) are likewise applied by the engine
+after the search. In both situations the rows returned are correct, but a query
+may return fewer than `limit` rows; use `full` mode or index every filtered
+column when the exact top-k matters.
+
+Primary-key full-text indexes do not support row filters yet; see
+[Primary-Key Indexes](../../primary-key-table/global-index#limitations).
+
 ## Query DSL
 
 The query DSL is a JSON object with one top-level query type. Use the examples

diff --git a/docs/docs/multimodal-table/global-index/hybrid-search.mdx b/docs/docs/multimodal-table/global-index/hybrid-search.mdx
@@ -397,3 +397,12 @@ increase route limits when a small candidate set restricts final recall.
 
 For Java and Python, the subsequent table scan does not preserve ranking order.
 See [Read Scored Results](./manage-indexes#read-scored-results) to access scores by row ID.
+
+## Row Filters
+
+A row filter given to a hybrid search (`WHERE` on the Spark `hybrid_search`
+result, or `withFilter` on the Java builder) is applied to every route before
+its own top-k, so vector and full-text candidates are drawn from the same
+filtered row set before the ranker merges them. Full-text routes follow the same
+rules as a standalone full-text search; see
+[Full-Text Row Filters](./full-text#row-filters).
diff --git a/docs/docs/primary-key-table/global-index.mdx b/docs/docs/primary-key-table/global-index.mdx
@@ -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
@@ -18,7 +18,10 @@
 
 package org.apache.paimon.table.source;
 
+import org.apache.paimon.CoreOptions.GlobalIndexSearchMode;
 import org.apache.paimon.Snapshot;
+import org.apache.paimon.globalindex.DataEvolutionGlobalIndexScanner;
+import org.apache.paimon.globalindex.GlobalIndexEvaluator;
 import org.apache.paimon.globalindex.GlobalIndexIOMeta;
 import org.apache.paimon.globalindex.GlobalIndexReadThreadPool;
 import org.apache.paimon.globalindex.GlobalIndexReader;
@@ -33,19 +36,27 @@
 import org.apache.paimon.index.IndexPathFactory;
 import org.apache.paimon.partition.PartitionPredicate;
 import org.apache.paimon.predicate.FullTextSearch;
+import org.apache.paimon.predicate.Predicate;
 import org.apache.paimon.table.FileStoreTable;
 import org.apache.paimon.types.DataField;
 import org.apache.paimon.utils.IOUtils;
 import org.apache.paimon.utils.Range;
 import org.apache.paimon.utils.RoaringNavigableMap64;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import javax.annotation.Nullable;
 
+import java.io.IOException;
 import java.util.ArrayList;
+import java.util.Comparator;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
+import java.util.Set;
+import java.util.TreeSet;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutorService;
 
@@ -55,8 +66,11 @@
 /** Implementation for {@link FullTextRead}. */
 public class DataEvolutionFullTextRead implements FullTextRead {
 
+    private static final Logger LOG = LoggerFactory.getLogger(DataEvolutionFullTextRead.class);
+
     private final FileStoreTable table;
     @Nullable private final PartitionPredicate partitionFilter;
+    @Nullable private final Predicate filter;
     private final int limit;
     private final DataField textColumn;
     private final String query;
@@ -67,8 +81,19 @@ public DataEvolutionFullTextRead(
             int limit,
             List<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,58 +125,170 @@ private GlobalIndexResult read(
         ExecutorService executor = GlobalIndexReadThreadPool.getExecutorService(parallelism);
 
         Map<String, List<IndexFullTextSearchSplit>> splitsByColumn = new HashMap<>();
+        List<IndexFullTextSearchSplit> indexSplits = new ArrayList<>();
         List<Range> rawRowRanges = new ArrayList<>();
+        List<RawFullTextSearchSplit> rawSplits = new ArrayList<>();
         for (FullTextSearchSplit split : splits) {
             if (split instanceof IndexFullTextSearchSplit) {
                 IndexFullTextSearchSplit indexSplit = (IndexFullTextSearchSplit) split;
+                indexSplits.add(indexSplit);
                 splitsByColumn
                         .computeIfAbsent(indexSplit.columnName(), k -> new ArrayList<>())
                         .add(indexSplit);
             } else if (split instanceof RawFullTextSearchSplit) {
-                rawRowRanges.addAll(((RawFullTextSearchSplit) split).rowRanges());
+                RawFullTextSearchSplit rawSplit = (RawFullTextSearchSplit) split;
+                rawSplits.add(rawSplit);
+                rawRowRanges.addAll(rawSplit.rowRanges());
             }
         }
 
         GlobalIndexFileReader indexFileReader = m -> table.fileIO().newInputStream(m.filePath());
         RoaringNavigableMap64 liveRows =
                 GlobalIndexLiveRowFilter.liveRows(table, planSnapshot, partitionFilter, null);
+        RoaringNavigableMap64 matchedRows = scalarMatchedRows(indexSplits, planSnapshot);
         ScoredGlobalIndexResult result =
-                evalQuery(splitsByColumn, indexPathFactory, indexFileReader, executor, liveRows);
+                evalQuery(
+                        splitsByColumn,
+                        indexPathFactory,
+                        indexFileReader,
+                        executor,
+                        liveRows,
+                        matchedRows);
         if (!rawRowRanges.isEmpty()) {
             result =
                     new RawFullTextReadImpl(
                                     table,
                                     planSnapshot,
                                     partitionFilter,
+                                    filter,
                                     limit,
                                     textColumn,
                                     this::evalQuery)
-                            .withRawSearch(result, rawRowRanges, splitsByColumn, executor);
+                            .withRawSearch(
+                                    result,
+                                    rawRowRanges,
+                                    rawPreFilter(rawSplits, planSnapshot),
+                                    splitsByColumn,
+                                    executor);
         }
         return result.topK(limit);
     }
 
+    /**
+     * Rows of the indexed splits that satisfy {@link #filter} according to the scalar global
+     * indexes attached to them, or {@code null} when there is no filter. Rows whose filter columns
+     * are not indexed are absent from the result: in {@code fast} scalar search mode they are
+     * excluded, in other modes the scan already routed them to a raw split where the predicate is
+     * evaluated on the data.
+     */
+    @Nullable
+    private RoaringNavigableMap64 scalarMatchedRows(
+            List<IndexFullTextSearchSplit> indexSplits, @Nullable Snapshot planSnapshot) {
+        if (filter == null || indexSplits.isEmpty()) {
+            return null;
+        }
+
+        Set<IndexFileMeta> scalarIndexFiles =
+                new TreeSet<>(Comparator.comparing(IndexFileMeta::fileName));
+        for (IndexFullTextSearchSplit split : indexSplits) {
+            scalarIndexFiles.addAll(split.scalarIndexFiles());
+        }
+
+        Optional<DataEvolutionGlobalIndexScanner> optionalScanner =
+                DataEvolutionGlobalIndexScanner.create(
+                        table, planSnapshot, partitionFilter, scalarIndexFiles);
+        if (!optionalScanner.isPresent()) {
+            warnUnindexedFilter();
+            return new RoaringNavigableMap64();
+        }
+
+        try (DataEvolutionGlobalIndexScanner scanner = optionalScanner.get()) {
+            Optional<GlobalIndexResult> result = scanner.scan(filter);
+            if (!result.isPresent()) {
+                warnUnindexedFilter();
+                return new RoaringNavigableMap64();
+            }
+            return result.get().results();
+        } catch (IOException e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+    /**
+     * Rows of the raw splits that may satisfy {@link #filter}: rows matched by the scalar indexes
+     * plus rows those indexes do not cover, or {@code null} when nothing can be pre-filtered. The
+     * raw read still evaluates the predicate row by row, so this only bounds the rows to scan.
+     */
+    @Nullable
+    private RoaringNavigableMap64 rawPreFilter(
+            List<RawFullTextSearchSplit> rawSplits, @Nullable Snapshot planSnapshot) {
+        if (filter == null || rawSplits.isEmpty()) {
+            return null;
+        }
+
+        Set<IndexFileMeta> scalarIndexFiles =
+                new TreeSet<>(Comparator.comparing(IndexFileMeta::fileName));
+        for (RawFullTextSearchSplit split : rawSplits) {
+            scalarIndexFiles.addAll(split.scalarIndexFiles());
+        }
+        Optional<DataEvolutionGlobalIndexScanner> optionalScanner =
+                DataEvolutionGlobalIndexScanner.create(
+                        table, planSnapshot, partitionFilter, scalarIndexFiles);
+        if (!optionalScanner.isPresent()) {
+            return null;
+        }
+
+        RoaringNavigableMap64 include = new RoaringNavigableMap64();
+        try (DataEvolutionGlobalIndexScanner scanner = optionalScanner.get()) {
+            Optional<GlobalIndexEvaluator.Evaluation> result = scanner.scanWithCoverage(filter);
+            if (!result.isPresent()) {
+                return null;
+            }
+            include.or(result.get().result().results());
+            include.or(
+                    scanner.unindexedRowsForContributingFields(result.get().contributingFieldIds())
+                            .results());
+        } catch (IOException e) {
+            throw new RuntimeException(e);
+        }
+        return include;
+    }
+
+    private void warnUnindexedFilter() {
+        if (table.coreOptions().scalarIndexSearchMode() == GlobalIndexSearchMode.FAST) {
+            LOG.warn(
+                    "Full-text search on table {} has a row filter {} that no scalar global index "
+                            + "can evaluate; indexed rows are excluded from the result because "
+                            + "scalar-index.search-mode is fast. Build a scalar index on the "
+                            + "filtered columns or use search mode full.",
+                    table.name(),
+                    filter);
+        }
+    }
+
     ScoredGlobalIndexResult evalQuery(
             Map<String, List<IndexFullTextSearchSplit>> splitsByColumn,
             IndexPathFactory indexPathFactory,
             GlobalIndexFileReader indexFileReader,
             ExecutorService executor) {
-        return evalQuery(splitsByColumn, indexPathFactory, indexFileReader, executor, null);
+        return evalQuery(splitsByColumn, indexPathFactory, indexFileReader, executor, null, null);
     }
 
     private ScoredGlobalIndexResult evalQuery(
             Map<String, List<IndexFullTextSearchSplit>> splitsByColumn,
             IndexPathFactory indexPathFactory,
             GlobalIndexFileReader indexFileReader,
             ExecutorService executor,
-            @Nullable RoaringNavigableMap64 liveRows) {
+            @Nullable RoaringNavigableMap64 liveRows,
+            @Nullable RoaringNavigableMap64 matchedRows) {
         return evalColumnQuery(
                 textColumn.name(),
                 splitsByColumn,
                 indexPathFactory,
                 indexFileReader,
                 executor,
-                liveRows);
+                liveRows,
+                matchedRows);
     }
 
     private ScoredGlobalIndexResult evalColumnQuery(
@@ -160,7 +297,8 @@ private ScoredGlobalIndexResult evalColumnQuery(
             IndexPathFactory indexPathFactory,
             GlobalIndexFileReader indexFileReader,
             ExecutorService executor,
-            @Nullable RoaringNavigableMap64 liveRows) {
+            @Nullable RoaringNavigableMap64 liveRows,
+            @Nullable RoaringNavigableMap64 matchedRows) {
         List<IndexFullTextSearchSplit> columnSplits = splitsByColumn.get(column);
         if (columnSplits == null || columnSplits.isEmpty()) {
             return ScoredGlobalIndexResult.createEmpty();
@@ -183,7 +321,7 @@ private ScoredGlobalIndexResult evalColumnQuery(
                             split.fullTextIndexFiles(),
                             indexFileReader,
                             executor,
-                            includeRowIds(split, liveRows)));
+                            includeRowIds(split, liveRows, matchedRows)));
         }
 
         CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();
@@ -201,14 +339,19 @@ private ScoredGlobalIndexResult evalColumnQuery(
 
     @Nullable
     private static RoaringNavigableMap64 includeRowIds(
-            IndexFullTextSearchSplit split, @Nullable RoaringNavigableMap64 liveRows) {
+            IndexFullTextSearchSplit split,
+            @Nullable RoaringNavigableMap64 liveRows,
+            @Nullable RoaringNavigableMap64 matchedRows) {
         RoaringNavigableMap64 include = new RoaringNavigableMap64();
         for (Range range : split.searchRowRanges()) {
             include.addRange(range);
         }
         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;
     }