fix(global-index): apply global top-k to full-text search across shards

docs/source/user_guide/global_index.rst

@@ -168,9 +168,12 @@ index shard, the engine ranks matching rows by BM25 score and returns at most ``
 the highest scores. ``include_row_ids`` holds global row ids and restricts the rows that are
 ranked; an empty set matches no rows.
 
-For both ``full-text`` and ``lucene-fts``, searches across multiple shards currently return the
-union of the shard candidates, which can exceed ``limit``. Applying a final global top-k is tracked
-in `issue #402 <https://github.com/apache/paimon-cpp/issues/402>`_.
+For both ``full-text`` and ``lucene-fts``, a search across multiple index shards keeps the
+``limit`` highest-scoring rows returned by the shards, breaking ties by ascending row id. Which rows
+tied at a shard's cutoff score the shard returns is up to its engine. A row in overlapping index
+ranges is ranked only by the range that starts first, or ends first if both start at the same row.
+A shard ignores the ``include_row_ids`` outside the rows it ranks and is not searched if none
+remain. The result iterates rows by ascending row id, not by score.
 
 **Configuration:**
 

include/paimon/global_index/global_index_reader.h

@@ -45,9 +45,12 @@ class PAIMON_EXPORT GlobalIndexReader : public FunctionVisitor<std::shared_ptr<G
     virtual Result<std::shared_ptr<ScoredGlobalIndexResult>> VisitVectorSearch(
         const std::shared_ptr<VectorSearch>& vector_search) = 0;
 
-    /// VisitFullTextSearch performs full text search and returns row ids with relevance scores.
-    /// Each index shard returns at most `full_text_search->limit` rows. Union readers currently
-    /// retain all shard candidates and can return more than that limit.
+    /// VisitFullTextSearch performs full text search and returns at most `full_text_search->limit`
+    /// row ids with the highest relevance scores, iterated by ascending row id rather than score.
+    /// A search over multiple index shards keeps the `limit` highest-scoring rows returned by the
+    /// shards, breaking ties by ascending row id; which rows tied at a shard's cutoff score the
+    /// shard returns is up to its index engine. A row in overlapping index ranges is ranked only by
+    /// the range that starts first, or ends first if both start at the same row.
     /// @warning `VisitFullTextSearch` may return error status when it is incorrectly invoked (e.g.,
     /// BTreeGlobalIndexReader call `VisitFullTextSearch`) or the query is invalid.
     virtual Result<std::shared_ptr<ScoredGlobalIndexResult>> VisitFullTextSearch(

include/paimon/predicate/full_text_search.h

@@ -17,6 +17,7 @@
  */
 
 #pragma once
+#include <algorithm>
 #include <cstdint>
 #include <functional>
 #include <map>
@@ -49,9 +50,23 @@ struct PAIMON_EXPORT FullTextSearch {
         }
         const RoaringBitmap64& global_row_ids = include_row_ids.value();
         RoaringBitmap64 local_row_ids;
-        for (auto iter = global_row_ids.EqualOrLarger(from), end = global_row_ids.End();
-             iter != end && *iter <= to; ++iter) {
-            local_row_ids.Add(*iter - from);
+        auto iter = global_row_ids.EqualOrLarger(from);
+        auto end = global_row_ids.End();
+        if (iter != end && *iter <= to) {
+            // Shift one run of consecutive row ids as a range, so that converting a range of rows
+            // to rank does not visit each row.
+            int64_t first = *global_row_ids.Begin();
+            int64_t last = first + (global_row_ids.Cardinality() - 1);
+            if (last == RoaringBitmap64::MAX_VALUE ||
+                global_row_ids.EqualOrLarger(last + 1) == end) {
+                int64_t local_to = std::min(last, to) - from;
+                local_row_ids.AddRange(*iter - from, local_to);
+                local_row_ids.Add(local_to);
+            } else {
+                for (; iter != end && *iter <= to; ++iter) {
+                    local_row_ids.Add(*iter - from);
+                }
+            }
         }
         return std::make_shared<FullTextSearch>(field_name, query, limit, std::move(local_row_ids));
     }
@@ -83,7 +98,8 @@ struct PAIMON_EXPORT FullTextSearch {
     /// and rejects `boost` queries and a `fuzziness` other than 0. An invalid query is reported as
     /// an error `Status`.
     std::string query;
-    /// Maximum number of highest-scoring rows requested from each index shard; must be positive.
+    /// Maximum number of highest-scoring rows to return; must be positive. Each index shard is
+    /// asked for this many rows before the shard results are merged.
     int32_t limit;
     /// The row ids to search within, aligned with Java `includeRowIds`. The row id domain is
     /// determined by the reader: global row ids for GlobalIndexReader and physical file-local row

src/paimon/common/global_index/offset_global_index_reader.cpp

@@ -24,8 +24,8 @@
 namespace paimon {
 
 OffsetGlobalIndexReader::OffsetGlobalIndexReader(std::shared_ptr<GlobalIndexReader>&& wrapped,
-                                                 int64_t offset)
-    : wrapped_(std::move(wrapped)), offset_(offset) {}
+                                                 int64_t offset, int64_t to)
+    : wrapped_(std::move(wrapped)), offset_(offset), to_(to) {}
 
 Result<std::shared_ptr<GlobalIndexResult>> OffsetGlobalIndexReader::VisitIsNotNull() {
     PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<GlobalIndexResult> result, wrapped_->VisitIsNotNull());
@@ -137,11 +137,10 @@ Result<std::shared_ptr<ScoredGlobalIndexResult>> OffsetGlobalIndexReader::VisitV
 
 Result<std::shared_ptr<ScoredGlobalIndexResult>> OffsetGlobalIndexReader::VisitFullTextSearch(
     const std::shared_ptr<FullTextSearch>& full_text_search) {
-    // Convert global include row ids to local row ids. The shard end is unknown here, so row ids
-    // past the shard are kept; they match no local row.
+    // Convert global include row ids to local row ids, dropping those outside [offset_, to_].
     std::shared_ptr<FullTextSearch> local_search = full_text_search;
     if (full_text_search && full_text_search->include_row_ids) {
-        local_search = full_text_search->OffsetRange(offset_, RoaringBitmap64::MAX_VALUE);
+        local_search = full_text_search->OffsetRange(offset_, to_);
     }
     PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<ScoredGlobalIndexResult> result,
                            wrapped_->VisitFullTextSearch(local_search));

src/paimon/common/global_index/offset_global_index_reader.h

@@ -34,8 +34,11 @@ class OffsetGlobalIndexReader : public GlobalIndexReader {
     /// Constructs an OffsetGlobalIndexReader.
     /// @param wrapped The inner reader to delegate queries to. Its results are expected to
     ///                contain local row ids.
-    /// @param offset  The offset to add to each row id in the results.
-    OffsetGlobalIndexReader(std::shared_ptr<GlobalIndexReader>&& wrapped, int64_t offset);
+    /// @param offset  The first global row id covered by the wrapped reader, added to each row id
+    ///                in the results.
+    /// @param to      The last global row id (inclusive) covered by the wrapped reader.
+    OffsetGlobalIndexReader(std::shared_ptr<GlobalIndexReader>&& wrapped, int64_t offset,
+                            int64_t to);
 
     Result<std::shared_ptr<GlobalIndexResult>> VisitIsNotNull() override;
     Result<std::shared_ptr<GlobalIndexResult>> VisitIsNull() override;
@@ -78,6 +81,7 @@ class OffsetGlobalIndexReader : public GlobalIndexReader {
  private:
     std::shared_ptr<GlobalIndexReader> wrapped_;
     int64_t offset_;
+    int64_t to_;
 };
 
 }  // namespace paimon