From 323f3e00c0d8655cd018c92b365867d960fcd44d Mon Sep 17 00:00:00 2001 From: Shivam Gupta Date: Thu, 22 Oct 2026 09:04:08 +0530 Subject: [PATCH] branch-4.2: [fix](inverted index) Share BM25 statistics across parallel scanners of a tablet #69352 A parallel scan splits one tablet into several scanners, and each scanner collected BM25 statistics (doc count, avgdl, term doc freqs) over only the rowsets in its own split. Rows with identical text therefore got different scores depending on which scanner read them, and because the split depends on the tablet's row count and the core count, a row's score and rank moved whenever unrelated data was loaded. Collect the statistics once per tablet over its entire read source and copy them into each scanner, so every row of a tablet is scored against the same idf and avgdl. Each scanner keeps its own copy, so the per-scanner idf/avgdl caches are not shared between threads. --- 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 +++++++ .../inverted/reader/collection_statistics.h | 32 ++++++++ .../reader/collection_statistics_test.cpp | 32 ++++++++ ...bm25_score_parallel_scan_statistics.groovy | 78 +++++++++++++++++++ 9 files changed, 234 insertions(+), 12 deletions(-) create mode 100644 regression-test/suites/inverted_index_p0/test_bm25_score_parallel_scan_statistics.groovy diff --git a/be/src/exec/operator/olap_scan_operator.cpp b/be/src/exec/operator/olap_scan_operator.cpp index b2cbc3b42131a2..0043a58fd9a405 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* scanners) { std::max(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* 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 fc808b1eac82b1..6ce8a2bcc5043f 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 117468bd66f3ee..5618b865b5e3c1 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 @@ class OlapScanner : public Scanner { 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 tablet_collection_statistics; }; OlapScanner(ScanLocalStateBase* parent, Params&& params); @@ -108,6 +112,7 @@ class OlapScanner : public Scanner { TabletReader::ReaderParams _tablet_reader_params; std::unique_ptr _tablet_reader; + std::shared_ptr _tablet_collection_statistics; public: std::vector _return_columns; diff --git a/be/src/exec/scan/parallel_scanner_builder.cpp b/be/src/exec/scan/parallel_scanner_builder.cpp index 825810106b241b..ba27c22d34a089 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& 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& .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& .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 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 .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 .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 +ParallelScannerBuilder::_create_tablet_collection_statistics( + const TabletReadSource& entire_read_source) const { + if (!_share_tablet_collection_statistics) { + return nullptr; + } + std::vector 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( + std::move(rowsets), std::make_unique()); +} + std::shared_ptr ParallelScannerBuilder::_build_scanner( BaseTabletSPtr tablet, int64_t version, const std::vector& key_ranges, - TabletReadSource&& read_source, io::FileCacheStatistics&& initial_file_cache_stats) { + TabletReadSource&& read_source, io::FileCacheStatistics&& initial_file_cache_stats, + std::shared_ptr tablet_collection_statistics) { OlapScanner::Params params { _state, _scanner_profile.get(), @@ -353,6 +379,7 @@ std::shared_ptr 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 a8507ad8cc5604..1f7dc44c44894f 100644 --- a/be/src/exec/scan/parallel_scanner_builder.h +++ b/be/src/exec/scan/parallel_scanner_builder.h @@ -64,6 +64,9 @@ class ParallelScannerBuilder { 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 @@ class ParallelScannerBuilder { // Build scanners so that each segment is handled by its own scanner. Status _build_scanners_by_segment(std::list& scanners); - std::shared_ptr _build_scanner(BaseTabletSPtr tablet, int64_t version, - const std::vector& key_ranges, - TabletReadSource&& read_source, - io::FileCacheStatistics&& initial_file_cache_stats); + std::shared_ptr _create_tablet_collection_statistics( + const TabletReadSource& entire_read_source) const; + + std::shared_ptr _build_scanner( + BaseTabletSPtr tablet, int64_t version, const std::vector& key_ranges, + TabletReadSource&& read_source, io::FileCacheStatistics&& initial_file_cache_stats, + std::shared_ptr tablet_collection_statistics); OlapScanLocalState* _parent; @@ -97,6 +103,8 @@ class ParallelScannerBuilder { // Force building one scanner per segment when true. bool _scan_parallelism_by_segment {false}; + bool _share_tablet_collection_statistics {false}; + std::shared_ptr _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 bcc21c1aec7234..ebcb2aa174df72 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 b05b9470426afc..75ad55d703497d 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 @@ class CollectionStatistics { 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 @@ class CollectionStatistics { }; using CollectionStatisticsPtr = std::shared_ptr; +// 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 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 _rowsets; + CollectionStatistics _statistics; +}; +using TabletCollectionStatisticsPtr = std::shared_ptr; + // 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 a4b3c5a54224ef..bcc4e3109d633a 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 @@ class CollectionStatisticsDetailedTest : public ::testing::Test { std::unique_ptr 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(); + + 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 00000000000000..09a27555c3a9b7 --- /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}" +}