This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.2 by this push:
new 9c1171700d6 branch-4.2: [improvement](be) Speed up dense unscored
inverted index conjunctions (#69172) (#69239)
9c1171700d6 is described below
commit 9c1171700d6399053474da2614ce792a0f19e032
Author: Jack <[email protected]>
AuthorDate: Mon Oct 19 11:36:46 2026 +0800
branch-4.2: [improvement](be) Speed up dense unscored inverted index
conjunctions (#69172) (#69239)
### What problem does this PR solve?
Issue Number: N/A
Related PR: #69172, #69206
Problem Summary: Backport #69172 after #69206 introduced the shared
inverted-index query engine to branch-4.2. Dense unscored conjunctions
now skip terms held by every row, write a lone term directly to the
result bitmap, use branch-free block intersection, and bound
streaming-source batches. Query results are unchanged.
Source PR: https://github.com/apache/doris/pull/69172
Source squash commit: `f937d950d45313f20864d287df453e1b90f5c615`
(cherry-picked with `-x`).
### Release note
None
### Check List (For Author)
- Test
- [x] Regression test: `test_match_format_parity` passed on an isolated
branch-4.2 FE/BE cluster after rebasing onto
`ecd322ad31193cda19cf4f3b08e0f3f9d12dd67c` (1 suite, 0 failures).
- [x] Unit Test: `ListedTermsTest.*:CursorChainedPostings.*` passed (41
tests in 2 suites); all 9 source-added tests were listed and executed.
No sanitizer report.
- [x] Manual test: `./build.sh --be -j64` and `./build.sh --fe -j8`
passed; isolated backend reported `Alive=true`, and `SELECT 1`
succeeded.
- Behavior changed:
- [x] No. Unscored conjunction execution is faster; query results are
unchanged.
- Does this need documentation?
- [x] No.
### Validation
- `build-support/check-format.sh`: PASS.
- Complete merged-tree format preflight against `origin/branch-4.2`:
PASS.
- Complete PR diff English description check: PASS (5 source files, no
exceptions).
- Branch-4.2 has no `build-support/run-clang-tidy.sh`; the branch's BE
build and format gates passed.
### Hunk audit
The source squash patch ID matches the final source PR head. The
cherry-pick was clean, and every changed file has a byte-identical
postimage to the source commit.
| Source file | Disposition |
| --- | --- |
|
`be/src/storage/index/inverted/query/engine/boolean_query/listed_terms.cpp`
| Ported (verbatim) |
| `be/src/storage/index/inverted/query/exec/cursor_chained_postings.cpp`
| Ported (verbatim) |
| `be/src/storage/index/inverted/query/exec/cursor_chained_postings.h` |
Ported (verbatim) |
|
`be/test/storage/index/inverted/query/engine/boolean_query/listed_terms_test.cpp`
| Ported (verbatim) |
|
`be/test/storage/index/inverted/query/exec/cursor_chained_postings_test.cpp`
| Ported (verbatim) |
`git range-diff` shows only the `-x` cherry-pick provenance line; the
code patch is unchanged.
---
.../query/engine/boolean_query/listed_terms.cpp | 35 ++++-
.../query/exec/cursor_chained_postings.cpp | 169 +++++++++++++++++++--
.../inverted/query/exec/cursor_chained_postings.h | 16 ++
.../engine/boolean_query/listed_terms_test.cpp | 92 +++++++++--
.../query/exec/cursor_chained_postings_test.cpp | 168 ++++++++++++++++++++
5 files changed, 460 insertions(+), 20 deletions(-)
diff --git
a/be/src/storage/index/inverted/query/engine/boolean_query/listed_terms.cpp
b/be/src/storage/index/inverted/query/engine/boolean_query/listed_terms.cpp
index 00a4ff1bd1e..59828cbae42 100644
--- a/be/src/storage/index/inverted/query/engine/boolean_query/listed_terms.cpp
+++ b/be/src/storage/index/inverted/query/engine/boolean_query/listed_terms.cpp
@@ -90,8 +90,41 @@ index_query::TruthSet ListedTerms::conjunction(const
std::vector<uint32_t>* cand
if (has_absent_term()) {
return result;
}
+ // A term held by every row the source lists cannot narrow the
conjunction, so the chain
+ // lists only the other terms.
+ const uint32_t doc_count = _source->doc_count();
+ std::vector<index_query::PostingsCursor*> narrowing;
+ for (const auto& cursor : _cursors) {
+ if (cursor->doc_freq() != doc_count) {
+ narrowing.push_back(cursor.get());
+ }
+ }
+ if (narrowing.empty()) {
+ if (candidates == nullptr) {
+ result.true_rows.addRange(0, doc_count);
+ } else {
+ const auto end = std::ranges::lower_bound(*candidates, doc_count);
+ result.true_rows.addMany(static_cast<size_t>(end -
candidates->begin()),
+ candidates->data());
+ }
+ return result;
+ }
+ // A lone term lists its blocks straight into the result instead of
through a list of its rows.
+ if (narrowing.size() == 1 && candidates == nullptr) {
+ THROW_IF_ERROR(narrowing.front()->prefetch(nullptr,
/*positions=*/false));
+ index_query::RoaringDocIdSink sink(result.true_rows);
+ THROW_IF_ERROR(index_query::list_blocks(*narrowing.front(), sink));
+ return result;
+ }
+ // A source reading on demand intersects the terms a batch of rows at a
time, straight into
+ // the result, instead of listing every row of the cheapest term first.
+ if (candidates == nullptr && !_source->batches_reads()) {
+ index_query::RoaringDocIdSink sink(result.true_rows);
+ THROW_IF_ERROR(index_query::stream_cursors(narrowing, sink));
+ return result;
+ }
std::vector<uint32_t> docs;
- THROW_IF_ERROR(index_query::chain_cursors(_cursors, candidates, &docs));
+ THROW_IF_ERROR(index_query::chain_cursors(narrowing, candidates, &docs));
result.true_rows.addMany(docs.size(), docs.data());
return result;
}
diff --git
a/be/src/storage/index/inverted/query/exec/cursor_chained_postings.cpp
b/be/src/storage/index/inverted/query/exec/cursor_chained_postings.cpp
index bb77e9b29a0..7905a291fde 100644
--- a/be/src/storage/index/inverted/query/exec/cursor_chained_postings.cpp
+++ b/be/src/storage/index/inverted/query/exec/cursor_chained_postings.cpp
@@ -22,6 +22,7 @@
#include <bit>
#include <numeric>
+#include "common/check.h"
#include "storage/index/inverted/query/exec/docid_sink.h"
namespace doris::index_query {
@@ -96,12 +97,18 @@ void intersect_through_bit_set(std::span<const uint32_t>
docs, std::span<const u
preceding += static_cast<uint32_t>(std::popcount(bits[word]));
}
} else {
+ // Every candidate is written and kept only when its bit is set, so no
branch depends on
+ // whether it matches.
+ std::vector<uint32_t>& rows = *out.rows;
+ const size_t begin = rows.size();
+ rows.resize(begin + candidates.size());
+ uint32_t* next = rows.data() + begin;
for (const uint32_t candidate : candidates) {
const uint32_t off = candidate - first;
- if ((bits[off >> 6] & (uint64_t {1} << (off & 63))) != 0) {
- out.append(candidate, 0);
- }
+ *next = candidate;
+ next += (bits[off >> 6] >> (off & 63)) & 1;
}
+ rows.resize(static_cast<size_t>(next - rows.data()));
}
}
@@ -192,10 +199,7 @@ Status
CursorChainedPostings::collect(std::vector<uint32_t>* out) {
if (_candidates == nullptr) {
out->reserve(out->size() + _cursor.doc_freq());
VectorDocIdSink sink(*out);
- return for_each_block(_cursor, [&sink](const PostingsBlock& block) {
- return block.dense ? sink.append_range(block.range_begin,
block.range_end)
- : sink.append_sorted(block.docs);
- });
+ return list_blocks(_cursor, sink);
}
out->reserve(out->size() + std::min<uint64_t>(_candidates->size(),
_cursor.doc_freq()));
return for_each_candidate_block(
@@ -219,8 +223,11 @@ Status
CursorChainedPostings::collect(std::vector<uint32_t>* out) {
});
}
-Status chain_cursors(std::span<const std::unique_ptr<PostingsCursor>> cursors,
- const std::vector<uint32_t>* candidates,
std::vector<uint32_t>* rows) {
+namespace {
+
+template <typename Cursors>
+Status chain(const Cursors& cursors, const std::vector<uint32_t>* candidates,
+ std::vector<uint32_t>* rows) {
std::vector<CursorChainedPostings> terms;
terms.reserve(cursors.size());
std::vector<ChainedPostings*> chain;
@@ -232,4 +239,148 @@ Status chain_cursors(std::span<const
std::unique_ptr<PostingsCursor>> cursors,
return chained_conjunction(chain, candidates, rows);
}
+} // namespace
+
+Status chain_cursors(std::span<const std::unique_ptr<PostingsCursor>> cursors,
+ const std::vector<uint32_t>* candidates,
std::vector<uint32_t>* rows) {
+ return chain(cursors, candidates, rows);
+}
+
+Status chain_cursors(std::span<PostingsCursor* const> cursors,
+ const std::vector<uint32_t>* candidates,
std::vector<uint32_t>* rows) {
+ return chain(cursors, candidates, rows);
+}
+
+namespace {
+
+// The most rows of the cheapest cursor a stream lists at once.
+constexpr size_t kStreamRows = size_t {1} << 16;
+
+// The cheapest cursor of a stream, handing on its rows a batch at a time.
+class StreamLead {
+public:
+ explicit StreamLead(PostingsCursor& cursor) : _cursor(cursor) {}
+
+ Status start() { return _cursor.next_block(&_block, &_eof); }
+ bool exhausted() const { return _eof; }
+
+ // Replaces `rows` by the cursor's next rows, at most kStreamRows of them;
the rest of the
+ // current block waits for the next batch.
+ Status take_batch(std::vector<uint32_t>* rows) {
+ rows->clear();
+ while (!_eof) {
+ const size_t size = _block.size();
+ const size_t stop = std::min(size, _taken + (kStreamRows -
rows->size()));
+ if (_block.dense) {
+ const size_t old = rows->size();
+ rows->resize(old + stop - _taken);
+ std::iota(rows->begin() + old, rows->end(),
+ static_cast<uint32_t>(_block.range_begin + _taken));
+ } else {
+ rows->insert(rows->end(), _block.docs.begin() + _taken,
_block.docs.begin() + stop);
+ }
+ if (stop < size) {
+ _taken = stop;
+ return Status::OK();
+ }
+ RETURN_IF_ERROR(_cursor.next_block(&_block, &_eof));
+ _taken = 0;
+ if (rows->size() == kStreamRows) {
+ return Status::OK();
+ }
+ }
+ return Status::OK();
+ }
+
+private:
+ PostingsCursor& _cursor;
+ PostingsBlock _block;
+ size_t _taken = 0;
+ bool _eof = false;
+};
+
+// A cursor narrowing a stream's rows; it keeps its current block for the next
batch's rows.
+class StreamFollower {
+public:
+ explicit StreamFollower(PostingsCursor& cursor) : _cursor(cursor) {}
+
+ // Appends the ascending `rows` the cursor holds to `out`; `ended` once it
holds no later row.
+ Status narrow(std::span<const uint32_t> rows, std::vector<uint32_t>* out,
bool* ended) {
+ size_t next = 0;
+ while (next < rows.size()) {
+ if (!_loaded || _last < rows[next]) {
+ bool eof = false;
+ RETURN_IF_ERROR(_cursor.seek_block(rows[next], &_block, &eof));
+ if (eof) {
+ *ended = true;
+ return Status::OK();
+ }
+ _loaded = true;
+ _last = _block.doc_at(_block.size() - 1);
+ }
+ const uint32_t first = _block.doc_at(0);
+ const size_t begin =
+ gallop_past(rows, next, [first](uint32_t doc) { return doc
< first; });
+ next = gallop_past(rows, begin, [this](uint32_t doc) { return doc
<= _last; });
+ if (next == begin) {
+ continue;
+ }
+ const auto slice = rows.subspan(begin, next - begin);
+ // A dense block holds every row in its span.
+ if (_block.dense) {
+ out->insert(out->end(), slice.begin(), slice.end());
+ } else {
+ intersect_block(_block.docs, slice, DocumentOutput {out});
+ }
+ }
+ return Status::OK();
+ }
+
+private:
+ PostingsCursor& _cursor;
+ PostingsBlock _block;
+ uint32_t _last = 0;
+ bool _loaded = false;
+};
+
+} // namespace
+
+Status stream_cursors(std::span<PostingsCursor* const> cursors, DocIdSink&
sink) {
+ DORIS_CHECK(!cursors.empty());
+ std::vector<PostingsCursor*> order(cursors.begin(), cursors.end());
+ std::ranges::stable_sort(order, {},
+ [](const PostingsCursor* cursor) { return
cursor->doc_freq(); });
+ for (PostingsCursor* cursor : order) {
+ RETURN_IF_ERROR(cursor->prefetch(nullptr, /*positions=*/false));
+ }
+ StreamLead lead(*order.front());
+ std::vector<StreamFollower> followers;
+ followers.reserve(order.size() - 1);
+ for (size_t i = 1; i < order.size(); ++i) {
+ followers.emplace_back(*order[i]);
+ }
+ std::vector<uint32_t> rows;
+ std::vector<uint32_t> narrowed;
+ RETURN_IF_ERROR(lead.start());
+ while (!lead.exhausted()) {
+ RETURN_IF_ERROR(lead.take_batch(&rows));
+ bool ended = false;
+ for (StreamFollower& follower : followers) {
+ narrowed.clear();
+ RETURN_IF_ERROR(follower.narrow(rows, &narrowed, &ended));
+ rows.swap(narrowed);
+ if (rows.empty()) {
+ break;
+ }
+ }
+ if (!rows.empty()) {
+ RETURN_IF_ERROR(sink.append_sorted(rows));
+ }
+ if (ended) {
+ return Status::OK();
+ }
+ }
+ return Status::OK();
+}
+
} // namespace doris::index_query
diff --git a/be/src/storage/index/inverted/query/exec/cursor_chained_postings.h
b/be/src/storage/index/inverted/query/exec/cursor_chained_postings.h
index 1c6e829c037..ae95fd35851 100644
--- a/be/src/storage/index/inverted/query/exec/cursor_chained_postings.h
+++ b/be/src/storage/index/inverted/query/exec/cursor_chained_postings.h
@@ -26,6 +26,7 @@
#include "common/status.h"
#include "storage/index/inverted/query/exec/chained_conjunction.h"
+#include "storage/index/inverted/query/exec/docid_sink.h"
#include "storage/index/inverted/query/spi/postings_cursor.h"
namespace doris::index_query {
@@ -60,6 +61,14 @@ Status for_each_block(PostingsCursor& cursor, Visit&& visit)
{
}
}
+// Appends every document the cursor lists from its position to `sink`, a
dense block as a range.
+inline Status list_blocks(PostingsCursor& cursor, DocIdSink& sink) {
+ return for_each_block(cursor, [&sink](const PostingsBlock& block) {
+ return block.dense ? sink.append_range(block.range_begin,
block.range_end)
+ : sink.append_sorted(block.docs);
+ });
+}
+
// Calls `visit(block, slice)` for every block of the cursor that can hold one
of the
// ascending `candidates`, with the candidates between its first and last
document. Only those
// blocks are decoded.
@@ -123,5 +132,12 @@ private:
// narrows the cheapest cursor's rows by the others.
Status chain_cursors(std::span<const std::unique_ptr<PostingsCursor>> cursors,
const std::vector<uint32_t>* candidates,
std::vector<uint32_t>* rows);
+Status chain_cursors(std::span<PostingsCursor* const> cursors,
+ const std::vector<uint32_t>* candidates,
std::vector<uint32_t>* rows);
+
+// Appends to `sink` the rows every one of `cursors` holds, a batch of at most
2^16 rows of the
+// cheapest cursor at a time, which the others narrow, so no list holds more
than a batch. Each
+// cursor is read in order, which suits a source that reads on demand.
+Status stream_cursors(std::span<PostingsCursor* const> cursors, DocIdSink&
sink);
} // namespace doris::index_query
diff --git
a/be/test/storage/index/inverted/query/engine/boolean_query/listed_terms_test.cpp
b/be/test/storage/index/inverted/query/engine/boolean_query/listed_terms_test.cpp
index 04ea00e21a3..fe3ec36a32f 100644
---
a/be/test/storage/index/inverted/query/engine/boolean_query/listed_terms_test.cpp
+++
b/be/test/storage/index/inverted/query/engine/boolean_query/listed_terms_test.cpp
@@ -23,6 +23,7 @@
#include <functional>
#include <map>
#include <memory>
+#include <numeric>
#include <roaring/roaring.hh>
#include <string>
#include <utility>
@@ -161,6 +162,13 @@ FakeIndexSource::Posting posting(uint32_t doc,
std::vector<uint32_t> positions)
return {.doc = doc, .positions = std::move(positions)};
}
+// Every row of a source holding `doc_count` rows.
+std::vector<uint32_t> every_row(uint32_t doc_count) {
+ std::vector<uint32_t> docs(doc_count);
+ std::iota(docs.begin(), docs.end(), 0U);
+ return docs;
+}
+
// More terms than a source batching its reads reads in one round, and their
names in the waves
// the engine reads them in.
constexpr size_t kManyTerms = 40;
@@ -476,6 +484,71 @@ TEST_F(ListedTermsTest,
AConjunctionChainsItsTermsFromTheCheapest) {
EXPECT_EQ(listed->fetches, 0U);
}
+// A term held by every row cannot narrow a conjunction, so the chain never
reads it.
+TEST_F(ListedTermsTest, AConjunctionSkipsATermHeldByEveryRow) {
+ for (const bool batches : {false, true}) {
+ auto input = source(batches);
+ input->add("every", every_row(input->doc_count()));
+ ListedTerms group(input, nullptr);
+ group.add(0, "a");
+ group.add(1, "every");
+ group.add(2, "b");
+ group.open(/*conjunctive=*/true);
+ EXPECT_EQ(group.conjunction(nullptr).true_rows,
roaring::Roaring::bitmapOf(3, 2U, 3U, 8U));
+ EXPECT_TRUE(input->prefetches["every"].empty());
+ EXPECT_EQ(input->prefetches["a"].size(), 1U);
+ EXPECT_EQ(input->prefetches["b"].size(), 1U);
+ }
+}
+
+// A conjunction of terms held by every row holds every row, or every
candidate, without reading
+// any of them.
+TEST_F(ListedTermsTest, AConjunctionOfTermsHeldByEveryRowHoldsEveryRow) {
+ auto input = source(true);
+ input->add("every", every_row(input->doc_count()));
+ input->add("each", every_row(input->doc_count()));
+ const auto both = [&] {
+ ListedTerms group(input, nullptr);
+ group.add(0, "every");
+ group.add(1, "each");
+ group.open(/*conjunctive=*/true);
+ return group;
+ };
+ roaring::Roaring all_rows;
+ all_rows.addRange(0, input->doc_count());
+ EXPECT_EQ(both().conjunction(nullptr).true_rows, all_rows);
+ const std::vector<uint32_t> candidates {3, 8, 20};
+ EXPECT_EQ(both().conjunction(&candidates).true_rows,
+ roaring::Roaring::bitmapOf(3, 3U, 8U, 20U));
+ EXPECT_TRUE(input->prefetches["every"].empty());
+ EXPECT_TRUE(input->prefetches["each"].empty());
+}
+
+// The one term left beside a term held by every row lists its own rows, or
those among the
+// candidates.
+TEST_F(ListedTermsTest, ALoneTermListsItsRowsOrThoseAmongTheCandidates) {
+ for (const bool batches : {false, true}) {
+ auto input = source(batches);
+ input->add("every", every_row(input->doc_count()));
+ const auto lone = [&] {
+ ListedTerms group(input, nullptr);
+ group.add(0, "every");
+ group.add(1, "c");
+ group.open(/*conjunctive=*/true);
+ return group;
+ };
+ EXPECT_EQ(lone().conjunction(nullptr).true_rows,
+ roaring::Roaring::bitmapOf(3, 3U, 8U, 20U));
+ const std::vector<uint32_t> candidates {2, 8, 20, 30};
+ EXPECT_EQ(lone().conjunction(&candidates).true_rows,
+ roaring::Roaring::bitmapOf(2, 8U, 20U));
+ ASSERT_EQ(input->prefetches["c"].size(), 2U);
+ EXPECT_TRUE(input->prefetches["c"][0].whole);
+ EXPECT_EQ(input->prefetches["c"][1].candidates, candidates);
+ EXPECT_TRUE(input->prefetches["every"].empty());
+ }
+}
+
// Collecting every row of a scored disjunction lists its terms one after
another; a TopK on a
// source that does not batch its reads streams them side by side instead of
holding every
// matching row and score first, and ranks the rows alike.
@@ -576,9 +649,10 @@ TEST_F(ListedTermsTest,
OtherClausesRunInCostOrderAroundTheChain) {
EXPECT_TRUE(listed->prefetches["c"][0].whole);
}
-// An unscored conjunction chains its terms on a source reading them one at a
time too: the
-// chain seeks the blocks the cheaper terms' rows fall in, and the source has
no round to fetch.
-TEST_F(ListedTermsTest, AConjunctionOnAStreamingSourceChainsItsTermsToo) {
+// An unscored conjunction on a source reading its terms one at a time streams
them a window of
+// rows at a time, so each term may be asked for any of its rows, and the
source has no round to
+// fetch.
+TEST_F(ListedTermsTest, AConjunctionOnAStreamingSourceStreamsItsTerms) {
auto streamed = source(false);
const auto rows = evaluate(
[&] {
@@ -588,14 +662,12 @@ TEST_F(ListedTermsTest,
AConjunctionOnAStreamingSourceChainsItsTermsToo) {
EXPECT_EQ(rows.true_rows, roaring::Roaring::bitmapOf(2, 3U, 8U));
EXPECT_EQ(rows.null_rows, roaring::Roaring::bitmapOf(2, 40U, 41U));
EXPECT_EQ(streamed->opened_together,
(std::vector<std::vector<std::string>> {{"a", "b", "c"}}));
- ASSERT_EQ(streamed->prefetches["c"].size(), 1U);
- EXPECT_TRUE(streamed->prefetches["c"][0].whole);
- ASSERT_EQ(streamed->prefetches["b"].size(), 1U);
- EXPECT_EQ(streamed->prefetches["b"][0].candidates, (std::vector<uint32_t>
{3, 8, 20}));
- ASSERT_EQ(streamed->prefetches["a"].size(), 1U);
- EXPECT_EQ(streamed->prefetches["a"][0].candidates, (std::vector<uint32_t>
{3, 8}));
+ for (const char* term : {"a", "b", "c"}) {
+ ASSERT_EQ(streamed->prefetches[term].size(), 1U);
+ EXPECT_TRUE(streamed->prefetches[term][0].whole);
+ }
EXPECT_EQ(streamed->fetches, 0U);
- // A disjunction on it still streams.
+ // A disjunction on it opens its terms one at a time.
auto any = source(false);
evaluate([&] { return boolean(OperatorType::OP_OR, {term("a"),
term("c")}); }, any);
EXPECT_TRUE(any->opened_together.empty());
diff --git
a/be/test/storage/index/inverted/query/exec/cursor_chained_postings_test.cpp
b/be/test/storage/index/inverted/query/exec/cursor_chained_postings_test.cpp
index c5487075a5f..40e3bb1ca81 100644
--- a/be/test/storage/index/inverted/query/exec/cursor_chained_postings_test.cpp
+++ b/be/test/storage/index/inverted/query/exec/cursor_chained_postings_test.cpp
@@ -20,8 +20,12 @@
#include <gtest/gtest.h>
#include <algorithm>
+#include <array>
#include <cstdint>
+#include <iterator>
#include <limits>
+#include <numeric>
+#include <random>
#include <span>
#include <utility>
#include <vector>
@@ -238,6 +242,142 @@ TEST(CursorChainedPostings,
ChainsCursorsThroughTheSharedConjunction) {
EXPECT_EQ(wide_prefetches[0].candidates, (std::vector<uint32_t> {3, 9,
12}));
}
+// The rows a sink is handed, batch by batch, a range as its rows.
+class BatchSink final : public DocIdSink {
+public:
+ Status append_sorted(std::span<const uint32_t> docids) override {
+ batches.emplace_back(docids.begin(), docids.end());
+ return Status::OK();
+ }
+ Status append_range(uint32_t first, uint64_t last_exclusive) override {
+ std::vector<uint32_t> rows(last_exclusive - first);
+ std::iota(rows.begin(), rows.end(), first);
+ batches.push_back(std::move(rows));
+ return Status::OK();
+ }
+
+ std::vector<uint32_t> rows() const {
+ std::vector<uint32_t> all;
+ for (const auto& batch : batches) {
+ all.insert(all.end(), batch.begin(), batch.end());
+ }
+ return all;
+ }
+
+ std::vector<std::vector<uint32_t>> batches;
+};
+
+constexpr uint32_t kBatch = 1U << 16;
+
+// A stream hands on at most 2^16 of the cheapest cursor's rows at a time, and
another cursor's
+// block reaching into the next batch is read once.
+TEST(CursorChainedPostings, AStreamHandsOnABatchOfRowsAtATime) {
+ BlockedCursor rare({BlockedCursor::dense(0, 40000),
BlockedCursor::dense(40000, 70000)});
+ BlockedCursor wide({BlockedCursor::listed({5, kBatch - 1, kBatch, 69999,
70001}),
+ BlockedCursor::dense(80000, 200000)});
+ const std::vector<PostingsCursor*> cursors = {&wide, &rare};
+ BatchSink sink;
+ ASSERT_TRUE(stream_cursors(cursors, sink).ok());
+ EXPECT_EQ(sink.batches,
+ (std::vector<std::vector<uint32_t>> {{5, kBatch - 1}, {kBatch,
69999}}));
+ EXPECT_EQ(rare.decoded, 2U);
+ EXPECT_EQ(wide.decoded, 1U);
+}
+
+// The other cursors of a stream read only the blocks the cheapest cursor's
rows reach.
+TEST(CursorChainedPostings, AStreamReadsOnlyTheBlocksTheCheapestRowsReach) {
+ BlockedCursor rare({BlockedCursor::listed({10, 3 * kBatch + 10})});
+ std::vector<BlockedCursor::Block> blocks;
+ for (uint32_t block = 0; block < 64; ++block) {
+ blocks.push_back(BlockedCursor::dense(block * 4096, block * 4096 +
4000));
+ }
+ BlockedCursor wide(blocks);
+ const std::vector<PostingsCursor*> cursors = {&wide, &rare};
+ BatchSink sink;
+ ASSERT_TRUE(stream_cursors(cursors, sink).ok());
+ EXPECT_EQ(sink.batches, (std::vector<std::vector<uint32_t>> {{10, 3 *
kBatch + 10}}));
+ EXPECT_EQ(wide.decoded, 2U);
+}
+
+// A stream ends once a cursor holds no further row, without reading the
cheapest cursor's later
+// blocks.
+TEST(CursorChainedPostings, AStreamEndsWhenACursorEnds) {
+ BlockedCursor rare({BlockedCursor::dense(0, kBatch),
+ BlockedCursor::dense(100000, 100000 + kBatch),
+ BlockedCursor::listed({300000}),
BlockedCursor::listed({400000})});
+ BlockedCursor wide({BlockedCursor::dense(0, 140000)});
+ const std::vector<PostingsCursor*> cursors = {&rare, &wide};
+ BatchSink sink;
+ ASSERT_TRUE(stream_cursors(cursors, sink).ok());
+ ASSERT_EQ(sink.batches.size(), 2U);
+ EXPECT_EQ(sink.batches[0].size(), kBatch);
+ EXPECT_EQ(sink.batches[1].front(), 100000U);
+ EXPECT_EQ(sink.batches[1].back(), 139999U);
+ EXPECT_EQ(sink.batches[1].size(), 40000U);
+ EXPECT_EQ(rare.decoded, 3U);
+}
+
+// Random cursors of dense and listed blocks, some cheapest ones holding
several batches: the
+// stream lists the rows the chain lists.
+TEST(CursorChainedPostings, AStreamListsTheRowsTheChainLists) {
+ std::mt19937 random(20261017);
+ for (int round = 0; round < 24; ++round) {
+ const size_t count = 2 + round % 3;
+ std::vector<std::vector<BlockedCursor::Block>> layouts(count);
+ std::vector<uint32_t> expected;
+ for (size_t c = 0; c < count; ++c) {
+ const double rate = std::array {0.002, 0.05, 0.5, 0.97}[(round +
c) % 4];
+ std::bernoulli_distribution holds(rate);
+ std::vector<uint32_t> docs;
+ for (uint32_t row = 0; row < 3 * kBatch + 4096; ++row) {
+ if (holds(random)) {
+ docs.push_back(row);
+ }
+ }
+ for (size_t begin = 0; begin < docs.size();) {
+ const size_t end = std::min(docs.size(), begin + 1 + random()
% 600);
+ if (docs[end - 1] - docs[begin] == end - begin - 1) {
+ layouts[c].push_back(BlockedCursor::dense(docs[begin],
docs[end - 1] + 1));
+ } else {
+ layouts[c].push_back(BlockedCursor::listed(
+ std::vector<uint32_t>(docs.begin() + begin,
docs.begin() + end)));
+ }
+ begin = end;
+ }
+ if (c == 0) {
+ expected = docs;
+ } else {
+ std::vector<uint32_t> both;
+ std::ranges::set_intersection(expected, docs,
std::back_inserter(both));
+ expected.swap(both);
+ }
+ }
+ std::vector<BlockedCursor> streamed;
+ std::vector<BlockedCursor> chained;
+ streamed.reserve(count);
+ chained.reserve(count);
+ for (const auto& layout : layouts) {
+ streamed.emplace_back(layout);
+ chained.emplace_back(layout);
+ }
+ std::vector<PostingsCursor*> stream_input;
+ std::vector<PostingsCursor*> chain_input;
+ for (size_t c = 0; c < count; ++c) {
+ stream_input.push_back(&streamed[c]);
+ chain_input.push_back(&chained[c]);
+ }
+ BatchSink sink;
+ ASSERT_TRUE(stream_cursors(stream_input, sink).ok());
+ std::vector<uint32_t> rows;
+ ASSERT_TRUE(chain_cursors(chain_input, nullptr, &rows).ok());
+ EXPECT_EQ(rows, expected) << round;
+ EXPECT_EQ(sink.rows(), expected) << round;
+ for (const auto& batch : sink.batches) {
+ EXPECT_LE(batch.size(), kBatch) << round;
+ }
+ }
+}
+
// A cursor without its own block positions reads each asked document through
its positions.
TEST(CursorChainedPostings, BlockPositionsByDefaultReadEachDocument) {
FakePostingsCursor cursor({{.doc = 2, .positions = {1, 4}},
@@ -320,6 +460,34 @@ TEST(CursorChainedPostings,
OrdinalIntersectionCoversSparseWideAndBoundedBlocks)
}
}
+// Random documents and candidates of a block's span, as unrelated terms of
several densities meet.
+TEST(CursorChainedPostings, IntersectionHandlesRandomDensities) {
+ std::mt19937 random(20261016);
+ for (const uint32_t span : {256U, 1024U, 16384U}) {
+ for (const double doc_rate : {0.05, 0.5, 0.95}) {
+ for (const double candidate_rate : {0.05, 0.5, 0.95}) {
+ std::bernoulli_distribution holds_doc(doc_rate);
+ std::bernoulli_distribution holds_candidate(candidate_rate);
+ std::vector<uint32_t> docs;
+ std::vector<uint32_t> candidates;
+ for (uint32_t doc = 0; doc < span; ++doc) {
+ if (holds_doc(random)) {
+ docs.push_back(doc);
+ }
+ if (holds_candidate(random)) {
+ candidates.push_back(doc);
+ }
+ }
+ ASSERT_FALSE(docs.empty());
+ std::erase_if(candidates, [&](uint32_t doc) {
+ return doc < docs.front() || doc > docs.back();
+ });
+ check_ordinals(docs, candidates);
+ }
+ }
+ }
+}
+
TEST(CursorChainedPostings, RetainsPartialSparseBlockOrdinals) {
BlockedCursor cursor({BlockedCursor::listed({1, 3, 8}),
BlockedCursor::dense(10, 13),
BlockedCursor::listed({20, 21}),
BlockedCursor::listed({30, 34, 40})});
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
