feat(global-index): support row filters in full-text search

docs/source/user_guide/global_index.rst

@@ -303,8 +303,7 @@ an index shard covers are searched, as in the default ``fast`` ``full-text-index
 Java Paimon, so rows written since the index was last built are not found until it is built for
 them. A search whose ``full-text-index.search-mode``, or else ``global-index.search-mode``, is
 ``full`` or ``detail``, which also search the rows that no index covers, fails with
-``NotImplemented``. Row filters, which Java Paimon evaluates with scalar indexes before ranking, are
-not supported yet.
+``NotImplemented``.
 
 .. code-block:: cpp
 
@@ -338,6 +337,8 @@ candidates only: which of the rows tied at the ``limit``-th score of a shard bec
 to its index.
 
 - ``SetPartitionFilter()`` restricts the search to some partitions, as in ``ScanContextBuilder``.
+- ``WithFilter()`` restricts the search to the rows matching a predicate; see
+  `Row Filters`_ below.
 - The search reads the snapshot that the time-travel options select, as Java Paimon does: snapshot
   ``scan.snapshot-id``, the snapshot of tag ``scan.tag-name``, the latest snapshot at or before
   ``scan.timestamp`` or ``scan.timestamp-millis``, or else the latest snapshot. At most one of these
@@ -359,6 +360,45 @@ to its index.
 - A column with a primary-key full-text index (``pk-full-text.index.columns``) cannot be searched
   yet.
 
+Row Filters
+~~~~~~~~~~~
+
+``WithFilter()`` adds a row filter, as Java Paimon's ``withFilter`` does. Only the matching rows are
+ranked, so the result is the top ``limit`` rows among them, not a filtered subset of the unfiltered
+top ``limit``. The matching rows are handed to the existing index as include row ids, so BM25
+statistics stay those of the whole index shard. Repeated calls are ANDed. Conjuncts over partition
+fields alone restrict the partitions instead, ANDed with ``SetPartitionFilter()``. As in Java
+Paimon, any other conjunct still filters rows but also restricts the partitions to those it
+implies: ``(p = 1 AND a > 0) OR (p = 2 AND a > 0)`` searches only partitions ``p = 1`` and
+``p = 2``.
+
+.. code-block:: cpp
+
+   auto filter = paimon::PredicateBuilder::Equal(/*field_index=*/1, "category",
+                                                 paimon::FieldType::INT, paimon::Literal(7));
+   search_builder.WithFilter(filter);
+
+The scan keeps the scalar global index files that index a field of the filter, and the read
+resolves the filter with them before ranking:
+
+- If the indexes decide the whole filter, the rows they return are searched.
+- If they can only narrow the filter down to candidates, because a conjunct has no index or the
+  filter has a ``Contains``, ``EndsWith`` or ``Like`` predicate, the candidates are excluded and a
+  warning is logged. The search then never returns a non-matching row, but may return fewer rows
+  than ``limit``. With ``global-index.filter.refine-from-data=true``, the read instead verifies the
+  candidates by reading their filter columns and ``_ROW_ID`` in the planned snapshot, on the caller
+  and possibly for every candidate.
+- Rows that no scalar index covers are excluded. If no index can evaluate the filter at all, no row
+  is searched and a warning is logged. This is the default ``fast`` ``scalar-index.search-mode``,
+  which falls back on ``global-index.search-mode``. A row filter in ``full`` or ``detail`` mode,
+  which decide such rows by reading their filter columns, fails with ``NotImplemented``.
+
+Only single-field indexes that do not support full-text search evaluate a filter. A filter with a
+``PredicateBuilder::FullTextMatch`` predicate fails with ``NotImplemented``.
+
+Pinning the Data Read
+~~~~~~~~~~~~~~~~~~~~~
+
 The row ids of the result refer to the snapshot the search was planned from. If the table may
 change before the rows are read, plan with ``NewFullTextScan()`` and pin the data scan to the
 snapshot of the plan, setting ``scan.mode`` too so that a ``scan.mode`` of the table cannot override
@@ -386,12 +426,13 @@ outlive the snapshot.
 ``ExecuteLocal()`` plans and searches in one process. To plan in one process and search in others,
 plan with ``NewFullTextScan()``, ship the splits with ``FullTextSearchSplit::Serialize()`` and
 ``FullTextSearchSplit::Deserialize()``, and search them with the ``NewFullTextRead()`` of a builder
-for the same table, column and query. The serialized form is versioned and specific to Paimon C++:
-Java Paimon serializes its splits with Java serialization, so splits cannot be exchanged with it.
-Each split records the snapshot it was planned from, and the read ranks the rows live in that
-snapshot. The read rejects a split whose index files do not index the column of the latest schema,
-such as after the column was replaced. A read returns the top rows of the splits it is given, so
-keep the ``limit`` highest-scoring rows of all reads if the splits are searched by several of them.
+for the same table, column, query and filter. The serialized form is versioned and specific to
+Paimon C++: Java Paimon serializes its splits with Java serialization, so splits cannot be exchanged
+with it. Each split records the snapshot it was planned from, and the read ranks the rows live in
+that snapshot. The read rejects a split whose index files do not index the column of the latest
+schema, such as after the column was replaced. A read returns the top rows of the splits it is
+given, so keep the ``limit`` highest-scoring rows of all reads if the splits are searched by several
+of them.
 
 .. code-block:: cpp
 

include/paimon/defs.h

@@ -623,6 +623,19 @@ struct PAIMON_EXPORT Options {
     /// index covers. Default value is "global-index.search-mode", or else "fast". Paimon C++
     /// supports only "fast" so far, and a full-text search in another mode fails.
     static const char FULL_TEXT_INDEX_SEARCH_MODE[];
+    /// "scalar-index.search-mode" - Search mode for scalar index queries: "fast" evaluates only the
+    /// rows that an index covers, while "full" and "detail" also evaluate the rows that no index
+    /// covers. Default value is "global-index.search-mode", or else "fast". Paimon C++ supports
+    /// only "fast" so far, and a full-text search with a row filter in another mode fails.
+    static const char SCALAR_INDEX_SEARCH_MODE[];
+    /// "global-index.filter.refine-from-data" - Whether a vector, hybrid or full-text search may
+    /// read the filter columns of candidate rows to verify a row filter that the scalar global
+    /// index can only answer with a superset, such as contains, ends-with or like on a BTree index
+    /// or a conjunction with a member no index can evaluate. When false, such candidates are
+    /// excluded from the search, which never returns a non-matching row but may return fewer than
+    /// the requested top-k. When true, the read runs on the caller and may cover every candidate
+    /// row. Default value is "false". Paimon C++ reads it only in full-text search so far.
+    static const char GLOBAL_INDEX_FILTER_REFINE_FROM_DATA[];
     /// "pk-btree.index.columns" - Comma-separated columns indexed by primary-key BTree indexes.
     /// No default value.
     static const char PK_BTREE_INDEX_COLUMNS[];

include/paimon/table/source/full_text_read.h

@@ -51,10 +51,18 @@ class PAIMON_EXPORT FullTextRead {
     /// ranks only its search row ranges, and with deletion vectors enabled only the rows that are
     /// live in the snapshot the splits were planned from; BM25 statistics still come from its whole
     /// index shard.
-    /// @param splits Splits planned by a `FullTextScan` of the same column, all from one snapshot.
+    ///
+    /// With a row filter (`FullTextSearchBuilder::WithFilter()`), a split ranks only the rows that
+    /// satisfy it, as resolved before ranking by the scalar indexes that the splits carry. Rows
+    /// that no scalar index can evaluate are not ranked. Candidates that the indexes cannot decide
+    /// exactly, as for `Contains`, `EndsWith` and `Like`, are excluded too, so the result may hold
+    /// fewer than `limit` rows, unless `global-index.filter.refine-from-data` verifies them against
+    /// the data of the planned snapshot on this thread.
+    /// @param splits Splits planned by a `FullTextScan` of the same column and filter, all from one
+    ///               snapshot.
     /// @return The rows found, or an error if a split is invalid, its index files do not index the
     ///         column of the latest schema (for example, after the column was replaced), or an
-    ///         index cannot be read.
+    ///         index or data file cannot be read.
     virtual Result<std::shared_ptr<ScoredGlobalIndexResult>> Read(
         const std::vector<std::shared_ptr<FullTextSearchSplit>>& splits) = 0;
 };

include/paimon/table/source/full_text_scan.h

@@ -76,7 +76,9 @@ class PAIMON_EXPORT FullTextScan {
     /// without the backend of a `full-text` or `lucene-fts` index it finds. When the row ranges of
     /// index shards overlap, each row is ranked by one shard only: an index whose indexed field is
     /// the column comes before one that carries it as an extra field, and further ties are broken
-    /// by index type and fields, then by the start and end of the shard.
+    /// by index type and fields, then by the start and end of the shard. With a row filter, each
+    /// split also carries the scalar index files that intersect its shard and index a field of the
+    /// filter.
     /// @return The plan, or an error if more than one time-travel option is set, the option does
     ///         not agree with `scan.mode`, the selected snapshot or tag does not exist, or reading
     ///         the index manifests fails; `NotImplemented` for an index whose backend is not built.

include/paimon/table/source/full_text_search_builder.h

@@ -25,6 +25,7 @@
 #include <vector>
 
 #include "paimon/global_index/global_index_result.h"
+#include "paimon/predicate/predicate.h"
 #include "paimon/result.h"
 #include "paimon/table/source/full_text_read.h"
 #include "paimon/table/source/full_text_scan.h"
@@ -49,7 +50,8 @@ class Executor;
 ///
 /// `ExecuteLocal()` plans and searches in this process. To run them in different processes, plan
 /// with `NewFullTextScan()`, serialize the splits with `FullTextSearchSplit::Serialize()`, and
-/// search them with the `NewFullTextRead()` of a builder for the same table, column and query.
+/// search them with the `NewFullTextRead()` of a builder for the same table, column, query and
+/// filter.
 ///
 /// The row ids of the result refer to the snapshot the search was planned from, which
 /// `FullTextScan::Plan::SnapshotId()` returns. If the table may change in between, pin the data
@@ -68,6 +70,17 @@ class PAIMON_EXPORT FullTextSearchBuilder {
     FullTextSearchBuilder& SetPartitionFilter(
         const std::vector<std::map<std::string, std::string>>& partition_filters);
 
+    /// Restricts the search to the rows matching `predicate`, as Java
+    /// `FullTextSearchBuilder#withFilter` does: only those rows are ranked, so the result is the
+    /// top `limit` rows among them, and BM25 statistics stay those of the whole index shard. The
+    /// fields of the predicate are matched by name. Repeated calls are ANDed. Conjuncts over
+    /// partition fields alone restrict the partitions, ANDed with `SetPartitionFilter()`, and the
+    /// rest is the row filter that `FullTextRead::Read()` resolves. A conjunct of the row filter
+    /// also restricts the partitions to those it implies, as `(p = 1 AND a > 0) OR (p = 2 AND
+    /// a > 0)` does to `p = 1 OR p = 2`.
+    /// @param predicate The filter; a null predicate adds no filter.
+    FullTextSearchBuilder& WithFilter(const std::shared_ptr<Predicate>& predicate);
+
     /// Sets the number of highest-scoring rows to return. It must be positive to create a read.
     FullTextSearchBuilder& SetLimit(int32_t limit);
 
@@ -100,9 +113,11 @@ class PAIMON_EXPORT FullTextSearchBuilder {
     /// Creates a scan that plans the search on the current settings. The query must be set, and
     /// its column must exist in the latest schema of the table.
     /// @return The scan, or an error if the settings are invalid or the table cannot be loaded;
-    ///         `NotImplemented` for a primary-key full-text index (`pk-full-text.index.columns`) or
-    ///         a search mode other than `fast` (`full-text-index.search-mode`, or else
-    ///         `global-index.search-mode`).
+    ///         `NotImplemented` for a primary-key full-text index (`pk-full-text.index.columns`), a
+    ///         search mode other than `fast` (`full-text-index.search-mode`, or else
+    ///         `global-index.search-mode`), a row filter whose scalar search mode is not `fast`
+    ///         (`scalar-index.search-mode`, or else `global-index.search-mode`), or a filter with a
+    ///         full-text match.
     Result<std::unique_ptr<FullTextScan>> NewFullTextScan() const;
 
     /// Creates a read that searches the splits of a scan on the current settings. In addition to

src/paimon/CMakeLists.txt

@@ -484,6 +484,7 @@ set(PAIMON_CORE_SRCS
     core/table/source/data_evolution_batch_scan.cpp
     core/table/source/data_evolution_full_text_read.cpp
     core/table/source/data_evolution_full_text_scan.cpp
+    core/table/source/filtered_row_id_reader.cpp
     core/table/source/full_text_search_builder.cpp
     core/table/source/full_text_search_split.cpp
     core/table/source/primary_key_sorted_index_scan.cpp
@@ -1033,6 +1034,7 @@ if(PAIMON_BUILD_TESTS)
                     core/table/source/data_evolution_batch_scan_test.cpp
                     core/table/source/data_evolution_full_text_read_test.cpp
                     core/table/source/data_evolution_full_text_scan_test.cpp
+                    core/table/source/filtered_row_id_reader_test.cpp
                     core/table/source/full_text_search_split_test.cpp
                     core/table/source/data_split_test.cpp
                     core/table/source/deletion_file_test.cpp

src/paimon/common/defs.cpp

@@ -164,6 +164,8 @@ const char Options::GLOBAL_INDEX_THREAD_NUM[] = "global-index.thread-num";
 const char Options::GLOBAL_INDEX_EXTERNAL_PATH[] = "global-index.external-path";
 const char Options::GLOBAL_INDEX_SEARCH_MODE[] = "global-index.search-mode";
 const char Options::FULL_TEXT_INDEX_SEARCH_MODE[] = "full-text-index.search-mode";
+const char Options::SCALAR_INDEX_SEARCH_MODE[] = "scalar-index.search-mode";
+const char Options::GLOBAL_INDEX_FILTER_REFINE_FROM_DATA[] = "global-index.filter.refine-from-data";
 const char Options::PK_BTREE_INDEX_COLUMNS[] = "pk-btree.index.columns";
 const char Options::PK_BITMAP_INDEX_COLUMNS[] = "pk-bitmap.index.columns";
 const char Options::PK_VECTOR_INDEX_COLUMNS[] = "pk-vector.index.columns";

src/paimon/core/core_options.cpp

@@ -178,11 +178,10 @@ class ConfigParser {
             "sort engine", sort_engine);
     }
 
-    // Parse GlobalIndexSearchMode of full-text search, as Java `CoreOptions#indexSearchMode` does
-    Status ParseFullTextIndexSearchMode(GlobalIndexSearchMode* search_mode) const {
-        const char* key = config_map_.count(Options::FULL_TEXT_INDEX_SEARCH_MODE) > 0
-                              ? Options::FULL_TEXT_INDEX_SEARCH_MODE
-                              : Options::GLOBAL_INDEX_SEARCH_MODE;
+    // Parse GlobalIndexSearchMode of an index family, as Java `CoreOptions#indexSearchMode` does
+    Status ParseIndexSearchMode(const char* family_key, GlobalIndexSearchMode* search_mode) const {
+        const char* key =
+            config_map_.count(family_key) > 0 ? family_key : Options::GLOBAL_INDEX_SEARCH_MODE;
         return ParseEnum(key,
                          {{"fast", GlobalIndexSearchMode::FAST},
                           {"full", GlobalIndexSearchMode::FULL},
@@ -450,6 +449,8 @@ struct CoreOptions::Impl {
     std::optional<int32_t> lookup_compact_max_interval;
     std::optional<int32_t> global_index_thread_num;
     GlobalIndexSearchMode full_text_index_search_mode = GlobalIndexSearchMode::FAST;
+    GlobalIndexSearchMode scalar_index_search_mode = GlobalIndexSearchMode::FAST;
+    bool global_index_filter_refine_from_data = false;
 
     bool realtime_enabled = false;
     bool realtime_spill_enabled = false;
@@ -927,7 +928,14 @@ struct CoreOptions::Impl {
         PAIMON_RETURN_NOT_OK(
             parser.Parse<int32_t>(Options::GLOBAL_INDEX_THREAD_NUM, &global_index_thread_num));
         // Parse full-text-index.search-mode, falling back to global-index.search-mode, default fast
-        PAIMON_RETURN_NOT_OK(parser.ParseFullTextIndexSearchMode(&full_text_index_search_mode));
+        PAIMON_RETURN_NOT_OK(parser.ParseIndexSearchMode(Options::FULL_TEXT_INDEX_SEARCH_MODE,
+                                                         &full_text_index_search_mode));
+        // Parse scalar-index.search-mode, falling back to global-index.search-mode, default fast
+        PAIMON_RETURN_NOT_OK(parser.ParseIndexSearchMode(Options::SCALAR_INDEX_SEARCH_MODE,
+                                                         &scalar_index_search_mode));
+        // Parse global-index.filter.refine-from-data, default false
+        PAIMON_RETURN_NOT_OK(parser.Parse<bool>(Options::GLOBAL_INDEX_FILTER_REFINE_FROM_DATA,
+                                                &global_index_filter_refine_from_data));
         // Parse global-index.external-path - global index root directory
         PAIMON_RETURN_NOT_OK(
             parser.Parse(Options::GLOBAL_INDEX_EXTERNAL_PATH, &global_index_external_path));
@@ -1920,6 +1928,14 @@ GlobalIndexSearchMode CoreOptions::GetFullTextIndexSearchMode() const {
     return impl_->full_text_index_search_mode;
 }
 
+GlobalIndexSearchMode CoreOptions::GetScalarIndexSearchMode() const {
+    return impl_->scalar_index_search_mode;
+}
+
+bool CoreOptions::GlobalIndexFilterRefineFromData() const {
+    return impl_->global_index_filter_refine_from_data;
+}
+
 std::optional<std::string> CoreOptions::GetGlobalIndexExternalPath() const {
     return impl_->global_index_external_path;
 }

src/paimon/core/core_options.h

@@ -307,6 +307,10 @@ class PAIMON_EXPORT CoreOptions {
     /// Returns `full-text-index.search-mode`, or else `global-index.search-mode`, or else
     /// `GlobalIndexSearchMode::FAST`, as Java `CoreOptions#fullTextIndexSearchMode` does.
     GlobalIndexSearchMode GetFullTextIndexSearchMode() const;
+    /// Returns `scalar-index.search-mode`, or else `global-index.search-mode`, or else
+    /// `GlobalIndexSearchMode::FAST`, as Java `CoreOptions#scalarIndexSearchMode` does.
+    GlobalIndexSearchMode GetScalarIndexSearchMode() const;
+    bool GlobalIndexFilterRefineFromData() const;
 
     const std::vector<std::string>& GetBlobFields() const;
     const std::vector<std::string>& GetBlobDescriptorFields() const;