diff --git a/be/src/exec/operator/olap_scan_operator.cpp b/be/src/exec/operator/olap_scan_operator.cpp index b2cbc3b42131a2..0043a58fd9a405 100644 --- a/be/src/exec/operator/olap_scan_operator.cpp +++ b/be/src/exec/operator/olap_scan_operator.cpp @@ -641,6 +641,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) { @@ -714,6 +716,9 @@ Status OlapScanLocalState::_init_scanners(std::list* 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)); diff --git a/be/src/exec/scan/olap_scanner.cpp b/be/src/exec/scan/olap_scanner.cpp index fc808b1eac82b1..6ce8a2bcc5043f 100644 --- a/be/src/exec/scan/olap_scanner.cpp +++ b/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; diff --git a/be/src/exec/scan/olap_scanner.h b/be/src/exec/scan/olap_scanner.h index 117468bd66f3ee..5618b865b5e3c1 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); @@ -108,6 +112,7 @@ class OlapScanner : public Scanner { TabletReader::ReaderParams _tablet_reader_params; std::unique_ptr _tablet_reader; + std::shared_ptr _tablet_collection_statistics; public: std::vector _return_columns; diff --git a/be/src/exec/scan/parallel_scanner_builder.cpp b/be/src/exec/scan/parallel_scanner_builder.cpp index 825810106b241b..ba27c22d34a089 100644 --- a/be/src/exec/scan/parallel_scanner_builder.cpp +++ b/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& 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& .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& .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 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 .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 .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 +ParallelScannerBuilder::_create_tablet_collection_statistics( + const TabletReadSource& entire_read_source) const { + if (!_share_tablet_collection_statistics) { + return nullptr; + } + std::vector 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( + std::move(rowsets), std::make_unique()); +} + 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 +379,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/index/inverted/reader/collection_statistics.cpp b/be/src/storage/index/inverted/reader/collection_statistics.cpp index bcc21c1aec7234..ebcb2aa174df72 100644 --- a/be/src/storage/index/inverted/reader/collection_statistics.cpp +++ b/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); diff --git a/be/src/storage/index/inverted/reader/collection_statistics.h b/be/src/storage/index/inverted/reader/collection_statistics.h index b05b9470426afc..75ad55d703497d 100644 --- a/be/src/storage/index/inverted/reader/collection_statistics.h +++ b/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; +// 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 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 _rowsets; + CollectionStatistics _statistics; +}; +using TabletCollectionStatisticsPtr = std::shared_ptr; + // 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. diff --git a/be/test/storage/index/inverted/reader/collection_statistics_test.cpp b/be/test/storage/index/inverted/reader/collection_statistics_test.cpp index a4b3c5a54224ef..bcc4e3109d633a 100644 --- a/be/test/storage/index/inverted/reader/collection_statistics_test.cpp +++ b/be/test/storage/index/inverted/reader/collection_statistics_test.cpp @@ -1227,6 +1227,38 @@ class CollectionStatisticsDetailedTest : public ::testing::Test { std::unique_ptr 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(); + + 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); 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}" +}