This is an automated email from the ASF dual-hosted git repository.
lxy-9602 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-cpp.git
The following commit(s) were added to refs/heads/main by this push:
new fff5a580 fix(global-index): apply global top-k to full-text search
across shards (#431)
fff5a580 is described below
commit fff5a580e6dad7c55c256eb7fb121d8c28f6d976
Author: Nicholas Jiang <[email protected]>
AuthorDate: Mon Oct 19 17:01:18 2026 +0700
fix(global-index): apply global top-k to full-text search across shards
(#431)
---
docs/source/user_guide/global_index.rst | 9 +-
include/paimon/global_index/global_index_reader.h | 9 +-
include/paimon/predicate/full_text_search.h | 24 +-
.../global_index/offset_global_index_reader.cpp | 9 +-
.../global_index/offset_global_index_reader.h | 8 +-
.../offset_global_index_reader_test.cpp | 67 +++--
.../global_index/union_global_index_reader.cpp | 179 ++++++++++++-
.../global_index/union_global_index_reader.h | 27 ++
.../union_global_index_reader_test.cpp | 277 +++++++++++++++++++--
.../common/predicate/full_text_search_test.cpp | 24 ++
.../core/global_index/global_index_scan_impl.cpp | 26 +-
test/inte/global_index_test.cpp | 138 ++++++++++
12 files changed, 712 insertions(+), 85 deletions(-)
diff --git a/docs/source/user_guide/global_index.rst
b/docs/source/user_guide/global_index.rst
index 956d28db..eed71866 100644
--- a/docs/source/user_guide/global_index.rst
+++ b/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:**
diff --git a/include/paimon/global_index/global_index_reader.h
b/include/paimon/global_index/global_index_reader.h
index 6155759b..94d9ecef 100644
--- a/include/paimon/global_index/global_index_reader.h
+++ b/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(
diff --git a/include/paimon/predicate/full_text_search.h
b/include/paimon/predicate/full_text_search.h
index 1adb0875..46e8024d 100644
--- a/include/paimon/predicate/full_text_search.h
+++ b/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
diff --git a/src/paimon/common/global_index/offset_global_index_reader.cpp
b/src/paimon/common/global_index/offset_global_index_reader.cpp
index 70b6afc6..b66a29b4 100644
--- a/src/paimon/common/global_index/offset_global_index_reader.cpp
+++ b/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));
diff --git a/src/paimon/common/global_index/offset_global_index_reader.h
b/src/paimon/common/global_index/offset_global_index_reader.h
index 08229466..de86fdee 100644
--- a/src/paimon/common/global_index/offset_global_index_reader.h
+++ b/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
diff --git a/src/paimon/common/global_index/offset_global_index_reader_test.cpp
b/src/paimon/common/global_index/offset_global_index_reader_test.cpp
index 39fec0b1..83d6079b 100644
--- a/src/paimon/common/global_index/offset_global_index_reader_test.cpp
+++ b/src/paimon/common/global_index/offset_global_index_reader_test.cpp
@@ -19,6 +19,7 @@
#include "paimon/common/global_index/offset_global_index_reader.h"
+#include <limits>
#include <string>
#include <vector>
@@ -65,7 +66,7 @@ TEST_F(OffsetGlobalIndexReaderTest, TestVisitEqualWithOffset)
{
auto fake_reader = std::make_shared<MockGlobalIndexReader>();
fake_reader->SetDefaultResult({0, 1, 3});
- auto offset_reader =
std::make_shared<OffsetGlobalIndexReader>(fake_reader, 100);
+ auto offset_reader =
std::make_shared<OffsetGlobalIndexReader>(fake_reader, 100, /*to=*/199);
Literal literal_5(5);
ASSERT_OK_AND_ASSIGN(auto result, offset_reader->VisitEqual(literal_5));
@@ -77,7 +78,7 @@ TEST_F(OffsetGlobalIndexReaderTest,
TestVisitIsNotNullWithOffset) {
auto fake_reader = std::make_shared<MockGlobalIndexReader>();
fake_reader->SetDefaultResult({0, 2, 4, 6});
- auto offset_reader =
std::make_shared<OffsetGlobalIndexReader>(fake_reader, 50);
+ auto offset_reader =
std::make_shared<OffsetGlobalIndexReader>(fake_reader, 50, /*to=*/149);
ASSERT_OK_AND_ASSIGN(auto result, offset_reader->VisitIsNotNull());
// row ids {0, 2, 4, 6} + offset 50 -> {50, 52, 54, 56}
@@ -88,7 +89,7 @@ TEST_F(OffsetGlobalIndexReaderTest,
TestVisitIsNullWithOffset) {
auto fake_reader = std::make_shared<MockGlobalIndexReader>();
fake_reader->SetDefaultResult({1, 3});
- auto offset_reader =
std::make_shared<OffsetGlobalIndexReader>(fake_reader, 200);
+ auto offset_reader =
std::make_shared<OffsetGlobalIndexReader>(fake_reader, 200, /*to=*/299);
ASSERT_OK_AND_ASSIGN(auto result, offset_reader->VisitIsNull());
// row ids {1, 3} + offset 200 -> {201, 203}
@@ -99,7 +100,7 @@ TEST_F(OffsetGlobalIndexReaderTest,
TestVisitLessThanWithOffset) {
auto fake_reader = std::make_shared<MockGlobalIndexReader>();
fake_reader->SetDefaultResult({0, 1});
- auto offset_reader =
std::make_shared<OffsetGlobalIndexReader>(fake_reader, 10);
+ auto offset_reader =
std::make_shared<OffsetGlobalIndexReader>(fake_reader, 10, /*to=*/109);
Literal literal_5(5);
ASSERT_OK_AND_ASSIGN(auto result, offset_reader->VisitLessThan(literal_5));
@@ -111,7 +112,7 @@ TEST_F(OffsetGlobalIndexReaderTest,
TestVisitGreaterOrEqualWithOffset) {
auto fake_reader = std::make_shared<MockGlobalIndexReader>();
fake_reader->SetDefaultResult({3, 4, 5});
- auto offset_reader =
std::make_shared<OffsetGlobalIndexReader>(fake_reader, 1000);
+ auto offset_reader =
std::make_shared<OffsetGlobalIndexReader>(fake_reader, 1000, /*to=*/1099);
Literal literal_10(10);
ASSERT_OK_AND_ASSIGN(auto result,
offset_reader->VisitGreaterOrEqual(literal_10));
@@ -123,7 +124,7 @@ TEST_F(OffsetGlobalIndexReaderTest, TestVisitInWithOffset) {
auto fake_reader = std::make_shared<MockGlobalIndexReader>();
fake_reader->SetDefaultResult({0, 2});
- auto offset_reader =
std::make_shared<OffsetGlobalIndexReader>(fake_reader, 5);
+ auto offset_reader =
std::make_shared<OffsetGlobalIndexReader>(fake_reader, 5, /*to=*/104);
std::vector<Literal> literals = {Literal(1), Literal(3)};
ASSERT_OK_AND_ASSIGN(auto result, offset_reader->VisitIn(literals));
@@ -135,7 +136,7 @@ TEST_F(OffsetGlobalIndexReaderTest,
TestVisitNotInWithOffset) {
auto fake_reader = std::make_shared<MockGlobalIndexReader>();
fake_reader->SetDefaultResult({1, 3, 5});
- auto offset_reader =
std::make_shared<OffsetGlobalIndexReader>(fake_reader, 20);
+ auto offset_reader =
std::make_shared<OffsetGlobalIndexReader>(fake_reader, 20, /*to=*/119);
std::vector<Literal> literals = {Literal(2)};
ASSERT_OK_AND_ASSIGN(auto result, offset_reader->VisitNotIn(literals));
@@ -147,7 +148,7 @@ TEST_F(OffsetGlobalIndexReaderTest,
TestVisitNotEqualWithOffset) {
auto fake_reader = std::make_shared<MockGlobalIndexReader>();
fake_reader->SetDefaultResult({0, 2, 4});
- auto offset_reader =
std::make_shared<OffsetGlobalIndexReader>(fake_reader, 7);
+ auto offset_reader =
std::make_shared<OffsetGlobalIndexReader>(fake_reader, 7, /*to=*/106);
Literal literal_1(1);
ASSERT_OK_AND_ASSIGN(auto result, offset_reader->VisitNotEqual(literal_1));
@@ -159,7 +160,7 @@ TEST_F(OffsetGlobalIndexReaderTest,
TestVisitLessOrEqualWithOffset) {
auto fake_reader = std::make_shared<MockGlobalIndexReader>();
fake_reader->SetDefaultResult({0, 1, 2});
- auto offset_reader =
std::make_shared<OffsetGlobalIndexReader>(fake_reader, 30);
+ auto offset_reader =
std::make_shared<OffsetGlobalIndexReader>(fake_reader, 30, /*to=*/129);
Literal literal_3(3);
ASSERT_OK_AND_ASSIGN(auto result,
offset_reader->VisitLessOrEqual(literal_3));
@@ -171,7 +172,7 @@ TEST_F(OffsetGlobalIndexReaderTest,
TestVisitGreaterThanWithOffset) {
auto fake_reader = std::make_shared<MockGlobalIndexReader>();
fake_reader->SetDefaultResult({4, 5});
- auto offset_reader =
std::make_shared<OffsetGlobalIndexReader>(fake_reader, 15);
+ auto offset_reader =
std::make_shared<OffsetGlobalIndexReader>(fake_reader, 15, /*to=*/114);
Literal literal_3(3);
ASSERT_OK_AND_ASSIGN(auto result,
offset_reader->VisitGreaterThan(literal_3));
@@ -183,7 +184,7 @@ TEST_F(OffsetGlobalIndexReaderTest, TestZeroOffset) {
auto fake_reader = std::make_shared<MockGlobalIndexReader>();
fake_reader->SetDefaultResult({0, 1, 2});
- auto offset_reader =
std::make_shared<OffsetGlobalIndexReader>(fake_reader, 0);
+ auto offset_reader =
std::make_shared<OffsetGlobalIndexReader>(fake_reader, 0, /*to=*/99);
ASSERT_OK_AND_ASSIGN(auto result, offset_reader->VisitIsNotNull());
// row ids {0, 1, 2} + offset 0 -> {0, 1, 2}
@@ -194,7 +195,7 @@ TEST_F(OffsetGlobalIndexReaderTest,
TestEmptyResultWithOffset) {
auto fake_reader = std::make_shared<MockGlobalIndexReader>();
fake_reader->SetDefaultResult({});
- auto offset_reader =
std::make_shared<OffsetGlobalIndexReader>(fake_reader, 100);
+ auto offset_reader =
std::make_shared<OffsetGlobalIndexReader>(fake_reader, 100, /*to=*/199);
ASSERT_OK_AND_ASSIGN(auto result, offset_reader->VisitIsNotNull());
// empty result + offset 100 -> still empty
@@ -203,14 +204,14 @@ TEST_F(OffsetGlobalIndexReaderTest,
TestEmptyResultWithOffset) {
TEST_F(OffsetGlobalIndexReaderTest, TestIsThreadSafeDelegated) {
auto fake_reader = std::make_shared<MockGlobalIndexReader>();
- auto offset_reader =
std::make_shared<OffsetGlobalIndexReader>(fake_reader, 100);
+ auto offset_reader =
std::make_shared<OffsetGlobalIndexReader>(fake_reader, 100, /*to=*/199);
// MockGlobalIndexReader returns true for IsThreadSafe
ASSERT_TRUE(offset_reader->IsThreadSafe());
}
TEST_F(OffsetGlobalIndexReaderTest, TestGetIndexTypeDelegated) {
auto fake_reader = std::make_shared<MockGlobalIndexReader>();
- auto offset_reader =
std::make_shared<OffsetGlobalIndexReader>(fake_reader, 100);
+ auto offset_reader =
std::make_shared<OffsetGlobalIndexReader>(fake_reader, 100, /*to=*/199);
ASSERT_EQ(offset_reader->GetIndexType(), "fake");
}
@@ -218,7 +219,7 @@ TEST_F(OffsetGlobalIndexReaderTest,
TestVisitFullTextSearchWithOffset) {
auto fake_reader = std::make_shared<MockGlobalIndexReader>();
fake_reader->SetScoredResult({0, 3, 5}, {0.5f, 0.9f, 0.1f});
- auto offset_reader =
std::make_shared<OffsetGlobalIndexReader>(fake_reader, 10);
+ auto offset_reader =
std::make_shared<OffsetGlobalIndexReader>(fake_reader, 10, /*to=*/109);
ASSERT_OK_AND_ASSIGN(auto result,
offset_reader->VisitFullTextSearch(nullptr));
// row ids {0, 3, 5} + offset 10 -> {10, 13, 15}
@@ -227,23 +228,23 @@ TEST_F(OffsetGlobalIndexReaderTest,
TestVisitFullTextSearchWithOffset) {
TEST_F(OffsetGlobalIndexReaderTest,
TestVisitFullTextSearchConvertsIncludeRowIds) {
auto fake_reader = std::make_shared<MockGlobalIndexReader>();
- auto offset_reader =
std::make_shared<OffsetGlobalIndexReader>(fake_reader, 10);
+ auto offset_reader =
std::make_shared<OffsetGlobalIndexReader>(fake_reader, 10, /*to=*/19);
{
auto search =
std::make_shared<FullTextSearch>("f0",
R"({"match":{"query":"paimon"}})", /*limit=*/7,
- RoaringBitmap64::From({5l, 10l,
13l, 15l}));
+ RoaringBitmap64::From({5l, 10l,
13l, 19l, 20l, 150l}));
ASSERT_OK(offset_reader->VisitFullTextSearch(search));
auto captured = fake_reader->CapturedFullTextSearch();
ASSERT_TRUE(captured);
ASSERT_EQ(captured->field_name, "f0");
ASSERT_EQ(captured->query, R"({"match":{"query":"paimon"}})");
ASSERT_EQ(captured->limit, 7);
- ASSERT_EQ(captured->include_row_ids, RoaringBitmap64::From({0l, 3l,
5l}));
- ASSERT_EQ(search->include_row_ids, RoaringBitmap64::From({5l, 10l,
13l, 15l}));
+ ASSERT_EQ(captured->include_row_ids, RoaringBitmap64::From({0l, 3l,
9l}));
+ ASSERT_EQ(search->include_row_ids, RoaringBitmap64::From({5l, 10l,
13l, 19l, 20l, 150l}));
}
{
- auto search = std::make_shared<FullTextSearch>("f0",
R"({"match":{"query":"paimon"}})",
- /*limit=*/7,
RoaringBitmap64::From({5l}));
+ auto search = std::make_shared<FullTextSearch>(
+ "f0", R"({"match":{"query":"paimon"}})", /*limit=*/7,
RoaringBitmap64::From({5l, 20l}));
ASSERT_OK(offset_reader->VisitFullTextSearch(search));
ASSERT_TRUE(fake_reader->CapturedFullTextSearch()->include_row_ids);
ASSERT_TRUE(fake_reader->CapturedFullTextSearch()->include_row_ids->IsEmpty());
@@ -254,13 +255,25 @@ TEST_F(OffsetGlobalIndexReaderTest,
TestVisitFullTextSearchConvertsIncludeRowIds
ASSERT_OK(offset_reader->VisitFullTextSearch(search));
ASSERT_EQ(fake_reader->CapturedFullTextSearch(), search);
}
+ {
+ constexpr int64_t kMax = std::numeric_limits<int64_t>::max();
+ auto last_reader = std::make_shared<MockGlobalIndexReader>();
+ auto last_offset_reader =
+ std::make_shared<OffsetGlobalIndexReader>(last_reader, kMax - 9,
/*to=*/kMax);
+ auto search =
+ std::make_shared<FullTextSearch>("f0",
R"({"match":{"query":"paimon"}})", /*limit=*/7,
+ RoaringBitmap64::From({kMax - 5,
kMax - 2, kMax}));
+ ASSERT_OK(last_offset_reader->VisitFullTextSearch(search));
+ ASSERT_EQ(last_reader->CapturedFullTextSearch()->include_row_ids,
+ RoaringBitmap64::From({4l, 7l, 9l}));
+ }
}
TEST_F(OffsetGlobalIndexReaderTest, TestVisitVectorSearchWithOffset) {
auto fake_reader = std::make_shared<MockGlobalIndexReader>();
fake_reader->SetScoredResult({0, 2, 5}, {0.9f, 0.7f, 0.3f});
- auto offset_reader =
std::make_shared<OffsetGlobalIndexReader>(fake_reader, 100);
+ auto offset_reader =
std::make_shared<OffsetGlobalIndexReader>(fake_reader, 100, /*to=*/199);
ASSERT_OK_AND_ASSIGN(auto result,
offset_reader->VisitVectorSearch(nullptr));
// row ids {0, 2, 5} + offset 100 -> {100, 102, 105}, scores unchanged
@@ -269,7 +282,7 @@ TEST_F(OffsetGlobalIndexReaderTest,
TestVisitVectorSearchWithOffset) {
TEST_F(OffsetGlobalIndexReaderTest, TestVisitVectorSearchNotSupported) {
auto fake_reader = std::make_shared<MockGlobalIndexReader>();
- auto offset_reader =
std::make_shared<OffsetGlobalIndexReader>(fake_reader, 10);
+ auto offset_reader =
std::make_shared<OffsetGlobalIndexReader>(fake_reader, 10, /*to=*/109);
// Propagate the wrapped reader's unsupported-search error.
fake_reader->SetReturnError("vector search not supported");
ASSERT_NOK_WITH_MSG(offset_reader->VisitVectorSearch(nullptr), "vector
search not supported");
@@ -279,7 +292,7 @@ TEST_F(OffsetGlobalIndexReaderTest,
TestVisitStartsWithWithOffset) {
auto fake_reader = std::make_shared<MockGlobalIndexReader>();
fake_reader->SetDefaultResult({0, 1, 4});
- auto offset_reader =
std::make_shared<OffsetGlobalIndexReader>(fake_reader, 50);
+ auto offset_reader =
std::make_shared<OffsetGlobalIndexReader>(fake_reader, 50, /*to=*/149);
Literal prefix(FieldType::STRING, "abc", 3);
ASSERT_OK_AND_ASSIGN(auto result, offset_reader->VisitStartsWith(prefix));
@@ -291,7 +304,7 @@ TEST_F(OffsetGlobalIndexReaderTest,
TestVisitEndsWithWithOffset) {
auto fake_reader = std::make_shared<MockGlobalIndexReader>();
fake_reader->SetDefaultResult({2, 3});
- auto offset_reader =
std::make_shared<OffsetGlobalIndexReader>(fake_reader, 25);
+ auto offset_reader =
std::make_shared<OffsetGlobalIndexReader>(fake_reader, 25, /*to=*/124);
Literal suffix(FieldType::STRING, "xyz", 3);
ASSERT_OK_AND_ASSIGN(auto result, offset_reader->VisitEndsWith(suffix));
@@ -303,7 +316,7 @@ TEST_F(OffsetGlobalIndexReaderTest,
TestVisitContainsWithOffset) {
auto fake_reader = std::make_shared<MockGlobalIndexReader>();
fake_reader->SetDefaultResult({1, 5, 6});
- auto offset_reader =
std::make_shared<OffsetGlobalIndexReader>(fake_reader, 300);
+ auto offset_reader =
std::make_shared<OffsetGlobalIndexReader>(fake_reader, 300, /*to=*/399);
Literal literal(FieldType::STRING, "foo", 3);
ASSERT_OK_AND_ASSIGN(auto result, offset_reader->VisitContains(literal));
@@ -315,7 +328,7 @@ TEST_F(OffsetGlobalIndexReaderTest,
TestVisitLikeWithOffset) {
auto fake_reader = std::make_shared<MockGlobalIndexReader>();
fake_reader->SetDefaultResult({0, 3, 7});
- auto offset_reader =
std::make_shared<OffsetGlobalIndexReader>(fake_reader, 40);
+ auto offset_reader =
std::make_shared<OffsetGlobalIndexReader>(fake_reader, 40, /*to=*/139);
Literal pattern(FieldType::STRING, "ab%", 3);
ASSERT_OK_AND_ASSIGN(auto result, offset_reader->VisitLike(pattern));
diff --git a/src/paimon/common/global_index/union_global_index_reader.cpp
b/src/paimon/common/global_index/union_global_index_reader.cpp
index 27966f6b..bb11cae3 100644
--- a/src/paimon/common/global_index/union_global_index_reader.cpp
+++ b/src/paimon/common/global_index/union_global_index_reader.cpp
@@ -19,16 +19,98 @@
#include "paimon/common/global_index/union_global_index_reader.h"
+#include <algorithm>
+#include <cassert>
#include <future>
+#include <unordered_set>
#include <utility>
#include "paimon/common/executor/future.h"
+#include "paimon/common/utils/fields_comparator.h"
+#include "paimon/global_index/bitmap_scored_global_index_result.h"
+#include "paimon/predicate/full_text_search.h"
+#include "paimon/utils/roaring_bitmap64.h"
namespace paimon {
+namespace {
+std::shared_ptr<ScoredGlobalIndexResult> EmptyScoredResult() {
+ return std::make_shared<BitmapScoredGlobalIndexResult>(RoaringBitmap64(),
std::vector<float>());
+}
+
+RoaringBitmap64 RowIdsOf(const Range& range) {
+ // `range.to` may be the largest int64 value, so add it apart from the
half-open range.
+ RoaringBitmap64 row_ids;
+ row_ids.AddRange(range.from, range.to);
+ row_ids.Add(range.to);
+ return row_ids;
+}
+
+/// Returns the row ids of `row_ids` within `range`. Trimming a copy costs
time in proportion to
+/// the containers of `row_ids`, not to the rows or the span of `range`.
+RoaringBitmap64 RowIdsWithin(const RoaringBitmap64& row_ids, const Range&
range) {
+ RoaringBitmap64 result = row_ids;
+ result.RemoveRange(0, range.from);
+ if (range.to < RoaringBitmap64::MAX_VALUE) {
+ // RemoveRange leaves out its end, so the largest int64 value is
removed on its own.
+ result.RemoveRange(range.to + 1, RoaringBitmap64::MAX_VALUE);
+ result -= RoaringBitmap64::From({RoaringBitmap64::MAX_VALUE});
+ }
+ return result;
+}
+
+/// Returns whether `left` ranks before `right`: a higher score first, then a
lower row id. Scores
+/// are ordered as -infinity < -0.0 < +0.0 < +infinity < NaN.
+bool RanksBefore(const std::pair<int64_t, float>& left, const
std::pair<int64_t, float>& right) {
+ int32_t score_order = FieldsComparator::CompareFloatingPoint(left.second,
right.second);
+ if (score_order != 0) {
+ return score_order > 0;
+ }
+ return left.first < right.first;
+}
+
+/// Merges scored results and keeps the `limit` rows that rank first. A row id
in more than one
+/// result keeps its first score.
+Result<std::shared_ptr<ScoredGlobalIndexResult>> MergeTopK(
+ const std::vector<std::shared_ptr<ScoredGlobalIndexResult>>& results,
int32_t limit) {
+ std::unordered_set<int64_t> merged_row_ids;
+ std::vector<std::pair<int64_t, float>> rows;
+ for (const auto& result : results) {
+
PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<ScoredGlobalIndexResult::ScoredIterator>
iter,
+ result->CreateScoredIterator());
+ while (iter->HasNext()) {
+ std::pair<int64_t, float> row = iter->NextWithScore();
+ if (merged_row_ids.insert(row.first).second) {
+ rows.push_back(row);
+ }
+ }
+ }
+ auto top_k = static_cast<size_t>(limit);
+ if (rows.size() > top_k) {
+ std::nth_element(rows.begin(), rows.begin() + top_k, rows.end(),
RanksBefore);
+ rows.resize(top_k);
+ }
+ std::sort(rows.begin(), rows.end());
+ RoaringBitmap64 bitmap;
+ std::vector<float> scores;
+ scores.reserve(rows.size());
+ for (const auto& [row_id, score] : rows) {
+ bitmap.Add(row_id);
+ scores.push_back(score);
+ }
+ return std::make_shared<BitmapScoredGlobalIndexResult>(std::move(bitmap),
std::move(scores));
+}
+} // namespace
+
UnionGlobalIndexReader::UnionGlobalIndexReader(
- std::vector<std::shared_ptr<GlobalIndexReader>>&& readers,
- const std::shared_ptr<Executor>& executor)
- : readers_(std::move(readers)), executor_(executor) {}
+ std::vector<std::shared_ptr<GlobalIndexReader>>&& readers,
std::vector<Range>&& row_ranges,
+ std::vector<std::optional<Range>>&& search_ranges, const
std::shared_ptr<Executor>& executor)
+ : readers_(std::move(readers)),
+ executor_(executor),
+ row_ranges_(std::move(row_ranges)),
+ search_ranges_(std::move(search_ranges)) {
+ assert(row_ranges_.size() == search_ranges_.size());
+ assert(search_ranges_.empty() || search_ranges_.size() == readers_.size());
+}
Result<std::shared_ptr<GlobalIndexResult>>
UnionGlobalIndexReader::VisitIsNotNull() {
return Union(
@@ -133,9 +215,75 @@ Result<std::shared_ptr<ScoredGlobalIndexResult>>
UnionGlobalIndexReader::VisitVe
Result<std::shared_ptr<ScoredGlobalIndexResult>>
UnionGlobalIndexReader::VisitFullTextSearch(
const std::shared_ptr<FullTextSearch>& full_text_search) {
- return UnionScored([full_text_search](const
std::shared_ptr<GlobalIndexReader>& reader) {
- return reader->VisitFullTextSearch(full_text_search);
- });
+ if (!full_text_search) {
+ return Status::Invalid("VisitFullTextSearch: null FullTextSearch
pointer");
+ }
+ if (full_text_search->limit <= 0) {
+ return Status::Invalid("full-text search requires a positive limit");
+ }
+ // If the readers rank disjoint rows, the top `limit` rows of each reader
include all of its
+ // rows scoring above the `limit`-th highest score of all readers, so
merging them loses none
+ // of those rows. Rows tied at that score may already have been cut by a
reader's engine.
+ auto reader_searches =
std::make_shared<std::vector<std::shared_ptr<FullTextSearch>>>();
+ reader_searches->reserve(readers_.size());
+ for (size_t index = 0; index < readers_.size(); ++index) {
+ reader_searches->push_back(FullTextSearchForReader(index,
full_text_search));
+ }
+ auto results =
ExecuteAllReadersWithIndex<Result<std::shared_ptr<ScoredGlobalIndexResult>>>(
+ [reader_searches](size_t index, const
std::shared_ptr<GlobalIndexReader>& reader)
+ -> Result<std::shared_ptr<ScoredGlobalIndexResult>> {
+ const std::shared_ptr<FullTextSearch>& reader_search =
(*reader_searches)[index];
+ if (!reader_search) {
+ return EmptyScoredResult();
+ }
+ return reader->VisitFullTextSearch(reader_search);
+ });
+ std::vector<std::shared_ptr<ScoredGlobalIndexResult>> scored_results;
+ scored_results.reserve(results.size());
+ for (auto& result_or_status : results) {
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<ScoredGlobalIndexResult> result,
+ std::move(result_or_status));
+ if (result != nullptr) {
+ scored_results.push_back(std::move(result));
+ }
+ }
+ // No reader answered the search, which differs from a search that matches
no rows.
+ if (scored_results.empty()) {
+ return std::shared_ptr<ScoredGlobalIndexResult>();
+ }
+ return MergeTopK(scored_results, full_text_search->limit);
+}
+
+std::shared_ptr<FullTextSearch>
UnionGlobalIndexReader::FullTextSearchForReader(
+ size_t index, const std::shared_ptr<FullTextSearch>& full_text_search)
const {
+ if (search_ranges_.empty()) {
+ return full_text_search;
+ }
+ const std::optional<Range>& search_range = search_ranges_[index];
+ if (!search_range) {
+ return nullptr;
+ }
+ bool ranks_all_rows = search_range.value() == row_ranges_[index];
+ if (!full_text_search->include_row_ids) {
+ if (ranks_all_rows) {
+ return full_text_search;
+ }
+ return std::make_shared<FullTextSearch>(full_text_search->field_name,
+ full_text_search->query,
full_text_search->limit,
+
RowIdsOf(search_range.value()));
+ }
+ const RoaringBitmap64& include_row_ids =
full_text_search->include_row_ids.value();
+ auto first = include_row_ids.EqualOrLarger(search_range->from);
+ if (first == include_row_ids.End() || *first > search_range->to) {
+ return nullptr;
+ }
+ // The reader drops include row ids outside its rows, so they need no
clipping here.
+ if (ranks_all_rows) {
+ return full_text_search;
+ }
+ return std::make_shared<FullTextSearch>(full_text_search->field_name,
full_text_search->query,
+ full_text_search->limit,
+ RowIdsWithin(include_row_ids,
search_range.value()));
}
Result<std::shared_ptr<ScoredGlobalIndexResult>>
UnionGlobalIndexReader::UnionScored(
@@ -198,11 +346,20 @@ bool UnionGlobalIndexReader::IsThreadSafe() const {
template <typename R>
std::vector<R> UnionGlobalIndexReader::ExecuteAllReaders(
const std::function<R(const std::shared_ptr<GlobalIndexReader>&)>& action)
{
+ return ExecuteAllReadersWithIndex<R>(
+ [action](size_t, const std::shared_ptr<GlobalIndexReader>& reader) ->
R {
+ return action(reader);
+ });
+}
+
+template <typename R>
+std::vector<R> UnionGlobalIndexReader::ExecuteAllReadersWithIndex(
+ const std::function<R(size_t, const std::shared_ptr<GlobalIndexReader>&)>&
action) {
if (executor_ == nullptr || readers_.size() == 1) {
std::vector<R> results;
results.reserve(readers_.size());
- for (const auto& reader : readers_) {
- results.push_back(action(reader));
+ for (size_t index = 0; index < readers_.size(); ++index) {
+ results.push_back(action(index, readers_[index]));
}
return results;
}
@@ -210,8 +367,10 @@ std::vector<R> UnionGlobalIndexReader::ExecuteAllReaders(
// Parallel: submit all tasks to executor, then collect results in
submission order
std::vector<std::future<R>> futures;
futures.reserve(readers_.size());
- for (const auto& reader : readers_) {
- futures.push_back(Via(executor_.get(), [action, reader]() -> R {
return action(reader); }));
+ for (size_t index = 0; index < readers_.size(); ++index) {
+ futures.push_back(Via(executor_.get(), [action, index, reader =
readers_[index]]() -> R {
+ return action(index, reader);
+ }));
}
return CollectAll(futures);
}
diff --git a/src/paimon/common/global_index/union_global_index_reader.h
b/src/paimon/common/global_index/union_global_index_reader.h
index 24992860..811e2462 100644
--- a/src/paimon/common/global_index/union_global_index_reader.h
+++ b/src/paimon/common/global_index/union_global_index_reader.h
@@ -21,11 +21,13 @@
#include <functional>
#include <memory>
+#include <optional>
#include <string>
#include <vector>
#include "paimon/executor.h"
#include "paimon/global_index/global_index_reader.h"
+#include "paimon/utils/range.h"
namespace paimon {
/// A GlobalIndexReader that combines results from multiple readers by
performing a union
@@ -35,7 +37,14 @@ namespace paimon {
/// collected in submission order. Otherwise, sub-readers are evaluated
sequentially.
class UnionGlobalIndexReader : public GlobalIndexReader {
public:
+ /// Constructs a UnionGlobalIndexReader.
+ /// @param row_ranges The global row ids covered by each reader, in the
order of `readers`.
+ /// @param search_ranges The global row ids each reader ranks in a
full-text search, within its
+ /// row range, in the order of `readers`;
std::nullopt if it ranks none.
+ /// If both are empty, each reader ranks all of its
rows.
UnionGlobalIndexReader(std::vector<std::shared_ptr<GlobalIndexReader>>&&
readers,
+ std::vector<Range>&& row_ranges,
+ std::vector<std::optional<Range>>&& search_ranges,
const std::shared_ptr<Executor>& executor);
Result<std::shared_ptr<GlobalIndexResult>> VisitIsNotNull() override;
@@ -58,6 +67,11 @@ class UnionGlobalIndexReader : public GlobalIndexReader {
Result<std::shared_ptr<ScoredGlobalIndexResult>> VisitVectorSearch(
const std::shared_ptr<VectorSearch>& vector_search) override;
+ /// Merges the scored results of all readers and keeps the
`full_text_search->limit` rows with
+ /// the highest scores among them, breaking ties by ascending row id. A
row id returned by more
+ /// than one reader keeps the score from the first of those readers.
Returns nullptr if no
+ /// reader returns a result, e.g. if the readers do not support full-text
search; a reader that
+ /// ranks none of the rows to search is not visited and counts as
returning an empty result.
Result<std::shared_ptr<ScoredGlobalIndexResult>> VisitFullTextSearch(
const std::shared_ptr<FullTextSearch>& full_text_search) override;
@@ -81,13 +95,26 @@ class UnionGlobalIndexReader : public GlobalIndexReader {
/// Executes the given action on all readers and merges the scored results
with Union.
Result<std::shared_ptr<ScoredGlobalIndexResult>>
UnionScored(ScoredReaderAction action);
+ /// Returns the full-text search to run on the reader at `index`, or
nullptr if the reader
+ /// ranks none of the rows to search.
+ std::shared_ptr<FullTextSearch> FullTextSearchForReader(
+ size_t index, const std::shared_ptr<FullTextSearch>& full_text_search)
const;
+
/// Executes the given action on all readers (parallel or sequential) and
collects results.
template <typename R>
std::vector<R> ExecuteAllReaders(
const std::function<R(const std::shared_ptr<GlobalIndexReader>&)>&
action);
+ /// Executes the given action on each reader and its index (parallel or
sequential) and
+ /// collects results in reader order.
+ template <typename R>
+ std::vector<R> ExecuteAllReadersWithIndex(
+ const std::function<R(size_t, const
std::shared_ptr<GlobalIndexReader>&)>& action);
+
std::vector<std::shared_ptr<GlobalIndexReader>> readers_;
std::shared_ptr<Executor> executor_;
+ std::vector<Range> row_ranges_;
+ std::vector<std::optional<Range>> search_ranges_;
};
} // namespace paimon
diff --git a/src/paimon/common/global_index/union_global_index_reader_test.cpp
b/src/paimon/common/global_index/union_global_index_reader_test.cpp
index 5399073b..8b83e134 100644
--- a/src/paimon/common/global_index/union_global_index_reader_test.cpp
+++ b/src/paimon/common/global_index/union_global_index_reader_test.cpp
@@ -19,18 +19,22 @@
#include "paimon/common/global_index/union_global_index_reader.h"
+#include <limits>
#include <map>
#include <memory>
+#include <optional>
#include <queue>
#include <stdexcept>
#include <string>
#include <vector>
#include "gtest/gtest.h"
+#include "paimon/common/global_index/offset_global_index_reader.h"
#include "paimon/common/global_index/sorted_file_global_index_reader.h"
#include "paimon/executor.h"
#include "paimon/global_index/bitmap_global_index_result.h"
#include "paimon/global_index/bitmap_scored_global_index_result.h"
+#include "paimon/predicate/full_text_search.h"
#include "paimon/predicate/literal.h"
#include "paimon/testing/mock/mock_global_index_reader.h"
#include "paimon/testing/utils/testharness.h"
@@ -145,6 +149,10 @@ class UnionGlobalIndexReaderTest : public ::testing::Test {
<< ", expected=" <<
RoaringBitmap64::From(expected_row_ids).ToString();
ASSERT_EQ(typed_result->GetScores(), expected_scores);
}
+
+ static std::shared_ptr<FullTextSearch> MakeFullTextSearch(int32_t limit) {
+ return std::make_shared<FullTextSearch>("f0",
R"({"match":{"query":"apple"}})", limit);
+ }
};
TEST_F(UnionGlobalIndexReaderTest, TestSingleReaderUnion) {
@@ -153,7 +161,8 @@ TEST_F(UnionGlobalIndexReaderTest, TestSingleReaderUnion) {
auto executor = std::make_shared<DeferAfterFirstExecutor>();
std::vector<std::shared_ptr<GlobalIndexReader>> readers = {reader};
- UnionGlobalIndexReader union_reader(std::move(readers), executor);
+ UnionGlobalIndexReader union_reader(std::move(readers), /*row_ranges=*/{},
+ /*search_ranges=*/{}, executor);
ASSERT_OK_AND_ASSIGN(auto result, union_reader.VisitIsNotNull());
CheckResult(result, {1, 2, 3});
@@ -170,7 +179,8 @@ TEST_F(UnionGlobalIndexReaderTest,
TestMultipleReadersUnionSequential) {
reader3->SetDefaultResult({5});
std::vector<std::shared_ptr<GlobalIndexReader>> readers = {reader1,
reader2, reader3};
- UnionGlobalIndexReader union_reader(std::move(readers), nullptr);
+ UnionGlobalIndexReader union_reader(std::move(readers), /*row_ranges=*/{},
+ /*search_ranges=*/{}, nullptr);
ASSERT_OK_AND_ASSIGN(auto result, union_reader.VisitIsNotNull());
// {1,2} OR {3,4} OR {5} -> {1,2,3,4,5}
@@ -187,7 +197,8 @@ TEST_F(UnionGlobalIndexReaderTest,
TestMultipleReadersUnionOverlappingIds) {
reader2->SetDefaultResult({2, 3, 4});
std::vector<std::shared_ptr<GlobalIndexReader>> readers = {reader1,
reader2};
- UnionGlobalIndexReader union_reader(std::move(readers), nullptr);
+ UnionGlobalIndexReader union_reader(std::move(readers), /*row_ranges=*/{},
+ /*search_ranges=*/{}, nullptr);
ASSERT_OK_AND_ASSIGN(auto result, union_reader.VisitIsNotNull());
// {1,2,3} OR {2,3,4} -> {1,2,3,4}
@@ -204,7 +215,8 @@ TEST_F(UnionGlobalIndexReaderTest,
TestMultipleReadersUnionWithExecutor) {
std::vector<std::shared_ptr<GlobalIndexReader>> readers = {reader1,
reader2, reader3};
std::shared_ptr<Executor> executor = CreateDefaultExecutor();
- UnionGlobalIndexReader union_reader(std::move(readers), executor);
+ UnionGlobalIndexReader union_reader(std::move(readers), /*row_ranges=*/{},
+ /*search_ranges=*/{}, executor);
ASSERT_OK_AND_ASSIGN(auto result, union_reader.VisitIsNotNull());
CheckResult(result, {10, 20, 30});
@@ -215,7 +227,8 @@ TEST_F(UnionGlobalIndexReaderTest,
TestMultipleReadersUnionWithExecutor) {
TEST_F(UnionGlobalIndexReaderTest, TestEmptyReaderList) {
std::vector<std::shared_ptr<GlobalIndexReader>> readers;
- UnionGlobalIndexReader union_reader(std::move(readers), nullptr);
+ UnionGlobalIndexReader union_reader(std::move(readers), /*row_ranges=*/{},
+ /*search_ranges=*/{}, nullptr);
// No readers means no results to merge -> nullptr
ASSERT_OK_AND_ASSIGN(auto result, union_reader.VisitIsNotNull());
@@ -229,7 +242,8 @@ TEST_F(UnionGlobalIndexReaderTest,
TestAllReadersReturnNullptr) {
reader2->SetReturnNullptr();
std::vector<std::shared_ptr<GlobalIndexReader>> readers = {reader1,
reader2};
- UnionGlobalIndexReader union_reader(std::move(readers), nullptr);
+ UnionGlobalIndexReader union_reader(std::move(readers), /*row_ranges=*/{},
+ /*search_ranges=*/{}, nullptr);
ASSERT_OK_AND_ASSIGN(auto result, union_reader.VisitIsNotNull());
// All readers return nullptr -> merged result is nullptr
@@ -245,7 +259,8 @@ TEST_F(UnionGlobalIndexReaderTest,
TestPartialReadersReturnNullptr) {
reader3->SetReturnNullptr();
std::vector<std::shared_ptr<GlobalIndexReader>> readers = {reader1,
reader2, reader3};
- UnionGlobalIndexReader union_reader(std::move(readers), nullptr);
+ UnionGlobalIndexReader union_reader(std::move(readers), /*row_ranges=*/{},
+ /*search_ranges=*/{}, nullptr);
ASSERT_OK_AND_ASSIGN(auto result, union_reader.VisitIsNotNull());
// A missing result from any reader makes the whole union unsupported.
@@ -259,7 +274,8 @@ TEST_F(UnionGlobalIndexReaderTest,
TestPartialReadersReturnNullptrWithExecutor)
reader2->SetReturnNullptr();
std::vector<std::shared_ptr<GlobalIndexReader>> readers = {reader1,
reader2};
- UnionGlobalIndexReader union_reader(std::move(readers),
CreateDefaultExecutor());
+ UnionGlobalIndexReader union_reader(std::move(readers), /*row_ranges=*/{},
+ /*search_ranges=*/{},
CreateDefaultExecutor());
ASSERT_OK_AND_ASSIGN(auto result, union_reader.VisitIsNotNull());
ASSERT_FALSE(result);
@@ -292,7 +308,8 @@ TEST_F(UnionGlobalIndexReaderTest,
TestErrorPropagationSequential) {
reader2->SetReturnError("Unknown error for reader2");
std::vector<std::shared_ptr<GlobalIndexReader>> readers = {reader1,
reader2};
- UnionGlobalIndexReader union_reader(std::move(readers), nullptr);
+ UnionGlobalIndexReader union_reader(std::move(readers), /*row_ranges=*/{},
+ /*search_ranges=*/{}, nullptr);
ASSERT_NOK_WITH_MSG(union_reader.VisitIsNotNull(), "Unknown error for
reader2");
}
@@ -305,7 +322,8 @@ TEST_F(UnionGlobalIndexReaderTest,
TestErrorPropagationWithExecutor) {
std::vector<std::shared_ptr<GlobalIndexReader>> readers = {reader1,
reader2};
std::shared_ptr<Executor> executor = CreateDefaultExecutor();
- UnionGlobalIndexReader union_reader(std::move(readers), executor);
+ UnionGlobalIndexReader union_reader(std::move(readers), /*row_ranges=*/{},
+ /*search_ranges=*/{}, executor);
ASSERT_NOK_WITH_MSG(union_reader.VisitIsNotNull(), "Unknown error for
reader2");
}
@@ -318,7 +336,8 @@ TEST_F(UnionGlobalIndexReaderTest,
TestDeferredTaskOwnsActionAfterEarlierFutureT
std::vector<std::shared_ptr<GlobalIndexReader>> readers =
{throwing_reader, deferred_reader};
auto executor = std::make_shared<DeferAfterFirstExecutor>();
- UnionGlobalIndexReader union_reader(std::move(readers), executor);
+ UnionGlobalIndexReader union_reader(std::move(readers), /*row_ranges=*/{},
+ /*search_ranges=*/{}, executor);
ASSERT_THROW(
{ [[maybe_unused]] auto result = union_reader.VisitIsNotNull(); },
std::runtime_error);
ASSERT_EQ(deferred_reader->InvocationCount(), 0);
@@ -334,7 +353,8 @@ TEST_F(UnionGlobalIndexReaderTest, TestVisitEqualUnion) {
reader2->SetDefaultResult({2});
std::vector<std::shared_ptr<GlobalIndexReader>> readers = {reader1,
reader2};
- UnionGlobalIndexReader union_reader(std::move(readers), nullptr);
+ UnionGlobalIndexReader union_reader(std::move(readers), /*row_ranges=*/{},
+ /*search_ranges=*/{}, nullptr);
Literal literal_42(42);
ASSERT_OK_AND_ASSIGN(auto result, union_reader.VisitEqual(literal_42));
@@ -348,7 +368,8 @@ TEST_F(UnionGlobalIndexReaderTest, TestVisitNotEqualUnion) {
reader2->SetDefaultResult({2});
std::vector<std::shared_ptr<GlobalIndexReader>> readers = {reader1,
reader2};
- UnionGlobalIndexReader union_reader(std::move(readers), nullptr);
+ UnionGlobalIndexReader union_reader(std::move(readers), /*row_ranges=*/{},
+ /*search_ranges=*/{}, nullptr);
Literal literal_42(42);
ASSERT_OK_AND_ASSIGN(auto result, union_reader.VisitNotEqual(literal_42));
@@ -362,7 +383,8 @@ TEST_F(UnionGlobalIndexReaderTest,
TestVisitRangeQueriesUnion) {
reader2->SetDefaultResult({2});
std::vector<std::shared_ptr<GlobalIndexReader>> readers = {reader1,
reader2};
- UnionGlobalIndexReader union_reader(std::move(readers), nullptr);
+ UnionGlobalIndexReader union_reader(std::move(readers), /*row_ranges=*/{},
+ /*search_ranges=*/{}, nullptr);
Literal literal_5(5);
ASSERT_OK_AND_ASSIGN(auto lt, union_reader.VisitLessThan(literal_5));
@@ -382,7 +404,8 @@ TEST_F(UnionGlobalIndexReaderTest, TestVisitInUnion) {
reader2->SetDefaultResult({2, 4});
std::vector<std::shared_ptr<GlobalIndexReader>> readers = {reader1,
reader2};
- UnionGlobalIndexReader union_reader(std::move(readers), nullptr);
+ UnionGlobalIndexReader union_reader(std::move(readers), /*row_ranges=*/{},
+ /*search_ranges=*/{}, nullptr);
std::vector<Literal> literals = {Literal(10), Literal(20)};
ASSERT_OK_AND_ASSIGN(auto in_result, union_reader.VisitIn(literals));
@@ -399,7 +422,8 @@ TEST_F(UnionGlobalIndexReaderTest,
TestVisitStringQueriesUnion) {
reader2->SetDefaultResult({2});
std::vector<std::shared_ptr<GlobalIndexReader>> readers = {reader1,
reader2};
- UnionGlobalIndexReader union_reader(std::move(readers), nullptr);
+ UnionGlobalIndexReader union_reader(std::move(readers), /*row_ranges=*/{},
+ /*search_ranges=*/{}, nullptr);
Literal literal_str(FieldType::STRING, "abc", 3);
ASSERT_OK_AND_ASSIGN(auto starts,
union_reader.VisitStartsWith(literal_str));
@@ -419,7 +443,8 @@ TEST_F(UnionGlobalIndexReaderTest, TestVisitIsNullUnion) {
reader2->SetDefaultResult({300});
std::vector<std::shared_ptr<GlobalIndexReader>> readers = {reader1,
reader2};
- UnionGlobalIndexReader union_reader(std::move(readers), nullptr);
+ UnionGlobalIndexReader union_reader(std::move(readers), /*row_ranges=*/{},
+ /*search_ranges=*/{}, nullptr);
ASSERT_OK_AND_ASSIGN(auto result, union_reader.VisitIsNull());
CheckResult(result, {100, 200, 300});
@@ -433,12 +458,202 @@ TEST_F(UnionGlobalIndexReaderTest,
TestVisitFullTextSearchUnion) {
reader3->SetScoredResult({2, 6}, {0.8f, 0.6f});
std::vector<std::shared_ptr<GlobalIndexReader>> readers = {reader1,
reader2, reader3};
- UnionGlobalIndexReader union_reader(std::move(readers), nullptr);
+ UnionGlobalIndexReader union_reader(std::move(readers), /*row_ranges=*/{},
+ /*search_ranges=*/{}, nullptr);
- ASSERT_OK_AND_ASSIGN(auto result,
union_reader.VisitFullTextSearch(nullptr));
+ ASSERT_OK_AND_ASSIGN(auto result,
union_reader.VisitFullTextSearch(MakeFullTextSearch(10)));
CheckScoredResult(result, {1, 2, 5, 6}, {0.9f, 0.8f, 0.5f, 0.6f});
}
+TEST_F(UnionGlobalIndexReaderTest,
TestVisitFullTextSearchKeepsTopKAcrossReaders) {
+ auto reader1 = std::make_shared<MockGlobalIndexReader>();
+ auto reader2 = std::make_shared<MockGlobalIndexReader>();
+ reader1->SetScoredResult({100, 110, 120}, {0.9f, 0.5f, 0.1f});
+ reader2->SetScoredResult({0, 10, 20}, {0.5f, 0.8f, 0.3f});
+
+ std::vector<std::shared_ptr<GlobalIndexReader>> readers = {reader1,
reader2};
+ std::shared_ptr<Executor> executor = CreateDefaultExecutor();
+ UnionGlobalIndexReader union_reader(std::move(readers), /*row_ranges=*/{},
+ /*search_ranges=*/{}, executor);
+
+ ASSERT_OK_AND_ASSIGN(auto result,
union_reader.VisitFullTextSearch(MakeFullTextSearch(3)));
+ CheckScoredResult(result, {0, 10, 100}, {0.5f, 0.8f, 0.9f});
+}
+
+TEST_F(UnionGlobalIndexReaderTest,
TestVisitFullTextSearchDuplicateRowIdKeepsFirstScore) {
+ auto reader1 = std::make_shared<MockGlobalIndexReader>();
+ auto reader2 = std::make_shared<MockGlobalIndexReader>();
+ reader1->SetScoredResult({1, 2}, {0.6f, 0.3f});
+ reader2->SetScoredResult({2, 3}, {0.9f, 0.5f});
+
+ std::vector<std::shared_ptr<GlobalIndexReader>> readers = {reader1,
reader2};
+ std::shared_ptr<Executor> executor = CreateDefaultExecutor();
+ UnionGlobalIndexReader union_reader(std::move(readers), /*row_ranges=*/{},
+ /*search_ranges=*/{}, executor);
+
+ ASSERT_OK_AND_ASSIGN(auto result,
union_reader.VisitFullTextSearch(MakeFullTextSearch(10)));
+ CheckScoredResult(result, {1, 2, 3}, {0.6f, 0.3f, 0.5f});
+ ASSERT_OK_AND_ASSIGN(result,
union_reader.VisitFullTextSearch(MakeFullTextSearch(2)));
+ CheckScoredResult(result, {1, 3}, {0.6f, 0.5f});
+}
+
+TEST_F(UnionGlobalIndexReaderTest, TestVisitFullTextSearchOrdersSpecialScores)
{
+ auto reader1 = std::make_shared<MockGlobalIndexReader>();
+ auto reader2 = std::make_shared<MockGlobalIndexReader>();
+ reader1->SetScoredResult({1, 2, 3}, {-0.0f,
std::numeric_limits<float>::quiet_NaN(), 1.0f});
+ reader2->SetScoredResult({4, 5}, {0.0f,
std::numeric_limits<float>::infinity()});
+
+ std::vector<std::shared_ptr<GlobalIndexReader>> readers = {reader1,
reader2};
+ UnionGlobalIndexReader union_reader(std::move(readers), /*row_ranges=*/{},
+ /*search_ranges=*/{}, nullptr);
+
+ std::vector<std::pair<int32_t, std::vector<int64_t>>> limit_and_row_ids = {
+ {1, {2}}, {2, {2, 5}}, {3, {2, 3, 5}}, {4, {2, 3, 4, 5}}};
+ for (const auto& [limit, expected_row_ids] : limit_and_row_ids) {
+ ASSERT_OK_AND_ASSIGN(auto result,
+
union_reader.VisitFullTextSearch(MakeFullTextSearch(limit)));
+ auto typed_result =
std::dynamic_pointer_cast<BitmapScoredGlobalIndexResult>(result);
+ ASSERT_TRUE(typed_result);
+ ASSERT_OK_AND_ASSIGN(const RoaringBitmap64* bitmap,
typed_result->GetBitmap());
+ ASSERT_EQ(*bitmap, RoaringBitmap64::From(expected_row_ids)) <<
"limit=" << limit;
+ }
+}
+
+TEST_F(UnionGlobalIndexReaderTest, TestVisitFullTextSearchAllNullptr) {
+ auto reader1 = std::make_shared<MockGlobalIndexReader>();
+ auto reader2 = std::make_shared<MockGlobalIndexReader>();
+
+ std::vector<std::shared_ptr<GlobalIndexReader>> readers = {reader1,
reader2};
+ UnionGlobalIndexReader union_reader(std::move(readers), /*row_ranges=*/{},
+ /*search_ranges=*/{}, nullptr);
+
+ ASSERT_OK_AND_ASSIGN(auto result,
union_reader.VisitFullTextSearch(MakeFullTextSearch(10)));
+ ASSERT_FALSE(result);
+}
+
+TEST_F(UnionGlobalIndexReaderTest, TestVisitFullTextSearchErrors) {
+ auto reader1 = std::make_shared<MockGlobalIndexReader>();
+ auto reader2 = std::make_shared<MockGlobalIndexReader>();
+ reader1->SetScoredResult({1}, {0.9f});
+ reader2->SetReturnError("full-text search failure");
+
+ std::vector<std::shared_ptr<GlobalIndexReader>> readers = {reader1,
reader2};
+ UnionGlobalIndexReader union_reader(std::move(readers), /*row_ranges=*/{},
+ /*search_ranges=*/{}, nullptr);
+
+ ASSERT_NOK_WITH_MSG(union_reader.VisitFullTextSearch(nullptr), "null
FullTextSearch pointer");
+
ASSERT_NOK_WITH_MSG(union_reader.VisitFullTextSearch(MakeFullTextSearch(0)),
+ "full-text search requires a positive limit");
+
ASSERT_NOK_WITH_MSG(union_reader.VisitFullTextSearch(MakeFullTextSearch(-1)),
+ "full-text search requires a positive limit");
+ ASSERT_EQ(reader1->InvocationCount(), 0);
+ ASSERT_EQ(reader2->InvocationCount(), 0);
+
ASSERT_NOK_WITH_MSG(union_reader.VisitFullTextSearch(MakeFullTextSearch(10)),
+ "full-text search failure");
+}
+
+TEST_F(UnionGlobalIndexReaderTest, TestVisitFullTextSearchWithinSearchRanges) {
+ auto reader1 = std::make_shared<MockGlobalIndexReader>();
+ auto reader2 = std::make_shared<MockGlobalIndexReader>();
+ auto reader3 = std::make_shared<MockGlobalIndexReader>();
+ reader1->SetScoredResult({0, 1, 2}, {0.2f, 0.5f, 0.9f});
+ reader2->SetScoredResult({1}, {1.0f});
+ reader3->SetScoredResult({3}, {0.7f});
+
+ std::vector<std::shared_ptr<GlobalIndexReader>> readers = {reader1,
reader2, reader3};
+ std::shared_ptr<Executor> executor = CreateDefaultExecutor();
+ UnionGlobalIndexReader union_reader(std::move(readers),
+ /*row_ranges=*/{Range(0, 2), Range(1,
1), Range(1, 3)},
+ /*search_ranges=*/{Range(0, 2),
std::nullopt, Range(3, 3)},
+ executor);
+
+ auto search = MakeFullTextSearch(2);
+ ASSERT_OK_AND_ASSIGN(auto result,
union_reader.VisitFullTextSearch(search));
+ CheckScoredResult(result, {2, 3}, {0.9f, 0.7f});
+ ASSERT_EQ(reader1->CapturedFullTextSearch(), search);
+ ASSERT_EQ(reader2->InvocationCount(), 0);
+ ASSERT_EQ(reader3->CapturedFullTextSearch()->include_row_ids,
RoaringBitmap64::From({3l}));
+
+ constexpr int64_t kMax = std::numeric_limits<int64_t>::max();
+ search->include_row_ids = RoaringBitmap64::From({1, 2, 3, 5, kMax});
+ ASSERT_OK(union_reader.VisitFullTextSearch(search));
+ ASSERT_EQ(reader1->CapturedFullTextSearch(), search);
+ ASSERT_EQ(reader2->InvocationCount(), 0);
+ ASSERT_EQ(reader3->CapturedFullTextSearch()->include_row_ids,
RoaringBitmap64::From({3l}));
+ ASSERT_EQ(search->include_row_ids, RoaringBitmap64::From({1, 2, 3, 5,
kMax}));
+
+ search->include_row_ids = RoaringBitmap64::From({5});
+ ASSERT_OK_AND_ASSIGN(result, union_reader.VisitFullTextSearch(search));
+ CheckScoredResult(result, {}, {});
+ ASSERT_EQ(reader1->InvocationCount(), 2);
+ ASSERT_EQ(reader2->InvocationCount(), 0);
+ ASSERT_EQ(reader3->InvocationCount(), 2);
+}
+
+TEST_F(UnionGlobalIndexReaderTest,
TestVisitFullTextSearchSparseIncludeInLargeRanges) {
+ constexpr int64_t kMax = std::numeric_limits<int64_t>::max();
+ auto reader1 = std::make_shared<MockGlobalIndexReader>();
+ auto reader2 = std::make_shared<MockGlobalIndexReader>();
+ std::vector<std::shared_ptr<GlobalIndexReader>> readers = {reader1,
reader2};
+ UnionGlobalIndexReader union_reader(
+ std::move(readers), /*row_ranges=*/{Range(0, kMax / 2), Range(10,
kMax)},
+ /*search_ranges=*/{Range(0, kMax / 2), Range(kMax / 2 + 1, kMax)},
/*executor=*/nullptr);
+
+ auto search = MakeFullTextSearch(10);
+ search->include_row_ids = RoaringBitmap64();
+ ASSERT_OK_AND_ASSIGN(auto result,
union_reader.VisitFullTextSearch(search));
+ CheckScoredResult(result, {}, {});
+ ASSERT_EQ(reader1->InvocationCount(), 0);
+ ASSERT_EQ(reader2->InvocationCount(), 0);
+
+ search->include_row_ids = RoaringBitmap64::From({20, kMax / 2 + 1, kMax});
+ ASSERT_OK(union_reader.VisitFullTextSearch(search));
+ ASSERT_EQ(reader1->CapturedFullTextSearch(), search);
+ ASSERT_EQ(reader2->CapturedFullTextSearch()->include_row_ids,
+ RoaringBitmap64::From({kMax / 2 + 1, kMax}));
+
+ search->include_row_ids = RoaringBitmap64::From({20l});
+ ASSERT_OK(union_reader.VisitFullTextSearch(search));
+ ASSERT_EQ(reader1->InvocationCount(), 2);
+ ASSERT_EQ(reader2->InvocationCount(), 1);
+}
+
+TEST_F(UnionGlobalIndexReaderTest,
TestVisitFullTextSearchOverlappingRangesThroughOffsetReaders) {
+ constexpr int64_t kMax = std::numeric_limits<int64_t>::max();
+ const int64_t to = int64_t{1} << 32;
+ std::vector<std::shared_ptr<MockGlobalIndexReader>> wrapped_readers;
+ std::vector<std::shared_ptr<GlobalIndexReader>> readers;
+ std::vector<Range> row_ranges = {Range(0, 99), Range(50, to), Range(kMax -
9, kMax - 3),
+ Range(kMax - 5, kMax)};
+ std::vector<std::optional<Range>> search_ranges = {
+ Range(0, 99), Range(100, to), Range(kMax - 9, kMax - 3), Range(kMax -
2, kMax)};
+ for (const Range& row_range : row_ranges) {
+ wrapped_readers.push_back(std::make_shared<MockGlobalIndexReader>());
+
readers.push_back(std::make_shared<OffsetGlobalIndexReader>(wrapped_readers.back(),
+
row_range.from, row_range.to));
+ }
+ UnionGlobalIndexReader union_reader(std::move(readers),
std::move(row_ranges),
+ std::move(search_ranges),
/*executor=*/nullptr);
+
+ auto search = MakeFullTextSearch(1);
+ ASSERT_OK(union_reader.VisitFullTextSearch(search));
+ ASSERT_EQ(wrapped_readers[0]->CapturedFullTextSearch(), search);
+ RoaringBitmap64 local_row_ids;
+ local_row_ids.AddRange(50, to - 49);
+ ASSERT_EQ(wrapped_readers[1]->CapturedFullTextSearch()->include_row_ids,
local_row_ids);
+ ASSERT_EQ(wrapped_readers[2]->CapturedFullTextSearch(), search);
+ ASSERT_EQ(wrapped_readers[3]->CapturedFullTextSearch()->include_row_ids,
+ RoaringBitmap64::From({3l, 4l, 5l}));
+
+ RoaringBitmap64 include_row_ids;
+ include_row_ids.AddRange(0, to + 1);
+ search->include_row_ids = include_row_ids;
+ ASSERT_OK(union_reader.VisitFullTextSearch(search));
+ ASSERT_EQ(wrapped_readers[1]->CapturedFullTextSearch()->include_row_ids,
local_row_ids);
+ ASSERT_EQ(wrapped_readers[2]->InvocationCount(), 1);
+ ASSERT_EQ(wrapped_readers[3]->InvocationCount(), 1);
+}
+
TEST_F(UnionGlobalIndexReaderTest, TestVisitVectorSearchAllNullptr) {
auto reader1 = std::make_shared<MockGlobalIndexReader>();
auto reader2 = std::make_shared<MockGlobalIndexReader>();
@@ -447,7 +662,8 @@ TEST_F(UnionGlobalIndexReaderTest,
TestVisitVectorSearchAllNullptr) {
reader2->SetDefaultResult({2});
std::vector<std::shared_ptr<GlobalIndexReader>> readers = {reader1,
reader2};
- UnionGlobalIndexReader union_reader(std::move(readers), nullptr);
+ UnionGlobalIndexReader union_reader(std::move(readers), /*row_ranges=*/{},
+ /*search_ranges=*/{}, nullptr);
ASSERT_OK_AND_ASSIGN(auto result, union_reader.VisitVectorSearch(nullptr));
ASSERT_FALSE(result);
@@ -458,7 +674,8 @@ TEST_F(UnionGlobalIndexReaderTest,
TestVisitVectorSearchSingleReader) {
reader->SetScoredResult({1, 3, 5}, {0.9f, 0.7f, 0.5f});
std::vector<std::shared_ptr<GlobalIndexReader>> readers = {reader};
- UnionGlobalIndexReader union_reader(std::move(readers), nullptr);
+ UnionGlobalIndexReader union_reader(std::move(readers), /*row_ranges=*/{},
+ /*search_ranges=*/{}, nullptr);
ASSERT_OK_AND_ASSIGN(auto result, union_reader.VisitVectorSearch(nullptr));
CheckScoredResult(result, {1, 3, 5}, {0.9f, 0.7f, 0.5f});
@@ -471,7 +688,8 @@ TEST_F(UnionGlobalIndexReaderTest,
TestVisitVectorSearchMultipleReadersUnion) {
reader2->SetScoredResult({2, 4}, {0.8f, 0.6f});
std::vector<std::shared_ptr<GlobalIndexReader>> readers = {reader1,
reader2};
- UnionGlobalIndexReader union_reader(std::move(readers), nullptr);
+ UnionGlobalIndexReader union_reader(std::move(readers), /*row_ranges=*/{},
+ /*search_ranges=*/{}, nullptr);
ASSERT_OK_AND_ASSIGN(auto result, union_reader.VisitVectorSearch(nullptr));
// {1,3} OR {2,4} -> {1,2,3,4}, scores merged in row id order
@@ -488,7 +706,8 @@ TEST_F(UnionGlobalIndexReaderTest,
TestVisitVectorSearchPartialNullptr) {
reader3->SetScoredResult({5, 6}, {0.5f, 0.4f});
std::vector<std::shared_ptr<GlobalIndexReader>> readers = {reader1,
reader2, reader3};
- UnionGlobalIndexReader union_reader(std::move(readers), nullptr);
+ UnionGlobalIndexReader union_reader(std::move(readers), /*row_ranges=*/{},
+ /*search_ranges=*/{}, nullptr);
ASSERT_OK_AND_ASSIGN(auto result, union_reader.VisitVectorSearch(nullptr));
// reader2 nullptr is skipped, {1,2} OR {5,6} -> {1,2,5,6}
@@ -502,7 +721,8 @@ TEST_F(UnionGlobalIndexReaderTest,
TestVisitVectorSearchErrorPropagation) {
reader2->SetReturnError("vector search failure");
std::vector<std::shared_ptr<GlobalIndexReader>> readers = {reader1,
reader2};
- UnionGlobalIndexReader union_reader(std::move(readers), nullptr);
+ UnionGlobalIndexReader union_reader(std::move(readers), /*row_ranges=*/{},
+ /*search_ranges=*/{}, nullptr);
ASSERT_NOK_WITH_MSG(union_reader.VisitVectorSearch(nullptr), "vector
search failure");
}
@@ -512,7 +732,8 @@ TEST_F(UnionGlobalIndexReaderTest,
TestIsThreadSafeReturnsTrueWhenAllReadersAreS
auto reader2 = std::make_shared<MockGlobalIndexReader>();
std::vector<std::shared_ptr<GlobalIndexReader>> readers = {reader1,
reader2};
- UnionGlobalIndexReader union_reader(std::move(readers), nullptr);
+ UnionGlobalIndexReader union_reader(std::move(readers), /*row_ranges=*/{},
+ /*search_ranges=*/{}, nullptr);
ASSERT_TRUE(union_reader.IsThreadSafe());
}
@@ -523,14 +744,16 @@ TEST_F(UnionGlobalIndexReaderTest,
TestIsThreadSafeReturnsFalseWhenAnyReaderIsNo
reader2->SetThreadSafe(false);
std::vector<std::shared_ptr<GlobalIndexReader>> readers = {reader1,
reader2};
- UnionGlobalIndexReader union_reader(std::move(readers), nullptr);
+ UnionGlobalIndexReader union_reader(std::move(readers), /*row_ranges=*/{},
+ /*search_ranges=*/{}, nullptr);
ASSERT_FALSE(union_reader.IsThreadSafe());
}
TEST_F(UnionGlobalIndexReaderTest, TestGetIndexTypeReturnsUnion) {
std::vector<std::shared_ptr<GlobalIndexReader>> readers;
- UnionGlobalIndexReader union_reader(std::move(readers), nullptr);
+ UnionGlobalIndexReader union_reader(std::move(readers), /*row_ranges=*/{},
+ /*search_ranges=*/{}, nullptr);
ASSERT_EQ(union_reader.GetIndexType(), "union");
}
diff --git a/src/paimon/common/predicate/full_text_search_test.cpp
b/src/paimon/common/predicate/full_text_search_test.cpp
index 2e14dbd3..d5489aed 100644
--- a/src/paimon/common/predicate/full_text_search_test.cpp
+++ b/src/paimon/common/predicate/full_text_search_test.cpp
@@ -60,4 +60,28 @@ TEST(FullTextSearchTest, TestOffsetRangeWithIncludeRowIds) {
RoaringBitmap64::From({3l, 7l}));
}
+TEST(FullTextSearchTest, TestOffsetRangeWithIncludeRowIdRun) {
+ auto row_ids_of = [](int64_t from, int64_t to) {
+ RoaringBitmap64 row_ids;
+ row_ids.AddRange(from, to);
+ row_ids.Add(to);
+ return row_ids;
+ };
+ FullTextSearch search("f0", R"({"match":{"query":"paimon"}})", /*limit=*/5,
+ row_ids_of(100, 199));
+ ASSERT_EQ(search.OffsetRange(150, 299)->include_row_ids, row_ids_of(0,
49));
+ ASSERT_EQ(search.OffsetRange(0, 149)->include_row_ids, row_ids_of(100,
149));
+ ASSERT_EQ(search.OffsetRange(120, 130)->include_row_ids, row_ids_of(0,
10));
+ ASSERT_EQ(search.OffsetRange(200, 299)->include_row_ids,
RoaringBitmap64());
+ ASSERT_EQ(search.include_row_ids, row_ids_of(100, 199));
+
+ constexpr int64_t kMax = RoaringBitmap64::MAX_VALUE;
+ FullTextSearch last_search("f0", R"({"match":{"query":"paimon"}})",
/*limit=*/5,
+ row_ids_of(kMax - 9, kMax));
+ ASSERT_EQ(last_search.OffsetRange(kMax - 5, kMax)->include_row_ids,
row_ids_of(0, 5));
+ FullTextSearch max_search("f0", R"({"match":{"query":"paimon"}})",
/*limit=*/5,
+ RoaringBitmap64::From({kMax}));
+ ASSERT_EQ(max_search.OffsetRange(kMax - 5, kMax)->include_row_ids,
RoaringBitmap64::From({5l}));
+}
+
} // namespace paimon::test
diff --git a/src/paimon/core/global_index/global_index_scan_impl.cpp
b/src/paimon/core/global_index/global_index_scan_impl.cpp
index c38c5037..eabc10e0 100644
--- a/src/paimon/core/global_index/global_index_scan_impl.cpp
+++ b/src/paimon/core/global_index/global_index_scan_impl.cpp
@@ -18,6 +18,9 @@
*/
#include "paimon/core/global_index/global_index_scan_impl.h"
+#include <algorithm>
+#include <limits>
+#include <optional>
#include <string>
#include <thread>
#include <utility>
@@ -184,8 +187,21 @@ Result<std::vector<std::shared_ptr<GlobalIndexReader>>>
GlobalIndexScanImpl::Cre
continue;
}
std::vector<std::shared_ptr<GlobalIndexReader>> union_readers;
+ std::vector<Range> row_ranges;
+ std::vector<std::optional<Range>> search_ranges;
union_readers.reserve(range_to_metas.size());
+ row_ranges.reserve(range_to_metas.size());
+ search_ranges.reserve(range_to_metas.size());
+ // Assign each full-text row to one reader before requesting
per-reader top-k: a row is
+ // ranked by the first range in (from, to) order that covers it, which
leaves each range
+ // only its rows after the largest `to` of earlier ones.
+ int64_t covered_to = std::numeric_limits<int64_t>::min();
for (const auto& [range, metas] : range_to_metas) {
+ std::optional<Range> search_range;
+ if (covered_to < range.to) {
+ search_range = Range(std::max(range.from, covered_to + 1),
range.to);
+ }
+ covered_to = std::max(covered_to, range.to);
if (row_range_index && !row_range_index->Intersects(range.from,
range.to)) {
continue;
}
@@ -199,14 +215,16 @@ Result<std::vector<std::shared_ptr<GlobalIndexReader>>>
GlobalIndexScanImpl::Cre
PAIMON_ASSIGN_OR_RAISE(
std::shared_ptr<GlobalIndexReader> index_reader,
indexer->CreateReader(&c_arrow_schema, index_file_manager_,
index_io_metas, pool_));
- union_readers.push_back(
-
std::make_shared<OffsetGlobalIndexReader>(std::move(index_reader), range.from));
+ union_readers.push_back(std::make_shared<OffsetGlobalIndexReader>(
+ std::move(index_reader), range.from, range.to));
+ row_ranges.push_back(range);
+ search_ranges.push_back(search_range);
}
if (union_readers.empty()) {
continue;
}
- readers.push_back(
- std::make_shared<UnionGlobalIndexReader>(std::move(union_readers),
executor_));
+ readers.push_back(std::make_shared<UnionGlobalIndexReader>(
+ std::move(union_readers), std::move(row_ranges),
std::move(search_ranges), executor_));
}
return readers;
}
diff --git a/test/inte/global_index_test.cpp b/test/inte/global_index_test.cpp
index e6c1abc5..191d4d1a 100644
--- a/test/inte/global_index_test.cpp
+++ b/test/inte/global_index_test.cpp
@@ -16,6 +16,7 @@
* limitations under the License.
*/
+#include <algorithm>
#include <atomic>
#include <limits>
@@ -74,6 +75,7 @@ class CountingGlobalIndexFileManager : public
GlobalIndexFileManager {
} // namespace
+using ScoredRows = std::vector<std::pair<int64_t, float>>;
// string: FileFormat, bool: UseSpecificFileSystem
using ParamType = std::tuple<std::string, bool>;
/// This is a sdk end-to-end test for global index.
@@ -185,6 +187,102 @@ class GlobalIndexTest : public ::testing::Test, public
::testing::WithParamInter
return Commit(table_path, {index_commit_msg});
}
+ static Result<ScoredRows> SearchFullText(
+ const std::shared_ptr<GlobalIndexReader>& reader, const std::string&
query, int32_t limit,
+ const std::optional<RoaringBitmap64>& include_row_ids) {
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<ScoredGlobalIndexResult> result,
+
reader->VisitFullTextSearch(std::make_shared<FullTextSearch>(
+ "f0", query, limit, include_row_ids)));
+ if (!result) {
+ return Status::Invalid("full-text search returned no result");
+ }
+
PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<ScoredGlobalIndexResult::ScoredIterator>
iter,
+ result->CreateScoredIterator());
+ ScoredRows rows;
+ while (iter->HasNext()) {
+ rows.push_back(iter->NextWithScore());
+ }
+ return rows;
+ }
+
+ static void CheckFullTextTopK(const std::shared_ptr<GlobalIndexReader>&
reader,
+ const std::string& query,
+ const std::optional<RoaringBitmap64>&
include_row_ids,
+ const std::vector<int64_t>&
expected_row_ids) {
+ ASSERT_OK_AND_ASSIGN(ScoredRows all_rows,
+ SearchFullText(reader, query, /*limit=*/100,
include_row_ids));
+ std::vector<int64_t> row_ids;
+ for (const auto& [row_id, score] : all_rows) {
+ row_ids.push_back(row_id);
+ }
+ ASSERT_EQ(row_ids, expected_row_ids);
+ ScoredRows ranked_rows = all_rows;
+ std::sort(
+ ranked_rows.begin(), ranked_rows.end(),
+ [](const std::pair<int64_t, float>& left, const std::pair<int64_t,
float>& right) {
+ return left.second != right.second ? left.second > right.second
+ : left.first < right.first;
+ });
+ for (size_t limit = 1; limit < ranked_rows.size(); ++limit) {
+ ScoredRows expected_rows(ranked_rows.begin(), ranked_rows.begin()
+ limit);
+ std::sort(expected_rows.begin(), expected_rows.end());
+ ASSERT_OK_AND_ASSIGN(
+ ScoredRows rows,
+ SearchFullText(reader, query, static_cast<int32_t>(limit),
include_row_ids));
+ ASSERT_EQ(rows, expected_rows) << "limit=" << limit;
+ }
+ }
+
+ void CheckSearchOverlappingIndexRanges(const std::string& index_type,
+ const std::map<std::string,
std::string>& index_options,
+ const std::string& query) {
+ arrow::FieldVector fields = {arrow::field("f0", arrow::utf8()),
+ arrow::field("f1", arrow::int32())};
+ auto schema = arrow::schema(fields);
+ std::map<std::string, std::string> options = {{Options::FILE_FORMAT,
file_format_},
+ {Options::FILE_SYSTEM,
"local"},
+
{Options::ROW_TRACKING_ENABLED, "true"},
+
{Options::DATA_EVOLUTION_ENABLED, "true"}};
+ CreateTable(/*partition_keys=*/{}, schema, options);
+ std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
+
+ auto src_array =
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), R"([
+["This is an test document.", 0],
+["This is an new document document document.", 1],
+["Document document document document test.", 2],
+["Document document document document document.", 3]
+ ])")
+ .ValueOrDie();
+ ASSERT_OK_AND_ASSIGN(auto commit_msgs,
+ WriteArray(table_path, schema->field_names(),
src_array));
+ ASSERT_OK(Commit(table_path, commit_msgs));
+ ASSERT_OK(WriteIndex(table_path, /*partition_filters=*/{}, "f0",
index_type, index_options,
+ Range(0, 2)));
+ ASSERT_OK(WriteIndex(table_path, /*partition_filters=*/{}, "f0",
index_type, index_options,
+ Range(1, 1)));
+ ASSERT_OK(WriteIndex(table_path, /*partition_filters=*/{}, "f0",
index_type, index_options,
+ Range(1, 3)));
+
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<GlobalIndexScan>
global_index_scan,
+ GlobalIndexScan::Create(table_path,
/*snapshot_id=*/std::nullopt,
+
/*partitions=*/std::nullopt, /*options=*/{},
+ fs_,
/*executor=*/nullptr, pool_));
+ ASSERT_OK_AND_ASSIGN(
+ auto readers, global_index_scan->CreateReaders("f0",
/*row_range_index=*/std::nullopt));
+ ASSERT_EQ(readers.size(), 1u);
+ CheckFullTextTopK(readers[0], query, std::nullopt, {0, 1, 2, 3});
+ ASSERT_OK_AND_ASSIGN(ScoredRows top_rows,
+ SearchFullText(readers[0], query, /*limit=*/1,
std::nullopt));
+ ASSERT_EQ(top_rows.size(), 1u);
+ ASSERT_EQ(top_rows[0].first, 3);
+ CheckFullTextTopK(readers[0], query, RoaringBitmap64::From({1, 2, 3}),
{1, 2, 3});
+
+ ASSERT_OK_AND_ASSIGN(RowRangeIndex row_range_index,
RowRangeIndex::Create({Range(3, 3)}));
+ ASSERT_OK_AND_ASSIGN(readers, global_index_scan->CreateReaders("f0",
row_range_index));
+ ASSERT_EQ(readers.size(), 1u);
+ CheckFullTextTopK(readers[0], query, std::nullopt, {3});
+ }
+
Result<std::shared_ptr<Plan>> ScanGlobalIndexAndData(
const std::string& table_path, const std::shared_ptr<Predicate>&
predicate,
const std::map<std::string, std::string>& options = {},
@@ -2803,6 +2901,28 @@ TEST_P(GlobalIndexTest,
TestWriteCommitScanReadLuceneIndexWithPartition) {
ASSERT_OK_AND_ASSIGN(RowRangeIndex row_range_index,
RowRangeIndex::Create({Range(2, 3)}));
scan_and_check_result({{"f1", "20"}}, row_range_index, filter, "row
ids: {2}");
}
+ {
+ ASSERT_OK_AND_ASSIGN(
+ std::shared_ptr<GlobalIndexScan> global_index_scan,
+ GlobalIndexScan::Create(table_path, /*snapshot_id=*/std::nullopt,
+ /*partitions=*/std::nullopt,
lucene_options, fs_,
+ /*executor=*/nullptr, pool_));
+ ASSERT_OK_AND_ASSIGN(
+ auto readers, global_index_scan->CreateReaders("f0",
/*row_range_index=*/std::nullopt));
+ ASSERT_EQ(readers.size(), 1u);
+ std::string query =
R"({"match":{"query":"document","operator":"And"}})";
+ CheckFullTextTopK(readers[0], query, std::nullopt, {0, 1, 2});
+ CheckFullTextTopK(readers[0], query, RoaringBitmap64::From({1, 2, 3}),
{1, 2});
+ }
+}
+
+TEST_P(GlobalIndexTest, TestLuceneSearchOverlappingIndexRanges) {
+ auto tmp_dir = paimon::test::UniqueTestDirectory::Create();
+ ASSERT_TRUE(tmp_dir);
+ CheckSearchOverlappingIndexRanges("lucene-fts",
+
{{"lucene-fts.write.omit-term-freq-and-position", "false"},
+ {"lucene-fts.write.tmp.directory",
tmp_dir->Str()}},
+
R"({"match":{"query":"document","operator":"And"}})");
}
#endif
@@ -2932,6 +3052,24 @@ TEST_P(GlobalIndexTest,
TestWriteCommitScanReadFullTextIndexWithPartition) {
ASSERT_OK_AND_ASSIGN(RowRangeIndex row_range_index,
RowRangeIndex::Create({Range(2, 3)}));
scan_and_check_result({{"f1", "20"}}, row_range_index, std::nullopt,
"{2}");
}
+ {
+ ASSERT_OK_AND_ASSIGN(
+ std::shared_ptr<GlobalIndexScan> global_index_scan,
+ GlobalIndexScan::Create(table_path, /*snapshot_id=*/std::nullopt,
+ /*partitions=*/std::nullopt,
/*options=*/{}, fs_,
+ /*executor=*/nullptr, pool_));
+ ASSERT_OK_AND_ASSIGN(
+ auto readers, global_index_scan->CreateReaders("f0",
/*row_range_index=*/std::nullopt));
+ ASSERT_EQ(readers.size(), 1u);
+ std::string query = R"({"match":{"query":"document"}})";
+ CheckFullTextTopK(readers[0], query, std::nullopt, {0, 1, 2});
+ CheckFullTextTopK(readers[0], query, RoaringBitmap64::From({1, 2, 3}),
{1, 2});
+ }
+}
+
+TEST_P(GlobalIndexTest, TestFullTextSearchOverlappingIndexRanges) {
+ CheckSearchOverlappingIndexRanges("full-text", {{"full-text.tokenizer",
"default"}},
+ R"({"match":{"query":"document"}})");
}
#endif
