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 06b990bcd7d branch-4.1: [fix](inverted index) Share BM25 statistics
across parallel scanners of a tablet #69352 (#69418)
06b990bcd7d is described below
commit 06b990bcd7da49e0027e25ad45c289076d39ee0c
Author: guptashivam04 <[email protected]>
AuthorDate: Fri Oct 23 08:11:08 2026 +0530
branch-4.1: [fix](inverted index) Share BM25 statistics across parallel
scanners of a tablet #69352 (#69418)
### What problem does this PR solve?
Issue Number: None
Related PR: #69352
Problem Summary:
Backport of #69352 to branch-4.1.
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.
The reproduction in #69352 (8 rowsets with a different share of matching
rows, forced split with `parallel_scan_min_rows_per_scanner=1024`) gives
the 2,280 identical matching rows **15 distinct scores (1.34 to 5.81)**
on 4.1.4, 4.1.5 and branch-4.1. With `enable_parallel_scan=false` they
all get 3.647686.
Fix (same design as #69352, adapted to branch-4.1):
- `TabletCollectionStatistics` (in
`storage/compaction/collection_statistics.h`) holds the `rs_splits` of a
tablet's entire read source. The first scanner of the tablet to prepare
collects the statistics once with `CollectionStatistics::collect()` over
all of them, 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: on branch-4.1, `CollectionStatistics` lives in
`storage/compaction/`, has no `SearchPredicateCollector` constructor
argument and no `collect_full_collection()`, so the holder keeps the
read source's `rs_splits` and calls `collect()` on them.
`OlapScanner::Params` is initialized positionally on this branch, so the
new field is appended at the end.
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.1 with this change.
- [x] Unit Test:
`CollectionStatisticsDetailedTest.CopyCollectedFromScoresLikeSource`,
`CollectionStatisticsTest.TabletCollectionStatisticsCollectsOnceAndCopiesToEveryCaller`
(not run locally; left to CI)
- [x] Manual test: ran the reproduction from #69352 against branch-4.1
builds on macOS arm64.
| Build | `enable_parallel_scan=true` (split forced) |
`enable_parallel_scan=false` |
|---|---|---|
| branch-4.1, without this change | 15 distinct scores (1.34 to 5.81) |
1 score (3.647686) |
| branch-4.1 `95846ddf` 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 | 28 ++++----
be/src/exec/scan/olap_scanner.cpp | 14 +++-
be/src/exec/scan/olap_scanner.h | 5 ++
be/src/exec/scan/parallel_scanner_builder.cpp | 30 ++++++--
be/src/exec/scan/parallel_scanner_builder.h | 16 +++--
.../storage/compaction/collection_statistics.cpp | 32 +++++++++
be/src/storage/compaction/collection_statistics.h | 33 +++++++++
.../compaction/collection_statistics_test.cpp | 39 ++++++++++
...test_bm25_score_parallel_scan_statistics.groovy | 82 ++++++++++++++++++++++
9 files changed, 255 insertions(+), 24 deletions(-)
diff --git a/be/src/exec/operator/olap_scan_operator.cpp
b/be/src/exec/operator/olap_scan_operator.cpp
index 5777afd1e73..540c89132be 100644
--- a/be/src/exec/operator/olap_scan_operator.cpp
+++ b/be/src/exec/operator/olap_scan_operator.cpp
@@ -564,6 +564,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) {
@@ -626,18 +628,20 @@ Status
OlapScanLocalState::_init_scanners(std::list<ScannerSPtr>* scanners) {
for (auto& split : _read_sources[scan_range_idx].rs_splits) {
split.rs_reader = split.rs_reader->clone();
}
- auto scanner =
- OlapScanner::create_shared(this, OlapScanner::Params {
- state(),
-
_scanner_profile.get(),
- scanner_ranges,
-
_tablets[scan_range_idx].tablet,
- version,
-
_read_sources[scan_range_idx],
- {},
- p._limit,
-
p._olap_scan_node.is_preaggregation,
- });
+ auto scanner = OlapScanner::create_shared(
+ this, OlapScanner::Params {
+ state(),
+ _scanner_profile.get(),
+ scanner_ranges,
+ _tablets[scan_range_idx].tablet,
+ version,
+ _read_sources[scan_range_idx],
+ {},
+ 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 06e6a08c771..7ebf0cba66a 100644
--- a/be/src/exec/scan/olap_scanner.cpp
+++ b/be/src/exec/scan/olap_scanner.cpp
@@ -72,6 +72,7 @@ using ReadSource = TabletReadSource;
OlapScanner::OlapScanner(ScanLocalStateBase* parent, OlapScanner::Params&&
params)
: Scanner(params.state, parent, params.limit, params.profile),
_key_ranges(std::move(params.key_ranges)),
+
_tablet_collection_statistics(std::move(params.tablet_collection_statistics)),
_tablet_reader_params({.tablet = std::move(params.tablet),
.tablet_schema {},
.aggregation = params.aggregation,
@@ -295,9 +296,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..9d7a439a60e 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);
@@ -105,6 +109,7 @@ private:
[[nodiscard]] Status _init_variant_columns();
std::vector<OlapScanRange*> _key_ranges;
+ std::shared_ptr<TabletCollectionStatistics> _tablet_collection_statistics;
TabletReader::ReaderParams _tablet_reader_params;
std::unique_ptr<TabletReader> _tablet_reader;
diff --git a/be/src/exec/scan/parallel_scanner_builder.cpp
b/be/src/exec/scan/parallel_scanner_builder.cpp
index 825810106b2..c6ae90b22ed 100644
--- a/be/src/exec/scan/parallel_scanner_builder.cpp
+++ b/be/src/exec/scan/parallel_scanner_builder.cpp
@@ -27,6 +27,7 @@
#include "exec/scan/olap_scanner.h"
#include "io/io_common.h"
#include "runtime/query_context.h"
+#include "storage/compaction/collection_statistics.h"
#include "storage/rowset/beta_rowset.h"
#include "storage/segment/segment_loader.h"
#include "storage/tablet/base_tablet.h"
@@ -83,6 +84,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 +150,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 +198,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 +216,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 +263,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 +301,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 +349,19 @@ 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;
+ }
+ return
std::make_shared<TabletCollectionStatistics>(entire_read_source.rs_splits);
+}
+
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 +372,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/compaction/collection_statistics.cpp
b/be/src/storage/compaction/collection_statistics.cpp
index 6b59c1cc1ec..413c8a84c66 100644
--- a/be/src/storage/compaction/collection_statistics.cpp
+++ b/be/src/storage/compaction/collection_statistics.cpp
@@ -237,6 +237,38 @@ Status CollectionStatistics::process_segment(const
RowsetSharedPtr& rowset, int3
return Status::OK();
}
+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();
+}
+
+TabletCollectionStatistics::TabletCollectionStatistics(std::vector<RowSetSplits>
rs_splits)
+ : _rs_splits(std::move(rs_splits)) {}
+
+TabletCollectionStatistics::~TabletCollectionStatistics() = default;
+
+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(state, _rs_splits, tablet_schema,
common_expr_ctxs_push_down,
+ io_ctx);
+ _collected = true;
+ // Do not hold the rowsets any longer than needed.
+ _rs_splits.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) {
if (!_term_doc_freqs.contains(lucene_col_name)) {
diff --git a/be/src/storage/compaction/collection_statistics.h
b/be/src/storage/compaction/collection_statistics.h
index e51d5750db6..5658438fb32 100644
--- a/be/src/storage/compaction/collection_statistics.h
+++ b/be/src/storage/compaction/collection_statistics.h
@@ -17,8 +17,10 @@
#pragma once
#include <cstdint>
+#include <mutex>
#include <string>
#include <unordered_map>
+#include <vector>
#include "common/be_mock_util.h"
#include "exprs/vexpr_fwd.h"
@@ -58,6 +60,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:
Status extract_collect_info(RuntimeState* state,
const VExprContextSPtrs&
common_expr_ctxs_push_down,
@@ -86,5 +92,32 @@ 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:
+ explicit TabletCollectionStatistics(std::vector<RowSetSplits> rs_splits);
+ ~TabletCollectionStatistics();
+
+ // 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<RowSetSplits> _rs_splits;
+ CollectionStatistics _statistics;
+};
+using TabletCollectionStatisticsPtr =
std::shared_ptr<TabletCollectionStatistics>;
+
#include "common/compile_check_end.h"
} // namespace doris
diff --git a/be/test/storage/compaction/collection_statistics_test.cpp
b/be/test/storage/compaction/collection_statistics_test.cpp
index ae605c15f2e..279ddf76787 100644
--- a/be/test/storage/compaction/collection_statistics_test.cpp
+++ b/be/test/storage/compaction/collection_statistics_test.cpp
@@ -680,6 +680,45 @@ TEST_F(CollectionStatisticsDetailedTest,
IdfCalculationWithDifferentFrequencies)
EXPECT_GT(rare_idf, 0);
}
+// 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_F(CollectionStatisticsTest,
TabletCollectionStatisticsCollectsOnceAndCopiesToEveryCaller) {
+ auto tablet_schema = create_tablet_schema_with_inverted_index();
+ auto expr_contexts = create_match_expr_contexts();
+ TabletCollectionStatistics tablet_statistics({});
+
+ 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());
+}
+
TEST_F(CollectionStatisticsTest, CollectWithCastWrappedSlotRef) {
auto tablet_schema = create_tablet_schema_with_inverted_index();
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..f6a69dfdb19
--- /dev/null
+++
b/regression-test/suites/inverted_index_p0/test_bm25_score_parallel_scan_statistics.groovy
@@ -0,0 +1,82 @@
+// 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"
+
+ // SEARCH() must be evaluated by the inverted index; fuzzy mode may turn
these off.
+ sql "set enable_common_expr_pushdown = true"
+ sql "set enable_common_expr_pushdown_for_inverted_index = true"
+
+ 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]
