feat(global-index): support table-level full-text search for data-evolution tables

docs/source/api/global_index.rst

@@ -64,6 +64,30 @@ Interface
    :members:
    :undoc-members:
 
+.. doxygenclass:: paimon::FullTextSearchBuilder
+   :members:
+   :undoc-members:
+
+.. doxygenclass:: paimon::FullTextScan
+   :members:
+   :undoc-members:
+
+.. doxygenclass:: paimon::FullTextScan::Plan
+   :members:
+   :undoc-members:
+
+.. doxygenclass:: paimon::FullTextRead
+   :members:
+   :undoc-members:
+
+.. doxygenclass:: paimon::FullTextSearchSplit
+   :members:
+   :undoc-members:
+
+.. doxygenclass:: paimon::IndexFullTextSearchSplit
+   :members:
+   :undoc-members:
+
 .. doxygenclass:: paimon::GlobalIndexWriteTask
    :members:
    :undoc-members:

docs/source/user_guide/global_index.rst

@@ -291,6 +291,131 @@ and readers use that copy.
 The prefix-stripped options are also stored as a flat JSON object in the metadata of each index
 file, as Java Paimon does.
 
+Table-Level Full-Text Search
+----------------------------
+
+``FullTextSearchBuilder`` (``paimon/table/source/full_text_search_builder.h``) searches the global
+full-text indexes of one column of a data-evolution table, as Java Paimon's
+``FullTextSearchBuilder`` does, so a caller does not create index readers or choose index types and
+shard ranges. It searches the index files whose indexer supports full-text search
+(``GlobalIndexer::SupportsFullTextSearch()``): ``full-text`` and ``lucene-fts``. Only the rows that
+an index shard covers are searched, as in the default ``fast`` ``full-text-index.search-mode`` of
+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.
+
+.. code-block:: cpp
+
+   paimon::FullTextSearchBuilder search_builder(table_path);
+   search_builder.SetQuery("content", R"({"match":{"query":"paimon lake"}})").SetLimit(10);
+   PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<paimon::ScoredGlobalIndexResult> result,
+                          search_builder.ExecuteLocal());
+
+   // Read the rows found, with their scores.
+   paimon::ScanContextBuilder scan_builder(table_path);
+   scan_builder.SetGlobalIndexResult(result);
+   PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<paimon::ScanContext> scan_context,
+                          scan_builder.Finish());
+   PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<paimon::TableScan> scan,
+                          paimon::TableScan::Create(std::move(scan_context)));
+   PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<paimon::Plan> data_plan, scan->CreatePlan());
+
+   paimon::ReadContextBuilder read_builder(table_path);
+   read_builder.SetReadFieldNames({"id", "content", "_INDEX_SCORE"});
+   PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<paimon::ReadContext> read_context,
+                          read_builder.Finish());
+   PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<paimon::TableRead> read,
+                          paimon::TableRead::Create(std::move(read_context)));
+   PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<paimon::BatchReader> reader,
+                          read->CreateReader(data_plan->Splits()));
+
+The result holds the ``limit`` highest-scoring rows among the candidates of all index shards, and
+a read of the rows returns their scores in the ``_INDEX_SCORE`` field. Each shard returns at most
+``limit`` candidates, chosen by its index, so ties are broken by ascending row id among the
+candidates only: which of the rows tied at the ``limit``-th score of a shard become candidates is up
+to its index.
+
+- ``SetPartitionFilter()`` restricts the search to some partitions, as in ``ScanContextBuilder``.
+- 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
+  options may be set, and it must agree with an explicit ``scan.mode``: ``from-snapshot`` needs
+  ``scan.snapshot-id`` or ``scan.tag-name``, ``from-snapshot-full`` needs ``scan.snapshot-id``,
+  ``from-timestamp`` needs ``scan.timestamp`` or ``scan.timestamp-millis``, and ``latest-full`` and
+  ``latest`` take none of them. The search still uses the latest schema of the table, not the
+  schema of that snapshot.
+- A shard whose indexed field is the column or that carries it as an extra field is searched.
+  When shards overlap, each row is ranked by one of them: an index whose indexed field is the
+  column comes first, then the order of the index type and fields, then the start and end of the
+  shard.
+- With deletion vectors, only the rows live in that snapshot are ranked, as described for
+  ``GlobalIndexScan`` above.
+- The shards are searched in parallel on the executor given to ``WithExecutor()``, or else on one
+  with ``global-index.thread-num`` threads, and each returns at most ``limit`` rows.
+- If Paimon C++ is built without the backend of a ``full-text`` or ``lucene-fts`` index, the
+  search fails instead of skipping the index and leaving its rows out of the result.
+- A column with a primary-key full-text index (``pk-full-text.index.columns``) cannot be searched
+  yet.
+
+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
+it. For a search of ``scan.tag-name``, pin the data scan with the same tag instead, as the tag may
+outlive the snapshot.
+
+.. code-block:: cpp
+
+   PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<paimon::FullTextScan> search_scan,
+                          search_builder.NewFullTextScan());
+   PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<paimon::FullTextScan::Plan> plan,
+                          search_scan->CreatePlan());
+   PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<paimon::FullTextRead> search_read,
+                          search_builder.NewFullTextRead());
+   PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<paimon::ScoredGlobalIndexResult> result,
+                          search_read->Read(*plan));
+
+   paimon::ScanContextBuilder scan_builder(table_path);
+   scan_builder.SetGlobalIndexResult(result);
+   if (plan->SnapshotId()) {
+       scan_builder.AddOption(paimon::Options::SCAN_MODE, "from-snapshot")
+           .AddOption(paimon::Options::SCAN_SNAPSHOT_ID, std::to_string(*plan->SnapshotId()));
+   }
+
+``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.
+
+.. code-block:: cpp
+
+   // Planning process.
+   PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<paimon::FullTextScan> scan,
+                          search_builder.NewFullTextScan());
+   PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<paimon::FullTextScan::Plan> plan, scan->CreatePlan());
+   for (const auto& split : plan->Splits()) {
+       PAIMON_ASSIGN_OR_RAISE(std::string bytes,
+                              paimon::FullTextSearchSplit::Serialize(split, pool));
+       // Send `bytes` to a searching process.
+   }
+
+   // Searching process.
+   std::vector<std::shared_ptr<paimon::FullTextSearchSplit>> splits;
+   PAIMON_ASSIGN_OR_RAISE(
+       std::shared_ptr<paimon::FullTextSearchSplit> split,
+       paimon::FullTextSearchSplit::Deserialize(bytes.data(), bytes.size(), pool));
+   splits.push_back(std::move(split));
+   PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<paimon::FullTextRead> read,
+                          search_builder.NewFullTextRead());
+   PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<paimon::ScoredGlobalIndexResult> result,
+                          read->Read(splits));
+
 Vector Index (Lumina)
 ---------------------
 

include/paimon/defs.h

@@ -614,6 +614,15 @@ struct PAIMON_EXPORT Options {
     /// "global-index.external-path" - Global index root directory, if not set, the global index
     /// files will be stored under the index directory.
     static const char GLOBAL_INDEX_EXTERNAL_PATH[];
+    /// "global-index.search-mode" - Fallback search mode for global index queries, used when the
+    /// search mode of the index family is not set: "fast", "full" or "detail". No default value.
+    /// Paimon C++ reads it only for full-text search so far.
+    static const char GLOBAL_INDEX_SEARCH_MODE[];
+    /// "full-text-index.search-mode" - Search mode for full-text index queries: "fast" searches
+    /// only the rows that an index covers, while "full" and "detail" also search 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 in another mode fails.
+    static const char FULL_TEXT_INDEX_SEARCH_MODE[];
     /// "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/global_index/global_indexer.h

@@ -83,6 +83,13 @@ class PAIMON_EXPORT GlobalIndexer {
     virtual bool SupportsCheckpoint() const {
         return false;
     }
+
+    /// Whether readers of this indexer answer `GlobalIndexReader::VisitFullTextSearch()`, as Java
+    /// `GlobalIndexerFactory#supportsFullTextSearch` says. `FullTextSearchBuilder` searches only
+    /// the index files whose indexer supports it.
+    virtual bool SupportsFullTextSearch() const {
+        return false;
+    }
 };
 
 }  // namespace paimon

include/paimon/table/source/full_text_read.h

@@ -0,0 +1,62 @@
+/*
+ * 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.
+ */
+
+#pragma once
+
+#include <memory>
+#include <vector>
+
+#include "paimon/global_index/global_index_result.h"
+#include "paimon/result.h"
+#include "paimon/table/source/full_text_scan.h"
+#include "paimon/table/source/full_text_search_split.h"
+#include "paimon/visibility.h"
+
+namespace paimon {
+/// Searches the splits of a table-level full-text search, aligned with Java
+/// `org.apache.paimon.table.source.FullTextRead`. Create one with
+/// `FullTextSearchBuilder::NewFullTextRead()`.
+class PAIMON_EXPORT FullTextRead {
+ public:
+    virtual ~FullTextRead() = default;
+
+    /// Searches the splits of `plan`; see `Read(splits)`.
+    Result<std::shared_ptr<ScoredGlobalIndexResult>> Read(const FullTextScan::Plan& plan) {
+        return Read(plan.Splits());
+    }
+
+    /// Searches `splits` and returns the `limit` highest-scoring rows among the candidates that
+    /// they return. The result holds global row ids and iterates them in ascending order; pass it
+    /// to `ScanContextBuilder::SetGlobalIndexResult()` to read the rows, whose scores the read
+    /// returns in the `_INDEX_SCORE` field.
+    ///
+    /// The splits are searched in parallel, and each one returns at most `limit` candidates, chosen
+    /// by its index. Ties among the candidates are broken by ascending row id, but which of the
+    /// rows tied at the `limit`-th score of a split become candidates is up to its index. A split
+    /// 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.
+    /// @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.
+    virtual Result<std::shared_ptr<ScoredGlobalIndexResult>> Read(
+        const std::vector<std::shared_ptr<FullTextSearchSplit>>& splits) = 0;
+};
+
+}  // namespace paimon

include/paimon/table/source/full_text_scan.h

@@ -0,0 +1,86 @@
+/*
+ * 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.
+ */
+
+#pragma once
+
+#include <cstdint>
+#include <memory>
+#include <optional>
+#include <utility>
+#include <vector>
+
+#include "paimon/result.h"
+#include "paimon/table/source/full_text_search_split.h"
+#include "paimon/visibility.h"
+
+namespace paimon {
+/// Plans a table-level full-text search into splits, aligned with Java
+/// `org.apache.paimon.table.source.FullTextScan`. Create one with
+/// `FullTextSearchBuilder::NewFullTextScan()`.
+class PAIMON_EXPORT FullTextScan {
+ public:
+    /// The splits of a full-text search and the snapshot they were planned from.
+    class Plan {
+     public:
+        Plan(const std::optional<int64_t>& snapshot_id,
+             std::vector<std::shared_ptr<FullTextSearchSplit>> splits)
+            : snapshot_id_(snapshot_id), splits_(std::move(splits)) {}
+
+        /// @return The splits to search. They can be read in this process or serialized and read
+        ///         in others.
+        const std::vector<std::shared_ptr<FullTextSearchSplit>>& Splits() const {
+            return splits_;
+        }
+
+        /// @return The id of the snapshot the plan was built from, or nullopt if the table has no
+        ///         snapshot.
+        const std::optional<int64_t>& SnapshotId() const {
+            return snapshot_id_;
+        }
+
+     private:
+        std::optional<int64_t> snapshot_id_;
+        std::vector<std::shared_ptr<FullTextSearchSplit>> splits_;
+    };
+
+    virtual ~FullTextScan() = default;
+
+    /// Plans the search on the snapshot that the options select, as Java
+    /// `TimeTravelUtil#tryTravelOrLatest` 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. As in Java, at most one of these
+    /// options may be set, and it must agree with an explicit `scan.mode`: `from-snapshot` needs
+    /// `scan.snapshot-id` or `scan.tag-name`, `from-snapshot-full` needs `scan.snapshot-id`,
+    /// `from-timestamp` needs `scan.timestamp` or `scan.timestamp-millis`, and `latest-full` and
+    /// `latest` take none of them.
+    ///
+    /// Each index shard of the searched column becomes one split. The scan considers only the
+    /// index files that are in the partitions of the partition filter, that index the column as
+    /// their indexed field or as one of their extra fields, and whose indexer supports full-text
+    /// search (`GlobalIndexer::SupportsFullTextSearch()`). The plan fails if Paimon C++ is built
+    /// 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.
+    /// @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.
+    virtual Result<std::shared_ptr<Plan>> CreatePlan() = 0;
+};
+
+}  // namespace paimon