branch-4.1: [fix](inverted index) Share BM25 statistics across parallel scanners of a tablet #69352

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));
         }

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;

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<TabletCollectionStatistics> tablet_collection_statistics;
     };
 
     OlapScanner(ScanLocalStateBase* parent, Params&& params);
@@ -105,6 +109,7 @@ class OlapScanner : public Scanner {
     [[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;

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));
 }

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<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 @@ 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<RuntimeProfile> _scanner_profile;
     OlapReaderStatistics _builder_stats;
     RuntimeState* _state;

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)) {

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 @@ 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<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

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();