This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.1 by this push:
new 3955a13c143 branch-4.1: [fix](inverted index) Count only visible rows
in BM25 statistics of merge-on-write tables #69438 (#69751)
3955a13c143 is described below
commit 3955a13c143d1eb1ce003f7628ab719feb2543bc
Author: guptashivam04 <[email protected]>
AuthorDate: Thu Oct 29 07:06:52 2026 +0530
branch-4.1: [fix](inverted index) Count only visible rows in BM25
statistics of merge-on-write tables #69438 (#69751)
### What problem does this PR solve?
Issue Number: None
Related PR: #69438 (this is its backport to branch-4.1), #69352 and its
branch-4.1 pick #69418 (BM25 statistics shared across the parallel
scanners of a tablet)
Problem Summary:
Backport of #69438 to branch-4.1.
On a merge-on-write table, BM25 `score()` gives the same row a different
score on each replica of its tablet. A query picks one replica per
tablet, so with `replication_num = 3` a row has up to three scores, and
pagination over `ORDER BY score()` repeats and skips rows from one page
to the next. `CollectionStatistics::process_segment()` takes the
document count, the token count and each term's document frequency
straight from every segment's inverted index. Rows a later load
overwrote or deleted (marked only in the delete bitmap) and rows written
with the delete sign (CDC deletes, `DELETE`) stay in the index and are
counted until compaction removes them, and each replica compacts on its
own schedule.
On branch-4.1 (3 BEs, `replication_num = 3`, 8 buckets, CDC-style
upserts and delete-sign deletes, the replicas on one BE fully
compacted), the 1,731 matching rows of a `SEARCH(... OR ...)` query had
1,294 rows with more than one score over 15 runs, and all 1,731 differed
between `use_fix_replica = 0/1/2`.
Fix (same as #69438):
When the scan reads a delete bitmap (merge-on-write),
`CollectionStatistics` describes only the rows the scan returns:
- `OlapScanner` passes the scan's delete bitmap and read version
(`MowHiddenRows`) to `collect()`, or to
`TabletCollectionStatistics::copy_to()` for a split tablet (#69418).
Every scanner of a tablet reads with the same bitmap and version.
- For each segment, the hidden rows are the rows the delete bitmap marks
at the read version (`get_agg`), plus the rows whose
`__DORIS_DELETE_SIGN__` is set. The delete-sign column is read only for
segments whose zone map shows a non-zero sign (or cannot tell),
following the query's `enable_segment_cache` and `enable_page_cache`.
- Document count: the index's document count minus the hidden rows.
- Term document frequency: the hidden rows a term's postings list are
subtracted. When the hidden rows are much rarer than the postings (32
times), the collector leaps from hidden row to hidden row; otherwise it
decodes the postings once and walks the hidden rows alongside, without a
bitset.
- Token count: the sum of the document lengths the norms of the returned
non-NULL rows stand for, or the returned row count for an index without
norms. A row's norm is the same in every segment it is written to, so
every replica counts the same, compacted or not. NULL rows are left out
of the token count through the index's null bitmap and still count as
documents.
- The statistics read CLucene norms one CLucene segment at a time into a
buffer of their own (`IndexReader::norms(field, bytes)`) and keep none.
What they keep is small and lives with the cached searcher: a field's
non-NULL norm length sum, and the length sums of the last four sets of
hidden rows of a field. The hidden and NULL row ID lists are charged to
the query's memory.
- A tablet whose rows are all hidden reports zero documents instead of
failing the query with "No data available for SimilarityCollector".
Tables without a delete bitmap (duplicate key, merge-on-read,
`skip_delete_bitmap`) collect the same statistics as before.
For every `score()` query: the statistics' I/O context follows the
query's `disable_file_cache`, and a CLucene searcher the statistics open
drops the query's I/O context once its meta is read, as the query path
does. A CLucene error while opening it now fails the query; before, it
escaped the scanner thread and aborted the backend (reproduced on
branch-4.1 by
`fault_injection_p0/test_bm25_score_statistics_searcher_io_context`).
Differences from #69438, for branch-4.1:
- branch-4.1 has no SNII format and no `IndexSource` layer. The
hidden-row counting works on `TermIterator` (`read_range` / `advance`)
and the norm reading on the CLucene reader directly; both are helpers in
`storage/compaction/collection_statistics.cpp`
(`collection_statistics_detail`).
- The norm length sums are kept on
`InvertedIndexSearcherCache::CacheValue`; branch-4.1 has no
`CluceneSearcherCacheValue`.
- `InvertedIndexReader::read_null_bitmap` is split so that the
statistics can read the null bitmap through the query cache without an
`InvertedIndexReader` (`read_inverted_index_null_bitmap()`).
- `inverted_index_snii_read_no_write_file_cache` does not exist on
branch-4.1, so only `disable_file_cache` is followed.
Not covered (as in #69438): merge-on-read unique tables and duplicate
tables with `DELETE` predicates; rows deleted by a delete predicate
(`enable_mow_light_delete`); rows written with the delete sign that the
query returns (`skip_delete_sign`, `show_hidden_columns`) are still left
out of the statistics; an analyzed empty string counts as one token on a
merge-on-write table. Statistics stay per tablet.
### Release note
Fix BM25 `score()` on merge-on-write tables differing between the
replicas of a tablet, and before and after compaction. Statistics now
count only the rows a query can return, and the average document length
comes from the indexed norms.
### Check List (For Author)
- Test
- [x] Regression test:
- New: `inverted_index_p0/test_bm25_score_replica_statistics` (3
replicas, overwrites, `DELETE`, delete-sign rows, one replica fully
compacted, scores compared with a freshly loaded table; NULL rows vs a
duplicate table; all rows deleted) and
`fault_injection_p0/test_bm25_score_statistics_searcher_io_context`
(same suites as #69438).
- Updated `.out`: `test_bm25_score.out` and the V2/V3 merge-on-write
cases of `test_score_filter_matrix.out`, to the same values as #69438.
- All 23 branch-4.1 suites that use `score()` (`inverted_index_p0`,
`search`, `ann_index_p0`, `variant_p0`, `nereids_rules_p0`,
`fault_injection_p0`) pass with this change on a 3-BE cluster. Without
it, the two new suites and the two updated ones fail; the
fault-injection suite aborts two of the three backends with an uncaught
`CLuceneError`.
- [x] Unit Test:
`CollectionStatisticsTest.HiddenRowsAreCollectedLikeCompactedAway` (real
CLucene segments: a segment with a hidden 45-token row collects the same
doc count, token count, df, avgdl and idf as the segment without it),
`CollectionStatisticsTest.AllRowsHiddenCollectsEmptyStatistics`,
`CollectionStatisticsTest.NullRowsCountAsDocumentsWithoutTokens` (real
segment with NULL rows, with and without a delete bitmap),
`CollectionStatisticsTest.CountListedDocsLeapsOrScansOverHiddenRows`
(both counting paths, dense blocks, norm length sums),
`CollectionStatisticsHiddenRowsTest.*` (`exclude_hidden_docs` with and
without norms and NULL rows, rows beyond the index),
`InvertedIndexSearcherCacheValueTest.*` (norm length sums computed once
per field and per set of hidden rows, failures not kept, oldest set
evicted), `MultiSegmentCollectorTest.NormLengthSumsReadEachSegment` (an
index of three CLucene segments sums its norms per segment, equal to the
combined norms),
`OlapScannerTest.BuildScoreRuntimeCollectionIoContextFollowsQueryFileCacheOptions`.
With the existing collection statistics, collection similarity, BM25,
searcher cache, searcher builder, olap scanner, multi-segment, boolean
query, null predicate and `SEARCH` function tests: 233 tests in 26
suites pass in an ASAN build on macOS arm64.
- [x] Manual test (3 BEs, macOS arm64, `replication_num = 3`, auto
compaction off, `enable_sql_cache = false`), base = branch-4.1 without
this change, same data and compaction state for both builds:
| Check | branch-4.1 | with this change |
|---|---|---|
| CDC-style table (8 buckets, upserts, delete-sign deletes, NULLs, 4
fields), one BE's replicas fully compacted with inverted index
compaction off: rows with more than one score over 15 runs / differing
between replicas 0, 1, 2 (of 1,731) | 1,294 / 1,731 | 0 / 0 |
| Same, scores differing from a freshly loaded copy of the visible rows
| — | 0 of 1,731 |
| Partial-column updates, sequence column, stream load with the delete
sign and `merge_type: DELETE`, group commit, keyword and
lowercase-normalizer indexes, partitions, variant: rows differing
between replicas after one replica compacted | 84–790 per table (7 of 8
checks) | 0 in every check, all equal to a fresh copy |
| Merge-on-write vs duplicate table with 50 % NULL rows (V3) | 0 of 250
differ | 0 of 250 differ |
Cost, query `SEARCH('content:jobs OR content:try') ORDER BY score() DESC
LIMIT 100`, one bucket, median of 15 warm runs:
| Table | `StatisticsCollectTime`, one scanner | Query total, default
parallel scan |
|---|---|---|
| V2, 8 M rows + 200 k overwritten + 50 k delete-sign rows | 0.3 ms →
7.7 ms | 78 ms → 82 ms |
| V3, 6 M rows + 2 M overwritten + 200 k delete-sign rows (2.2 M hidden)
| 0.3 ms → 62 ms | 71 ms → 125 ms |
The first query after a restart also sums each segment's norms (8 M
rows: 5 ms → 32 ms; 2.2 M hidden: 6 ms → 99 ms). The cost grows with the
rows compaction has not removed yet.
Memory (BE allocated bytes, sum of 3 BEs, a 4-field merge-on-write table
of 41 segments): after 4 queries +16 MB → +22 MB with key pruning and
+109 MB → +75 MB scoring all segments, within the noise of the
measurement.
- Behavior changed:
- [x] Yes. On merge-on-write tables, BM25 statistics leave out rows
hidden by the delete bitmap and rows with the delete sign, and avgdl is
computed from the norms of the visible rows, so scores no longer depend
on the replica or on compaction. Scores of merge-on-write tables shift
once, also without deletes, because norms store lengths above 40 tokens
approximately.
- Does this need documentation?
- [x] No.
### Check List (For Reviewer who merge this PR)
- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label
---------
Co-authored-by: Shivam Gupta <[email protected]>
---
be/src/exec/scan/olap_scanner.cpp | 22 +-
.../storage/compaction/collection_statistics.cpp | 505 ++++++++++++++++++++-
be/src/storage/compaction/collection_statistics.h | 97 +++-
.../storage/index/inverted/inverted_index_cache.h | 67 +++
.../index/inverted/inverted_index_reader.cpp | 23 +-
.../storage/index/inverted/inverted_index_reader.h | 7 +
be/test/exec/scan/olap_scanner_test.cpp | 14 +
.../compaction/collection_statistics_test.cpp | 451 +++++++++++++++++-
.../query_v2/multi_segment_collector_test.cpp | 74 ++-
.../segment/inverted_index_searcher_cache_test.cpp | 74 +++
...t_bm25_score_statistics_searcher_io_context.out | 103 +++++
.../data/inverted_index_p0/test_bm25_score.out | 2 +-
.../test_bm25_score_replica_statistics.out | 6 +
.../inverted_index_p0/test_score_filter_matrix.out | 48 +-
...m25_score_statistics_searcher_io_context.groovy | 61 +++
.../test_bm25_score_replica_statistics.groovy | 178 ++++++++
16 files changed, 1669 insertions(+), 63 deletions(-)
diff --git a/be/src/exec/scan/olap_scanner.cpp
b/be/src/exec/scan/olap_scanner.cpp
index 7ebf0cba66a..28f35ba2ab2 100644
--- a/be/src/exec/scan/olap_scanner.cpp
+++ b/be/src/exec/scan/olap_scanner.cpp
@@ -170,6 +170,11 @@ io::IOContext
build_score_runtime_collection_io_context(RuntimeState* state, Rea
if (auto* query_ctx = state->get_query_ctx(); query_ctx != nullptr) {
io_ctx.remote_scan_cache_write_limiter =
query_ctx->remote_scan_cache_write_limiter();
}
+ // The statistics read the index before the scan does, so they follow the
query's file cache
+ // choice as the scan's reads do.
+ if (reader_type == ReaderType::READER_QUERY) {
+ io_ctx.is_disposable = state->query_options().disable_file_cache;
+ }
return io_ctx;
}
@@ -296,15 +301,28 @@ Status OlapScanner::prepare() {
_state, ReaderType::READER_QUERY,
tablet->file_cache_ttl_expiration_time(),
&_tablet_reader->mutable_stats()->file_cache_stats);
+ // A merge-on-write scan leaves out rows its segments still hold,
which replicas drop
+ // at different times, so the statistics count only the rows the scan
returns.
+ std::optional<MowHiddenRows> hidden_rows;
+ if (_tablet_reader_params.delete_bitmap != nullptr) {
+ const auto& query_options = _state->query_options();
+ hidden_rows = MowHiddenRows {
+ .delete_bitmap = _tablet_reader_params.delete_bitmap,
+ .version = _tablet_reader_params.version.second,
+ .use_segment_cache =
!query_options.__isset.enable_segment_cache ||
+ query_options.enable_segment_cache,
+ .use_page_cache = _state->enable_page_cache()};
+ }
+ const MowHiddenRows* hidden_rows_ptr = hidden_rows ? &*hidden_rows :
nullptr;
if (_tablet_collection_statistics != nullptr) {
RETURN_IF_ERROR(_tablet_collection_statistics->copy_to(
_state, _tablet_reader_params.tablet_schema,
- _tablet_reader_params.common_expr_ctxs_push_down, &io_ctx,
+ _tablet_reader_params.common_expr_ctxs_push_down, &io_ctx,
hidden_rows_ptr,
_tablet_reader_params.collection_statistics.get()));
} else {
RETURN_IF_ERROR(_tablet_reader_params.collection_statistics->collect(
_state, _tablet_reader_params.rs_splits,
_tablet_reader_params.tablet_schema,
- _tablet_reader_params.common_expr_ctxs_push_down,
&io_ctx));
+ _tablet_reader_params.common_expr_ctxs_push_down, &io_ctx,
hidden_rows_ptr));
}
}
diff --git a/be/src/storage/compaction/collection_statistics.cpp
b/be/src/storage/compaction/collection_statistics.cpp
index dda0ef1efc5..94ab5cefec6 100644
--- a/be/src/storage/compaction/collection_statistics.cpp
+++ b/be/src/storage/compaction/collection_statistics.cpp
@@ -17,11 +17,37 @@
#include "storage/compaction/collection_statistics.h"
+#ifdef __clang__
+#pragma clang diagnostic push
+#pragma clang diagnostic ignored "-Wshadow-field"
+#pragma clang diagnostic ignored "-Woverloaded-virtual"
+#pragma clang diagnostic ignored "-Winconsistent-missing-override"
+#pragma clang diagnostic ignored "-Wreorder-ctor"
+#pragma clang diagnostic ignored "-Wshorten-64-to-32"
+#elif defined(__GNUC__)
+#pragma GCC diagnostic push
+#pragma GCC diagnostic ignored "-Woverloaded-virtual"
+#endif
+#include <CLucene.h>
+#include <CLucene/index/MultiReader.h>
+#include <CLucene/index/_MultiSegmentReader.h>
+#ifdef __clang__
+#pragma clang diagnostic pop
+#elif defined(__GNUC__)
+#pragma GCC diagnostic pop
+#endif
+
+#include <algorithm>
+#include <array>
+#include <iterator>
#include <set>
#include <sstream>
#include <stack>
#include "common/exception.h"
+#include "core/column/column_vector.h"
+#include "core/custom_allocator.h"
+#include "core/data_type/data_type_factory.hpp"
#include "exprs/vexpr.h"
#include "exprs/vexpr_context.h"
#include "exprs/vliteral.h"
@@ -29,9 +55,20 @@
#include "storage/index/index_file_reader.h"
#include "storage/index/index_reader_helper.h"
#include "storage/index/inverted/analyzer/analyzer.h"
+#include "storage/index/inverted/inverted_index_cache.h"
+#include "storage/index/inverted/inverted_index_reader.h"
+#include "storage/index/inverted/similarity/bm25_similarity.h"
#include "storage/index/inverted/util/string_helper.h"
+#include "storage/index/inverted/util/term_iterator.h"
+#include "storage/index/zone_map/zone_map_index.h"
+#include "storage/iterators.h"
+#include "storage/rowset/beta_rowset.h"
#include "storage/rowset/rowset.h"
#include "storage/rowset/rowset_reader.h"
+#include "storage/segment/column_reader.h"
+#include "storage/segment/segment.h"
+#include "storage/segment/segment_loader.h"
+#include "storage/tablet/tablet_meta.h"
#include "util/uid_util.h"
namespace doris {
@@ -51,7 +88,19 @@ Result<std::shared_ptr<lucene::search::IndexSearcher>>
open_optional_clucene_sta
}
return ResultError(status);
}
- auto* reader = lucene::index::IndexReader::open(compound.value().get());
+ lucene::index::IndexReader* reader = nullptr;
+ try {
+ reader = lucene::index::IndexReader::open(compound.value().get());
+ } catch (const CLuceneError& e) {
+ return
ResultError(Status::Error<ErrorCode::INVERTED_INDEX_CLUCENE_ERROR>(
+ "CLuceneError occurred opening the index for BM25 statistics:
{}", e.what()));
+ }
+ // As the query path does once the index meta is read: the searcher may be
cached and read
+ // by later queries, which must not use this query's I/O context, and
later reads are of
+ // index data, not meta.
+ auto* stream = compound.value()->getDorisIndexInput();
+ stream->setIoContext(nullptr);
+ stream->setIndexFile(false);
return std::make_shared<lucene::search::IndexSearcher>(reader, true);
}
@@ -59,11 +108,378 @@ Result<std::shared_ptr<lucene::search::IndexSearcher>>
open_optional_clucene_sta
#include "common/compile_check_begin.h"
+namespace collection_statistics_detail {
+
+namespace {
+
+// How many postings a scan decodes in the time a leap costs.
+constexpr uint64_t kPostingsPerSeek = 32;
+
+// Leaps to each of the ascending `docs` in turn.
+uint64_t count_listed_docs_by_leaping(const segment_v2::TermIterator& postings,
+ std::span<const uint32_t> docs) {
+ uint64_t listed = 0;
+ int64_t doc = -1;
+ size_t next = 0;
+ while (next < docs.size()) {
+ if (doc < docs[next]) {
+ // skipTo always moves past the current document, so only leap
forward.
+ doc = postings.advance(cast_set<int32_t>(docs[next]));
+ if (doc == INT_MAX) {
+ break;
+ }
+ }
+ if (doc == docs[next]) {
+ ++listed;
+ ++next;
+ } else {
+ next = static_cast<size_t>(
+ std::lower_bound(docs.begin() + next, docs.end(),
static_cast<uint32_t>(doc)) -
+ docs.begin());
+ }
+ }
+ return listed;
+}
+
+// Decodes every posting once and walks the ascending `docs` alongside.
+uint64_t count_listed_docs_by_scan(const segment_v2::TermIterator& postings,
+ std::span<const uint32_t> docs) {
+ uint64_t listed = 0;
+ size_t next = 0;
+ DocRange range;
+ // A block without documents only comes from a corrupt image and ends the
term.
+ while (next < docs.size() && postings.read_range(&range) &&
range.doc_many_size_ != 0) {
+ if (range.type_ == DocRangeType::kRange) {
+ const auto begin =
+ std::lower_bound(docs.begin() + next, docs.end(),
range.doc_range.first);
+ const auto end = std::lower_bound(begin, docs.end(),
range.doc_range.second);
+ listed += static_cast<uint64_t>(end - begin);
+ next = static_cast<size_t>(end - docs.begin());
+ continue;
+ }
+ for (uint32_t i = 0; i < range.doc_many_size_; ++i) {
+ const uint32_t doc = (*range.doc_many)[i];
+ while (next < docs.size() && docs[next] < doc) {
+ ++next;
+ }
+ if (next == docs.size()) {
+ break;
+ }
+ if (docs[next] == doc) {
+ ++listed;
+ ++next;
+ }
+ }
+ }
+ return listed;
+}
+
+// The length each encoded norm stands for, as BM25 scoring decodes it.
+const std::array<uint64_t, 256>& norm_length_table() {
+ static const std::array<uint64_t, 256> table = [] {
+ std::array<uint64_t, 256> lengths {};
+ for (size_t norm = 0; norm < lengths.size(); ++norm) {
+ lengths[norm] = static_cast<uint64_t>(static_cast<float>(
+
segment_v2::BM25Similarity::byte4_to_int(static_cast<uint8_t>(norm))));
+ }
+ return lengths;
+ }();
+ return table;
+}
+
+// A CLucene segment of a reader and the range of the reader's documents it
holds.
+struct ReaderPartition {
+ lucene::index::IndexReader* reader;
+ uint32_t begin;
+ uint32_t end;
+};
+
+const lucene::util::ArrayBase<lucene::index::IndexReader*>* sub_readers(
+ lucene::index::IndexReader* reader) {
+ if (auto* multi_segment =
dynamic_cast<lucene::index::MultiSegmentReader*>(reader)) {
+ return multi_segment->getSubReaders();
+ }
+ if (auto* multi = dynamic_cast<lucene::index::MultiReader*>(reader)) {
+ return multi->getSubReaders();
+ }
+ return nullptr;
+}
+
+void append_partitions(lucene::index::IndexReader* reader, uint32_t begin,
+ std::vector<ReaderPartition>* partitions) {
+ if (const auto* subs = sub_readers(reader); subs != nullptr &&
subs->length != 0) {
+ for (size_t i = 0; i < subs->length; ++i) {
+ auto* child = (*subs)[i];
+ append_partitions(child, begin, partitions);
+ begin += static_cast<uint32_t>(child->maxDoc());
+ }
+ } else {
+ partitions->push_back({reader, begin, begin +
static_cast<uint32_t>(reader->maxDoc())});
+ }
+}
+
+// The statistics read a field's norms into a buffer of their own, freed after
use: asking a reader
+// for its norms makes it keep them, uncharged to the searcher cache, for
every segment the
+// statistics visit, scored or not. Reading per segment also avoids the copy
of all segments' norms
+// a reader of several segments would keep. A segment without the field's
norms reads as the
+// default norm, as that reader would.
+std::vector<ReaderPartition> norm_partitions(lucene::index::IndexReader*
reader) {
+ std::vector<ReaderPartition> partitions;
+ append_partitions(reader, 0, &partitions);
+ return partitions;
+}
+
+void read_norms(const ReaderPartition& partition, const std::wstring& field,
+ DorisVector<uint8_t>* norms) {
+ norms->assign(partition.end - partition.begin, 0);
+ partition.reader->norms(field.c_str(), norms->data());
+}
+
+Status clucene_norms_error(const char* what) {
+ return Status::Error<ErrorCode::INVERTED_INDEX_CLUCENE_ERROR>(
+ "CLuceneError occurred reading norms for BM25 statistics: {}",
what);
+}
+
+} // namespace
+
+uint64_t count_listed_docs(const segment_v2::TermIterator& postings,
+ std::span<const uint32_t> docs) {
+ const auto doc_freq = postings.doc_freq();
+ if (docs.empty() || doc_freq <= 0) {
+ return 0;
+ }
+ // Leaping costs a seek per wanted document, scanning a decode per
posting: leap only over
+ // documents much rarer than the postings.
+ if (docs.size() * kPostingsPerSeek >= static_cast<uint64_t>(doc_freq)) {
+ return count_listed_docs_by_scan(postings, docs);
+ }
+ return count_listed_docs_by_leaping(postings, docs);
+}
+
+Status sum_norm_lengths(lucene::index::IndexReader* reader, const
std::wstring& field,
+ uint64_t* out) {
+ *out = 0;
+ try {
+ if (!reader->hasNorms(field.c_str())) {
+ return Status::Error<ErrorCode::INVERTED_INDEX_NOT_SUPPORTED>(
+ "field {} has no norms", StringHelper::to_string(field));
+ }
+ const auto& lengths = norm_length_table();
+ DorisVector<uint8_t> norms;
+ for (const ReaderPartition& partition : norm_partitions(reader)) {
+ read_norms(partition, field, &norms);
+ for (const uint8_t norm : norms) {
+ *out += lengths[norm];
+ }
+ }
+ } catch (CLuceneError& e) {
+ return clucene_norms_error(e.what());
+ }
+ return Status::OK();
+}
+
+Status sum_norm_lengths_of(lucene::index::IndexReader* reader, const
std::wstring& field,
+ std::span<const uint32_t> docs, uint64_t* out) {
+ *out = 0;
+ if (docs.empty()) {
+ return Status::OK();
+ }
+ try {
+ if (!reader->hasNorms(field.c_str())) {
+ return Status::Error<ErrorCode::INVERTED_INDEX_NOT_SUPPORTED>(
+ "field {} has no norms", StringHelper::to_string(field));
+ }
+ const auto& lengths = norm_length_table();
+ const auto count = static_cast<uint32_t>(reader->maxDoc());
+ if (docs.back() >= count) {
+ return Status::Error<ErrorCode::INVERTED_INDEX_FILE_CORRUPTED>(
+ "document {} is beyond the {} documents of the index",
docs.back(), count);
+ }
+ DorisVector<uint8_t> norms;
+ // `docs` ascend, so each partition's documents are contiguous in them.
+ size_t next = 0;
+ for (const ReaderPartition& partition : norm_partitions(reader)) {
+ if (next == docs.size()) {
+ break;
+ }
+ if (docs[next] >= partition.end) {
+ continue;
+ }
+ read_norms(partition, field, &norms);
+ for (; next < docs.size() && docs[next] < partition.end; ++next) {
+ *out += lengths[norms[docs[next] - partition.begin]];
+ }
+ }
+ } catch (CLuceneError& e) {
+ return clucene_norms_error(e.what());
+ }
+ return Status::OK();
+}
+
+Status sum_non_null_norm_lengths(lucene::index::IndexReader* reader, const
std::wstring& field,
+ std::span<const uint32_t> null_docs,
uint64_t* out) {
+ const auto doc_count = static_cast<uint32_t>(reader->maxDoc());
+ if (!null_docs.empty() && null_docs.back() >= doc_count) {
+ return Status::Error<ErrorCode::INVERTED_INDEX_FILE_CORRUPTED>(
+ "NULL row {} is beyond the {} documents of the index",
null_docs.back(), doc_count);
+ }
+ uint64_t all_lengths = 0;
+ RETURN_IF_ERROR(sum_norm_lengths(reader, field, &all_lengths));
+ uint64_t null_lengths = 0;
+ RETURN_IF_ERROR(sum_norm_lengths_of(reader, field, null_docs,
&null_lengths));
+ DORIS_CHECK(null_lengths <= all_lengths);
+ *out = all_lengths - null_lengths;
+ return Status::OK();
+}
+
+Status exclude_hidden_docs(std::span<const uint32_t> hidden_docs, const
roaring::Roaring& null_docs,
+ uint64_t non_null_length_sum,
+ const SumNormLengthsOf& sum_norm_lengths_of,
+ SegmentScoringStats* stats) {
+ DORIS_CHECK(stats != nullptr);
+ if (!hidden_docs.empty() && hidden_docs.back() >= stats->doc_count) {
+ return Status::Error<ErrorCode::INVERTED_INDEX_FILE_CORRUPTED>(
+ "hidden row {} is beyond the {} documents of the index",
hidden_docs.back(),
+ stats->doc_count);
+ }
+ stats->doc_count -= hidden_docs.size();
+ if (!stats->has_norms) {
+ // Without norms every document stands for one token.
+ stats->token_count = stats->doc_count;
+ return Status::OK();
+ }
+ // The exact token count of the hidden rows is not stored anywhere, and
norms hold lengths
+ // only approximately, so count the lengths the norms of the returned rows
stand for. A
+ // replica that has compacted the hidden rows away counts the same rows
with the same norms.
+ // Most segments have no NULL rows, and then the hidden rows are summed as
they are.
+ DorisVector<uint32_t> hidden_non_null_docs;
+ std::span<const uint32_t> summed_docs = hidden_docs;
+ if (!null_docs.isEmpty()) {
+ std::ranges::copy_if(hidden_docs,
std::back_inserter(hidden_non_null_docs),
+ [&null_docs](uint32_t doc) { return
!null_docs.contains(doc); });
+ summed_docs = hidden_non_null_docs;
+ }
+ uint64_t hidden_lengths = 0;
+ if (!summed_docs.empty()) {
+ RETURN_IF_ERROR(sum_norm_lengths_of(summed_docs, &hidden_lengths));
+ }
+ DORIS_CHECK(hidden_lengths <= non_null_length_sum);
+ stats->token_count = non_null_length_sum - hidden_lengths;
+ return Status::OK();
+}
+
+Status exclude_hidden_term_docs(const segment_v2::TermIterator& postings,
+ std::span<const uint32_t> hidden_docs,
uint64_t* doc_frequency) {
+ DORIS_CHECK(doc_frequency != nullptr);
+ const uint64_t hidden = count_listed_docs(postings, hidden_docs);
+ if (hidden > *doc_frequency) {
+ return Status::Error<ErrorCode::INVERTED_INDEX_FILE_CORRUPTED>(
+ "term lists {} hidden rows but only {} documents", hidden,
*doc_frequency);
+ }
+ *doc_frequency -= hidden;
+ return Status::OK();
+}
+
+} // namespace collection_statistics_detail
+
+namespace {
+
+// The rows of a segment that a scan with `hidden_rows` does not return,
ascending.
+Status collect_hidden_docs(const RowsetSharedPtr& rowset, int32_t seg_id,
+ const TabletSchema& tablet_schema, const
MowHiddenRows& hidden_rows,
+ const io::IOContext* io_ctx, DorisVector<uint32_t>*
hidden_docs) {
+ DORIS_CHECK(hidden_rows.delete_bitmap != nullptr);
+ DORIS_CHECK(hidden_rows.version >= 0);
+ roaring::Roaring hidden = *hidden_rows.delete_bitmap->get_agg(
+ {rowset->rowset_id(), cast_set<uint32_t>(seg_id),
hidden_rows.version});
+
+ // A row written with the delete sign stays readable until base compaction
removes it, while
+ // queries filter it out by predicate.
+ if (tablet_schema.delete_sign_idx() >= 0) {
+ OlapReaderStatistics stats;
+ SegmentCacheHandle cache_handle;
+ // The delete sign is column data, read as the query reads its columns.
+ io::IOContext data_io_ctx;
+ if (io_ctx != nullptr) {
+ data_io_ctx = *io_ctx;
+ }
+ data_io_ctx.is_inverted_index = false;
+ RETURN_IF_ERROR(SegmentLoader::instance()->load_segment(
+ std::static_pointer_cast<BetaRowset>(rowset), seg_id,
&cache_handle,
+ hidden_rows.use_segment_cache, false, &stats, &data_io_ctx));
+ DORIS_CHECK(cache_handle.get_segments().size() == 1);
+ const auto& segment = cache_handle.get_segments()[0];
+ const TabletColumn& delete_sign =
tablet_schema.column(tablet_schema.delete_sign_idx());
+
+ bool may_hold_deletes = true;
+ std::shared_ptr<segment_v2::ColumnReader> column_reader;
+ auto status = segment->get_column_reader(delete_sign, &column_reader,
&stats, &data_io_ctx);
+ if (status.is<ErrorCode::NOT_FOUND>()) {
+ // A segment written before the column existed reads its default,
0.
+ may_hold_deletes = false;
+ } else {
+ RETURN_IF_ERROR(status);
+ DORIS_CHECK(column_reader != nullptr);
+ if (column_reader->has_zone_map()) {
+ segment_v2::ZoneMap zone_map;
+
RETURN_IF_ERROR(column_reader->get_segment_zone_map(&zone_map));
+ // Any sign other than 0 deletes the row. Bounds the zone map
cannot vouch for say
+ // nothing.
+ may_hold_deletes =
+ zone_map.pass_all ||
+ (zone_map.has_not_null &&
(zone_map.min_value.get<TYPE_TINYINT>() != 0 ||
+
zone_map.max_value.get<TYPE_TINYINT>() != 0));
+ }
+ }
+ if (may_hold_deletes) {
+ StorageReadOptions read_options;
+ read_options.stats = &stats;
+ read_options.io_ctx = data_io_ctx;
+ std::unique_ptr<segment_v2::ColumnIterator> iterator;
+ RETURN_IF_ERROR(segment->new_column_iterator(delete_sign,
&iterator, &read_options));
+ segment_v2::ColumnIteratorOptions iterator_options {
+ .use_page_cache =
+ hidden_rows.use_page_cache &&
!config::disable_storage_page_cache,
+ .file_reader = segment->file_reader().get(),
+ .stats = &stats,
+ .io_ctx = read_options.io_ctx,
+ };
+ RETURN_IF_ERROR(iterator->init(iterator_options));
+ RETURN_IF_ERROR(iterator->seek_to_ordinal(0));
+ auto data_type =
DataTypeFactory::instance().create_data_type(delete_sign);
+ constexpr size_t kBatch = 4096;
+ for (uint32_t begin = 0; begin < segment->num_rows();) {
+ size_t rows = std::min<size_t>(kBatch, segment->num_rows() -
begin);
+ MutableColumnPtr column = data_type->create_column();
+ bool has_null = false;
+ RETURN_IF_ERROR(iterator->next_batch(&rows, column,
&has_null));
+ if (rows == 0) {
+ return Status::Corruption("segment {} ended after {} of
its {} rows",
+ segment->id(), begin,
segment->num_rows());
+ }
+ const auto& signs = assert_cast<const
ColumnInt8&>(*column).get_data();
+ for (size_t i = 0; i < rows; ++i) {
+ if (signs[i] != 0) {
+ hidden.add(begin + cast_set<uint32_t>(i));
+ }
+ }
+ begin += cast_set<uint32_t>(rows);
+ }
+ }
+ }
+
+ hidden_docs->resize(hidden.cardinality());
+ hidden.toUint32Array(hidden_docs->data());
+ return Status::OK();
+}
+
+} // namespace
+
Status CollectionStatistics::collect(RuntimeState* state,
const std::vector<RowSetSplits>&
rs_splits,
const TabletSchemaSPtr& tablet_schema,
const VExprContextSPtrs&
common_expr_ctxs_push_down,
- io::IOContext* io_ctx) {
+ io::IOContext* io_ctx, const
MowHiddenRows* hidden_rows) {
std::unordered_map<std::wstring, CollectInfo> collect_infos;
RETURN_IF_ERROR(
extract_collect_info(state, common_expr_ctxs_push_down,
tablet_schema, &collect_infos));
@@ -77,8 +493,8 @@ Status CollectionStatistics::collect(RuntimeState* state,
auto rowset = rs_reader->rowset();
auto num_segments = rowset->num_segments();
for (int32_t seg_id = 0; seg_id < num_segments; ++seg_id) {
- auto status =
- process_segment(rowset, seg_id, tablet_schema.get(),
collect_infos, io_ctx);
+ auto status = process_segment(rowset, seg_id, tablet_schema.get(),
collect_infos,
+ hidden_rows, io_ctx);
if (!status.ok()) {
if (status.code() == ErrorCode::INVERTED_INDEX_FILE_NOT_FOUND
||
status.code() == ErrorCode::INVERTED_INDEX_BYPASS) {
@@ -179,6 +595,7 @@ Status CollectionStatistics::extract_collect_info(
Status CollectionStatistics::process_segment(const RowsetSharedPtr& rowset,
int32_t seg_id,
const TabletSchema* tablet_schema,
const CollectInfoMap&
collect_infos,
+ const MowHiddenRows* hidden_rows,
io::IOContext* io_ctx) {
auto seg_path = DORIS_TRY(rowset->segment_path(seg_id));
auto rowset_meta = rowset->rowset_meta();
@@ -190,6 +607,14 @@ Status CollectionStatistics::process_segment(const
RowsetSharedPtr& rowset, int3
rowset_meta->inverted_index_file_info(seg_id),
rowset_meta->tablet_id());
RETURN_IF_ERROR(idx_file_reader->init(config::inverted_index_read_buffer_size,
io_ctx));
+ // Row ID lists are as large as the segment's hidden or NULL rows, so they
are charged to the
+ // query's memory.
+ DorisVector<uint32_t> hidden_docs;
+ if (hidden_rows != nullptr) {
+ RETURN_IF_ERROR(collect_hidden_docs(rowset, seg_id, *tablet_schema,
*hidden_rows, io_ctx,
+ &hidden_docs));
+ }
+
int32_t total_seg_num_docs = 0;
for (const auto& [ws_field_name, collect_info] : collect_infos) {
@@ -229,8 +654,6 @@ Status CollectionStatistics::process_segment(const
RowsetSharedPtr& rowset, int3
index_searcher = index_searcher_ptr.get();
index_reader = index_searcher->getReader();
#endif
- total_seg_num_docs = std::max(total_seg_num_docs,
index_reader->maxDoc());
-
// BM25 on an analyzed index needs the record length of every row, and
CLucene keeps
// them, together with the field's token count, in the norms. A
segment written without
// norms would feed a zero avgdl, or rank its rows as zero-length
documents next to the
@@ -249,22 +672,77 @@ Status CollectionStatistics::process_segment(const
RowsetSharedPtr& rowset, int3
seg_path, StringHelper::to_string(ws_field_name));
}
// Missing term totals imply one token per row for keyword and
lowercase indexes.
- _total_num_tokens[ws_field_name] +=
token_count.value_or(index_reader->maxDoc());
+ collection_statistics_detail::SegmentScoringStats scoring_stats {
+ .doc_count = static_cast<uint64_t>(index_reader->maxDoc()),
+ .token_count = token_count.value_or(index_reader->maxDoc()),
+ .has_norms = token_count.has_value()};
+ if (hidden_rows != nullptr) {
+ std::shared_ptr<roaring::Roaring> null_docs =
std::make_shared<roaring::Roaring>();
+ uint64_t non_null_length_sum = 0;
+ if (scoring_stats.has_norms) {
+ segment_v2::InvertedIndexQueryCacheHandle
null_bitmap_cache_handle;
+ RETURN_IF_ERROR(segment_v2::read_inverted_index_null_bitmap(
+ *idx_file_reader, *collect_info.index_meta, io_ctx,
+ &null_bitmap_cache_handle));
+ null_docs = null_bitmap_cache_handle.get_bitmap();
+ DORIS_CHECK(null_docs != nullptr);
+ auto sum_lengths = [index_reader, &ws_field_name,
&null_docs](uint64_t* out) {
+ DorisVector<uint32_t>
null_doc_ids(null_docs->cardinality());
+ null_docs->toUint32Array(null_doc_ids.data());
+ return
collection_statistics_detail::sum_non_null_norm_lengths(
+ index_reader, ws_field_name, null_doc_ids, out);
+ };
+#ifdef BE_TEST
+ RETURN_IF_ERROR(sum_lengths(&non_null_length_sum));
+#else
+
RETURN_IF_ERROR(inverted_index_cache_handle.get_index_cache_value()
+
->non_null_norm_length_sum(ws_field_name, sum_lengths,
+
&non_null_length_sum));
+#endif
+ }
+ // The statistics keep no norms, so the lengths of a segment's
hidden rows are summed
+ // once per set of hidden rows and kept with the searcher.
+ auto sum_hidden_lengths = [&](std::span<const uint32_t> docs,
uint64_t* out) {
+ auto sum = [index_reader, &ws_field_name, docs](uint64_t*
lengths) {
+ return collection_statistics_detail::sum_norm_lengths_of(
+ index_reader, ws_field_name, docs, lengths);
+ };
+#ifdef BE_TEST
+ return sum(out);
+#else
+ return
inverted_index_cache_handle.get_index_cache_value()->hidden_norm_length_sum(
+ ws_field_name, docs, sum, out);
+#endif
+ };
+ RETURN_IF_ERROR(collection_statistics_detail::exclude_hidden_docs(
+ hidden_docs, *null_docs, non_null_length_sum,
sum_hidden_lengths,
+ &scoring_stats));
+ }
+ total_seg_num_docs =
+ std::max(total_seg_num_docs,
cast_set<int32_t>(scoring_stats.doc_count));
+ _total_num_tokens[ws_field_name] += scoring_stats.token_count;
for (const auto& term_info : collect_info.term_infos) {
auto iter = TermIterator::create(io_ctx, false, index_reader,
ws_field_name,
term_info.get_single_term());
- _term_doc_freqs[ws_field_name][iter->term()] += iter->doc_freq();
+ auto doc_frequency = static_cast<uint64_t>(iter->doc_freq());
+ if (!hidden_docs.empty()) {
+
RETURN_IF_ERROR(collection_statistics_detail::exclude_hidden_term_docs(
+ *iter, hidden_docs, &doc_frequency));
+ }
+ _term_doc_freqs[ws_field_name][iter->term()] += doc_frequency;
}
}
_total_num_docs += total_seg_num_docs;
+ _counted_hidden_docs = _counted_hidden_docs || !hidden_docs.empty();
return Status::OK();
}
void CollectionStatistics::copy_collected_from(const CollectionStatistics&
other) {
_total_num_docs = other._total_num_docs;
+ _counted_hidden_docs = other._counted_hidden_docs;
_total_num_tokens = other._total_num_tokens;
_term_doc_freqs = other._term_doc_freqs;
_avg_dl_by_col.clear();
@@ -279,13 +757,14 @@ TabletCollectionStatistics::~TabletCollectionStatistics()
= default;
Status TabletCollectionStatistics::copy_to(RuntimeState* state,
const TabletSchemaSPtr&
tablet_schema,
const VExprContextSPtrs&
common_expr_ctxs_push_down,
- io::IOContext* io_ctx,
+ io::IOContext* io_ctx, const
MowHiddenRows* hidden_rows,
CollectionStatistics* statistics) {
DORIS_CHECK(statistics != nullptr);
std::lock_guard lock(_mutex);
if (!_collected) {
+ // Every scanner of the tablet reads with the same delete bitmap and
version.
_status = _statistics.collect(state, _rs_splits, tablet_schema,
common_expr_ctxs_push_down,
- io_ctx);
+ io_ctx, hidden_rows);
_collected = true;
// Do not hold the rowsets any longer than needed.
_rs_splits.clear();
@@ -323,7 +802,9 @@ uint64_t
CollectionStatistics::get_total_term_cnt_by_col(const std::wstring& luc
}
uint64_t CollectionStatistics::get_doc_num() const {
- if (_total_num_docs == 0) {
+ // A tablet whose rows are all hidden has no document a query can match,
so its statistics
+ // are empty rather than missing.
+ if (_total_num_docs == 0 && !_counted_hidden_docs) {
throw Exception(
ErrorCode::INVERTED_INDEX_CLUCENE_ERROR,
"Index statistics collection failed: No data available for
SimilarityCollector");
@@ -340,7 +821,7 @@ float CollectionStatistics::get_or_calculate_avg_dl(const
std::wstring& lucene_c
const uint64_t total_term_cnt = get_total_term_cnt_by_col(lucene_col_name);
const uint64_t total_doc_cnt = get_doc_num();
- float avg_dl = total_doc_cnt > 0 ? float((double)total_term_cnt /
(double)total_doc_cnt) : 0.0F;
+ float avg_dl = total_doc_cnt > 0 ? float((double)total_term_cnt /
(double)total_doc_cnt) : 1.0F;
_avg_dl_by_col[lucene_col_name] = avg_dl;
return avg_dl;
}
diff --git a/be/src/storage/compaction/collection_statistics.h
b/be/src/storage/compaction/collection_statistics.h
index 5658438fb32..986e5053516 100644
--- a/be/src/storage/compaction/collection_statistics.h
+++ b/be/src/storage/compaction/collection_statistics.h
@@ -17,7 +17,9 @@
#pragma once
#include <cstdint>
+#include <functional>
#include <mutex>
+#include <span>
#include <string>
#include <unordered_map>
#include <vector>
@@ -29,6 +31,14 @@
#include "storage/olap_common.h"
#include "storage/predicate_collector.h"
+namespace lucene::index {
+class IndexReader;
+} // namespace lucene::index
+
+namespace roaring {
+class Roaring;
+} // namespace roaring
+
namespace doris {
#include "common/compile_check_begin.h"
@@ -40,6 +50,8 @@ struct IOContext;
struct RowSetSplits;
+class DeleteBitmap;
+
class Rowset;
using RowsetSharedPtr = std::shared_ptr<Rowset>;
@@ -47,14 +59,41 @@ class TabletIndex;
class TabletSchema;
using TabletSchemaSPtr = std::shared_ptr<TabletSchema>;
+namespace segment_v2 {
+class TermIterator;
+} // namespace segment_v2
+
+// The rows of a merge-on-write tablet that its segments still hold but a scan
at `version` does
+// not return: rows a later load overwrote or deleted, which the delete bitmap
marks, and rows
+// written with the delete sign, which queries filter out. Compaction removes
both, on each
+// replica at its own time, so statistics counting them would differ between
the replicas of a
+// tablet, and so would every BM25 score. Rows written with the delete sign
are left out even when
+// the query returns them (`skip_delete_sign`, `show_hidden_columns`), and
rows deleted by a delete
+// predicate (`enable_mow_light_delete`) are still counted, as the delete
bitmap does not mark them.
+//
+// The cache flags carry the query's choices to the segment and column reads
of the delete sign;
+// the file cache choices come with the I/O context.
+struct MowHiddenRows {
+ std::shared_ptr<DeleteBitmap> delete_bitmap;
+ int64_t version = -1;
+ bool use_segment_cache = true;
+ bool use_page_cache = true;
+};
+
class CollectionStatistics {
public:
CollectionStatistics() = default;
virtual ~CollectionStatistics() = default;
+ // With `hidden_rows`, the statistics describe only the rows a
merge-on-write scan returns,
+ // so every replica of a tablet scores alike whatever it has compacted:
the hidden rows are
+ // left out of the document count and the term document frequencies, and
the token count is
+ // the sum of the document lengths the norms of the returned non-NULL rows
stand for, or the
+ // returned row count for an index without norms.
Status collect(RuntimeState* state, const std::vector<RowSetSplits>&
rs_splits,
const TabletSchemaSPtr& tablet_schema,
- const VExprContextSPtrs& common_expr_ctxs_push_down,
io::IOContext* io_ctx);
+ const VExprContextSPtrs& common_expr_ctxs_push_down,
io::IOContext* io_ctx,
+ const MowHiddenRows* hidden_rows = nullptr);
MOCK_FUNCTION float get_or_calculate_idf(const std::wstring&
lucene_col_name,
const std::wstring& term);
@@ -71,7 +110,7 @@ private:
CollectInfoMap* collect_infos);
Status process_segment(const RowsetSharedPtr& rowset, int32_t seg_id,
const TabletSchema* tablet_schema, const
CollectInfoMap& collect_infos,
- io::IOContext* io_ctx);
+ const MowHiddenRows* hidden_rows, io::IOContext*
io_ctx);
uint64_t get_term_doc_freq_by_col(const std::wstring& lucene_col_name,
const std::wstring& term);
@@ -79,6 +118,9 @@ private:
uint64_t get_doc_num() const;
uint64_t _total_num_docs = 0;
+ // Whether a collected segment held rows the scan does not return, so that
no documents can
+ // mean every row is hidden.
+ bool _counted_hidden_docs = false;
std::unordered_map<std::wstring, uint64_t> _total_num_tokens;
std::unordered_map<std::wstring, std::unordered_map<std::wstring,
uint64_t>> _term_doc_freqs;
@@ -108,7 +150,7 @@ public:
// Later callers wait for the first collection and receive its result or
error.
Status copy_to(RuntimeState* state, const TabletSchemaSPtr& tablet_schema,
const VExprContextSPtrs& common_expr_ctxs_push_down,
io::IOContext* io_ctx,
- CollectionStatistics* statistics);
+ const MowHiddenRows* hidden_rows, CollectionStatistics*
statistics);
private:
std::mutex _mutex;
@@ -119,5 +161,54 @@ private:
};
using TabletCollectionStatisticsPtr =
std::shared_ptr<TabletCollectionStatistics>;
+// Implementation details of the visible-row statistics, surfaced for
collection_statistics_test.
+namespace collection_statistics_detail {
+
+// How many of the ascending `docs` the term's postings list: leaping over the
postings between
+// them when the docs are much rarer than the postings, otherwise decoding
every posting once.
+uint64_t count_listed_docs(const segment_v2::TermIterator& postings,
+ std::span<const uint32_t> docs);
+
+// The sum of the lengths the norms of `field` stand for, of every document of
`reader`, read one
+// CLucene segment at a time into a buffer of the statistics' own.
+Status sum_norm_lengths(lucene::index::IndexReader* reader, const
std::wstring& field,
+ uint64_t* out);
+
+// The sum of the lengths the norms of `field` stand for, of the ascending
`docs` of `reader`.
+Status sum_norm_lengths_of(lucene::index::IndexReader* reader, const
std::wstring& field,
+ std::span<const uint32_t> docs, uint64_t* out);
+
+// The sum of the lengths the norms of a segment's documents other than the
ascending `null_docs`
+// stand for. A NULL row is a document without tokens, though its norm stands
for one. Rows that are
+// indexed without tokens but are not NULL keep a norm that cannot be told
from a one-token row's:
+// an analyzed empty string.
+Status sum_non_null_norm_lengths(lucene::index::IndexReader* reader, const
std::wstring& field,
+ std::span<const uint32_t> null_docs,
uint64_t* out);
+
+// The scoring statistics of one field of a segment's index.
+struct SegmentScoringStats {
+ uint64_t doc_count = 0;
+ uint64_t token_count = 0;
+ bool has_norms = false;
+};
+
+// Sums the lengths the norms of ascending rows of a segment's index stand for.
+using SumNormLengthsOf = std::function<Status(std::span<const uint32_t> docs,
uint64_t* out)>;
+
+// Turns a segment's scoring statistics into those of its documents other than
the ascending
+// `hidden_docs`: their count, and the sum of the lengths the norms of the
non-NULL ones stand
+// for, or their count when the index has no norms. `non_null_length_sum` is
the segment's
+// sum_non_null_norm_lengths(), and `null_docs` its NULL rows; both are unused
without norms. The
+// lengths of the hidden non-NULL rows are summed by `sum_norm_lengths_of`.
+Status exclude_hidden_docs(std::span<const uint32_t> hidden_docs, const
roaring::Roaring& null_docs,
+ uint64_t non_null_length_sum,
+ const SumNormLengthsOf& sum_norm_lengths_of,
SegmentScoringStats* stats);
+
+// Takes from a term's document frequency the documents of `hidden_docs` its
postings list.
+Status exclude_hidden_term_docs(const segment_v2::TermIterator& postings,
+ std::span<const uint32_t> hidden_docs,
uint64_t* doc_frequency);
+
+} // namespace collection_statistics_detail
+
#include "common/compile_check_end.h"
} // namespace doris
diff --git a/be/src/storage/index/inverted/inverted_index_cache.h
b/be/src/storage/index/inverted/inverted_index_cache.h
index 0e33fe747b3..f4e9341015c 100644
--- a/be/src/storage/index/inverted/inverted_index_cache.h
+++ b/be/src/storage/index/inverted/inverted_index_cache.h
@@ -23,9 +23,15 @@
#include <stdint.h>
#include <atomic>
+#include <functional>
#include <memory>
+#include <mutex>
#include <roaring/roaring.hh>
+#include <span>
#include <string>
+#include <unordered_map>
+#include <utility>
+#include <vector>
#include "common/config.h"
#include "common/status.h"
@@ -35,6 +41,7 @@
#include "runtime/memory/lru_cache_policy.h"
#include "runtime/memory/mem_tracker.h"
#include "storage/index/inverted/inverted_index_searcher.h"
+#include "util/hash_util.hpp"
#include "util/lru_cache.h"
#include "util/slice.h"
#include "util/time.h"
@@ -65,6 +72,66 @@ public:
size = mem_size;
last_visit_time = visit_time;
}
+
+ // The sum of the lengths the norms of the non-NULL documents of
`field` stand for,
+ // computed by `compute` the first time. The index never changes, and
merge-on-write
+ // scoring needs the sum for every query.
+ Status non_null_norm_length_sum(const std::wstring& field,
+ const
std::function<Status(uint64_t*)>& compute,
+ uint64_t* out) {
+ std::lock_guard lock(_norm_length_sums_mutex);
+ if (auto it = _norm_length_sums.find(field); it !=
_norm_length_sums.end()) {
+ *out = it->second;
+ return Status::OK();
+ }
+ RETURN_IF_ERROR(compute(out));
+ _norm_length_sums.emplace(field, *out);
+ return Status::OK();
+ }
+
+ // The sum of the lengths the norms of `docs`, ascending rows of the
index, stand for,
+ // computed by `compute` the first time the field meets that set of
rows. Queries at one
+ // read version hide the same rows, so they read no norms; a few sets
per field are kept
+ // for queries at different versions.
+ Status hidden_norm_length_sum(const std::wstring& field,
std::span<const uint32_t> docs,
+ const std::function<Status(uint64_t*)>&
compute,
+ uint64_t* out) {
+ const RowSetKey key {
+ .fingerprint = HashUtil::xxHash64WithSeed(
+ reinterpret_cast<const char*>(docs.data()),
docs.size_bytes(), 0),
+ .count = docs.size(),
+ .first = docs.empty() ? 0 : docs.front(),
+ .last = docs.empty() ? 0 : docs.back()};
+ std::lock_guard lock(_norm_length_sums_mutex);
+ auto& sums = _hidden_length_sums[field];
+ for (const auto& [cached_key, sum] : sums) {
+ if (cached_key == key) {
+ *out = sum;
+ return Status::OK();
+ }
+ }
+ RETURN_IF_ERROR(compute(out));
+ if (sums.size() == kHiddenSetsPerField) {
+ sums.erase(sums.begin());
+ }
+ sums.emplace_back(key, *out);
+ return Status::OK();
+ }
+
+ private:
+ struct RowSetKey {
+ uint64_t fingerprint;
+ size_t count;
+ uint32_t first;
+ uint32_t last;
+ bool operator==(const RowSetKey&) const = default;
+ };
+ static constexpr size_t kHiddenSetsPerField = 4;
+
+ std::mutex _norm_length_sums_mutex;
+ std::unordered_map<std::wstring, uint64_t> _norm_length_sums;
+ std::unordered_map<std::wstring, std::vector<std::pair<RowSetKey,
uint64_t>>>
+ _hidden_length_sums;
};
// Create global instance of this class.
// "capacity" is the capacity of lru cache.
diff --git a/be/src/storage/index/inverted/inverted_index_reader.cpp
b/be/src/storage/index/inverted/inverted_index_reader.cpp
index d315d5c377c..12c382fd062 100644
--- a/be/src/storage/index/inverted/inverted_index_reader.cpp
+++ b/be/src/storage/index/inverted/inverted_index_reader.cpp
@@ -155,15 +155,15 @@ std::string InvertedIndexReader::get_index_file_path() {
return _index_file_reader->get_index_file_path(&_index_meta);
}
-Status InvertedIndexReader::read_null_bitmap(const IndexQueryContextPtr&
context,
- InvertedIndexQueryCacheHandle*
cache_handle,
- lucene::store::Directory* dir) {
- SCOPED_RAW_TIMER(&context->stats->inverted_index_query_null_bitmap_timer);
+Status read_inverted_index_null_bitmap(IndexFileReader& file_reader, const
TabletIndex& index_meta,
+ const io::IOContext* io_ctx,
+ InvertedIndexQueryCacheHandle*
cache_handle,
+ lucene::store::Directory* dir) {
lucene::store::IndexInput* null_bitmap_in = nullptr;
bool owned_dir = false;
try {
// try to get query bitmap result from cache and return immediately on
cache hit
- auto index_file_key =
_index_file_reader->get_index_file_cache_key(&_index_meta);
+ auto index_file_key =
file_reader.get_index_file_cache_key(&index_meta);
InvertedIndexQueryCache::CacheKey cache_key {
index_file_key, "", InvertedIndexQueryType::UNKNOWN_QUERY,
"null_bitmap"};
auto* cache = InvertedIndexQueryCache::instance();
@@ -173,13 +173,12 @@ Status InvertedIndexReader::read_null_bitmap(const
IndexQueryContextPtr& context
if (!dir) {
// TODO: ugly code here, try to refact.
- auto st =
_index_file_reader->init(config::inverted_index_read_buffer_size,
- context->io_ctx);
+ auto st =
file_reader.init(config::inverted_index_read_buffer_size, io_ctx);
if (!st.ok()) {
LOG(WARNING) << st;
return st;
}
- auto directory = DORIS_TRY(_index_file_reader->open(&_index_meta,
context->io_ctx));
+ auto directory = DORIS_TRY(file_reader.open(&index_meta, io_ctx));
dir = directory.release();
owned_dir = true;
}
@@ -216,6 +215,14 @@ Status InvertedIndexReader::read_null_bitmap(const
IndexQueryContextPtr& context
return Status::OK();
}
+Status InvertedIndexReader::read_null_bitmap(const IndexQueryContextPtr&
context,
+ InvertedIndexQueryCacheHandle*
cache_handle,
+ lucene::store::Directory* dir) {
+ SCOPED_RAW_TIMER(&context->stats->inverted_index_query_null_bitmap_timer);
+ return read_inverted_index_null_bitmap(*_index_file_reader, _index_meta,
context->io_ctx,
+ cache_handle, dir);
+}
+
bool InvertedIndexReader::handle_query_cache(const IndexQueryContextPtr&
context,
InvertedIndexQueryCache* cache,
const
InvertedIndexQueryCache::CacheKey& cache_key,
diff --git a/be/src/storage/index/inverted/inverted_index_reader.h
b/be/src/storage/index/inverted/inverted_index_reader.h
index 0e2f6a120d4..32c14991913 100644
--- a/be/src/storage/index/inverted/inverted_index_reader.h
+++ b/be/src/storage/index/inverted/inverted_index_reader.h
@@ -213,6 +213,13 @@ private:
}
};
+// The null bitmap of an inverted index, through the inverted index query
cache. Reads it from
+// `dir` when that is the index's open directory, otherwise opens the index.
+Status read_inverted_index_null_bitmap(IndexFileReader& file_reader, const
TabletIndex& index_meta,
+ const io::IOContext* io_ctx,
+ InvertedIndexQueryCacheHandle*
cache_handle,
+ lucene::store::Directory* dir =
nullptr);
+
class InvertedIndexReader : public IndexReader {
public:
explicit InvertedIndexReader(const TabletIndex* index_meta,
diff --git a/be/test/exec/scan/olap_scanner_test.cpp
b/be/test/exec/scan/olap_scanner_test.cpp
index a3f20ef6b57..f5cca0a759f 100644
--- a/be/test/exec/scan/olap_scanner_test.cpp
+++ b/be/test/exec/scan/olap_scanner_test.cpp
@@ -102,6 +102,20 @@ TEST(OlapScannerTest,
BuildScoreRuntimeCollectionIoContextAllowsMissingQueryCont
EXPECT_EQ(io_ctx.file_cache_stats, &stats);
EXPECT_TRUE(io_ctx.is_inverted_index);
EXPECT_EQ(io_ctx.remote_scan_cache_write_limiter, nullptr);
+ EXPECT_FALSE(io_ctx.is_disposable);
+}
+
+// The statistics read the index before the scan, with the query's file cache
choice.
+TEST(OlapScannerTest,
BuildScoreRuntimeCollectionIoContextFollowsQueryFileCacheOptions) {
+ MockRuntimeState state;
+ state._query_id = make_query_id();
+ state._query_ctx = nullptr;
+ state._query_options.__set_disable_file_cache(true);
+
+ io::FileCacheStatistics stats;
+ auto io_ctx = build_score_runtime_collection_io_context(&state,
ReaderType::READER_QUERY, 3600,
+ &stats);
+ EXPECT_TRUE(io_ctx.is_disposable);
}
} // namespace doris
diff --git a/be/test/storage/compaction/collection_statistics_test.cpp
b/be/test/storage/compaction/collection_statistics_test.cpp
index e39eb107813..7fa78f1e6a6 100644
--- a/be/test/storage/compaction/collection_statistics_test.cpp
+++ b/be/test/storage/compaction/collection_statistics_test.cpp
@@ -21,7 +21,10 @@
#include <gmock/gmock.h>
#include <gtest/gtest.h>
+#include <cmath>
#include <memory>
+#include <roaring/roaring.hh>
+#include <set>
#include <string>
#include "common/exception.h"
@@ -41,11 +44,13 @@
#include "storage/index/index_file_reader.h"
#include "storage/index/index_file_writer.h"
#include "storage/index/index_writer.h"
+#include "storage/index/inverted/inverted_index_cache.h"
#include "storage/index/inverted/inverted_index_desc.h"
#include "storage/index/inverted/util/term_iterator.h"
#include "storage/rowset/rowset.h"
#include "storage/rowset/rowset_meta.h"
#include "storage/rowset/rowset_reader.h"
+#include "storage/tablet/tablet_meta.h"
#include "storage/tablet/tablet_schema.h"
#include "testutil/mock/mock_runtime_state.h"
#include "util/defer_op.h"
@@ -272,11 +277,18 @@ protected:
test_dir_ = "./collection_statistics_test_" +
std::to_string(::testing::UnitTest::GetInstance()->random_seed());
ASSERT_TRUE(io::global_local_filesystem()->create_directory(test_dir_).ok());
+ // Merge-on-write collection reads the null bitmap through the query
cache.
+ previous_query_cache_ =
ExecEnv::GetInstance()->get_inverted_index_query_cache();
+ query_cache_.reset(
+ segment_v2::InvertedIndexQueryCache::create_global_cache(1024
* 1024, 1));
+
ExecEnv::GetInstance()->set_inverted_index_query_cache(query_cache_.get());
}
void TearDown() override {
stats_.reset();
runtime_state_.reset();
+
ExecEnv::GetInstance()->set_inverted_index_query_cache(previous_query_cache_);
+ query_cache_.reset();
(void)io::global_local_filesystem()->delete_directory(test_dir_);
}
@@ -343,6 +355,8 @@ protected:
std::unique_ptr<CollectionStatistics> stats_;
std::shared_ptr<MockRuntimeState> runtime_state_;
std::string test_dir_;
+ segment_v2::InvertedIndexQueryCache* previous_query_cache_ = nullptr;
+ std::unique_ptr<segment_v2::InvertedIndexQueryCache> query_cache_;
};
TEST_F(CollectionStatisticsTest, CollectWithEmptyRowsetSplits) {
@@ -526,7 +540,8 @@ TEST_F(CollectionStatisticsTest,
MissingFieldIndexPreservesOtherFieldStatistics)
other_info.index_meta = tablet_schema->inverted_indexs(2).front();
infos.emplace(L"2", std::move(other_info));
for (int i = 0; i < 2; ++i) {
- auto status = stats_->process_segment(rowset, i, tablet_schema.get(),
infos, nullptr);
+ auto status =
+ stats_->process_segment(rowset, i, tablet_schema.get(), infos,
nullptr, nullptr);
ASSERT_TRUE(status.ok()) << status;
}
EXPECT_EQ(2u, stats_->get_doc_num());
@@ -803,14 +818,432 @@ TEST_F(CollectionStatisticsTest,
TabletCollectionStatisticsCollectsOnceAndCopies
CollectionStatistics first;
CollectionStatistics second;
- EXPECT_TRUE(
- tablet_statistics
- .copy_to(runtime_state_.get(), tablet_schema,
expr_contexts, nullptr, &first)
- .ok());
- EXPECT_TRUE(
- tablet_statistics
- .copy_to(runtime_state_.get(), tablet_schema,
expr_contexts, nullptr, &second)
- .ok());
+ EXPECT_TRUE(tablet_statistics
+ .copy_to(runtime_state_.get(), tablet_schema,
expr_contexts, nullptr,
+ nullptr, &first)
+ .ok());
+ EXPECT_TRUE(tablet_statistics
+ .copy_to(runtime_state_.get(), tablet_schema,
expr_contexts, nullptr,
+ nullptr, &second)
+ .ok());
+}
+
+namespace {
+
+using collection_statistics_detail::count_listed_docs;
+using collection_statistics_detail::exclude_hidden_docs;
+using collection_statistics_detail::exclude_hidden_term_docs;
+using collection_statistics_detail::SegmentScoringStats;
+using collection_statistics_detail::sum_non_null_norm_lengths;
+using collection_statistics_detail::sum_norm_lengths;
+using collection_statistics_detail::sum_norm_lengths_of;
+
+TabletSchemaSPtr create_english_index_schema() {
+ TabletSchemaPB schema_pb;
+ schema_pb.set_keys_type(DUP_KEYS);
+
schema_pb.set_inverted_index_storage_format(InvertedIndexStorageFormatPB::V3);
+ auto tablet_schema = std::make_shared<TabletSchema>();
+ tablet_schema->init_from_pb(schema_pb);
+
+ TabletColumn column;
+ column.set_unique_id(1);
+ column.set_name("content");
+ column.set_type(FieldType::OLAP_FIELD_TYPE_STRING);
+ column.set_is_nullable(true);
+ tablet_schema->append_column(column);
+
+ TabletIndex index;
+ index._index_id = 1;
+ index._index_type = IndexType::INVERTED;
+ index._col_unique_ids.push_back(1);
+ index._properties["parser"] = "english";
+ index._properties["support_phrase"] = "true";
+ tablet_schema->append_index(std::move(index));
+ return tablet_schema;
+}
+
+// Writes the inverted index of a segment holding `rows`, one document each,
the rows of
+// `null_rows` as NULL whatever their value.
+Status write_index_segment(const TabletSchemaSPtr& tablet_schema, const
std::string& test_dir,
+ const std::string& segment_path, const
std::vector<std::string>& rows,
+ const std::set<size_t>& null_rows = {}) {
+ std::vector<StorePath> paths;
+ paths.emplace_back(test_dir, 1024);
+ auto tmp_file_dirs = std::make_unique<segment_v2::TmpFileDirs>(paths);
+ RETURN_IF_ERROR(tmp_file_dirs->init());
+ ExecEnv::GetInstance()->set_tmp_file_dir(std::move(tmp_file_dirs));
+
+ const std::string prefix {
+
segment_v2::InvertedIndexDescriptor::get_index_file_path_prefix(segment_path)};
+ io::FileWriterPtr compound_file;
+ io::FileWriterOptions options;
+ auto fs = io::global_local_filesystem();
+ RETURN_IF_ERROR(
+
fs->create_file(segment_v2::InvertedIndexDescriptor::get_index_file_path_v2(prefix),
+ &compound_file, &options));
+ segment_v2::IndexFileWriter file_writer(fs, prefix, "hidden_rows", 0,
+ InvertedIndexStorageFormatPB::V3,
+ std::move(compound_file));
+ const auto indexes = tablet_schema->inverted_indexs(1);
+ DORIS_CHECK(indexes.size() == 1);
+ std::unique_ptr<segment_v2::IndexColumnWriter> column_writer;
+
RETURN_IF_ERROR(segment_v2::IndexColumnWriter::create(&tablet_schema->column(0),
&column_writer,
+ &file_writer,
indexes[0]));
+ for (size_t row = 0; row < rows.size(); ++row) {
+ if (null_rows.contains(row)) {
+ RETURN_IF_ERROR(column_writer->add_nulls(1));
+ } else {
+ const Slice value(rows[row]);
+ RETURN_IF_ERROR(column_writer->add_values("content", &value, 1));
+ }
+ }
+ RETURN_IF_ERROR(column_writer->finish());
+ RETURN_IF_ERROR(file_writer.begin_close());
+ return file_writer.finish_close();
+}
+
+std::vector<RowSetSplits> single_segment_splits(const TabletSchemaSPtr&
tablet_schema,
+ const std::string&
segment_path) {
+ auto rowset_meta =
std::make_shared<collection_statistics::MockRowsetMeta>();
+ auto rowset =
std::make_shared<collection_statistics::MockRowset>(tablet_schema, rowset_meta);
+ rowset->set_num_segments(1);
+ rowset->set_segment_path(0, segment_path);
+ auto reader =
std::make_shared<collection_statistics::MockRowsetReader>(rowset);
+ return {RowSetSplits(reader)};
+}
+
+// A delete bitmap of its own tablet, marking `rows` of the first segment of
the splits' rowset at
+// `version`. The bitmap's aggregation cache is keyed by tablet and rowset,
and the mock rowsets
+// share their ids, so each bitmap takes its own tablet id.
+MowHiddenRows hide_rows(const std::vector<RowSetSplits>& splits, int64_t
tablet_id, int64_t version,
+ const std::vector<uint32_t>& rows) {
+ MowHiddenRows hidden {.delete_bitmap =
std::make_shared<DeleteBitmap>(tablet_id),
+ .version = version};
+ const auto rowset_id = splits[0].rs_reader->rowset()->rowset_id();
+ for (const uint32_t row : rows) {
+ hidden.delete_bitmap->add({rowset_id, 0, version}, row);
+ }
+ return hidden;
+}
+
+// A reader of the CLucene index of a segment written by write_index_segment.
+class SegmentIndexReader {
+public:
+ SegmentIndexReader(const TabletSchemaSPtr& tablet_schema, const
std::string& segment_path)
+ : _file_reader(
+ io::global_local_filesystem(),
+ std::string
{segment_v2::InvertedIndexDescriptor::get_index_file_path_prefix(
+ segment_path)},
+ InvertedIndexStorageFormatPB::V3) {
+ DORIS_CHECK(_file_reader.init().ok());
+ auto compound =
_file_reader.open(tablet_schema->inverted_indexs(1)[0]);
+ DORIS_CHECK(compound.has_value());
+ _searcher = std::make_unique<lucene::search::IndexSearcher>(
+ lucene::index::IndexReader::open(compound->get()), true);
+ }
+
+ lucene::index::IndexReader* get() const { return _searcher->getReader(); }
+
+private:
+ segment_v2::IndexFileReader _file_reader;
+ std::unique_ptr<lucene::search::IndexSearcher> _searcher;
+};
+
+} // namespace
+
+// A replica whose segment still holds a row that a later load overwrote must
collect the same
+// statistics as a replica that compacted the row away, or the two score every
row differently.
+TEST_F(CollectionStatisticsTest, HiddenRowsAreCollectedLikeCompactedAway) {
+ IndexPolicyMgr policy_mgr;
+ auto* exec_env = ExecEnv::GetInstance();
+ auto* original_policy_mgr = exec_env->index_policy_mgr();
+ exec_env->_index_policy_mgr = &policy_mgr;
+ Defer restore_policy_mgr([&] { exec_env->_index_policy_mgr =
original_policy_mgr; });
+
+ auto tablet_schema = create_english_index_schema();
+ // Forty-five tokens: CLucene's norm holds lengths exactly only up to 40.
+ std::string overwritten;
+ for (int i = 0; i < 44; ++i) {
+ overwritten += "gamma ";
+ }
+ overwritten += "alpha";
+ const std::string uncompacted_path = test_dir_ +
"/hidden_rows_uncompacted_0.dat";
+ const std::string compacted_path = test_dir_ +
"/hidden_rows_compacted_0.dat";
+ ASSERT_TRUE(write_index_segment(tablet_schema, test_dir_, uncompacted_path,
+ {"alpha beta", overwritten, "beta gamma
delta"})
+ .ok());
+ ASSERT_TRUE(write_index_segment(tablet_schema, test_dir_, compacted_path,
+ {"alpha beta", "beta gamma delta"})
+ .ok());
+
+ struct Collected {
+ uint64_t doc_count;
+ uint64_t token_count;
+ uint64_t alpha_doc_frequency;
+ uint64_t gamma_doc_frequency;
+ float avg_dl;
+ float alpha_idf;
+ };
+ int64_t next_tablet_id = 9100;
+ auto collect = [&](const std::string& segment_path,
+ const std::vector<uint32_t>& hidden_rows) -> Collected {
+ auto splits = single_segment_splits(tablet_schema, segment_path);
+ MowHiddenRows hidden = hide_rows(splits, next_tablet_id++, 5,
hidden_rows);
+ CollectionStatistics stats;
+ const Status status =
+ stats.collect(runtime_state_.get(), splits, tablet_schema,
+ create_match_expr_contexts("alpha gamma"),
nullptr, &hidden);
+ EXPECT_TRUE(status.ok()) << status;
+ return {.doc_count = stats.get_doc_num(),
+ .token_count = stats.get_total_term_cnt_by_col(L"1"),
+ .alpha_doc_frequency = stats.get_term_doc_freq_by_col(L"1",
L"alpha"),
+ .gamma_doc_frequency = stats.get_term_doc_freq_by_col(L"1",
L"gamma"),
+ .avg_dl = stats.get_or_calculate_avg_dl(L"1"),
+ .alpha_idf = stats.get_or_calculate_idf(L"1", L"alpha")};
+ };
+
+ const Collected uncompacted = collect(uncompacted_path, {1});
+ const Collected compacted = collect(compacted_path, {});
+ EXPECT_EQ(uncompacted.doc_count, 2);
+ EXPECT_EQ(uncompacted.token_count, 5);
+ EXPECT_EQ(uncompacted.alpha_doc_frequency, 1);
+ EXPECT_EQ(uncompacted.gamma_doc_frequency, 1);
+ EXPECT_EQ(uncompacted.doc_count, compacted.doc_count);
+ EXPECT_EQ(uncompacted.token_count, compacted.token_count);
+ EXPECT_EQ(uncompacted.alpha_doc_frequency, compacted.alpha_doc_frequency);
+ EXPECT_EQ(uncompacted.gamma_doc_frequency, compacted.gamma_doc_frequency);
+ EXPECT_EQ(uncompacted.avg_dl, compacted.avg_dl);
+ EXPECT_EQ(uncompacted.alpha_idf, compacted.alpha_idf);
+
+ // Without the delete bitmap the overwritten row is counted, with its
exact length.
+ ASSERT_TRUE(stats_->collect(runtime_state_.get(),
+ single_segment_splits(tablet_schema,
uncompacted_path),
+ tablet_schema,
create_match_expr_contexts("alpha gamma"), nullptr)
+ .ok());
+ EXPECT_EQ(stats_->get_doc_num(), 3);
+ EXPECT_EQ(stats_->get_total_term_cnt_by_col(L"1"), 50);
+ EXPECT_EQ(stats_->get_term_doc_freq_by_col(L"1", L"alpha"), 2);
+}
+
+// A tablet whose rows were all deleted or overwritten still has segments to
collect from, but no
+// document a query can match. Its statistics are empty, not missing.
+TEST_F(CollectionStatisticsTest, AllRowsHiddenCollectsEmptyStatistics) {
+ IndexPolicyMgr policy_mgr;
+ auto* exec_env = ExecEnv::GetInstance();
+ auto* original_policy_mgr = exec_env->index_policy_mgr();
+ exec_env->_index_policy_mgr = &policy_mgr;
+ Defer restore_policy_mgr([&] { exec_env->_index_policy_mgr =
original_policy_mgr; });
+
+ auto tablet_schema = create_english_index_schema();
+ const std::string segment_path = test_dir_ + "/all_rows_hidden_0.dat";
+ ASSERT_TRUE(write_index_segment(tablet_schema, test_dir_, segment_path,
{"alpha beta", "alpha"})
+ .ok());
+
+ auto splits = single_segment_splits(tablet_schema, segment_path);
+ MowHiddenRows hidden = hide_rows(splits, 9200, 3, {0, 1});
+ const Status status = stats_->collect(runtime_state_.get(), splits,
tablet_schema,
+ create_match_expr_contexts("alpha"),
nullptr, &hidden);
+ ASSERT_TRUE(status.ok()) << status;
+ EXPECT_EQ(stats_->get_doc_num(), 0);
+ EXPECT_EQ(stats_->get_term_doc_freq_by_col(L"1", L"alpha"), 0);
+ EXPECT_TRUE(std::isfinite(stats_->get_or_calculate_avg_dl(L"1")));
+ EXPECT_TRUE(std::isfinite(stats_->get_or_calculate_idf(L"1", L"alpha")));
+
+ // Without any collected segment the statistics are still reported missing.
+ CollectionStatistics nothing_collected;
+ EXPECT_THROW(nothing_collected.get_doc_num(), Exception);
+}
+
+// NULL rows have no tokens. A merge-on-write tablet, whose token count comes
from the norms, counts
+// them as documents without tokens, as a table without a delete bitmap does.
+TEST_F(CollectionStatisticsTest, NullRowsCountAsDocumentsWithoutTokens) {
+ IndexPolicyMgr policy_mgr;
+ auto* exec_env = ExecEnv::GetInstance();
+ auto* original_policy_mgr = exec_env->index_policy_mgr();
+ exec_env->_index_policy_mgr = &policy_mgr;
+ Defer restore_policy_mgr([&] { exec_env->_index_policy_mgr =
original_policy_mgr; });
+
+ auto tablet_schema = create_english_index_schema();
+ const std::string segment_path = test_dir_ + "/null_rows_0.dat";
+ ASSERT_TRUE(write_index_segment(tablet_schema, test_dir_, segment_path,
+ {"alpha beta", "", "alpha", "", "beta
gamma delta"}, {1, 3})
+ .ok());
+ auto splits = single_segment_splits(tablet_schema, segment_path);
+ auto collect = [&](const MowHiddenRows* hidden) {
+ CollectionStatistics stats;
+ const Status status = stats.collect(runtime_state_.get(), splits,
tablet_schema,
+
create_match_expr_contexts("alpha"), nullptr, hidden);
+ EXPECT_TRUE(status.ok()) << status;
+ return std::make_pair(stats.get_doc_num(),
stats.get_total_term_cnt_by_col(L"1"));
+ };
+
+ const auto without_bitmap = collect(nullptr);
+ EXPECT_EQ(without_bitmap.first, 5);
+ EXPECT_EQ(without_bitmap.second, 6);
+ MowHiddenRows nothing_hidden = hide_rows(splits, 9300, 2, {});
+ EXPECT_EQ(collect(¬hing_hidden), without_bitmap);
+
+ // A hidden NULL row leaves the document count and still adds no tokens.
+ MowHiddenRows null_hidden = hide_rows(splits, 9301, 2, {1});
+ EXPECT_EQ(collect(&null_hidden), std::make_pair(uint64_t {4}, uint64_t
{6}));
+
+ // The norms of the NULL rows stand for one token each; their lengths are
left out.
+ SegmentIndexReader reader(tablet_schema, segment_path);
+ uint64_t all_lengths = 0;
+ ASSERT_TRUE(sum_norm_lengths(reader.get(), L"1", &all_lengths).ok());
+ EXPECT_EQ(all_lengths, 8);
+ const std::vector<uint32_t> null_docs {1, 3};
+ uint64_t non_null_lengths = 0;
+ ASSERT_TRUE(sum_non_null_norm_lengths(reader.get(), L"1", null_docs,
&non_null_lengths).ok());
+ EXPECT_EQ(non_null_lengths, 6);
+ const std::vector<uint32_t> beyond {5};
+ EXPECT_EQ(sum_non_null_norm_lengths(reader.get(), L"1", beyond,
&non_null_lengths).code(),
+ ErrorCode::INVERTED_INDEX_FILE_CORRUPTED);
+ EXPECT_EQ(sum_norm_lengths_of(reader.get(), L"1", beyond,
&non_null_lengths).code(),
+ ErrorCode::INVERTED_INDEX_FILE_CORRUPTED);
+}
+
+// The hidden rows a term lists are counted by leaping over the postings
between them when they
+// are rare, and otherwise by walking them alongside the decoded postings.
+TEST_F(CollectionStatisticsTest, CountListedDocsLeapsOrScansOverHiddenRows) {
+ IndexPolicyMgr policy_mgr;
+ auto* exec_env = ExecEnv::GetInstance();
+ auto* original_policy_mgr = exec_env->index_policy_mgr();
+ exec_env->_index_policy_mgr = &policy_mgr;
+ Defer restore_policy_mgr([&] { exec_env->_index_policy_mgr =
original_policy_mgr; });
+
+ auto tablet_schema = create_english_index_schema();
+ std::vector<std::string> rows;
+ for (int i = 0; i < 300; ++i) {
+ rows.emplace_back(i % 3 == 0 ? "alpha beta" : "beta");
+ }
+ const std::string segment_path = test_dir_ + "/count_listed_0.dat";
+ ASSERT_TRUE(write_index_segment(tablet_schema, test_dir_, segment_path,
rows).ok());
+ SegmentIndexReader reader(tablet_schema, segment_path);
+ auto postings = [&](const std::string& term) {
+ return segment_v2::TermIterator::create(nullptr, false, reader.get(),
L"1", term);
+ };
+
+ // Eight rows among the 100 postings of alpha are walked alongside them;
among the 300 of
+ // beta, leapt to.
+ const std::vector<uint32_t> hidden {0, 1, 2, 3, 150, 151, 297, 299};
+ // 0, 3, 150 and 297 hold alpha.
+ EXPECT_EQ(count_listed_docs(*postings("alpha"), hidden), 4);
+ EXPECT_EQ(count_listed_docs(*postings("beta"), hidden), hidden.size());
+ EXPECT_EQ(count_listed_docs(*postings("omega"), hidden), 0);
+ EXPECT_EQ(count_listed_docs(*postings("alpha"), std::span<const uint32_t>
{}), 0);
+
+ // Many rows among the contiguous postings of beta are walked alongside
them, a block at a
+ // time.
+ std::vector<uint32_t> many;
+ for (uint32_t row = 5; row < 300; row += 7) {
+ many.push_back(row);
+ }
+ EXPECT_EQ(count_listed_docs(*postings("beta"), many), many.size());
+ // Those that are multiples of 3 hold alpha.
+ uint64_t alpha_rows = 0;
+ for (const uint32_t row : many) {
+ alpha_rows += row % 3 == 0 ? 1 : 0;
+ }
+ EXPECT_EQ(count_listed_docs(*postings("alpha"), many), alpha_rows);
+
+ uint64_t doc_frequency = 100;
+ ASSERT_TRUE(exclude_hidden_term_docs(*postings("alpha"), hidden,
&doc_frequency).ok());
+ EXPECT_EQ(doc_frequency, 96);
+ uint64_t too_small = 3;
+ EXPECT_EQ(exclude_hidden_term_docs(*postings("alpha"), hidden,
&too_small).code(),
+ ErrorCode::INVERTED_INDEX_FILE_CORRUPTED);
+
+ uint64_t lengths = 0;
+ ASSERT_TRUE(sum_norm_lengths(reader.get(), L"1", &lengths).ok());
+ EXPECT_EQ(lengths, 400);
+ ASSERT_TRUE(sum_norm_lengths_of(reader.get(), L"1", hidden,
&lengths).ok());
+ EXPECT_EQ(lengths, 12);
+ // A field without norms has no lengths to sum.
+ EXPECT_EQ(sum_norm_lengths_of(reader.get(), L"no_such_field", hidden,
&lengths).code(),
+ ErrorCode::INVERTED_INDEX_NOT_SUPPORTED);
+}
+
+TEST(CollectionStatisticsHiddenRowsTest,
ExcludeHiddenDocsSumsNormsOfReturnedRows) {
+ const std::vector<uint64_t> norm_lengths {3, 5, 2, 7};
+ auto sum_of = [&norm_lengths](std::span<const uint32_t> docs, uint64_t*
out) {
+ *out = 0;
+ for (const uint32_t doc : docs) {
+ *out += norm_lengths[doc];
+ }
+ return Status::OK();
+ };
+ const roaring::Roaring no_nulls;
+ SegmentScoringStats stats {.doc_count = 4, .token_count = 1000, .has_norms
= true};
+ const std::vector<uint32_t> hidden {1, 3};
+ ASSERT_TRUE(exclude_hidden_docs(hidden, no_nulls, 17, sum_of,
&stats).ok());
+ EXPECT_EQ(stats.doc_count, 2);
+ EXPECT_EQ(stats.token_count, 5);
+
+ // With nothing hidden the token count still comes from the norms, as on a
replica that
+ // compacted the hidden rows away.
+ stats = {.doc_count = 4, .token_count = 1000, .has_norms = true};
+ ASSERT_TRUE(exclude_hidden_docs({}, no_nulls, 17, sum_of, &stats).ok());
+ EXPECT_EQ(stats.doc_count, 4);
+ EXPECT_EQ(stats.token_count, 17);
+}
+
+TEST(CollectionStatisticsHiddenRowsTest,
ExcludeHiddenDocsWithoutNormsCountsOneTokenPerRow) {
+ auto never = [](std::span<const uint32_t>, uint64_t*) {
+ ADD_FAILURE() << "an index without norms reads none";
+ return Status::OK();
+ };
+ SegmentScoringStats stats {.doc_count = 5, .token_count = 9, .has_norms =
false};
+ const std::vector<uint32_t> hidden {0, 4};
+ roaring::Roaring nulls;
+ nulls.add(1);
+ ASSERT_TRUE(exclude_hidden_docs(hidden, nulls, 0, never, &stats).ok());
+ EXPECT_EQ(stats.doc_count, 3);
+ EXPECT_EQ(stats.token_count, 3);
+}
+
+TEST(CollectionStatisticsHiddenRowsTest,
ExcludeHiddenDocsRejectsRowBeyondIndex) {
+ auto sum_of = [](std::span<const uint32_t>, uint64_t* out) {
+ *out = 0;
+ return Status::OK();
+ };
+ SegmentScoringStats stats {.doc_count = 2, .token_count = 2, .has_norms =
true};
+ const std::vector<uint32_t> hidden {2};
+ const Status status = exclude_hidden_docs(hidden, roaring::Roaring {}, 2,
sum_of, &stats);
+ EXPECT_EQ(status.code(), ErrorCode::INVERTED_INDEX_FILE_CORRUPTED) <<
status;
+}
+
+// A NULL row is a document without tokens, though its norm stands for one: it
stays in the
+// document count and its length is left out, hidden or not.
+TEST(CollectionStatisticsHiddenRowsTest,
ExcludeHiddenDocsLeavesOutLengthsOfNullRows) {
+ const std::vector<uint64_t> norm_lengths {3, 5, 1, 7, 1};
+ std::vector<uint32_t> summed;
+ auto sum_of = [&](std::span<const uint32_t> docs, uint64_t* out) {
+ *out = 0;
+ for (const uint32_t doc : docs) {
+ summed.push_back(doc);
+ *out += norm_lengths[doc];
+ }
+ return Status::OK();
+ };
+ roaring::Roaring nulls;
+ nulls.add(2);
+ nulls.add(4);
+ // 3 + 5 + 7, the lengths of the non-NULL rows.
+ constexpr uint64_t kNonNullLengths = 15;
+
+ SegmentScoringStats stats {.doc_count = 5, .token_count = 1000, .has_norms
= true};
+ ASSERT_TRUE(exclude_hidden_docs({}, nulls, kNonNullLengths, sum_of,
&stats).ok());
+ EXPECT_EQ(stats.doc_count, 5);
+ EXPECT_EQ(stats.token_count, 15);
+
+ stats = {.doc_count = 5, .token_count = 1000, .has_norms = true};
+ const std::vector<uint32_t> hidden {1, 2};
+ ASSERT_TRUE(exclude_hidden_docs(hidden, nulls, kNonNullLengths, sum_of,
&stats).ok());
+ EXPECT_EQ(stats.doc_count, 3);
+ EXPECT_EQ(stats.token_count, 10);
+ // Only the hidden non-NULL row's norm is read.
+ EXPECT_EQ(summed, std::vector<uint32_t> {1});
}
TEST_F(CollectionStatisticsTest, CollectWithCastWrappedSlotRef) {
diff --git
a/be/test/storage/index/inverted/query_v2/multi_segment_collector_test.cpp
b/be/test/storage/index/inverted/query_v2/multi_segment_collector_test.cpp
index 28acadf0d92..6a9ab114bf4 100644
--- a/be/test/storage/index/inverted/query_v2/multi_segment_collector_test.cpp
+++ b/be/test/storage/index/inverted/query_v2/multi_segment_collector_test.cpp
@@ -17,6 +17,23 @@
#include <CLucene.h>
#include <CLucene/index/MultiReader.h>
+#ifdef __clang__
+#pragma clang diagnostic push
+#pragma clang diagnostic ignored "-Wshadow-field"
+#pragma clang diagnostic ignored "-Woverloaded-virtual"
+#pragma clang diagnostic ignored "-Winconsistent-missing-override"
+#pragma clang diagnostic ignored "-Wreorder-ctor"
+#pragma clang diagnostic ignored "-Wshorten-64-to-32"
+#elif defined(__GNUC__)
+#pragma GCC diagnostic push
+#pragma GCC diagnostic ignored "-Woverloaded-virtual"
+#endif
+#include <CLucene/index/_MultiSegmentReader.h>
+#ifdef __clang__
+#pragma clang diagnostic pop
+#elif defined(__GNUC__)
+#pragma GCC diagnostic pop
+#endif
#include <gtest/gtest.h>
#include <memory>
@@ -25,6 +42,7 @@
#include <vector>
#include "io/fs/local_file_system.h"
+#include "storage/compaction/collection_statistics.h"
#include "storage/index/index_query_context.h"
#include "storage/index/inverted/analyzer/custom_analyzer.h"
#include "storage/index/inverted/query_v2/bit_set_query/bit_set_query.h"
@@ -33,6 +51,7 @@
#include "storage/index/inverted/query_v2/collect/top_k_collector.h"
#include "storage/index/inverted/query_v2/null_masked_query.h"
#include "storage/index/inverted/query_v2/term_query/term_query.h"
+#include "storage/index/inverted/similarity/bm25_similarity.h"
#include "storage/index/inverted/util/string_helper.h"
CL_NS_USE(index)
@@ -66,9 +85,11 @@ public:
EXPECT_TRUE(io::global_local_filesystem()->delete_directory(kTestDir).ok());
}
-private:
+protected:
static void create_test_index(const std::string& dir, const
std::vector<std::string>& docs,
- int32_t max_buffered_docs = 100) {
+ int32_t max_buffered_docs = 100,
+ const std::string& field_name = "title",
+ bool store_norms = false) {
CustomAnalyzerConfig::Builder builder;
builder.with_tokenizer_config("standard", {});
auto custom_analyzer_config = builder.build();
@@ -85,9 +106,11 @@ private:
auto char_string_reader =
std::make_shared<lucene::util::SStringReader<char>>();
auto* doc = _CLNEW lucene::document::Document();
int32_t field_config = lucene::document::Field::STORE_NO;
- field_config |= lucene::document::Field::INDEX_NONORMS;
+ if (!store_norms) {
+ field_config |= lucene::document::Field::INDEX_NONORMS;
+ }
field_config |= lucene::document::Field::INDEX_TOKENIZED;
- auto field_name_w = StringHelper::to_wstring("title");
+ auto field_name_w = StringHelper::to_wstring(field_name);
auto* field = _CLNEW lucene::document::Field(field_name_w.c_str(),
field_config);
field->setOmitTermFreqAndPositions(false);
doc->add(*field);
@@ -251,4 +274,47 @@ TEST_F(MultiSegmentCollectorTest,
CollectDocSetWithSingleReaderBinding) {
_CLDECDELETE(dir1);
}
+// An index of several CLucene segments opens as a reader of several segments.
The BM25 statistics
+// read its norm length sums per segment, and they equal those of the reader's
combined norms.
+TEST_F(MultiSegmentCollectorTest, NormLengthSumsReadEachSegment) {
+ const std::string dir = kTestDir + "/segmented_norms";
+ ASSERT_TRUE(io::global_local_filesystem()->create_directory(dir).ok());
+ std::string long_doc;
+ for (int i = 0; i < 30; ++i) {
+ long_doc += "word ";
+ }
+ create_test_index(dir, {"a b", "c", long_doc, "d e f", "g", "h i"}, 2,
"title", true);
+
+ auto reader =
make_shared_reader(lucene::index::IndexReader::open(dir.c_str()));
+ auto* multi_segment =
dynamic_cast<lucene::index::MultiSegmentReader*>(reader.get());
+ ASSERT_NE(multi_segment, nullptr);
+ ASSERT_EQ(multi_segment->getSubReaders()->length, 3);
+ uint64_t all = 0;
+ ASSERT_TRUE(collection_statistics_detail::sum_norm_lengths(reader.get(),
L"title", &all).ok());
+ const std::vector<uint32_t> docs {0, 2, 3, 5};
+ uint64_t some = 0;
+ ASSERT_TRUE(
+ collection_statistics_detail::sum_norm_lengths_of(reader.get(),
L"title", docs, &some)
+ .ok());
+
+ auto combined =
make_shared_reader(lucene::index::IndexReader::open(dir.c_str()));
+ const uint8_t* norms = combined->norms(L"title");
+ ASSERT_NE(norms, nullptr);
+ auto length = [](uint8_t norm) {
+ return
static_cast<uint64_t>(static_cast<float>(BM25Similarity::byte4_to_int(norm)));
+ };
+ uint64_t expected_all = 0;
+ for (int32_t doc = 0; doc < combined->maxDoc(); ++doc) {
+ expected_all += length(norms[doc]);
+ }
+ uint64_t expected_some = 0;
+ for (const uint32_t doc : docs) {
+ expected_some += length(norms[doc]);
+ }
+ EXPECT_EQ(all, expected_all);
+ EXPECT_EQ(some, expected_some);
+ // Lengths below 24 are exact; the 30-token row reads back approximately.
+ EXPECT_EQ(expected_all - length(norms[2]), 9);
+}
+
} // namespace doris::segment_v2
diff --git a/be/test/storage/segment/inverted_index_searcher_cache_test.cpp
b/be/test/storage/segment/inverted_index_searcher_cache_test.cpp
index 222d016bc46..7ac2d2091f8 100644
--- a/be/test/storage/segment/inverted_index_searcher_cache_test.cpp
+++ b/be/test/storage/segment/inverted_index_searcher_cache_test.cpp
@@ -24,6 +24,7 @@
#include <memory>
#include <string>
#include <thread>
+#include <vector>
#include "common/status.h"
#include "gtest/gtest_pred_impl.h"
@@ -330,5 +331,78 @@ TEST_F(InvertedIndexSearcherCacheTest,
remove_element_only_in_table) {
delete index_searcher_cache;
}
+// The non-NULL norm length sum of a field is computed once per cached
searcher, kept apart from
+// the other fields' sums, and not kept when computing it fails.
+TEST(InvertedIndexSearcherCacheValueTest,
NonNullNormLengthSumIsComputedOncePerField) {
+ InvertedIndexSearcherCache::CacheValue
value(FulltextIndexSearcherPtr(nullptr), 0, 0);
+ int computed = 0;
+ auto compute = [&computed](uint64_t sum) {
+ return [&computed, sum](uint64_t* out) {
+ ++computed;
+ *out = sum;
+ return Status::OK();
+ };
+ };
+ uint64_t sum = 0;
+ ASSERT_TRUE(value.non_null_norm_length_sum(L"a", compute(7), &sum).ok());
+ EXPECT_EQ(sum, 7);
+ ASSERT_TRUE(value.non_null_norm_length_sum(L"a", compute(9), &sum).ok());
+ EXPECT_EQ(sum, 7);
+ ASSERT_TRUE(value.non_null_norm_length_sum(L"b", compute(9), &sum).ok());
+ EXPECT_EQ(sum, 9);
+ EXPECT_EQ(computed, 2);
+
+ auto fail = [&computed](uint64_t*) {
+ ++computed;
+ return Status::Corruption("bad norms");
+ };
+ EXPECT_FALSE(value.non_null_norm_length_sum(L"c", fail, &sum).ok());
+ ASSERT_TRUE(value.non_null_norm_length_sum(L"c", compute(3), &sum).ok());
+ EXPECT_EQ(sum, 3);
+ EXPECT_EQ(computed, 4);
+}
+
+// The norm length sum of a set of hidden rows is computed once per field and
set, and a few sets
+// per field are kept, the oldest given up first.
+TEST(InvertedIndexSearcherCacheValueTest,
HiddenNormLengthSumIsComputedOncePerRowSet) {
+ InvertedIndexSearcherCache::CacheValue
value(FulltextIndexSearcherPtr(nullptr), 0, 0);
+ int computed = 0;
+ auto compute = [&computed](uint64_t sum) {
+ return [&computed, sum](uint64_t* out) {
+ ++computed;
+ *out = sum;
+ return Status::OK();
+ };
+ };
+ const std::vector<uint32_t> rows {1, 5, 9};
+ const std::vector<uint32_t> other_rows {1, 5, 10};
+ uint64_t sum = 0;
+ ASSERT_TRUE(value.hidden_norm_length_sum(L"a", rows, compute(7),
&sum).ok());
+ EXPECT_EQ(sum, 7);
+ ASSERT_TRUE(value.hidden_norm_length_sum(L"a", rows, compute(8),
&sum).ok());
+ EXPECT_EQ(sum, 7);
+ ASSERT_TRUE(value.hidden_norm_length_sum(L"a", other_rows, compute(8),
&sum).ok());
+ EXPECT_EQ(sum, 8);
+ ASSERT_TRUE(value.hidden_norm_length_sum(L"b", rows, compute(9),
&sum).ok());
+ EXPECT_EQ(sum, 9);
+ EXPECT_EQ(computed, 3);
+
+ auto fail = [](uint64_t*) { return Status::Corruption("bad norms"); };
+ const std::vector<uint32_t> failing_rows {2};
+ EXPECT_FALSE(value.hidden_norm_length_sum(L"a", failing_rows, fail,
&sum).ok());
+ ASSERT_TRUE(value.hidden_norm_length_sum(L"a", failing_rows, compute(4),
&sum).ok());
+ EXPECT_EQ(sum, 4);
+ EXPECT_EQ(computed, 4);
+
+ // Field "a" holds {1,5,9}, {1,5,10} and {2}; two more sets give up
{1,5,9}.
+ const std::vector<uint32_t> fourth {3};
+ const std::vector<uint32_t> fifth {4};
+ ASSERT_TRUE(value.hidden_norm_length_sum(L"a", fourth, compute(1),
&sum).ok());
+ ASSERT_TRUE(value.hidden_norm_length_sum(L"a", fifth, compute(1),
&sum).ok());
+ ASSERT_TRUE(value.hidden_norm_length_sum(L"a", rows, compute(6),
&sum).ok());
+ EXPECT_EQ(sum, 6);
+ EXPECT_EQ(computed, 7);
+}
+
} // namespace segment_v2
} // namespace doris
diff --git
a/regression-test/data/fault_injection_p0/test_bm25_score_statistics_searcher_io_context.out
b/regression-test/data/fault_injection_p0/test_bm25_score_statistics_searcher_io_context.out
new file mode 100644
index 00000000000..bcd881009df
--- /dev/null
+++
b/regression-test/data/fault_injection_p0/test_bm25_score_statistics_searcher_io_context.out
@@ -0,0 +1,103 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !scores --
+0 0.6931471
+1 0.6931471
+10 0.6931471
+101 0.6931471
+105 0.6931471
+109 0.6931471
+113 0.6931471
+117 0.6931471
+12 0.6931471
+121 0.6931471
+125 0.6931471
+129 0.6931471
+13 0.6931471
+133 0.6931471
+137 0.6931471
+14 0.6931471
+141 0.6931471
+145 0.6931471
+149 0.6931471
+153 0.6931471
+157 0.6931471
+16 0.6931471
+161 0.6931471
+165 0.6931471
+169 0.6931471
+17 0.6931471
+173 0.6931471
+177 0.6931471
+18 0.6931471
+181 0.6931471
+185 0.6931471
+189 0.6931471
+193 0.6931471
+197 0.6931471
+2 0.6931471
+20 0.6931471
+21 0.6931471
+22 0.6931471
+24 0.6931471
+25 0.6931471
+26 0.6931471
+28 0.6931471
+29 0.6931471
+30 0.6931471
+32 0.6931471
+33 0.6931471
+34 0.6931471
+36 0.6931471
+37 0.6931471
+38 0.6931471
+4 0.6931471
+40 0.6931471
+41 0.6931471
+42 0.6931471
+44 0.6931471
+45 0.6931471
+46 0.6931471
+48 0.6931471
+49 0.6931471
+5 0.6931471
+50 0.6931471
+52 0.6931471
+53 0.6931471
+54 0.6931471
+56 0.6931471
+57 0.6931471
+58 0.6931471
+6 0.6931471
+60 0.6931471
+61 0.6931471
+62 0.6931471
+64 0.6931471
+65 0.6931471
+66 0.6931471
+68 0.6931471
+69 0.6931471
+70 0.6931471
+72 0.6931471
+73 0.6931471
+74 0.6931471
+76 0.6931471
+77 0.6931471
+78 0.6931471
+8 0.6931471
+80 0.6931471
+81 0.6931471
+82 0.6931471
+84 0.6931471
+85 0.6931471
+86 0.6931471
+88 0.6931471
+89 0.6931471
+9 0.6931471
+90 0.6931471
+92 0.6931471
+93 0.6931471
+94 0.6931471
+96 0.6931471
+97 0.6931471
+98 0.6931471
+
diff --git a/regression-test/data/inverted_index_p0/test_bm25_score.out
b/regression-test/data/inverted_index_p0/test_bm25_score.out
index 5b8ede10f56..4dfdb19d126 100644
--- a/regression-test/data/inverted_index_p0/test_bm25_score.out
+++ b/regression-test/data/inverted_index_p0/test_bm25_score.out
@@ -42,7 +42,7 @@
3 3 this is a test
-- !sql --
-4 4 hello world 1.0925692
+4 4 hello world 0.8025915
-- !sql --
4 4 hello world
diff --git
a/regression-test/data/inverted_index_p0/test_bm25_score_replica_statistics.out
b/regression-test/data/inverted_index_p0/test_bm25_score_replica_statistics.out
new file mode 100644
index 00000000000..90163f13821
--- /dev/null
+++
b/regression-test/data/inverted_index_p0/test_bm25_score_replica_statistics.out
@@ -0,0 +1,6 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !emptied --
+
+-- !refilled --
+4 0.2876821
+
diff --git
a/regression-test/data/inverted_index_p0/test_score_filter_matrix.out
b/regression-test/data/inverted_index_p0/test_score_filter_matrix.out
index 0a69f5e8815..cd932f57e8a 100644
--- a/regression-test/data/inverted_index_p0/test_score_filter_matrix.out
+++ b/regression-test/data/inverted_index_p0/test_score_filter_matrix.out
@@ -148,32 +148,32 @@
6 0.7187315
-- !V2_match_mow --
-2 0.64072424
-3 0.90232176
+2 0.60996956
+3 0.8713851
-- !V2_match_mow_topn_control --
-2 0.64072424
-3 0.90232176
+2 0.60996956
+3 0.8713851
-- !V2_search_mow --
-2 0.64072424
-3 0.90232176
+2 0.60996956
+3 0.8713851
-- !V2_search_mow_topn_control --
-2 0.64072424
-3 0.90232176
+2 0.60996956
+3 0.8713851
-- !V2_match_mow_after_delete --
-3 1.0217376
+3 1.1823694
-- !V2_match_mow_after_delete_topn_control --
-3 1.0217376
+3 1.1823694
-- !V2_search_mow_after_delete --
-3 1.0217376
+3 1.1823694
-- !V2_search_mow_after_delete_topn_control --
-3 1.0217376
+3 1.1823694
-- !V3_match_range --
1 0.4700036
@@ -324,32 +324,32 @@
6 0.7187315
-- !V3_match_mow --
-2 0.64072424
-3 0.90232176
+2 0.60996956
+3 0.8713851
-- !V3_match_mow_topn_control --
-2 0.64072424
-3 0.90232176
+2 0.60996956
+3 0.8713851
-- !V3_search_mow --
-2 0.64072424
-3 0.90232176
+2 0.60996956
+3 0.8713851
-- !V3_search_mow_topn_control --
-2 0.64072424
-3 0.90232176
+2 0.60996956
+3 0.8713851
-- !V3_match_mow_after_delete --
-3 1.0217376
+3 1.1823694
-- !V3_match_mow_after_delete_topn_control --
-3 1.0217376
+3 1.1823694
-- !V3_search_mow_after_delete --
-3 1.0217376
+3 1.1823694
-- !V3_search_mow_after_delete_topn_control --
-3 1.0217376
+3 1.1823694
-- !ann_control --
1 0.0
diff --git
a/regression-test/suites/fault_injection_p0/test_bm25_score_statistics_searcher_io_context.groovy
b/regression-test/suites/fault_injection_p0/test_bm25_score_statistics_searcher_io_context.groovy
new file mode 100644
index 00000000000..4e7f3f71ed9
--- /dev/null
+++
b/regression-test/suites/fault_injection_p0/test_bm25_score_statistics_searcher_io_context.groovy
@@ -0,0 +1,61 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+// BM25 statistics open the CLucene searcher of a segment the searcher cache
does not hold yet.
+// Once its meta is read, the searcher's stream must drop the opening query's
I/O context and stop
+// reading as index meta, as the query path's searcher does:
CSIndexInput.readInternal fails any
+// read of index data that still reads as meta. Loads must not put the
searchers into the cache
+// first, or the statistics would not open them.
+suite("test_bm25_score_statistics_searcher_io_context", "nonConcurrent") {
+ sql "DROP TABLE IF EXISTS test_bm25_score_statistics_searcher_io_context"
+ sql """
+ CREATE TABLE test_bm25_score_statistics_searcher_io_context (
+ id BIGINT,
+ content TEXT NULL,
+ INDEX idx_content (content) USING INVERTED PROPERTIES("parser" =
"english")
+ ) ENGINE=OLAP
+ UNIQUE KEY(id)
+ DISTRIBUTED BY HASH(id) BUCKETS 1
+ PROPERTIES (
+ "replication_num" = "1",
+ "enable_unique_key_merge_on_write" = "true",
+ "disable_auto_compaction" = "true"
+ );
+ """
+ setBeConfigTemporary([enable_write_index_searcher_cache: false]) {
+ sql """
+ INSERT INTO test_bm25_score_statistics_searcher_io_context
+ SELECT number, concat_ws(' ', if(number % 4 = 1, 'jobs', 'work'),
'alpha') FROM numbers('number' = '200')
+ """
+ // Overwritten rows make the statistics read postings of the hidden
rows.
+ sql """
+ INSERT INTO test_bm25_score_statistics_searcher_io_context
+ SELECT number * 2, 'jobs again' FROM numbers('number' = '50')
+ """
+ sql "sync"
+ }
+
+ try {
+ GetDebugPoint().enableDebugPointForAllBEs("CSIndexInput.readInternal")
+ order_qt_scores """
+ SELECT id, score() FROM
test_bm25_score_statistics_searcher_io_context
+ WHERE content MATCH_ANY 'jobs' ORDER BY score() DESC LIMIT 1000
+ """
+ } finally {
+ GetDebugPoint().disableDebugPointForAllBEs("CSIndexInput.readInternal")
+ }
+}
diff --git
a/regression-test/suites/inverted_index_p0/test_bm25_score_replica_statistics.groovy
b/regression-test/suites/inverted_index_p0/test_bm25_score_replica_statistics.groovy
new file mode 100644
index 00000000000..45ea13d1acc
--- /dev/null
+++
b/regression-test/suites/inverted_index_p0/test_bm25_score_replica_statistics.groovy
@@ -0,0 +1,178 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+import java.util.concurrent.TimeUnit
+import org.awaitility.Awaitility
+
+// A merge-on-write replica keeps overwritten rows and rows written with the
delete sign in its
+// segments until it compacts them, and each replica compacts on its own
schedule. BM25 statistics
+// must count only the rows a query sees, so a row scores the same on every
replica, before and
+// after compaction, and the same as in a table freshly loaded with the
visible rows.
+suite("test_bm25_score_replica_statistics", "p0") {
+ def aliveBackends = sql_return_maparray("SHOW BACKENDS").findAll {
it.Alive.toString() == "true" }
+ def replicaNum = aliveBackends.size() >= 3 ? 3 : 1
+ sql "set enable_common_expr_pushdown = true"
+ sql "set enable_common_expr_pushdown_for_inverted_index = true"
+
+ def createTable = { String name, String keys, int replicas ->
+ sql "DROP TABLE IF EXISTS ${name}"
+ def mergeOnWrite = keys.startsWith("UNIQUE") ?
'"enable_unique_key_merge_on_write" = "true",' : ""
+ sql """
+ CREATE TABLE ${name} (
+ id BIGINT,
+ content TEXT NULL,
+ INDEX idx_content (content) USING INVERTED
+ PROPERTIES("parser" = "english", "support_phrase" = "true")
+ ) ENGINE=OLAP
+ ${keys}
+ DISTRIBUTED BY HASH(id) BUCKETS 4
+ PROPERTIES (
+ "replication_num" = "${replicas}",
+ ${mergeOnWrite}
+ "disable_auto_compaction" = "true"
+ );
+ """
+ }
+ createTable("test_bm25_score_replica_statistics", "UNIQUE KEY(id)",
replicaNum)
+ createTable("test_bm25_score_replica_statistics_fresh", "UNIQUE KEY(id)",
1)
+
+ // Documents of different lengths, some longer than the lengths norms hold
exactly, and NULLs.
+ def text = { int round ->
+ """if((number + ${round}) % 7 = 0, NULL, concat_ws(' ',
+ repeat('alpha beta gamma ', cast((number * 7 + ${round} * 13) %
15 AS INT)),
+ if((number + ${round}) % 3 = 0, 'jobs', 'work'),
+ if((number * 5 + ${round}) % 4 = 0, 'try hard', 'rest'),
+ repeat('delta ', cast((number + ${round}) % 5 AS INT))))"""
+ }
+ for (int batch = 0; batch < 6; batch++) {
+ sql "INSERT INTO test_bm25_score_replica_statistics SELECT number +
${batch} * 500, ${text(batch)} FROM numbers('number' = '500')"
+ }
+ // Overwrite a quarter of the rows three times.
+ for (int round = 1; round <= 3; round++) {
+ sql "INSERT INTO test_bm25_score_replica_statistics SELECT number * 4
+ ${round}, ${text(round + 20)} FROM numbers('number' = '750')"
+ }
+ // Delete rows with DELETE, and with full rows carrying the delete sign,
as CDC loads write them.
+ sql "DELETE FROM test_bm25_score_replica_statistics WHERE id % 17 = 3"
+ sql "SET show_hidden_columns = true"
+ sql """
+ INSERT INTO test_bm25_score_replica_statistics (id, content,
__DORIS_DELETE_SIGN__)
+ SELECT number * 13 + 5, ${text(40)}, 1 FROM numbers('number' = '200')
+ """
+ sql "SET show_hidden_columns = false"
+ sql "INSERT INTO test_bm25_score_replica_statistics_fresh SELECT id,
content FROM test_bm25_score_replica_statistics"
+ sql "sync"
+
+ def scores = { String table, String vars ->
+ def rows = sql """
+ SELECT /*+ SET_VAR(${vars}) */ id, score()
+ FROM ${table}
+ WHERE SEARCH('content:jobs OR content:try')
+ ORDER BY score() DESC
+ LIMIT 100000
+ """
+ def byId = [:]
+ rows.each { byId[it[0]] = it[1] }
+ return byId
+ }
+ def assertSameScores = { Map expected, Map actual, String state ->
+ assertTrue(expected.size() > 0)
+ assertEquals(expected.size(), actual.size())
+ def differing = expected.findAll { id, score -> actual[id] != score }
+ assertTrue(differing.isEmpty(),
+ "${state}: ${differing.size()} rows score differently, "
+ + "e.g. ${differing.take(3).collect { id, score -> "id
${id}: ${actual[id]} vs ${score}" }}")
+ }
+
+ def expected = scores("test_bm25_score_replica_statistics_fresh",
"use_fix_replica = -1")
+ def assertScoresLikeFreshTable = { String vars, String state ->
+ assertSameScores(expected,
scores("test_bm25_score_replica_statistics", vars),
+ "${state} (${vars}), compared with a freshly loaded table")
+ }
+
+ assertScoresLikeFreshTable("use_fix_replica = -1", "before compaction")
+
+ // Cloud mode keeps one shared copy of each tablet, so there are no
replicas to compact apart,
+ // and its compaction is driven differently.
+ if (!isCloudMode() && replicaNum == 3) {
+ // Compact the replicas on one backend only, so the replicas hold
different rows.
+ def backendIdToIp = [:]
+ def backendIdToHttpPort = [:]
+ getBackendIpHttpPort(backendIdToIp, backendIdToHttpPort)
+ def tablets = sql_return_maparray("SHOW TABLETS FROM
test_bm25_score_replica_statistics")
+ def compactedBackend = tablets[0].BackendId
+ def compactedTablets = tablets.findAll { it.BackendId ==
compactedBackend }
+ for (tablet in compactedTablets) {
+ def (code, out, err) =
be_run_full_compaction(backendIdToIp["${compactedBackend}"],
+ backendIdToHttpPort["${compactedBackend}"],
tablet.TabletId)
+ assertEquals(0, code)
+ assertTrue(out.toLowerCase().contains("success"), out)
+ }
+ Awaitility.await().atMost(300, TimeUnit.SECONDS).pollInterval(1,
TimeUnit.SECONDS).until {
+ compactedTablets.every { tablet ->
+ def (code, out, err) =
be_show_tablet_status(backendIdToIp["${compactedBackend}"],
+ backendIdToHttpPort["${compactedBackend}"],
tablet.TabletId)
+ code == 0 && parseJson(out.trim()).rowsets.size() == 1
+ }
+ }
+
+ for (int run = 0; run < 10; run++) {
+ assertScoresLikeFreshTable("use_fix_replica = -1", "one replica
compacted, run ${run}")
+ }
+ for (int replica = 0; replica < 3; replica++) {
+ assertScoresLikeFreshTable("use_fix_replica = ${replica}", "one
replica compacted")
+ }
+ }
+
+ if (!isCloudMode()) {
+ trigger_and_wait_compaction("test_bm25_score_replica_statistics",
"full")
+ for (int replica = 0; replica < replicaNum; replica++) {
+ assertScoresLikeFreshTable("use_fix_replica = ${replica}", "all
replicas compacted")
+ }
+ }
+
+ // NULL rows have no tokens, on a merge-on-write table as on a duplicate
table: with lengths the
+ // norms hold exactly, both score alike.
+ createTable("test_bm25_score_replica_statistics_nulls_mow", "UNIQUE
KEY(id)", 1)
+ createTable("test_bm25_score_replica_statistics_nulls_dup", "DUPLICATE
KEY(id)", 1)
+ for (table in ["test_bm25_score_replica_statistics_nulls_mow",
"test_bm25_score_replica_statistics_nulls_dup"]) {
+ sql """
+ INSERT INTO ${table}
+ SELECT number, if(number % 2 = 0, NULL,
+ concat_ws(' ', if(number % 4 = 1, 'jobs',
'work'), repeat('alpha ', cast(number % 6 AS INT))))
+ FROM numbers('number' = '1000')
+ """
+ }
+ sql "sync"
+ assertSameScores(scores("test_bm25_score_replica_statistics_nulls_dup",
"use_fix_replica = -1"),
+ scores("test_bm25_score_replica_statistics_nulls_mow",
"use_fix_replica = -1"),
+ "merge-on-write table with NULL rows, compared with a duplicate
table")
+
+ // A bucket whose rows are all deleted still has segments but no row a
query can match. Scoring
+ // must return nothing for it instead of failing.
+ createTable("test_bm25_score_replica_statistics_emptied", "UNIQUE
KEY(id)", 1)
+ sql "INSERT INTO test_bm25_score_replica_statistics_emptied VALUES (1, 'we
should try these jobs'), (2, 'more jobs'), (3, 'other')"
+ sql "DELETE FROM test_bm25_score_replica_statistics_emptied WHERE id > 0"
+ order_qt_emptied """
+ SELECT id, score() FROM test_bm25_score_replica_statistics_emptied
+ WHERE SEARCH('content:jobs') ORDER BY score() DESC LIMIT 10
+ """
+ sql "INSERT INTO test_bm25_score_replica_statistics_emptied VALUES (4,
'jobs again')"
+ order_qt_refilled """
+ SELECT id, score() FROM test_bm25_score_replica_statistics_emptied
+ WHERE SEARCH('content:jobs') ORDER BY score() DESC LIMIT 10
+ """
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
