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..f6a69dfdb19fa7 --- /dev/null +++ b/regression-test/suites/inverted_index_p0/test_bm25_score_parallel_scan_statistics.groovy @@ -0,0 +1,82 @@ +// 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" + + // 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 + 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}" +}