From 6a0b2e20338a87a89cb63df58e7ef828186e596f Mon Sep 17 00:00:00 2001 From: Shivam Gupta Date: Tue, 20 Oct 2026 23:09:30 +0530 Subject: [PATCH 1/2] branch-4.1: [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 | 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 +++- .../compaction/collection_statistics.cpp | 32 ++++++++ .../compaction/collection_statistics.h | 33 ++++++++ .../compaction/collection_statistics_test.cpp | 39 ++++++++++ ...bm25_score_parallel_scan_statistics.groovy | 78 +++++++++++++++++++ 9 files changed, 251 insertions(+), 24 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 5777afd1e732f5..540c89132be15d 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* 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) { @@ -626,18 +628,20 @@ Status OlapScanLocalState::_init_scanners(std::list* 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 06e6a08c771e32..7ebf0cba66a30a 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 117468bd66f3ee..9d7a439a60e15d 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); @@ -105,6 +109,7 @@ class OlapScanner : public Scanner { [[nodiscard]] Status _init_variant_columns(); std::vector _key_ranges; + std::shared_ptr _tablet_collection_statistics; TabletReader::ReaderParams _tablet_reader_params; std::unique_ptr _tablet_reader; diff --git a/be/src/exec/scan/parallel_scanner_builder.cpp b/be/src/exec/scan/parallel_scanner_builder.cpp index 825810106b241b..c6ae90b22ed74f 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& 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& .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& .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 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 .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 .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 +ParallelScannerBuilder::_create_tablet_collection_statistics( + const TabletReadSource& entire_read_source) const { + if (!_share_tablet_collection_statistics) { + return nullptr; + } + return std::make_shared(entire_read_source.rs_splits); +} + 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 +372,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/compaction/collection_statistics.cpp b/be/src/storage/compaction/collection_statistics.cpp index 6b59c1cc1ec48a..413c8a84c6660d 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 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 e51d5750db64d2..5658438fb327a6 100644 --- a/be/src/storage/compaction/collection_statistics.h +++ b/be/src/storage/compaction/collection_statistics.h @@ -17,8 +17,10 @@ #pragma once #include +#include #include #include +#include #include "common/be_mock_util.h" #include "exprs/vexpr_fwd.h" @@ -58,6 +60,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: Status extract_collect_info(RuntimeState* state, const VExprContextSPtrs& common_expr_ctxs_push_down, @@ -86,5 +92,32 @@ 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: + explicit TabletCollectionStatistics(std::vector 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 _rs_splits; + CollectionStatistics _statistics; +}; +using TabletCollectionStatisticsPtr = std::shared_ptr; + #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 ae605c15f2ec6a..279ddf7678725f 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 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}" +} From 98be8b8d56506288da256f14c1379dc83ac6c345 Mon Sep 17 00:00:00 2001 From: Shivam Gupta Date: Thu, 22 Oct 2026 16:02:44 +0530 Subject: [PATCH 2/2] branch-4.1: [fix](inverted index) Enable expression pushdown in the BM25 parallel scan regression Fuzzy mode on branch-4.1 may turn enable_common_expr_pushdown off, and SEARCH() then fails with "SearchExpr should not be executed without inverted index". Enable it explicitly, as the other BM25 suites do. --- .../test_bm25_score_parallel_scan_statistics.groovy | 4 ++++ 1 file changed, 4 insertions(+) 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 index 09a27555c3a9b7..f6a69dfdb19fa7 100644 --- 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 @@ -51,6 +51,10 @@ suite("test_bm25_score_parallel_scan_statistics", "p0") { } 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