diff --git a/be/src/exec/operator/olap_scan_operator.cpp b/be/src/exec/operator/olap_scan_operator.cpp index 50d41d3dcc979a..d04d37f371dad9 100644 --- a/be/src/exec/operator/olap_scan_operator.cpp +++ b/be/src/exec/operator/olap_scan_operator.cpp @@ -46,6 +46,7 @@ #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" @@ -53,6 +54,22 @@ #include "util/to_string.h" namespace doris { +namespace { +std::shared_ptr 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 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(std::move(rowsets)); +} +} // namespace Status OlapScanLocalState::init(RuntimeState* state, LocalStateInfo& info) { const TOlapScanNode& olap_scan_node = _parent->cast()._olap_scan_node; @@ -699,6 +716,18 @@ Status OlapScanLocalState::_init_scanners(std::list* 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* 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* 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)); diff --git a/be/src/exec/scan/olap_scanner.cpp b/be/src/exec/scan/olap_scanner.cpp index 834ee012737ecd..7c265a793637f2 100644 --- a/be/src/exec/scan/olap_scanner.cpp +++ b/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(); + 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& 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; diff --git a/be/src/exec/scan/olap_scanner.h b/be/src/exec/scan/olap_scanner.h index 308a619475674f..38683b69a631d6 100644 --- a/be/src/exec/scan/olap_scanner.h +++ b/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 start_tso; std::optional end_tso; + // Shared per tablet so the collection-wide statistics are built once, not once per scanner. + std::shared_ptr collection_statistics_build_state; }; OlapScanner(ScanLocalStateBase* parent, Params&& params); @@ -124,6 +127,7 @@ class OlapScanner : public Scanner { std::unique_ptr _tablet_reader; std::optional _start_tso; std::optional _end_tso; + std::shared_ptr _collection_statistics_build_state; 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 9ae9a8bb40342d..65f41df198e0cd 100644 --- a/be/src/exec/scan/parallel_scanner_builder.cpp +++ b/be/src/exec/scan/parallel_scanner_builder.cpp @@ -296,6 +296,10 @@ Status ParallelScannerBuilder::_load() { 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) { + std::shared_ptr 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 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)); } diff --git a/be/src/exec/scan/parallel_scanner_builder.h b/be/src/exec/scan/parallel_scanner_builder.h index 82b63b07824c3a..d8c8732fd982f2 100644 --- a/be/src/exec/scan/parallel_scanner_builder.h +++ b/be/src/exec/scan/parallel_scanner_builder.h @@ -38,6 +38,10 @@ class Scanner; using ScannerSPtr = std::shared_ptr; +class CollectionStatisticsBuildState; +using CollectionStatisticsBuildStateMap = + std::unordered_map>; + class ParallelScannerBuilder { public: ParallelScannerBuilder(OlapScanLocalState* parent, @@ -45,7 +49,8 @@ class ParallelScannerBuilder { std::vector& read_sources, const std::shared_ptr& profile, const std::vector& 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& scanners); @@ -113,6 +119,7 @@ class ParallelScannerBuilder { std::vector _key_ranges; std::unordered_map _all_read_sources; std::vector& _read_sources; + CollectionStatisticsBuildStateMap _collection_statistics_build_states; }; } // namespace doris diff --git a/be/src/storage/compaction/collection_statistics.cpp b/be/src/storage/compaction/collection_statistics.cpp index 9afbe5f0944d43..151f1569539009 100644 --- a/be/src/storage/compaction/collection_statistics.cpp +++ b/be/src/storage/compaction/collection_statistics.cpp @@ -17,8 +17,10 @@ #include "storage/compaction/collection_statistics.h" +#include #include #include +#include #include "common/exception.h" #include "exprs/vexpr.h" @@ -36,6 +38,27 @@ namespace doris { +Status CollectionStatistics::_collect_rowset(const RowsetSharedPtr& rowset, + const TabletSchemaSPtr& tablet_schema, + const CollectInfoMap& collect_infos, + io::IOContext* io_ctx) { + const auto num_segments = rowset->num_segments(); + for (int32_t seg_id = 0; seg_id < num_segments; ++seg_id) { + auto status = process_segment(rowset, seg_id, tablet_schema.get(), collect_infos, io_ctx); + if (!status.ok()) { + // A missing or bypassed index means this segment contributes nothing; the collection is + // still usable, so log and keep going rather than failing the query. + if (status.code() == ErrorCode::INVERTED_INDEX_FILE_NOT_FOUND || + status.code() == ErrorCode::INVERTED_INDEX_BYPASS) { + LOG(ERROR) << "Index statistics collection failed: " << status.to_string(); + } else { + return status; + } + } + } + return Status::OK(); +} + Status CollectionStatistics::collect(RuntimeState* state, const std::vector& rs_splits, const TabletSchemaSPtr& tablet_schema, @@ -50,21 +73,8 @@ Status CollectionStatistics::collect(RuntimeState* state, } for (const auto& rs_split : rs_splits) { - const auto& rs_reader = rs_split.rs_reader; - auto rowset = rs_reader->rowset(); - auto num_segments = rowset->num_segments(); - for (int32_t seg_id = 0; seg_id < num_segments; ++seg_id) { - auto status = - process_segment(rowset, seg_id, tablet_schema.get(), collect_infos, io_ctx); - if (!status.ok()) { - if (status.code() == ErrorCode::INVERTED_INDEX_FILE_NOT_FOUND || - status.code() == ErrorCode::INVERTED_INDEX_BYPASS) { - LOG(ERROR) << "Index statistics collection failed: " << status.to_string(); - } else { - return status; - } - } - } + RETURN_IF_ERROR(_collect_rowset(rs_split.rs_reader->rowset(), tablet_schema, collect_infos, + io_ctx)); } // Build a single-line log with query_id, tablet_ids, and per-field term statistics @@ -219,41 +229,153 @@ Status CollectionStatistics::process_segment(const RowsetSharedPtr& rowset, int3 return Status::OK(); } +Status CollectionStatistics::collect_full_collection( + RuntimeState* state, const std::vector& rowsets, + const TabletSchemaSPtr& tablet_schema, + const VExprContextSPtrs& common_expr_ctxs_push_down, io::IOContext* io_ctx) { + CollectInfoMap collect_infos; + RETURN_IF_ERROR( + extract_collect_info(state, common_expr_ctxs_push_down, tablet_schema, &collect_infos)); + if (collect_infos.empty()) { + LOG(WARNING) << "Index statistics collection: no collect info extracted."; + return Status::OK(); + } + + for (const auto& rowset : rowsets) { + RETURN_IF_ERROR(_collect_rowset(rowset, tablet_schema, collect_infos, io_ctx)); + } + return Status::OK(); +} + +CollectionStatisticsPtr CollectionStatistics::clone_for_scanner() const { + DCHECK(_shared_base != nullptr); + auto clone = std::make_shared(); + clone->_shared_base = _shared_base; + // Derived BM25 caches stay empty so every scanner owns its lazy mutations. + return clone; +} + +void CollectionStatistics::freeze_for_scanners() { + DCHECK(_shared_base == nullptr); + auto base = std::make_shared(); + base->total_num_docs = _total_num_docs; + base->total_num_tokens = std::move(_total_num_tokens); + base->term_doc_freqs = std::move(_term_doc_freqs); + _shared_base = std::move(base); +} + +Status CollectionStatisticsBuildState::get_or_build(const Builder& builder, + CollectionStatisticsPtr* statistics) { + DCHECK(static_cast(builder)); + DCHECK(statistics != nullptr); + *statistics = nullptr; + + bool build = false; + CollectionStatisticsPtr ready_prototype; + { + std::unique_lock lock(_mutex); + if (_state == State::EMPTY) { + _state = State::BUILDING; + build = true; + } else if (_state == State::BUILDING) { + _condition.wait(lock, + [this] { return _state == State::READY || _state == State::FAILED; }); + } + + if (!build) { + if (_state == State::FAILED) { + return _build_status; + } + DCHECK(_state == State::READY); + DCHECK(_prototype != nullptr); + ready_prototype = _prototype; + } + } + + if (!build) { + *statistics = ready_prototype->clone_for_scanner(); + return Status::OK(); + } + + CollectionStatisticsPtr prototype; + Status status; + try { + prototype = std::make_shared(); + status = builder(prototype.get(), _rowsets); + if (status.ok()) { + prototype->freeze_for_scanners(); + } + } catch (const Exception& exception) { + // collect() throws on malformed statistics; a throwing builder must still release the + // waiters below instead of leaving them blocked in BUILDING forever. + status = exception.to_status(); + } catch (const std::exception& exception) { + status = Status::InternalError("Collection statistics build failed: {}", exception.what()); + } catch (...) { + status = Status::InternalError("Collection statistics build failed with unknown error"); + } + + { + std::lock_guard lock(_mutex); + DCHECK(_state == State::BUILDING); + if (status.ok()) { + _prototype = std::move(prototype); + ready_prototype = _prototype; + _state = State::READY; + } else { + _build_status = status; + _state = State::FAILED; + } + } + _condition.notify_all(); + + RETURN_IF_ERROR(status); + DCHECK(ready_prototype != nullptr); + *statistics = ready_prototype->clone_for_scanner(); + 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)) { + const auto& term_doc_freqs = + _shared_base == nullptr ? _term_doc_freqs : _shared_base->term_doc_freqs; + if (!term_doc_freqs.contains(lucene_col_name)) { throw Exception(ErrorCode::INVERTED_INDEX_CLUCENE_ERROR, "Index statistics collection failed: Not such column {}", StringHelper::to_string(lucene_col_name)); } - if (!_term_doc_freqs[lucene_col_name].contains(term)) { + if (!term_doc_freqs.at(lucene_col_name).contains(term)) { throw Exception(ErrorCode::INVERTED_INDEX_CLUCENE_ERROR, "Index statistics collection failed: Not such term {}", StringHelper::to_string(term)); } - return _term_doc_freqs[lucene_col_name][term]; + return term_doc_freqs.at(lucene_col_name).at(term); } uint64_t CollectionStatistics::get_total_term_cnt_by_col(const std::wstring& lucene_col_name) { - if (!_total_num_tokens.contains(lucene_col_name)) { + const auto& total_num_tokens = + _shared_base == nullptr ? _total_num_tokens : _shared_base->total_num_tokens; + if (!total_num_tokens.contains(lucene_col_name)) { throw Exception(ErrorCode::INVERTED_INDEX_CLUCENE_ERROR, "Index statistics collection failed: Not such column {}", StringHelper::to_string(lucene_col_name)); } - return _total_num_tokens[lucene_col_name]; + return total_num_tokens.at(lucene_col_name); } uint64_t CollectionStatistics::get_doc_num() const { - if (_total_num_docs == 0) { + const uint64_t total_num_docs = + _shared_base == nullptr ? _total_num_docs : _shared_base->total_num_docs; + if (total_num_docs == 0) { throw Exception( ErrorCode::INVERTED_INDEX_CLUCENE_ERROR, "Index statistics collection failed: No data available for SimilarityCollector"); } - return _total_num_docs; + return total_num_docs; } float CollectionStatistics::get_or_calculate_avg_dl(const std::wstring& lucene_col_name) { diff --git a/be/src/storage/compaction/collection_statistics.h b/be/src/storage/compaction/collection_statistics.h index b93d5a424ae1ab..12a568d011274e 100644 --- a/be/src/storage/compaction/collection_statistics.h +++ b/be/src/storage/compaction/collection_statistics.h @@ -16,9 +16,13 @@ // under the License. #pragma once +#include #include +#include +#include #include #include +#include #include "common/be_mock_util.h" #include "exprs/vexpr_fwd.h" @@ -53,15 +57,39 @@ class CollectionStatistics { const TabletSchemaSPtr& tablet_schema, const VExprContextSPtrs& common_expr_ctxs_push_down, io::IOContext* io_ctx); + // Collect over every rowset of the tablet rather than one scanner's slice of them. BM25 idf and + // avg_dl are collection-level quantities, so they must not depend on how the read source was + // divided among parallel scanners. + Status collect_full_collection(RuntimeState* state, const std::vector& rowsets, + const TabletSchemaSPtr& tablet_schema, + const VExprContextSPtrs& common_expr_ctxs_push_down, + io::IOContext* io_ctx); + + // Hand a scanner its own view of an already-collected prototype. The collected counts are + // shared read-only; the lazily memoized idf/avg_dl maps start empty so each scanner mutates + // only its own. + std::shared_ptr clone_for_scanner() const; + MOCK_FUNCTION float get_or_calculate_idf(const std::wstring& lucene_col_name, const std::wstring& term); MOCK_FUNCTION float get_or_calculate_avg_dl(const std::wstring& lucene_col_name); private: + struct SharedBaseStatistics { + uint64_t total_num_docs = 0; + std::unordered_map total_num_tokens; + std::unordered_map> term_doc_freqs; + }; + + // Move the collected counts into an immutable block that every clone shares. + void freeze_for_scanners(); + Status extract_collect_info(RuntimeState* state, const VExprContextSPtrs& common_expr_ctxs_push_down, const TabletSchemaSPtr& tablet_schema, CollectInfoMap* collect_infos); + Status _collect_rowset(const RowsetSharedPtr& rowset, const TabletSchemaSPtr& tablet_schema, + const CollectInfoMap& collect_infos, io::IOContext* io_ctx); Status process_segment(const RowsetSharedPtr& rowset, int32_t seg_id, const TabletSchema* tablet_schema, const CollectInfoMap& collect_infos, io::IOContext* io_ctx); @@ -74,6 +102,8 @@ class CollectionStatistics { uint64_t _total_num_docs = 0; std::unordered_map _total_num_tokens; std::unordered_map> _term_doc_freqs; + // Set only on clones; when present it supersedes the three members above. + std::shared_ptr _shared_base; std::unordered_map _avg_dl_by_col; std::unordered_map> _idf_by_col_term; @@ -82,7 +112,34 @@ class CollectionStatistics { MOCK_DEFINE(friend class CollectionStatisticsTest;) MOCK_DEFINE(friend class BooleanQueryTest;) MOCK_DEFINE(friend class OccurBooleanQueryTest;) + friend class CollectionStatisticsBuildState; }; using CollectionStatisticsPtr = std::shared_ptr; +// One tablet's collection statistics, built at most once and shared by every scanner reading that +// tablet. Without this each parallel scanner would repeat the full-collection walk. +class CollectionStatisticsBuildState { +public: + using Builder = + std::function&)>; + + explicit CollectionStatisticsBuildState(std::vector rowsets) + : _rowsets(std::move(rowsets)) {} + + // The first caller builds while the others wait, then everyone gets its own clone. A failed + // build is remembered and returned to the waiters rather than retried per scanner. + Status get_or_build(const Builder& builder, CollectionStatisticsPtr* statistics); + +private: + enum class State { EMPTY, BUILDING, READY, FAILED }; + + const std::vector _rowsets; + std::mutex _mutex; + std::condition_variable _condition; + State _state = State::EMPTY; + Status _build_status; + CollectionStatisticsPtr _prototype; +}; +using CollectionStatisticsBuildStatePtr = std::shared_ptr; + } // namespace doris diff --git a/be/test/storage/compaction/collection_statistics_test.cpp b/be/test/storage/compaction/collection_statistics_test.cpp index b78355b316ebdd..d5328105e4c507 100644 --- a/be/test/storage/compaction/collection_statistics_test.cpp +++ b/be/test/storage/compaction/collection_statistics_test.cpp @@ -20,6 +20,9 @@ #include #include #include +#include +#include +#include #include #include @@ -320,6 +323,23 @@ class CollectionStatisticsTest : public ::testing::Test { return splits; } + void seed_statistics(CollectionStatistics* statistics, const std::wstring& field_name, + const std::wstring& term, uint64_t doc_count, uint64_t token_count, + uint64_t doc_freq) { + statistics->_total_num_docs = doc_count; + statistics->_total_num_tokens[field_name] = token_count; + statistics->_term_doc_freqs[field_name][term] = doc_freq; + } + + void set_cached_avg_dl(CollectionStatistics* statistics, const std::wstring& field_name, + float avg_dl) { + statistics->_avg_dl_by_col[field_name] = avg_dl; + } + + const void* shared_base_identity(const CollectionStatistics* statistics) { + return statistics->_shared_base.get(); + } + std::unique_ptr stats_; std::shared_ptr runtime_state_; std::string test_dir_; @@ -1307,4 +1327,120 @@ TEST(TermInfoComparerTest, OrdersByTermAndDedups) { EXPECT_THAT(ordered, ::testing::ElementsAre("apple", "banana", "cherry")); } +TEST_F(CollectionStatisticsTest, BuildStateBuildsFullCollectionOnceAcrossConcurrentScanners) { + constexpr size_t kScannerCount = 8; + auto build_state = + std::make_shared(std::vector {}); + std::atomic build_count = 0; + std::barrier start_barrier(kScannerCount); + std::vector statuses(kScannerCount); + std::vector scanner_statistics(kScannerCount); + std::vector scanners; + scanners.reserve(kScannerCount); + + for (size_t scanner_index = 0; scanner_index < kScannerCount; ++scanner_index) { + scanners.emplace_back([&, scanner_index] { + start_barrier.arrive_and_wait(); + statuses[scanner_index] = build_state->get_or_build( + [&](CollectionStatistics* statistics, + const std::vector& rowsets) { + EXPECT_TRUE(rowsets.empty()); + ++build_count; + seed_statistics(statistics, L"1", L"term", 10, 20, 3); + return Status::OK(); + }, + &scanner_statistics[scanner_index]); + }); + } + for (auto& scanner : scanners) { + scanner.join(); + } + + EXPECT_EQ(build_count.load(), 1u); + for (size_t scanner_index = 0; scanner_index < kScannerCount; ++scanner_index) { + ASSERT_TRUE(statuses[scanner_index].ok()) << statuses[scanner_index].to_string(); + ASSERT_NE(scanner_statistics[scanner_index], nullptr); + EXPECT_FLOAT_EQ(scanner_statistics[scanner_index]->get_or_calculate_avg_dl(L"1"), 2.0F); + } +} + +TEST_F(CollectionStatisticsTest, BuildStateBroadcastsOneFailureToConcurrentScanners) { + constexpr size_t kScannerCount = 8; + auto build_state = + std::make_shared(std::vector {}); + std::atomic build_count = 0; + std::atomic calls_started = 0; + std::promise builder_started; + auto builder_started_future = builder_started.get_future(); + std::promise release_builder; + auto release_builder_future = release_builder.get_future().share(); + std::barrier start_barrier(kScannerCount); + std::vector statuses(kScannerCount); + std::vector scanner_statistics(kScannerCount); + std::vector scanners; + scanners.reserve(kScannerCount); + + for (size_t scanner_index = 0; scanner_index < kScannerCount; ++scanner_index) { + scanners.emplace_back([&, scanner_index] { + start_barrier.arrive_and_wait(); + ++calls_started; + statuses[scanner_index] = build_state->get_or_build( + [&](CollectionStatistics*, const std::vector&) { + if (++build_count == 1) { + builder_started.set_value(); + } + release_builder_future.wait(); + return Status::Error( + "incompatible scoring segment"); + }, + &scanner_statistics[scanner_index]); + }); + } + + builder_started_future.wait(); + while (calls_started.load() != kScannerCount) { + std::this_thread::yield(); + } + release_builder.set_value(); + for (auto& scanner : scanners) { + scanner.join(); + } + + EXPECT_EQ(build_count.load(), 1u); + for (size_t scanner_index = 0; scanner_index < kScannerCount; ++scanner_index) { + EXPECT_EQ(statuses[scanner_index].code(), ErrorCode::INVERTED_INDEX_NOT_SUPPORTED); + EXPECT_THAT(statuses[scanner_index].msg(), + ::testing::HasSubstr("incompatible scoring segment")); + EXPECT_EQ(scanner_statistics[scanner_index], nullptr); + } +} + +TEST_F(CollectionStatisticsTest, BuildStateReturnsScannerLocalLazyCacheClones) { + auto build_state = + std::make_shared(std::vector {}); + size_t build_count = 0; + auto builder = [&](CollectionStatistics* statistics, const std::vector&) { + ++build_count; + seed_statistics(statistics, L"1", L"term", 10, 20, 3); + return Status::OK(); + }; + + CollectionStatisticsPtr first; + ASSERT_TRUE(build_state->get_or_build(builder, &first).ok()); + set_cached_avg_dl(first.get(), L"1", 5.0F); + EXPECT_FLOAT_EQ(first->get_or_calculate_avg_dl(L"1"), 5.0F); + + CollectionStatisticsPtr second; + ASSERT_TRUE(build_state->get_or_build(builder, &second).ok()); + + EXPECT_EQ(build_count, 1u); + ASSERT_NE(first, nullptr); + ASSERT_NE(second, nullptr); + EXPECT_NE(first.get(), second.get()); + EXPECT_NE(shared_base_identity(first.get()), nullptr); + EXPECT_EQ(shared_base_identity(first.get()), shared_base_identity(second.get())); + EXPECT_FLOAT_EQ(second->get_or_calculate_avg_dl(L"1"), 2.0F); + EXPECT_FLOAT_EQ(first->get_or_calculate_avg_dl(L"1"), 5.0F); +} + } // namespace doris diff --git a/regression-test/suites/inverted_index_p0/test_bm25_score_parallel_scan.groovy b/regression-test/suites/inverted_index_p0/test_bm25_score_parallel_scan.groovy new file mode 100644 index 00000000000000..5296d2980a5824 --- /dev/null +++ b/regression-test/suites/inverted_index_p0/test_bm25_score_parallel_scan.groovy @@ -0,0 +1,94 @@ +// 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. + +// BM25 idf and avg_dl are properties of the whole collection, so score() must not depend on how +// many scanners the read source was divided into. Collection statistics used to be collected from +// each scanner's own rs_splits, which made the scores drift as soon as parallel scan produced more +// than one scanner per tablet. +suite("test_bm25_score_parallel_scan", "p0") { + def tableName = "test_bm25_score_parallel_scan" + + sql "DROP TABLE IF EXISTS ${tableName}" + sql """ + CREATE TABLE ${tableName} ( + id INT NOT NULL, + body STRING NULL, + INDEX idx_body (body) USING INVERTED PROPERTIES ( + "parser" = "english", + "support_phrase" = "true" + ) + ) ENGINE=OLAP + DUPLICATE KEY(id) + DISTRIBUTED BY HASH(id) BUCKETS 1 + PROPERTIES ( + "replication_num" = "1", + "disable_auto_compaction" = "true", + "inverted_index_storage_format" = "V3" + ) + """ + + // Several separate loads so the tablet ends up with several rowsets. One rowset per INSERT is + // what lets the parallel scanner builder hand different rowsets to different scanners -- with a + // single rowset every scanner would see the whole collection and the bug would stay hidden. + def rareDocs = (0..<8).collect { "(${it}, 'alpha rare_term beta')" }.join(", ") + sql "INSERT INTO ${tableName} VALUES ${rareDocs}" + for (int batch = 0; batch < 6; batch++) { + def filler = (0..<200).collect { + def docId = 1000 + batch * 200 + it + "(${docId}, 'alpha common filler document number ${docId}')" + }.join(", ") + sql "INSERT INTO ${tableName} VALUES ${filler}" + } + sql "SYNC" + + def scoreRows = { -> + sql """ + SELECT id, ROUND(s, 6) FROM ( + SELECT id, score() AS s FROM ${tableName} + WHERE body MATCH_ANY 'rare_term alpha' + ORDER BY s DESC LIMIT 20 + ) ranked ORDER BY s DESC, id + """ + } + + // Capture rather than hardcode: the defaults for these have changed before, and restoring a + // wrong value would leak into whatever runs next on this connection. + def savedVars = ["enable_parallel_scan", "parallel_scan_max_scanners_count", + "parallel_scan_min_rows_per_scanner"].collectEntries { name -> + [(name): sql("SHOW VARIABLES LIKE '${name}'")[0][1].toString()] + } + + try { + // Baseline: one scanner per tablet, so its slice already is the whole collection. + sql "SET enable_parallel_scan = false" + def serialScores = scoreRows() + assertFalse(serialScores.isEmpty(), "serial scan must return rows") + + // Force many scanners over the same data. Every scanner has to derive idf from the whole + // collection, not from the rowsets it happens to own. + sql "SET enable_parallel_scan = true" + sql "SET parallel_scan_max_scanners_count = 16" + sql "SET parallel_scan_min_rows_per_scanner = 16" + def parallelScores = scoreRows() + + assertEquals(serialScores, parallelScores, + "score() changed when the read source was split across scanners, which means " + + "collection statistics were collected per scanner instead of per collection") + } finally { + savedVars.each { name, value -> sql "SET ${name} = ${value}" } + } +}