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

be/src/exec/operator/olap_scan_operator.cpp

@@ -641,6 +641,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) {
@@ -714,6 +716,9 @@ Status OlapScanLocalState::_init_scanners(std::list<ScannerSPtr>* 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));

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;

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);
@@ -108,6 +112,7 @@ class OlapScanner : public Scanner {
 
     TabletReader::ReaderParams _tablet_reader_params;
     std::unique_ptr<TabletReader> _tablet_reader;
+    std::shared_ptr<TabletCollectionStatistics> _tablet_collection_statistics;
 
 public:
     std::vector<ColumnId> _return_columns;

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<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 +151,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 +199,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 +217,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 +264,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 +302,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 +350,25 @@ 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,
-        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 +379,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/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);

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

be/test/storage/index/inverted/reader/collection_statistics_test.cpp

@@ -1227,6 +1227,38 @@ class CollectionStatisticsDetailedTest : public ::testing::Test {
     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);