This is an automated email from the ASF dual-hosted git repository.
airborne12 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new f937d950d45 [improvement](be) Speed up dense unscored inverted index
conjunctions (#69172)
f937d950d45 is described below
commit f937d950d45313f20864d287df453e1b90f5c615
Author: Jack <[email protected]>
AuthorDate: Sat Oct 17 22:25:49 2026 +0800
[improvement](be) Speed up dense unscored inverted index conjunctions
(#69172)
### What problem does this PR solve?
Issue Number: N/A
Related PR: #68758
Problem Summary:
#68758 runs an unscored AND of terms on one field, such as MATCH_ALL, as
a chain: the rarest term's rows are listed into an array and the other
terms narrow them into another. On dense terms this was slower than the
bitmap intersection it replaced on V2 segments, and its arrays grew with
the segment (review comment M2 on #68758):
1. A term held by every row cannot remove any row, but the chain still
read its postings. For `MATCH_ALL 'http status'`, where both terms are
held by every row, it copied every row id twice and rebuilt the result
bitmap from the array: 14.4 ms on a 2M-row V2 segment, about 10 times
what it took before #68758.
2. A term left alone, such as `get` in `get http` once `http` is
skipped, was listed into an array of all its rows before the result
bitmap was built from it. For a term in 90% of 2M rows that cost about
1,200 extra page faults per query.
3. Intersecting a block through a bit set tested each candidate with a
branch. For two unrelated terms each in half the rows (`alpha beta`),
the branch mispredicted about half the time, and the intersection took
30% of the scan thread's cycles.
4. The arrays held up to 8 bytes per row of the segment in each scanner.
This PR:
- skips a term whose document frequency equals the source's document
count, so a conjunction whose terms are all held by every row returns
every row, or every candidate, without reading postings;
- lists a lone term's blocks straight into the result bitmap, a dense
block as a range;
- intersects a block through its bit set without a branch per candidate:
every candidate is written and kept only when its bit is set;
- streams a conjunction of several terms on a source that reads on
demand, such as a V2 segment: the cheapest term lists at most 2^16 of
its rows at a time, the others narrow them, and the rows left go
straight into the result, so no array holds more than a batch. The other
terms keep their current block from one batch to the next, so each block
is still decoded once. A batching source, such as SNII, keeps the chain,
which reads each term's blocks in one round.
Scored queries keep their path, and results are unchanged.
Index filter time of MATCH_ALL on a 2M-row V2 segment, as the median
ratio of 8 paired runs on fresh backends; each column is a separate
comparison and lower is faster:
| Query | Rows holding each term | vs #68758 | vs before #68758 |
|---|---|---|---|
| `http status` | 100%, 100% | 0.01 | 0.10 |
| `rare http` | 0.05%, 100% | 0.31 | 0.10 |
| `sparse http` | 0.5%, 100% | 0.23 | 0.31 |
| `get http` | 90%, 100% | 0.60 | 0.95 |
| `alpha beta` | 50%, 50% | 0.56 | 0.72 |
In the comparison with #68758, the `http status` filter drops from 16.9
ms to 0.15 ms and its end-to-end time to 0.63, and the end-to-end time
of `alpha beta` to 0.81. SNII segments take the same paths except the
stream; there these queries take 0.02 to 0.83 of #68758 and 0.02 to 0.93
of their time before #68758. Seven more shapes, from `rare alpha` and
`sparse alpha` to `alpha beta get` and the empty `get post`, take 0.20
to 0.89 of their time before #68758 on V2 and 0.57 to 1.00 on SNII.
Two-term MATCH_ALL queries on short text take 0.86 to 1.01 of #68758,
and the 18 queries used to accept #68758 (MATCH_PHRASE, MATCH_ANY and
SEARCH phrases on V2 and SNII) stay between 0.97 and 1.03 of it, every
95% interval including 1.
On V2 the stream holds at most two arrays of 2^16 rows per scanner, and
it cuts the page faults per query of `alpha beta` from 458 to 192 and
those of `get alpha` from 484 to 47. Its scan threads run within 1% of
the chain's instructions, and a cheapest term of at most 2^16 rows is
listed in one batch, as the chain lists it.
### Release note
None
### Check List (For Author)
- Test <!-- At least one of them must be included. -->
- [ ] Regression test
- [x] Unit Test
- [x] Manual test (add detailed scripts or steps below)
- [ ] No need to test or manual test. Explain why:
- [ ] This is a refactor/code format and no logic has been changed.
- [ ] Previous test can cover this change.
- [ ] No code files have been changed.
- [ ] Other reason <!-- Add your reason? -->
Unit test: `ListedTermsTest.AConjunctionSkipsATermHeldByEveryRow` and
`ListedTermsTest.AConjunctionOfTermsHeldByEveryRowHoldsEveryRow` fail
before the first commit, because the chain read the term held by every
row, and pass after it.
`ListedTermsTest.AConjunctionOnAStreamingSourceStreamsItsTerms` replaces
`AConjunctionOnAStreamingSourceChainsItsTermsToo`, which recorded that
such a source chained its terms, and fails before the last commit.
`ListedTermsTest.ALoneTermListsItsRowsOrThoseAmongTheCandidates`,
`CursorChainedPostings.IntersectionHandlesRandomDensities` and the
`CursorChainedPostings.AStream*` tests cover the lone-term listing, the
intersection and the stream; each fails when the code it covers is
broken on purpose.
Manual test: a 2M-row corpus with terms held by 100%, 90%, 50%, 10%,
0.5% and 0.05% of the rows was loaded into a V2 table and an SNII table.
Each of 8 pairs started two fresh backends, one per build, swapping
which backend ran which build on alternate pairs; each query ran 4
warm-up and 16 measured times on each build, alternating between the
builds. Every result matched aggregates computed independently from the
corpus. Scan-thread CPU time, cycles, instructions and page faults were
sampled with `perf stat` over 200 runs of each query on each build.
- Behavior changed:
- [x] No.
- [ ] Yes. <!-- Explain the behavior change -->
- Does this need documentation?
- [x] No.
- [ ] Yes. <!-- Add document PR link here. eg:
https://github.com/apache/doris-website/pull/1214 -->
### Check List (For Reviewer who merge this PR)
- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label <!-- Add branch pick label that this PR
should merge into -->
---
.../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]
