This is an automated email from the ASF dual-hosted git repository.
airborne12 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new d30058a8033 [fix](inverted index) Share BM25 statistics across
parallel scanners of a tablet (#69352)
d30058a8033 is described below
commit d30058a80337e56f0f4930d7a75c0df50aa39ffc
Author: guptashivam04 <[email protected]>
AuthorDate: Thu Oct 22 08:14:49 2026 +0530
[fix](inverted index) Share BM25 statistics across parallel scanners of a
tablet (#69352)
### What problem does this PR solve?
Issue Number: None
Related PR: #69313 (BM25 statistics after compaction), #66979 (deleted
documents in score TopN)
Problem Summary:
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.
The split depends on the tablet's total row count,
`parallel_scan_max_scanners_count` (defaulting to the core count) and
`parallel_scan_min_rows_per_scanner`. A row's score and its rank in
`ORDER BY score()` therefore change when unrelated rows are loaded into
the tablet, and differ between BEs with different core counts.
Pagination over a score-ordered search returns repeated and missing
rows.
Reproduction (one bucket, one replica, no concurrent writes):
```sql
CREATE TABLE t (
id BIGINT,
content TEXT,
INDEX idx_content (content) USING INVERTED PROPERTIES("parser" =
"english")
) UNIQUE KEY(id) DISTRIBUTED BY HASH(id) BUCKETS 1
PROPERTIES ("replication_num" = "1", "enable_unique_key_merge_on_write" =
"true",
"disable_auto_compaction" = "true");
-- 8 loads (8 rowsets), each with a different share of matching rows
INSERT INTO t SELECT 0 * 2000 + number, IF(number % 2 = 0, 'we should try
these jobs today', 'call the candidate about salary') FROM numbers("number" =
"2000");
INSERT INTO t SELECT 1 * 2000 + number, IF(number % 5 = 0, 'we should try
these jobs today', 'call the candidate about salary') FROM numbers("number" =
"2000");
-- ... batches 2..7 with modulus 8, 11, 14, 17, 20, 23
SELECT /*+ SET_VAR(enable_parallel_scan=true,
parallel_scan_min_rows_per_scanner=1024, parallel_scan_max_scanners_count=16) */
score() AS s, count(*)
FROM (SELECT score() FROM t WHERE SEARCH('content:jobs OR content:try')
ORDER BY score() DESC LIMIT 100000) x
GROUP BY s;
```
On 4.1.4, the 2,280 matching rows all have the same text but get **15
distinct scores (1.34 to 5.81)**. With `enable_parallel_scan=false` they
all get 3.647686. On a production-sized table, the default thresholds
give the same behavior.
Fix:
- `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 per-segment build
paths pass it to their scanners.
- The non-parallel path is unchanged. Each of its scanners already
receives the tablet's full read source.
Statistics remain per tablet. Making them consistent across tablets and
BEs (as with Elasticsearch's `dfs_query_then_fetch`) is out of scope.
---
be/src/exec/operator/olap_scan_operator.cpp | 4 ++
be/src/exec/scan/olap_scanner.cpp | 14 +++-
be/src/exec/scan/olap_scanner.h | 5 ++
be/src/exec/scan/parallel_scanner_builder.cpp | 34 ++++++++--
be/src/exec/scan/parallel_scanner_builder.h | 18 +++--
.../inverted/reader/collection_statistics.cpp | 27 ++++++++
.../index/inverted/reader/collection_statistics.h | 32 +++++++++
be/test/exec/scan/scanner_late_arrival_rf_test.cpp | 2 +-
.../inverted/reader/collection_statistics_test.cpp | 32 +++++++++
...test_bm25_score_parallel_scan_statistics.groovy | 78 ++++++++++++++++++++++
10 files changed, 233 insertions(+), 13 deletions(-)
diff --git a/be/src/exec/operator/olap_scan_operator.cpp
b/be/src/exec/operator/olap_scan_operator.cpp
index a287d1d479b..a61fa641819 100644
--- a/be/src/exec/operator/olap_scan_operator.cpp
+++ b/be/src/exec/operator/olap_scan_operator.cpp
@@ -833,6 +833,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) {
@@ -934,6 +936,8 @@ Status
OlapScanLocalState::_init_scanners(std::list<ScannerSPtr>* scanners) {
palo_scan_range.__isset.end_tso
?
std::make_optional(palo_scan_range.end_tso)
: std::nullopt,
+ // 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 d99a641c717..eb0d1604f60 100644
--- a/be/src/exec/scan/olap_scanner.cpp
+++ b/be/src/exec/scan/olap_scanner.cpp
@@ -108,6 +108,7 @@ OlapScanner::OlapScanner(ScanLocalStateBase* parent,
OlapScanner::Params&& param
_end_tso(params.end_tso),
_bucket_seq(params.bucket_seq),
_bucket_num(params.bucket_num),
+
_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());
@@ -286,9 +287,16 @@ Status OlapScanner::_prepare_impl() {
_state, _tablet_reader_params.reader_type,
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 0d46ce31996..4a4f45e0a88 100644
--- a/be/src/exec/scan/olap_scanner.h
+++ b/be/src/exec/scan/olap_scanner.h
@@ -59,6 +59,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,
@@ -84,6 +85,9 @@ public:
int32_t bucket_num = 0;
std::optional<int64_t> start_tso;
std::optional<int64_t> end_tso;
+ // 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);
@@ -127,6 +131,7 @@ private:
std::optional<int64_t> _end_tso;
int32_t _bucket_seq;
int32_t _bucket_num;
+ std::shared_ptr<TabletCollectionStatistics> _tablet_collection_statistics;
public:
io::FileCacheStatistics _initial_file_cache_stats;
diff --git a/be/src/exec/scan/parallel_scanner_builder.cpp
b/be/src/exec/scan/parallel_scanner_builder.cpp
index 119caad6c5c..ec624a92749 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"
@@ -89,6 +91,8 @@ Status
ParallelScannerBuilder::_build_scanners_by_rowid(std::list<ScannerSPtr>&
const auto& scan_range = *_scan_ranges[tablet_idx];
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
@@ -153,7 +157,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());
@@ -200,7 +205,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));
}
}
@@ -223,6 +229,8 @@ Status
ParallelScannerBuilder::_build_scanners_by_per_segment(std::list<ScannerS
const auto& scan_range = *_scan_ranges[tablet_idx];
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
@@ -257,7 +265,8 @@ Status
ParallelScannerBuilder::_build_scanners_by_per_segment(std::list<ScannerS
.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));
}
}
}
@@ -307,10 +316,26 @@ 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,
const TPaloScanRange& scan_range, TabletReadSource&& read_source,
- io::FileCacheStatistics&& initial_file_cache_stats) {
+ io::FileCacheStatistics&& initial_file_cache_stats,
+ std::shared_ptr<TabletCollectionStatistics>
tablet_collection_statistics) {
OlapScanner::Params params {
.state = _state,
.profile = _scanner_profile.get(),
@@ -327,6 +352,7 @@ std::shared_ptr<OlapScanner>
ParallelScannerBuilder::_build_scanner(
.bucket_num = scan_range.bucket_num,
.start_tso = std::nullopt,
.end_tso = std::nullopt,
+ .tablet_collection_statistics =
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 7a8a704c488..525ae78f39b 100644
--- a/be/src/exec/scan/parallel_scanner_builder.h
+++ b/be/src/exec/scan/parallel_scanner_builder.h
@@ -74,6 +74,9 @@ public:
void set_scan_parallelism_by_per_segment(bool v) {
_scan_parallelism_by_per_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:
@@ -84,11 +87,14 @@ private:
// Build scanners so that each segment is handled by its own scanner.
Status _build_scanners_by_per_segment(std::list<ScannerSPtr>& scanners);
- std::shared_ptr<OlapScanner> _build_scanner(BaseTabletSPtr tablet, int64_t
version,
- const
std::vector<OlapScanRange*>& key_ranges,
- const TPaloScanRange&
scan_range,
- 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,
+ const TPaloScanRange& scan_range, TabletReadSource&& read_source,
+ io::FileCacheStatistics&& initial_file_cache_stats,
+ std::shared_ptr<TabletCollectionStatistics>
tablet_collection_statistics);
OlapScanLocalState* _parent;
@@ -108,6 +114,8 @@ private:
// Force building one scanner per segment when true.
bool _scan_parallelism_by_per_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 07e3fa866d6..1e691caaf15 100644
--- a/be/src/storage/index/inverted/reader/collection_statistics.cpp
+++ b/be/src/storage/index/inverted/reader/collection_statistics.cpp
@@ -420,6 +420,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 27b61b6ad75..0da3daa2dd3 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/exec/scan/scanner_late_arrival_rf_test.cpp
b/be/test/exec/scan/scanner_late_arrival_rf_test.cpp
index 5989ce1e7f0..53197920dc7 100644
--- a/be/test/exec/scan/scanner_late_arrival_rf_test.cpp
+++ b/be/test/exec/scan/scanner_late_arrival_rf_test.cpp
@@ -597,7 +597,7 @@ TEST_F(ScannerLateArrivalRfTest,
parallel_scanner_factory_preserves_bucket_ident
tablet_key_ranges, state, -1, true, true);
auto scanner = builder._build_scanner(tablet, 1, {}, *scan_ranges.front(),
TabletReadSource {},
- io::FileCacheStatistics {});
+ io::FileCacheStatistics {}, nullptr);
EXPECT_EQ(scanner->_bucket_seq, bucket_seq);
EXPECT_EQ(scanner->_bucket_num, bucket_num);
EXPECT_TRUE(scanner->is_pruned_by_runtime_filter());
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 c5c454bd0b3..02e04d046db 100644
--- a/be/test/storage/index/inverted/reader/collection_statistics_test.cpp
+++ b/be/test/storage/index/inverted/reader/collection_statistics_test.cpp
@@ -1218,6 +1218,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]
