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, 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