This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.2 by this push:
new 7ad136799f2 branch-4.2: [fix](inverted index) Share BM25 statistics
across parallel scanners of a tablet #69352 (#69440)
7ad136799f2 is described below
commit 7ad136799f2a36f4085ad7107195979a1250a788
Author: guptashivam04 <[email protected]>
AuthorDate: Fri Oct 23 08:12:34 2026 +0530
branch-4.2: [fix](inverted index) Share BM25 statistics across parallel
scanners of a tablet #69352 (#69440)
### What problem does this PR solve?
Issue Number: None
Related PR: #69352
Problem Summary:
Backport of #69352 to branch-4.2.
A parallel scan (`ParallelScannerBuilder`) splits one tablet into
several scanners. Each `OlapScanner` then calls
`CollectionStatistics::collect()` with only the `rs_splits` of its own
split, so the BM25 doc count, avgdl and term doc frequencies come from a
subset of the tablet's rowsets. Rows with identical text get different
scores depending on which scanner reads them. Because the split depends
on the tablet's row count and the core count, a row's score and its rank
in `ORDER BY score()` change when unrelated rows are loaded, and
pagination over a score-ordered search returns repeated and missing
rows.
Fix (same as #69352):
- `TabletCollectionStatistics` holds the rowsets of a tablet's entire
read source. The first scanner of the tablet to prepare collects the
statistics once with `collect_full_collection()`, and every scanner of
that tablet copies them into its own `CollectionStatistics`. Later
scanners wait for the first collection and get its result or error.
- `CollectionStatistics::copy_collected_from()` copies the collected
counters and resets the derived idf/avgdl caches, so no cache is shared
between scanner threads.
- `ParallelScannerBuilder` creates one holder per tablet, only when the
scan has a `score()` runtime. Both the by-rowid and by-segment build
paths pass it to their scanners through `OlapScanner::Params`.
- The non-parallel path is unchanged. Each of its scanners already
receives the tablet's full read source.
Differences from #69352: branch-4.2 does not have the changelog, TSO and
bucket fields of `OlapScanner::Params` or the `scan_range` argument of
`_build_scanner`, so the scanner changes were applied by hand on top of
the branch-4.2 code. `OlapScanner::Params` is initialized positionally
on this branch, so the new field is appended at the end. branch-4.2 has
no test that calls `_build_scanner` directly, so the follow-up test fix
from #69352 is not needed here.
Statistics remain per tablet. Making them consistent across tablets and
BEs is out of scope.
### Release note
Fix BM25 `score()` returning different scores for identical rows of the
same tablet when a parallel scan split the tablet into several scanners.
### Check List (For Author)
- Test
- [x] Regression test:
`inverted_index_p0/test_bm25_score_parallel_scan_statistics` (same suite
as #69352). Passes on branch-4.2 with this change.
- [x] Unit Test:
`CollectionStatisticsDetailedTest.CopyCollectedFromScoresLikeSource`,
`TabletCollectionStatisticsTest.CollectsOnceAndCopiesToEveryCaller`
(compiled locally; run left to CI)
- [x] Manual test: ran the reproduction from #69352 against branch-4.2
builds on macOS arm64.
| Build | `enable_parallel_scan=true` (split forced) |
`enable_parallel_scan=false` |
|---|---|---|
| branch-4.2 `a5d2661d`, without this change | 15 distinct scores (1.34
to 5.81) | 1 score (3.647686) |
| branch-4.2 `a5d2661d` with this change | 1 score (3.647686), same over
5 repeated runs | 1 score (3.647686) |
- [ ] No need to test or manual test.
- Behavior changed:
- [x] Yes. Score queries that use a parallel scan now score every row of
a tablet against statistics of the whole tablet. Scores can differ from
earlier releases but are now independent of the scanner split.
- 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/operator/olap_scan_operator.cpp | 5 ++
be/src/exec/scan/olap_scanner.cpp | 14 +++-
be/src/exec/scan/olap_scanner.h | 5 ++
be/src/exec/scan/parallel_scanner_builder.cpp | 37 ++++++++--
be/src/exec/scan/parallel_scanner_builder.h | 16 +++--
.../inverted/reader/collection_statistics.cpp | 27 ++++++++
.../index/inverted/reader/collection_statistics.h | 32 +++++++++
.../inverted/reader/collection_statistics_test.cpp | 32 +++++++++
...test_bm25_score_parallel_scan_statistics.groovy | 78 ++++++++++++++++++++++
9 files changed, 234 insertions(+), 12 deletions(-)
diff --git a/be/src/exec/operator/olap_scan_operator.cpp
b/be/src/exec/operator/olap_scan_operator.cpp
index b2cbc3b4213..0043a58fd9a 100644
--- a/be/src/exec/operator/olap_scan_operator.cpp
+++ b/be/src/exec/operator/olap_scan_operator.cpp
@@ -641,6 +641,8 @@ Status
OlapScanLocalState::_init_scanners(std::list<ScannerSPtr>* scanners) {
std::max<int64_t>(1024,
state()->parallel_scan_min_rows_per_scanner());
scanner_builder.set_max_scanners_count(max_scanners_count);
scanner_builder.set_min_rows_per_scanner(min_rows_per_scanner);
+ // BM25 statistics must not depend on how a tablet is split into
scanners.
+ scanner_builder.set_share_tablet_collection_statistics(_score_runtime
!= nullptr);
// If the session variable is set, force one scanner per segment.
if (state()->query_options().__isset.optimize_index_scan_parallelism &&
state()->query_options().optimize_index_scan_parallelism) {
@@ -714,6 +716,9 @@ Status
OlapScanLocalState::_init_scanners(std::list<ScannerSPtr>* scanners) {
{},
p._limit,
p._olap_scan_node.is_preaggregation,
+ // Each scanner
reads the tablet's
+ // entire read
source.
+ nullptr,
});
RETURN_IF_ERROR(scanner->init(state(), _conjuncts));
scanners->push_back(std::move(scanner));
diff --git a/be/src/exec/scan/olap_scanner.cpp
b/be/src/exec/scan/olap_scanner.cpp
index fc808b1eac8..6ce8a2bcc50 100644
--- a/be/src/exec/scan/olap_scanner.cpp
+++ b/be/src/exec/scan/olap_scanner.cpp
@@ -99,6 +99,7 @@ OlapScanner::OlapScanner(ScanLocalStateBase* parent,
OlapScanner::Params&& param
.collection_statistics {},
.ann_topn_runtime {},
.condition_cache_digest =
parent->get_condition_cache_digest()}),
+
_tablet_collection_statistics(std::move(params.tablet_collection_statistics)),
_initial_file_cache_stats(std::move(params.initial_file_cache_stats)) {
_tablet_reader_params.set_read_source(std::move(params.read_source),
_state->skip_delete_bitmap());
@@ -300,9 +301,16 @@ Status OlapScanner::prepare() {
_state, ReaderType::READER_QUERY,
tablet->file_cache_ttl_expiration_time(),
&_tablet_reader->mutable_stats()->file_cache_stats);
- 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));
+ 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.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));
+ }
}
_has_prepared = true;
diff --git a/be/src/exec/scan/olap_scanner.h b/be/src/exec/scan/olap_scanner.h
index 117468bd66f..5618b865b5e 100644
--- a/be/src/exec/scan/olap_scanner.h
+++ b/be/src/exec/scan/olap_scanner.h
@@ -56,6 +56,7 @@ struct IOContext;
} // namespace io
class Block;
+class TabletCollectionStatistics;
io::IOContext build_score_runtime_collection_io_context(RuntimeState* state,
ReaderType reader_type,
int64_t
expiration_time,
@@ -75,6 +76,9 @@ public:
io::FileCacheStatistics initial_file_cache_stats;
int64_t limit;
bool aggregation;
+ // Set when the tablet is split across several scanners, so that they
all score with
+ // statistics of the entire tablet instead of only the rowsets each
one reads.
+ std::shared_ptr<TabletCollectionStatistics>
tablet_collection_statistics;
};
OlapScanner(ScanLocalStateBase* parent, Params&& params);
@@ -108,6 +112,7 @@ private:
TabletReader::ReaderParams _tablet_reader_params;
std::unique_ptr<TabletReader> _tablet_reader;
+ std::shared_ptr<TabletCollectionStatistics> _tablet_collection_statistics;
public:
std::vector<ColumnId> _return_columns;
diff --git a/be/src/exec/scan/parallel_scanner_builder.cpp
b/be/src/exec/scan/parallel_scanner_builder.cpp
index 825810106b2..ba27c22d34a 100644
--- a/be/src/exec/scan/parallel_scanner_builder.cpp
+++ b/be/src/exec/scan/parallel_scanner_builder.cpp
@@ -25,8 +25,10 @@
#include "common/status.h"
#include "exec/operator/olap_scan_operator.h"
#include "exec/scan/olap_scanner.h"
+#include "exprs/function/search/search_predicate_collector.h"
#include "io/io_common.h"
#include "runtime/query_context.h"
+#include "storage/index/inverted/reader/collection_statistics.h"
#include "storage/rowset/beta_rowset.h"
#include "storage/segment/segment_loader.h"
#include "storage/tablet/base_tablet.h"
@@ -83,6 +85,8 @@ Status
ParallelScannerBuilder::_build_scanners_by_rowid(std::list<ScannerSPtr>&
for (auto&& [tablet, version] : _tablets) {
DCHECK(_all_read_sources.contains(tablet->tablet_id()));
auto& entire_read_source = _all_read_sources[tablet->tablet_id()];
+ auto tablet_collection_statistics =
+ _create_tablet_collection_statistics(entire_read_source);
if (config::is_cloud_mode()) {
// FIXME(plat1ko): Avoid pointer cast
@@ -147,7 +151,8 @@ Status
ParallelScannerBuilder::_build_scanners_by_rowid(std::list<ScannerSPtr>&
.delete_predicates =
entire_read_source.delete_predicates,
.delete_bitmap =
entire_read_source.delete_bitmap},
take_initial_file_cache_stats(&_tablet_preload_file_cache_stats,
-
tablet->tablet_id())));
+
tablet->tablet_id()),
+ tablet_collection_statistics));
partitial_read_source = {};
split = RowSetSplits(reader->clone());
@@ -194,7 +199,8 @@ Status
ParallelScannerBuilder::_build_scanners_by_rowid(std::list<ScannerSPtr>&
.delete_predicates =
entire_read_source.delete_predicates,
.delete_bitmap =
entire_read_source.delete_bitmap},
take_initial_file_cache_stats(&_tablet_preload_file_cache_stats,
-
tablet->tablet_id())));
+
tablet->tablet_id()),
+ tablet_collection_statistics));
}
}
@@ -211,6 +217,8 @@ Status
ParallelScannerBuilder::_build_scanners_by_segment(std::list<ScannerSPtr>
for (auto&& [tablet, version] : _tablets) {
DCHECK(_all_read_sources.contains(tablet->tablet_id()));
auto& entire_read_source = _all_read_sources[tablet->tablet_id()];
+ auto tablet_collection_statistics =
+ _create_tablet_collection_statistics(entire_read_source);
if (config::is_cloud_mode()) {
// FIXME(plat1ko): Avoid pointer cast
@@ -256,7 +264,8 @@ Status
ParallelScannerBuilder::_build_scanners_by_segment(std::list<ScannerSPtr>
.delete_predicates =
entire_read_source.delete_predicates,
.delete_bitmap =
entire_read_source.delete_bitmap},
take_initial_file_cache_stats(&_tablet_preload_file_cache_stats,
-
tablet->tablet_id())));
+ tablet->tablet_id()),
+ tablet_collection_statistics));
// Reset for next scanner
partitial_read_source = TabletReadSource();
@@ -293,7 +302,8 @@ Status
ParallelScannerBuilder::_build_scanners_by_segment(std::list<ScannerSPtr>
.delete_predicates =
entire_read_source.delete_predicates,
.delete_bitmap =
entire_read_source.delete_bitmap},
take_initial_file_cache_stats(&_tablet_preload_file_cache_stats,
-
tablet->tablet_id())));
+
tablet->tablet_id()),
+ tablet_collection_statistics));
}
}
@@ -340,9 +350,25 @@ Status ParallelScannerBuilder::_load() {
return Status::OK();
}
+std::shared_ptr<TabletCollectionStatistics>
+ParallelScannerBuilder::_create_tablet_collection_statistics(
+ const TabletReadSource& entire_read_source) const {
+ if (!_share_tablet_collection_statistics) {
+ return nullptr;
+ }
+ std::vector<RowsetSharedPtr> rowsets;
+ rowsets.reserve(entire_read_source.rs_splits.size());
+ for (const auto& rs_split : entire_read_source.rs_splits) {
+ rowsets.emplace_back(rs_split.rs_reader->rowset());
+ }
+ return std::make_shared<TabletCollectionStatistics>(
+ std::move(rowsets), std::make_unique<SearchPredicateCollector>());
+}
+
std::shared_ptr<OlapScanner> ParallelScannerBuilder::_build_scanner(
BaseTabletSPtr tablet, int64_t version, const
std::vector<OlapScanRange*>& key_ranges,
- TabletReadSource&& read_source, io::FileCacheStatistics&&
initial_file_cache_stats) {
+ TabletReadSource&& read_source, io::FileCacheStatistics&&
initial_file_cache_stats,
+ std::shared_ptr<TabletCollectionStatistics>
tablet_collection_statistics) {
OlapScanner::Params params {
_state,
_scanner_profile.get(),
@@ -353,6 +379,7 @@ std::shared_ptr<OlapScanner>
ParallelScannerBuilder::_build_scanner(
std::move(initial_file_cache_stats),
_limit,
_is_preaggregation,
+ std::move(tablet_collection_statistics),
};
return OlapScanner::create_shared(_parent, std::move(params));
}
diff --git a/be/src/exec/scan/parallel_scanner_builder.h
b/be/src/exec/scan/parallel_scanner_builder.h
index a8507ad8cc5..1f7dc44c448 100644
--- a/be/src/exec/scan/parallel_scanner_builder.h
+++ b/be/src/exec/scan/parallel_scanner_builder.h
@@ -64,6 +64,9 @@ public:
void set_scan_parallelism_by_segment(bool v) {
_scan_parallelism_by_segment = v; }
+ // Scanners of one tablet share BM25 statistics collected over the entire
tablet.
+ void set_share_tablet_collection_statistics(bool v) {
_share_tablet_collection_statistics = v; }
+
const OlapReaderStatistics* builder_stats() const { return
&_builder_stats; }
private:
@@ -74,10 +77,13 @@ private:
// Build scanners so that each segment is handled by its own scanner.
Status _build_scanners_by_segment(std::list<ScannerSPtr>& scanners);
- std::shared_ptr<OlapScanner> _build_scanner(BaseTabletSPtr tablet, int64_t
version,
- const
std::vector<OlapScanRange*>& key_ranges,
- TabletReadSource&& read_source,
- io::FileCacheStatistics&&
initial_file_cache_stats);
+ std::shared_ptr<TabletCollectionStatistics>
_create_tablet_collection_statistics(
+ const TabletReadSource& entire_read_source) const;
+
+ std::shared_ptr<OlapScanner> _build_scanner(
+ BaseTabletSPtr tablet, int64_t version, const
std::vector<OlapScanRange*>& key_ranges,
+ TabletReadSource&& read_source, io::FileCacheStatistics&&
initial_file_cache_stats,
+ std::shared_ptr<TabletCollectionStatistics>
tablet_collection_statistics);
OlapScanLocalState* _parent;
@@ -97,6 +103,8 @@ private:
// Force building one scanner per segment when true.
bool _scan_parallelism_by_segment {false};
+ bool _share_tablet_collection_statistics {false};
+
std::shared_ptr<RuntimeProfile> _scanner_profile;
OlapReaderStatistics _builder_stats;
RuntimeState* _state;
diff --git a/be/src/storage/index/inverted/reader/collection_statistics.cpp
b/be/src/storage/index/inverted/reader/collection_statistics.cpp
index bcc21c1aec7..ebcb2aa174d 100644
--- a/be/src/storage/index/inverted/reader/collection_statistics.cpp
+++ b/be/src/storage/index/inverted/reader/collection_statistics.cpp
@@ -442,6 +442,33 @@ void CollectionStatistics::clear() {
_idf_by_col_term.clear();
}
+void CollectionStatistics::copy_collected_from(const CollectionStatistics&
other) {
+ _total_num_docs = other._total_num_docs;
+ _total_num_tokens = other._total_num_tokens;
+ _term_doc_freqs = other._term_doc_freqs;
+ _avg_dl_by_col.clear();
+ _idf_by_col_term.clear();
+}
+
+Status TabletCollectionStatistics::copy_to(RuntimeState* state,
+ const TabletSchemaSPtr&
tablet_schema,
+ const VExprContextSPtrs&
common_expr_ctxs_push_down,
+ io::IOContext* io_ctx,
+ CollectionStatistics* statistics) {
+ DORIS_CHECK(statistics != nullptr);
+ std::lock_guard lock(_mutex);
+ if (!_collected) {
+ _status = _statistics.collect_full_collection(state, _rowsets,
tablet_schema,
+
common_expr_ctxs_push_down, io_ctx);
+ _collected = true;
+ // Do not hold the rowsets any longer than needed.
+ _rowsets.clear();
+ }
+ RETURN_IF_ERROR(_status);
+ statistics->copy_collected_from(_statistics);
+ return Status::OK();
+}
+
uint64_t CollectionStatistics::get_term_doc_freq_by_col(const std::wstring&
lucene_col_name,
const std::wstring&
term) {
const auto field = _term_doc_freqs.find(lucene_col_name);
diff --git a/be/src/storage/index/inverted/reader/collection_statistics.h
b/be/src/storage/index/inverted/reader/collection_statistics.h
index b05b9470426..75ad55d7034 100644
--- a/be/src/storage/index/inverted/reader/collection_statistics.h
+++ b/be/src/storage/index/inverted/reader/collection_statistics.h
@@ -70,6 +70,10 @@ public:
const std::wstring& term);
MOCK_FUNCTION float get_or_calculate_avg_dl(const std::wstring&
lucene_col_name);
+ // Replaces the collected counters with those of `other`. The derived
idf/avgdl caches are
+ // reset rather than copied, so the copy can be used by another thread
without sharing them.
+ void copy_collected_from(const CollectionStatistics& other);
+
private:
struct SniiScoringSegmentAccumulator {
uint64_t doc_count = 0;
@@ -112,6 +116,34 @@ private:
};
using CollectionStatisticsPtr = std::shared_ptr<CollectionStatistics>;
+// BM25 statistics of a whole tablet, shared by every scanner the tablet is
split into.
+//
+// A parallel scan divides a tablet's rowsets among several scanners. If each
scanner
+// collected statistics over only its own rowsets, rows with identical content
would score
+// differently depending on which scanner read them, and the split itself
depends on the
+// tablet's row count and the number of cores. Collecting once over the entire
read source
+// gives every row of the tablet the same idf and avgdl.
+class TabletCollectionStatistics {
+public:
+ TabletCollectionStatistics(std::vector<RowsetSharedPtr> rowsets,
+ PredicateCollectorPtr search_collector)
+ : _rowsets(std::move(rowsets)),
_statistics(std::move(search_collector)) {}
+
+ // Collects the statistics on the first call, then copies them into
`statistics`.
+ // 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);
+
+private:
+ std::mutex _mutex;
+ bool _collected = false;
+ Status _status;
+ std::vector<RowsetSharedPtr> _rowsets;
+ CollectionStatistics _statistics;
+};
+using TabletCollectionStatisticsPtr =
std::shared_ptr<TabletCollectionStatistics>;
+
// Implementation details of the SNII scoring-segment admission math, surfaced
so
// collection_statistics_test.cpp can exercise them without compiling the .cpp
a
// second time via #include.
diff --git
a/be/test/storage/index/inverted/reader/collection_statistics_test.cpp
b/be/test/storage/index/inverted/reader/collection_statistics_test.cpp
index a4b3c5a5422..bcc4e3109d6 100644
--- a/be/test/storage/index/inverted/reader/collection_statistics_test.cpp
+++ b/be/test/storage/index/inverted/reader/collection_statistics_test.cpp
@@ -1227,6 +1227,38 @@ protected:
std::unique_ptr<TestableCollectionStatistics> stats_;
};
+// Scanners of one tablet each copy the tablet's statistics, so they must
score identically.
+TEST_F(CollectionStatisticsDetailedTest, CopyCollectedFromScoresLikeSource) {
+ stats_->set_total_num_docs(1000);
+ stats_->set_total_num_tokens(L"content", 5000);
+ stats_->set_term_doc_freq(L"content", L"jobs", 37);
+
+ CollectionStatistics copy;
+ copy.copy_collected_from(*stats_);
+
+ EXPECT_FLOAT_EQ(copy.get_or_calculate_idf(L"content", L"jobs"),
+ stats_->get_or_calculate_idf(L"content", L"jobs"));
+ EXPECT_FLOAT_EQ(copy.get_or_calculate_avg_dl(L"content"),
+ stats_->get_or_calculate_avg_dl(L"content"));
+
+ // The copy does not share the source's counters or derived caches.
+ stats_->set_term_doc_freq(L"content", L"jobs", 900);
+ CollectionStatistics recopied;
+ recopied.copy_collected_from(*stats_);
+ EXPECT_NE(recopied.get_or_calculate_idf(L"content", L"jobs"),
+ copy.get_or_calculate_idf(L"content", L"jobs"));
+}
+
+TEST(TabletCollectionStatisticsTest, CollectsOnceAndCopiesToEveryCaller) {
+ TabletCollectionStatistics tablet_statistics({}, nullptr);
+ auto tablet_schema = std::make_shared<TabletSchema>();
+
+ CollectionStatistics first;
+ CollectionStatistics second;
+ EXPECT_TRUE(tablet_statistics.copy_to(nullptr, tablet_schema, {}, nullptr,
&first).ok());
+ EXPECT_TRUE(tablet_statistics.copy_to(nullptr, tablet_schema, {}, nullptr,
&second).ok());
+}
+
// SNII scoring requires only positions and norms; statistics come directly
from the stats block.
TEST(CollectionStatisticsSniiScoringTest,
ResolveUsesPhysicalDocAndTokenCounts) {
auto result = resolve_snii_scoring_segment(3, 7, /*has_positions=*/true,
/*has_norms=*/true);
diff --git
a/regression-test/suites/inverted_index_p0/test_bm25_score_parallel_scan_statistics.groovy
b/regression-test/suites/inverted_index_p0/test_bm25_score_parallel_scan_statistics.groovy
new file mode 100644
index 00000000000..09a27555c3a
--- /dev/null
+++
b/regression-test/suites/inverted_index_p0/test_bm25_score_parallel_scan_statistics.groovy
@@ -0,0 +1,78 @@
+// 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.
+
+// A parallel scan splits one tablet into several scanners. BM25 statistics
must
+// still cover the whole tablet, so rows with identical text get identical
scores
+// no matter which scanner reads them.
+suite("test_bm25_score_parallel_scan_statistics", "p0") {
+ def tableName = "test_bm25_score_parallel_scan_statistics"
+
+ sql "DROP TABLE IF EXISTS ${tableName}"
+ sql """
+ CREATE TABLE ${tableName} (
+ id BIGINT,
+ content TEXT,
+ INDEX idx_content (content) USING INVERTED PROPERTIES("parser" =
"english")
+ ) ENGINE=OLAP
+ UNIQUE KEY(id)
+ DISTRIBUTED BY HASH(id) BUCKETS 1
+ PROPERTIES (
+ "replication_allocation" = "tag.location.default: 1",
+ "enable_unique_key_merge_on_write" = "true",
+ "disable_auto_compaction" = "true"
+ );
+ """
+
+ // Each load is one rowset with a different share of matching rows, so
statistics
+ // collected per scanner would differ between scanners.
+ for (int batch = 0; batch < 8; batch++) {
+ sql """
+ INSERT INTO ${tableName}
+ SELECT ${batch} * 2000 + number,
+ IF(number % ${batch * 3 + 2} = 0,
+ 'we should try these jobs today',
+ 'call the candidate about salary')
+ FROM numbers("number" = "2000");
+ """
+ }
+ sql "sync"
+
+ def scoreQuery = { String vars ->
+ sql """
+ SELECT /*+ SET_VAR(${vars}) */ score() AS s
+ FROM ${tableName}
+ WHERE SEARCH('content:jobs OR content:try')
+ ORDER BY score() DESC
+ LIMIT 100000
+ """
+ }
+
+ def parallel = scoreQuery("enable_parallel_scan=true,
parallel_scan_min_rows_per_scanner=1024, parallel_scan_max_scanners_count=16")
+ def serial = scoreQuery("enable_parallel_scan=false")
+
+ def parallelScores = parallel.collect { it[0] }.unique()
+ def serialScores = serial.collect { it[0] }.unique()
+
+ assertTrue(parallel.size() > 0)
+ assertEquals(serial.size(), parallel.size())
+ assertEquals(1, serialScores.size())
+ assertEquals(1, parallelScores.size(),
+ "identical rows got different BM25 scores across scanners:
${parallelScores}")
+ assertEquals(serialScores[0], parallelScores[0])
+
+ sql "DROP TABLE IF EXISTS ${tableName}"
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
