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

UnionGlobalIndexReader::VisitFullTextSearch ORed shard results, so a
search with limit k could return up to shards x k rows, and it failed
when two shards returned the same row id. It now merges shard results,
where a duplicate row id keeps its first score, and keeps only the k
highest-scoring rows. Scores are ordered with NaN highest and -0.0 below
+0.0, and ties among merged rows are broken by ascending row id. A
shard's engine still picks which rows tied at its cutoff score it
returns, so the API comment and user guide promise row id order only
among rows the shards return. A null search or a non-positive limit is
now rejected.

OffsetGlobalIndexReader shifted include row ids by the shard offset
without clipping them, so row ids before a shard became negative local
ids and row ids past it were kept. It now takes the end of its row range
and clips include row ids to it.

GlobalIndexScan now ranks a row covered by overlapping index ranges
only in the first of them in (from, to) order, so the top k rows of
each shard include all of its rows that score above the global k-th
score. UnionGlobalIndexReader receives the rows each shard ranks, limits
the shard's include row ids to them and skips a shard with none left.

UnionGlobalIndexReader trims a copy of the include row ids, which costs
time in proportion to the containers of the include set rather than to
the rows of the range. FullTextSearch::OffsetRange shifts an include set
of consecutive row ids as one range instead of one row id at a time.

Author: SteNicholas

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