[fix](bm25) Collect collection statistics over the whole collection, not one scanner's slice

be/src/exec/operator/olap_scan_operator.cpp

@@ -46,13 +46,30 @@
 #include "runtime/runtime_profile.h"
 #include "runtime/runtime_state.h"
 #include "service/backend_options.h"
+#include "storage/compaction/collection_statistics.h"
 #include "storage/index/ann/ann_topn_runtime.h"
 #include "storage/storage_engine.h"
 #include "storage/tablet/tablet.h"
 #include "storage/tablet/tablet_manager.h"
 #include "util/to_string.h"
 
 namespace doris {
+namespace {
+std::shared_ptr<CollectionStatisticsBuildState> create_collection_statistics_build_state(
+        const TabletReadSource& read_source) {
+    // Capture the tablet read-source scope before parallel scanners slice it into segments/rows:
+    // BM25 idf must be computed over the whole collection, not over one scanner's share of it.
+    std::vector<RowsetSharedPtr> rowsets;
+    rowsets.reserve(read_source.rs_splits.size());
+    for (const auto& split : read_source.rs_splits) {
+        DCHECK(split.rs_reader != nullptr);
+        auto rowset = split.rs_reader->rowset();
+        DCHECK(rowset != nullptr);
+        rowsets.emplace_back(std::move(rowset));
+    }
+    return std::make_shared<CollectionStatisticsBuildState>(std::move(rowsets));
+}
+} // namespace
 
 Status OlapScanLocalState::init(RuntimeState* state, LocalStateInfo& info) {
     const TOlapScanNode& olap_scan_node = _parent->cast<OlapScanOperatorX>()._olap_scan_node;
@@ -699,6 +716,18 @@ Status OlapScanLocalState::_init_scanners(std::list<ScannerSPtr>* scanners) {
             p._olap_scan_node.__isset.read_row_binlog && p._olap_scan_node.read_row_binlog;
     bool has_tso_predicate = _scan_ranges[0]->__isset.start_tso || _scan_ranges[0]->__isset.end_tso;
 
+    CollectionStatisticsBuildStateMap collection_statistics_build_states;
+    if (_score_runtime) {
+        DCHECK_EQ(_tablets.size(), _read_sources.size());
+        collection_statistics_build_states.reserve(_read_sources.size());
+        for (size_t i = 0; i < _read_sources.size(); ++i) {
+            const auto insert_result = collection_statistics_build_states.emplace(
+                    _tablets[i].tablet->tablet_id(),
+                    create_collection_statistics_build_state(_read_sources[i]));
+            DCHECK(insert_result.second);
+        }
+    }
+
     // The flag of preagg's meaning is whether return pre agg data(or partial agg data)
     // PreAgg ON: The storage layer returns partially aggregated data without additional processing. (Fast data reading)
     // for example, if a table is select userid,count(*) from base table.
@@ -731,7 +760,8 @@ Status OlapScanLocalState::_init_scanners(std::list<ScannerSPtr>* scanners) {
 
         ParallelScannerBuilder scanner_builder(this, _tablets, _read_sources, _scanner_profile,
                                                key_ranges, state(), p._limit, true,
-                                               p._olap_scan_node.is_preaggregation);
+                                               p._olap_scan_node.is_preaggregation,
+                                               collection_statistics_build_states);
 
         int max_scanners_count = state()->parallel_scan_max_scanners_count();
 
@@ -826,6 +856,9 @@ Status OlapScanLocalState::_init_scanners(std::list<ScannerSPtr>* scanners) {
                                   palo_scan_range.__isset.end_tso
                                           ? std::make_optional(palo_scan_range.end_tso)
                                           : std::nullopt,
+                                  _score_runtime ? collection_statistics_build_states.at(
+                                                           _tablets[scan_range_idx].tablet->tablet_id())
+                                                 : nullptr,
                           });
             RETURN_IF_ERROR(scanner->init(state(), _conjuncts));
             scanners->push_back(std::move(scanner));

be/src/exec/scan/olap_scanner.cpp

@@ -106,6 +106,7 @@ OlapScanner::OlapScanner(ScanLocalStateBase* parent, OlapScanner::Params&& param
                                  .binlog_scan_type = params.binlog_scan_type}),
           _start_tso(params.start_tso),
           _end_tso(params.end_tso),
+          _collection_statistics_build_state(std::move(params.collection_statistics_build_state)),
           _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());
@@ -292,15 +293,23 @@ Status OlapScanner::_prepare_impl() {
 
     if (_tablet_reader_params.score_runtime) {
         SCOPED_TIMER(local_state->_statistics_collect_timer);
-        _tablet_reader_params.collection_statistics = std::make_shared<CollectionStatistics>();
+        DCHECK(_collection_statistics_build_state != nullptr);
 
         auto io_ctx = build_score_runtime_collection_io_context(
                 _state, _tablet_reader_params.reader_type, tablet->ttl_seconds(),
                 &_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));
+        // Collect over the tablet's whole read source, shared with the other scanners of this
+        // tablet. Collecting from _tablet_reader_params.rs_splits instead would give each parallel
+        // scanner a different idf, so score() would depend on how the read source was split.
+        RETURN_IF_ERROR(_collection_statistics_build_state->get_or_build(
+                [&](CollectionStatistics* statistics,
+                    const std::vector<RowsetSharedPtr>& full_collection_rowsets) {
+                    return statistics->collect_full_collection(
+                            _state, full_collection_rowsets, _tablet_reader_params.tablet_schema,
+                            _tablet_reader_params.common_expr_ctxs_push_down, &io_ctx);
+                },
+                &_tablet_reader_params.collection_statistics));
     }
 
     _has_prepared = true;

be/src/exec/scan/olap_scanner.h

@@ -49,6 +49,7 @@ class RuntimeProfile;
 class RuntimeState;
 class TPaloScanRange;
 class ScanLocalStateBase;
+class CollectionStatisticsBuildState;
 struct FilterPredicates;
 #ifndef NDEBUG
 struct OlapReaderStatistics;
@@ -83,6 +84,8 @@ class OlapScanner : public Scanner {
         TBinlogScanType::type binlog_scan_type = TBinlogScanType::NONE;
         std::optional<int64_t> start_tso;
         std::optional<int64_t> end_tso;
+        // Shared per tablet so the collection-wide statistics are built once, not once per scanner.
+        std::shared_ptr<CollectionStatisticsBuildState> collection_statistics_build_state;
     };
 
     OlapScanner(ScanLocalStateBase* parent, Params&& params);
@@ -124,6 +127,7 @@ class OlapScanner : public Scanner {
     std::unique_ptr<TabletReader> _tablet_reader;
     std::optional<int64_t> _start_tso;
     std::optional<int64_t> _end_tso;
+    std::shared_ptr<CollectionStatisticsBuildState> _collection_statistics_build_state;
 
 public:
     std::vector<ColumnId> _return_columns;

be/src/exec/scan/parallel_scanner_builder.cpp

@@ -296,6 +296,10 @@ Status ParallelScannerBuilder::_load() {
 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) {
+    std::shared_ptr<CollectionStatisticsBuildState> collection_statistics_build_state;
+    if (!_collection_statistics_build_states.empty()) {
+        collection_statistics_build_state = _collection_statistics_build_states.at(tablet->tablet_id());
+    }
     OlapScanner::Params params {
             .state = _state,
             .profile = _scanner_profile.get(),
@@ -310,6 +314,7 @@ std::shared_ptr<OlapScanner> ParallelScannerBuilder::_build_scanner(
             .binlog_scan_type = TBinlogScanType::NONE,
             .start_tso = std::nullopt,
             .end_tso = std::nullopt,
+            .collection_statistics_build_state = std::move(collection_statistics_build_state),
     };
     return OlapScanner::create_shared(_parent, std::move(params));
 }

be/src/exec/scan/parallel_scanner_builder.h

@@ -38,14 +38,19 @@ class Scanner;
 
 using ScannerSPtr = std::shared_ptr<Scanner>;
 
+class CollectionStatisticsBuildState;
+using CollectionStatisticsBuildStateMap =
+        std::unordered_map<int64_t, std::shared_ptr<CollectionStatisticsBuildState>>;
+
 class ParallelScannerBuilder {
 public:
     ParallelScannerBuilder(OlapScanLocalState* parent,
                            const std::vector<TabletWithVersion>& tablets,
                            std::vector<TabletReadSource>& read_sources,
                            const std::shared_ptr<RuntimeProfile>& profile,
                            const std::vector<OlapScanRange*>& key_ranges, RuntimeState* state,
-                           int64_t limit, bool is_dup_mow_key, bool is_preaggregation)
+                           int64_t limit, bool is_dup_mow_key, bool is_preaggregation,
+                           CollectionStatisticsBuildStateMap collection_statistics_build_states = {})
             : _parent(parent),
               _scanner_profile(profile),
               _state(state),
@@ -54,7 +59,8 @@ class ParallelScannerBuilder {
               _is_preaggregation(is_preaggregation),
               _tablets(tablets.cbegin(), tablets.cend()),
               _key_ranges(key_ranges.cbegin(), key_ranges.cend()),
-              _read_sources(read_sources) {}
+              _read_sources(read_sources),
+              _collection_statistics_build_states(std::move(collection_statistics_build_states)) {}
 
     Status build_scanners(std::list<ScannerSPtr>& scanners);
 
@@ -113,6 +119,7 @@ class ParallelScannerBuilder {
     std::vector<OlapScanRange*> _key_ranges;
     std::unordered_map<int64_t, TabletReadSource> _all_read_sources;
     std::vector<TabletReadSource>& _read_sources;
+    CollectionStatisticsBuildStateMap _collection_statistics_build_states;
 };
 
 } // namespace doris