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 as
Java ScoredGlobalIndexResult#merge and #topK do: a duplicate row id
keeps its first score, and only the k highest-scoring rows are kept.
Scores follow Java Double#compare ordering, and ties among merged rows
are broken by ascending row id. Which rows tied at a shard's cutoff the
shard returns is still up to its engine, so the API comment and user
guide promise row id order only among rows the shards return.

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 that range before shifting them. As in Java
DataEvolutionFullTextScan#chooseIndexRanges, GlobalIndexScan ranks a row
covered by overlapping index ranges only in the first of them, so the
top k rows of each shard include all of its rows that score above the
global k-th score. Each reader is given the last row that earlier ranges
rank, which does not overflow when a range ends at INT64_MAX. A shard
with no row left to rank is not searched, but it still rejects a
non-positive limit.

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, as Java
+Paimon does. 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

@@ -83,7 +83,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

@@ -19,13 +19,28 @@
 
 #include "paimon/common/global_index/offset_global_index_reader.h"
 
+#include <limits>
 #include <utility>
+#include <vector>
+
+#include "paimon/global_index/bitmap_scored_global_index_result.h"
+#include "paimon/utils/roaring_bitmap64.h"
 
 namespace paimon {
+namespace {
+std::shared_ptr<ScoredGlobalIndexResult> EmptyScoredResult() {
+    return std::make_shared<BitmapScoredGlobalIndexResult>(RoaringBitmap64(), std::vector<float>());
+}
+}  // namespace
+
+OffsetGlobalIndexReader::OffsetGlobalIndexReader(std::shared_ptr<GlobalIndexReader>&& wrapped,
+                                                 int64_t offset, int64_t to)
+    : OffsetGlobalIndexReader(std::move(wrapped), offset, to,
+                              /*skip_to=*/std::numeric_limits<int64_t>::min()) {}
 
 OffsetGlobalIndexReader::OffsetGlobalIndexReader(std::shared_ptr<GlobalIndexReader>&& wrapped,
-                                                 int64_t offset)
-    : wrapped_(std::move(wrapped)), offset_(offset) {}
+                                                 int64_t offset, int64_t to, int64_t skip_to)
+    : wrapped_(std::move(wrapped)), offset_(offset), to_(to), skip_to_(skip_to) {}
 
 Result<std::shared_ptr<GlobalIndexResult>> OffsetGlobalIndexReader::VisitIsNotNull() {
     PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<GlobalIndexResult> result, wrapped_->VisitIsNotNull());
@@ -137,11 +152,34 @@ 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.
     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);
+    if (full_text_search) {
+        if (full_text_search->limit <= 0) {
+            return Status::Invalid("full-text search requires a positive limit");
+        }
+        // Rows up to `skip_to_` are ranked by a reader over an overlapping row range.
+        if (skip_to_ >= to_) {
+            return EmptyScoredResult();
+        }
+        bool skips_rows = skip_to_ >= offset_;
+        if (full_text_search->include_row_ids) {
+            local_search = full_text_search->OffsetRange(offset_, to_);
+            if (skips_rows) {
+                local_search->include_row_ids->RemoveRange(0, skip_to_ - offset_ + 1);
+            }
+            if (local_search->include_row_ids->IsEmpty()) {
+                return EmptyScoredResult();
+            }
+        } else if (skips_rows) {
+            // `local_to` may be the largest int64 value, so add it apart from the half-open range.
+            int64_t local_to = to_ - offset_;
+            RoaringBitmap64 local_row_ids;
+            local_row_ids.AddRange(skip_to_ - offset_ + 1, local_to);
+            local_row_ids.Add(local_to);
+            local_search = std::make_shared<FullTextSearch>(
+                full_text_search->field_name, full_text_search->query, full_text_search->limit,
+                std::move(local_row_ids));
+        }
     }
     PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<ScoredGlobalIndexResult> result,
                            wrapped_->VisitFullTextSearch(local_search));

src/paimon/common/global_index/offset_global_index_reader.h

@@ -31,11 +31,22 @@ namespace paimon {
 /// results. This is used to convert local row IDs into global row IDs.
 class OffsetGlobalIndexReader : public GlobalIndexReader {
  public:
-    /// Constructs an OffsetGlobalIndexReader.
+    /// Constructs an OffsetGlobalIndexReader whose full-text searches rank all rows it covers.
     /// @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);
+
+    /// Constructs an OffsetGlobalIndexReader whose full-text searches rank rows in [`offset`,
+    /// `to`], excluding rows up to `skip_to` (inclusive) assigned to earlier readers.
+    /// No row is ranked if `skip_to` >= `to`. Storing the inclusive end avoids computing
+    /// INT64_MAX + 1 when an earlier range covers the largest row id.
+    /// @param skip_to The last global row id already assigned to an earlier reader.
+    OffsetGlobalIndexReader(std::shared_ptr<GlobalIndexReader>&& wrapped, int64_t offset,
+                            int64_t to, int64_t skip_to);
 
     Result<std::shared_ptr<GlobalIndexResult>> VisitIsNotNull() override;
     Result<std::shared_ptr<GlobalIndexResult>> VisitIsNull() override;
@@ -78,6 +89,8 @@ class OffsetGlobalIndexReader : public GlobalIndexReader {
  private:
     std::shared_ptr<GlobalIndexReader> wrapped_;
     int64_t offset_;
+    int64_t to_;
+    int64_t skip_to_;
 };
 
 }  // namespace paimon