From f32b11b2235bb0bcd73296442732b4ce3f9f5bf7 Mon Sep 17 00:00:00 2001 From: airborne12 Date: Sun, 8 Nov 2026 08:59:41 +0000 Subject: [PATCH] [opt](inverted index) Share index runs among the scanners of a split segment ### Background ParallelScannerBuilder splits a large segment into row ranges, one per scanner, and each scanner evaluates the segment's index predicates over the whole segment before it keeps its own range. TextIndexReader can make concurrent identical full-segment queries share one run (#68758); only SNII turns that on. ### What problem does this PR solve? Issue Number: None Related PR: #68758 (added the shared run for SNII) PR Type: Performance Problem Summary: With the default 2M rows per scanner, an index predicate on a 6.3M-row segment runs three times: MATCH, keyword and BKD predicates on CLucene indexes and every search() call. CLucene text readers now share runs as SNII does, the BKD reader shares them through the same helper, moved into InvertedIndexReader, and search(), which reads indexes without the readers' query path, shares its whole-segment result the same way. ### Details - **Why it was slow:** each scanner of a split segment repeats the segment's index work. Over 100M rows in 6.3M-row segments, split scanners took 1.6 to 3.1 times the CPU of whole-segment scanners for MATCH on CLucene and 1.4 to 2.3 times for BKD ranges; search() repeated its work on both formats. - **Optimization:** a run is shared by concurrent identical queries under the result cache's key, whether or not the result cache is on. Scoring runs and runs restricted to candidate rows run alone. The leader publishes on every exit path without allocating, and a follower whose leader ended without a result runs for itself. search() hands every follower its own copy, since SQL composition mutates its input. - **Results:** a branch-4.2 build with these changes, 100M rows in 6.3M-row segments, one client, alternating runs on one BE: CLucene MATCH_REGEXP 'cach.*' CPU 18.4 to 6.0 s and MATCH_PHRASE 1.98 to 0.77 s, both equal to an unsplit scan; search() `*cache*` 14.8 to 5.3 s (V3) and 11.3 to 4.8 s (SNII). At 64 clients, with search() runs shared only within a query, its workload served 57 to 139 queries per second (V3) and 77 to 166 (SNII), p99 3315 to 1046 ms and 1625 to 707 ms; sharing across queries as well, as here, measured up to 7% more. Rows identical. - **Applicability and cost:** segments larger than the rows per scanner. Followers wait for the leader instead of repeating its work; concurrent identical queries of different sessions share a run as well, as they already did on SNII. Scanner layout, writing and storage are unchanged. - **Behavior change:** CLucene readers now share concurrent identical queries; `FulltextIndependentQueries`, which asserted they did not, becomes `FulltextConcurrentQueriesShareRuns`. ### Testing Local ASAN build: - `InvertedIndexReaderTest.FulltextIdenticalQueriesShareOneRun` and `BkdIdenticalQueriesShareOneRun` hold the leader of two identical queries until the follower joins, then check that both get the same rows from one run; `FulltextConcurrentQueriesShareRuns` checks every one of three concurrent queries leads, joins or hits the cache. - `SearchFlightTest`: scanners of one search share a run and each gets its own copy; a different search runs alone and completes while the first is held; a follower whose leader fails runs for itself; a scoring run never joins. - With each shared run switched off in turn, every sharing test above fails. - 2832 inverted-index, search, match, SNII and BKD unit tests pass; clang-tidy reports nothing on the changed lines. - Regression `inverted_index_p0/test_split_scanners_share_index_runs` on a local ASAN cluster: one 16384-row CLucene segment split among 8 scanners (NumScanners 8 in the profile) gives the same rows as a whole-segment scan for MATCH, keyword, BKD, search() and compound predicates. ### Release note Queries with inverted index predicates on large segments use less CPU when the segment is read by several scanners. ### Check List (For Author) - Behavior changed: Yes (see Details) - Does this need documentation: No - Pre-submission status: Partially verified: local ASAN unit and regression tests only; CI not run ### AI Disclosure - Tool: Claude Code (Claude Opus 5.5) - Generated: code, tests, this description - Human review: not yet reviewed Co-Authored-By: Claude Opus 5.5 --- .../exprs/function/search/function_search.cpp | 76 +++++- .../exprs/function/search/function_search.h | 9 + .../inverted/clucene/clucene_index_reader.cpp | 13 +- .../index/inverted/inverted_index_reader.cpp | 47 ++++ .../index/inverted/inverted_index_reader.h | 8 + .../inverted/reader/text_index_reader.cpp | 67 +----- .../function/search/function_search_test.cpp | 227 ++++++++++++++++++ .../segment/inverted_index_reader_test.cpp | 166 ++++++++++++- .../test_split_scanners_share_index_runs.out | 43 ++++ ...est_split_scanners_share_index_runs.groovy | 86 +++++++ 10 files changed, 672 insertions(+), 70 deletions(-) create mode 100644 regression-test/data/inverted_index_p0/test_split_scanners_share_index_runs.out create mode 100644 regression-test/suites/inverted_index_p0/test_split_scanners_share_index_runs.groovy diff --git a/be/src/exprs/function/search/function_search.cpp b/be/src/exprs/function/search/function_search.cpp index 0dc39328bf56d0..f2a78a5eb56d1f 100644 --- a/be/src/exprs/function/search/function_search.cpp +++ b/be/src/exprs/function/search/function_search.cpp @@ -48,6 +48,7 @@ #include "storage/index/index_file_reader.h" #include "storage/index/index_query_context.h" #include "storage/index/inverted/analysis/analyzer/analyzer.h" +#include "storage/index/inverted/common/single_flight.h" #include "storage/index/inverted/inverted_index_iterator.h" #include "storage/index/inverted/inverted_index_reader.h" #include "storage/index/inverted/query/engine/all_query/all_query.h" @@ -62,6 +63,7 @@ #include "storage/index/inverted/reader/collection_statistics.h" #include "storage/olap_common.h" #include "storage/segment/variant/nested_group_provider.h" +#include "util/defer_op.h" #include "util/thrift_util.h" namespace doris { @@ -143,6 +145,24 @@ void insert_dsl_result(InvertedIndexQueryCache* cache, const std::string& seg_pr std::make_shared(*result.get_data_bitmap()), &rows_handle); } +// Concurrent identical SEARCH runs over one segment share one execution, as concurrent identical +// index queries do in the index readers. The leader publishes rows nobody mutates, or null when it +// ends without a result, and each follower takes a copy, since SQL bitmap composition mutates its +// input. +using SearchFlight = + segment_v2::inverted_index::SingleFlight>; + +SearchFlight& search_single_flight() { + static SearchFlight flight; + return flight; +} + +#ifdef BE_TEST +FunctionSearch::SingleFlightObserver g_leader_before_compute_observer = nullptr; +FunctionSearch::SingleFlightObserver g_follower_joined_observer = nullptr; +void* g_single_flight_observer_opaque = nullptr; +#endif + bool is_nested_group_search_supported() { auto provider = segment_v2::create_nested_group_read_provider(); return provider != nullptr && provider->should_enable_nested_group_read_path(); @@ -398,6 +418,16 @@ Status FunctionSearch::evaluate_inverted_index( return Status::OK(); } +#ifdef BE_TEST +void FunctionSearch::set_single_flight_observers_for_test( + SingleFlightObserver leader_before_compute, SingleFlightObserver follower_joined, + void* opaque) { + g_leader_before_compute_observer = leader_before_compute; + g_follower_joined_observer = follower_joined; + g_single_flight_observer_opaque = opaque; +} +#endif + Status FunctionSearch::evaluate_inverted_index_with_search_param( const TSearchParam& search_param, const std::unordered_map& data_type_with_names, @@ -459,12 +489,21 @@ Status FunctionSearch::_evaluate_search_param( index_query_context != nullptr && index_query_context->collection_similarity != nullptr; auto* dsl_cache = enable_cache && !enable_scoring ? InvertedIndexQueryCache::instance() : nullptr; + // ParallelScannerBuilder splits a large segment into rowid ranges, and every scanner of the + // segment evaluates SEARCH over the whole segment before it intersects its own range: a + // 6.3M-row segment split at the default 2M rows ran the same SEARCH three times at once. + // Concurrent identical runs over one segment therefore share one execution, whether or not the + // result cache is on, under the result cache's key. A scoring run records the scores of its + // own scanner and runs alone, as does NESTED with its own evaluator. + const bool coalesce = !enable_scoring && !is_nested_query; std::string seg_prefix; std::string dsl_sig; bool cache_usable = false; - if (dsl_cache) { + if (dsl_cache != nullptr || coalesce) { seg_prefix = extract_segment_prefix(iterators); dsl_sig = dsl_cache_signature(search_param, index_query_context.get()); + } + if (dsl_cache) { if (!seg_prefix.empty() && !dsl_sig.empty()) { cache_usable = true; bool dsl_hit = false; @@ -486,6 +525,38 @@ Status FunctionSearch::_evaluate_search_param( } } + std::string flight_key; + bool flight_leader = false; + std::shared_ptr flight_rows; + if (coalesce && !seg_prefix.empty() && !dsl_sig.empty()) { + // A shared run must cover the whole segment. SegmentIterator never hands SEARCH candidate + // rows, since SEARCH combines its leaves into whole-segment answers. + DORIS_CHECK(index_query_context == nullptr || + index_query_context->candidate_rows == nullptr); + flight_key = dsl_cache_key(seg_prefix, dsl_sig).encode(); + auto follower = search_single_flight().join_or_lead(flight_key); + flight_leader = !follower.has_value(); +#ifdef BE_TEST + const auto observer = + flight_leader ? g_leader_before_compute_observer : g_follower_joined_observer; + if (observer != nullptr) { + observer(g_single_flight_observer_opaque); + } +#endif + if (!flight_leader) { + if (const auto& rows = follower->get(); rows != nullptr) { + bitmap_result = InvertedIndexResultBitmap(std::make_shared(*rows), + std::make_shared()); + return Status::OK(); + } + // A leader that ended without rows, by an error, an exception or an empty query + // tree, leaves each follower to run for itself. + } + } + // The publish allocates nothing, so it reaches the followers on every exit path, exceptions + // and memory-limit failures included; a waiting follower is never left behind. + DEFER(if (flight_leader) { search_single_flight().publish(flight_key, flight_rows); }); + std::shared_ptr context; if (index_query_context) { context = index_query_context; @@ -601,6 +672,9 @@ Status FunctionSearch::_evaluate_search_param( // SEARCH never returns UNKNOWN; outer SQL composes this non-nullable match set. bitmap_result = InvertedIndexResultBitmap(std::move(roaring), std::make_shared()); + if (flight_leader) { + flight_rows = std::make_shared(*bitmap_result.get_data_bitmap()); + } if (dsl_cache && cache_usable) { insert_dsl_result(dsl_cache, seg_prefix, dsl_sig, bitmap_result); diff --git a/be/src/exprs/function/search/function_search.h b/be/src/exprs/function/search/function_search.h index 8912970a692e74..764db225a2ed3b 100644 --- a/be/src/exprs/function/search/function_search.h +++ b/be/src/exprs/function/search/function_search.h @@ -115,6 +115,15 @@ class FunctionSearch : public IFunction { static std::string dsl_cache_signature(const TSearchParam& search_param, const IndexQueryContext* context); +#ifdef BE_TEST + using SingleFlightObserver = void (*)(void*) noexcept; + // Called when an evaluation leads a shared SEARCH run, before it computes, and when one joins + // a run another evaluation leads. They are process-wide: a function carries no member data. + static void set_single_flight_observers_for_test(SingleFlightObserver leader_before_compute, + SingleFlightObserver follower_joined, + void* opaque); +#endif + private: Status _evaluate_search_param( const TSearchParam& search_param, diff --git a/be/src/storage/index/inverted/clucene/clucene_index_reader.cpp b/be/src/storage/index/inverted/clucene/clucene_index_reader.cpp index 17e75604484e88..fd1d56f2fd4b22 100644 --- a/be/src/storage/index/inverted/clucene/clucene_index_reader.cpp +++ b/be/src/storage/index/inverted/clucene/clucene_index_reader.cpp @@ -785,9 +785,16 @@ Status BkdIndexReader::query(const IndexQueryContextPtr& context, const std::str return Status::OK(); } - RETURN_IF_ERROR(invoke_bkd_query(context, query_value, query_type, r, bit_map)); - bit_map->runOptimize(); - cache->insert(cache_key, bit_map, &cache_handler); + // Every scanner of a split segment asks for the whole segment at once, so they share one + // run, as text queries do. + RETURN_IF_ERROR(run_query_single_flight( + cache_key.encode(), &bit_map, [&](std::shared_ptr* result) { + *result = std::make_shared(); + RETURN_IF_ERROR(invoke_bkd_query(context, query_value, query_type, r, *result)); + (*result)->runOptimize(); + cache->insert(cache_key, *result, &cache_handler); + return Status::OK(); + })); VLOG_DEBUG << "BKD index search column: " << column_name << " result: " << bit_map->cardinality(); diff --git a/be/src/storage/index/inverted/inverted_index_reader.cpp b/be/src/storage/index/inverted/inverted_index_reader.cpp index ccc3710910802d..b0299a869ee086 100644 --- a/be/src/storage/index/inverted/inverted_index_reader.cpp +++ b/be/src/storage/index/inverted/inverted_index_reader.cpp @@ -24,9 +24,56 @@ #include "runtime/runtime_profile.h" #include "runtime/runtime_state.h" #include "storage/index/index_file_reader.h" +#include "storage/index/inverted/common/single_flight.h" +#include "util/defer_op.h" namespace doris::segment_v2 { +namespace { + +// Null tells a follower that its leader ended without a bitmap. +using QueryFlight = inverted_index::SingleFlight>; + +QueryFlight& query_single_flight() { + static QueryFlight flight; + return flight; +} + +} // namespace + +Status InvertedIndexReader::run_query_single_flight( + const std::string& key, std::shared_ptr* result, + const std::function*)>& compute) { + auto& flight = query_single_flight(); + auto follower = flight.join_or_lead(key); + if (follower.has_value()) { +#ifdef BE_TEST + if (_single_flight_follower_joined_observer != nullptr) { + _single_flight_follower_joined_observer(_single_flight_follower_joined_opaque); + } +#endif + if (const auto& leader_bitmap = follower->get(); leader_bitmap != nullptr) { + *result = leader_bitmap; + return Status::OK(); + } + } + const bool is_leader = !follower.has_value(); +#ifdef BE_TEST + if (is_leader && _single_flight_leader_before_compute_observer != nullptr) { + _single_flight_leader_before_compute_observer(_single_flight_leader_before_compute_opaque); + } +#endif + std::shared_ptr published; + // Followers learn the outcome on every exit path, errors and exceptions included. The publish + // allocates nothing, so it cannot fail on the way out and leave a follower waiting. + DEFER(if (is_leader) { flight.publish(key, published); }); + std::shared_ptr bitmap; + RETURN_IF_ERROR(compute(&bitmap)); + published = bitmap; + *result = std::move(bitmap); + return Status::OK(); +} + std::string InvertedIndexReader::get_index_file_path() { return _index_file_reader->get_index_file_path(&_index_meta); } diff --git a/be/src/storage/index/inverted/inverted_index_reader.h b/be/src/storage/index/inverted/inverted_index_reader.h index b16f4dbab16fe7..78a67f616854ba 100644 --- a/be/src/storage/index/inverted/inverted_index_reader.h +++ b/be/src/storage/index/inverted/inverted_index_reader.h @@ -347,6 +347,14 @@ class InvertedIndexReader : public IndexReader { const TabletIndex& get_index_meta() const { return _index_meta; } protected: + // Runs `compute` once for concurrent identical full-segment queries, such as those of the + // scanners a large segment is split into, each of which asks for the whole segment: the first + // caller leads and the others take its bitmap. A leader that ends without one leaves each + // follower to compute for itself. `key` names the query as the result cache does. + Status run_query_single_flight( + const std::string& key, std::shared_ptr* result, + const std::function*)>& compute); + friend class InvertedIndexIterator; std::shared_ptr _index_file_reader; TabletIndex _index_meta; diff --git a/be/src/storage/index/inverted/reader/text_index_reader.cpp b/be/src/storage/index/inverted/reader/text_index_reader.cpp index d9d2abe9700610..043694c795cc6f 100644 --- a/be/src/storage/index/inverted/reader/text_index_reader.cpp +++ b/be/src/storage/index/inverted/reader/text_index_reader.cpp @@ -39,7 +39,6 @@ #include "storage/index/inverted/analysis/inverted_index_parser.h" #include "storage/index/inverted/common/inverted_index_cache.h" #include "storage/index/inverted/common/inverted_index_query_type.h" -#include "storage/index/inverted/common/single_flight.h" #include "storage/index/inverted/common/string_helper.h" #include "storage/index/inverted/inverted_index_iterator.h" #include "storage/index/inverted/query/engine/all_query/all_query.h" @@ -60,7 +59,6 @@ #include "storage/index/inverted/query/logical/match_lowering.h" #include "storage/index/inverted/query/scoring/bm25_similarity.h" #include "storage/olap_common.h" -#include "util/defer_op.h" namespace doris::segment_v2 { @@ -234,57 +232,6 @@ Status plan_query(const logical::Node& leaf, const IndexQueryContextPtr& context namespace { -using QueryFlight = - inverted_index::SingleFlight>>; - -QueryFlight& query_single_flight() { - static QueryFlight flight; - return flight; -} - -// Runs `compute` once for concurrent identical queries: the first caller leads and the others -// take its bitmap. A leader's failure leaves each follower to compute for itself. -template -Status run_query_single_flight( - QueryFlight& flight, const std::string& key, std::shared_ptr* result, -#ifdef BE_TEST - InvertedIndexReader::SingleFlightFollowerJoinedObserver follower_joined_observer, - void* follower_joined_opaque, - InvertedIndexReader::SingleFlightLeaderBeforeComputeObserver leader_before_compute_observer, - void* leader_before_compute_opaque, -#endif - Compute&& compute) { - auto follower = flight.join_or_lead(key); - if (follower.has_value()) { -#ifdef BE_TEST - if (follower_joined_observer != nullptr) { - follower_joined_observer(follower_joined_opaque); - } -#endif - auto [leader_status, leader_bitmap] = follower->get(); - if (leader_status.ok() && leader_bitmap != nullptr) { - *result = std::move(leader_bitmap); - return Status::OK(); - } - } - const bool is_leader = !follower.has_value(); -#ifdef BE_TEST - if (is_leader && leader_before_compute_observer != nullptr) { - leader_before_compute_observer(leader_before_compute_opaque); - } -#endif - Status status = Status::OK(); - std::shared_ptr bitmap; - { - // Followers learn the outcome on every exit path, errors included. - DEFER(if (is_leader) { flight.publish(key, std::make_pair(status, bitmap)); }); - status = compute(&bitmap); - } - RETURN_IF_ERROR(status); - *result = std::move(bitmap); - return Status::OK(); -} - // The one exact term a COUNT_ON_INDEX scan of `leaf` can answer from the dictionary. const std::string* single_exact_term(const logical::Node& leaf) { if (const auto* term = leaf.as()) { @@ -388,13 +335,17 @@ Status TextIndexReader::_query_raw(const IndexQueryContextPtr& context, Status TextIndexReader::_admit(const IndexQueryContextPtr& /*context*/, const std::string& column_name, const LeafRequest& request, - Admission* /*admission*/) { + Admission* admission) { // CLucene indexes do not persist the gram tokenizer contract needed to compile a pattern. if (is_gram_query(request.query_type)) { return Status::Error( "{} requires SNII storage format for column {}", query_type_to_string(request.query_type), column_name); } + // Every scanner of a split segment asks for the whole segment at once, so concurrent identical + // queries share one run, as SNII's do: with 6.3M-row segments cut at 2M rows per scanner, + // MATCH_REGEXP 'cach.*' over 100M rows took 18 s of CPU instead of 6. + admission->coalesce = true; return Status::OK(); } @@ -546,13 +497,7 @@ Status TextIndexReader::_run_request(const IndexQueryContextPtr& context, std::shared_ptr result; const bool coalesce = allow_result_cache && admission.coalesce; const auto run_coalesced = [&](const auto& run) { - return run_query_single_flight(query_single_flight(), cache_key.encode(), &result, -#ifdef BE_TEST - _single_flight_follower_joined_observer, - _single_flight_follower_joined_opaque, - _single_flight_leader_before_compute_observer, - _single_flight_leader_before_compute_opaque, -#endif + return run_query_single_flight(cache_key.encode(), &result, [&](std::shared_ptr* shared) { RETURN_IF_ERROR(run(shared)); insert_query_cache(context, cache, cache_key, *shared, diff --git a/be/test/exprs/function/search/function_search_test.cpp b/be/test/exprs/function/search/function_search_test.cpp index cec322d67ef2d4..55a9c06e4de4e4 100644 --- a/be/test/exprs/function/search/function_search_test.cpp +++ b/be/test/exprs/function/search/function_search_test.cpp @@ -40,12 +40,16 @@ #include #include +#include +#include #include #include #include +#include #include #include #include +#include #include #include #include @@ -65,6 +69,7 @@ #include "io/fs/local_file_system.h" #include "runtime/exec_env.h" #include "runtime/index_policy/index_policy_mgr.h" +#include "runtime/memory/mem_tracker_limiter.h" #include "runtime/runtime_state.h" #include "storage/index/index_file_reader.h" #include "storage/index/index_iterator.h" @@ -4497,4 +4502,226 @@ TEST_F(FunctionSearchTest, SearchTopKAboveCollectorCapacityPreservesAllMatches) EXPECT_EQ(15000, fixture.context->collection_similarity->_bm25_scores.size()); } +// Holds the first leader of a shared SEARCH run before it computes, until the test releases it, +// and counts the evaluations that lead or join a run. Released from the start, it only counts. +class SearchFlightGate { +public: + static void leader(void* opaque) noexcept { + auto& gate = *static_cast(opaque); + std::unique_lock lock(gate._mutex); + ++gate._leaders; + gate._cv.notify_all(); + if (gate._leaders == 1) { + gate._cv.wait(lock, [&] { return gate._released; }); + } + } + + static void follower(void* opaque) noexcept { + auto& gate = *static_cast(opaque); + std::lock_guard lock(gate._mutex); + ++gate._followers; + gate._cv.notify_all(); + } + + void install() { + FunctionSearch::set_single_flight_observers_for_test(&leader, &follower, this); + } + + bool wait_for(size_t leaders, size_t followers) { + std::unique_lock lock(_mutex); + return _cv.wait_for(lock, std::chrono::seconds(10), + [&] { return _leaders == leaders && _followers == followers; }); + } + + void release() { + std::lock_guard lock(_mutex); + _released = true; + _cv.notify_all(); + } + + // Read only after every evaluation has returned. + size_t leaders() const { return _leaders; } + size_t followers() const { return _followers; } + +private: + std::mutex _mutex; + std::condition_variable _cv; + size_t _leaders = 0; + size_t _followers = 0; + bool _released = false; +}; + +// body:*lpha over one SNII segment whose index answers rows {0, 2}, evaluated the way each +// scanner of a split segment evaluates it. Only one evaluation computes at a time in these tests, +// since the recording reader is not thread-safe: a leader is held before it computes. +class SearchFlightTest : public FunctionSearchTest { +protected: + void TearDown() override { + FunctionSearch::set_single_flight_observers_for_test(nullptr, nullptr, nullptr); + FunctionSearchTest::TearDown(); + } + + void SetUp() override { + FunctionSearchTest::SetUp(); + _reader->fake_source->add("alpha", {0, 2}); + _reader->fake_source->add("beta", {1}); + _iterator.add_reader(segment_v2::InvertedIndexReaderType::FULLTEXT, _reader); + _data_type_with_names.emplace( + "body", IndexFieldNameAndTypePair {"body", std::make_shared()}); + _iterators["body"] = &_iterator; + TSearchFieldBinding field_binding; + field_binding.field_name = "body"; + field_binding.index_properties = _index_meta.properties(); + field_binding.__isset.index_properties = true; + _search_param.original_dsl = "body:*lpha"; + _search_param.root = make_leaf_clause("WILDCARD", "*lpha"); + _search_param.field_bindings = {field_binding}; + } + + // A query whose scanners run the SEARCH. + static std::unique_ptr make_query(int64_t id) { + TUniqueId query_id; + query_id.__set_hi(id); + query_id.__set_lo(id); + return std::make_unique( + query_id, 0, TQueryOptions(), TQueryGlobals(), ExecEnv::GetInstance(), + MemTrackerLimiter::create_shared(MemTrackerLimiter::Type::OTHER, + "search_flight_test")); + } + + // Runs the SEARCH, or `root` in its place, as one scanner of `query` does, with the result + // cache off. + Status run(const FunctionSearch& function, RuntimeState* query, + InvertedIndexResultBitmap* result, bool scoring = false, + std::optional root = std::nullopt) { + auto context = std::make_shared(); + context->runtime_state = query; + if (scoring) { + context->collection_statistics = std::make_shared(); + context->collection_similarity = std::make_shared(); + } + TSearchParam search_param = _search_param; + if (root.has_value()) { + search_param.original_dsl = "body:" + root->value; + search_param.root = *root; + } + return function.evaluate_inverted_index_with_search_param( + search_param, _data_type_with_names, _iterators, 4, *result, + /*enable_cache=*/false, nullptr, {}, context); + } + + TabletIndex _index_meta = make_test_inverted_index( + 32, {{INVERTED_INDEX_PARSER_KEY, INVERTED_INDEX_PARSER_STANDARD}}); + std::shared_ptr _index_file_reader = + std::make_shared(InvertedIndexStorageFormatPB::SNII, + "/tmp/search_flight_idx"); + std::shared_ptr _reader = + std::make_shared(&_index_meta, _index_file_reader); + segment_v2::InvertedIndexIterator _iterator; + std::unordered_map _data_type_with_names; + std::unordered_map _iterators; + TSearchParam _search_param; +}; + +// The scanners a segment is split into each evaluate the SEARCH over the whole segment at once. +// Those of one query share one run, and each takes its own copy of the rows. +TEST_F(SearchFlightTest, ScannersOfOneQueryShareOneRun) { + auto query = make_query(1); + SearchFlightGate gate; + gate.install(); + InvertedIndexResultBitmap leader_result; + InvertedIndexResultBitmap follower_result; + Status leader_status; + Status follower_status; + std::thread leader([&] { leader_status = run(*function_search, query.get(), &leader_result); }); + EXPECT_TRUE(gate.wait_for(1, 0)); + std::thread follower( + [&] { follower_status = run(*function_search, query.get(), &follower_result); }); + EXPECT_TRUE(gate.wait_for(1, 1)); + gate.release(); + leader.join(); + follower.join(); + + ASSERT_TRUE(leader_status.ok()) << leader_status; + ASSERT_TRUE(follower_status.ok()) << follower_status; + expect_bitmap_eq(*leader_result.get_data_bitmap(), {0, 2}); + expect_bitmap_eq(*follower_result.get_data_bitmap(), {0, 2}); + EXPECT_EQ(1, _reader->open_source_calls); + EXPECT_EQ(std::vector {""}, _reader->fake_source->expanded); + // SQL composition mutates the rows it is handed, so the follower's are its own. + follower_result.get_data_bitmap()->add(3); + expect_bitmap_eq(*leader_result.get_data_bitmap(), {0, 2}); +} + +// A different search over the same segment never joins the run in progress: it leads its own and +// completes while the first search is held before it computes. +TEST_F(SearchFlightTest, DifferentSearchesRunSeparately) { + auto query = make_query(1); + SearchFlightGate gate; + gate.install(); + + InvertedIndexResultBitmap first_result; + Status first_status; + std::thread first([&] { first_status = run(*function_search, query.get(), &first_result); }); + EXPECT_TRUE(gate.wait_for(1, 0)); + InvertedIndexResultBitmap second_result; + auto second = std::async(std::launch::async, [&] { + return run(*function_search, query.get(), &second_result, /*scoring=*/false, + make_leaf_clause("WILDCARD", "bet*")); + }); + const bool second_done = second.wait_for(std::chrono::seconds(10)) == std::future_status::ready; + gate.release(); + first.join(); + const Status second_status = second.get(); + + EXPECT_TRUE(second_done); + ASSERT_TRUE(first_status.ok()) << first_status; + ASSERT_TRUE(second_status.ok()) << second_status; + expect_bitmap_eq(*first_result.get_data_bitmap(), {0, 2}); + expect_bitmap_eq(*second_result.get_data_bitmap(), {1}); + EXPECT_EQ(2U, gate.leaders()); + EXPECT_EQ(0U, gate.followers()); + EXPECT_EQ(2, _reader->open_source_calls); +} + +// A leader that ends without rows, here on an error its expansion raises, publishes none; the +// follower then runs the SEARCH itself instead of waiting on rows that never come. +TEST_F(SearchFlightTest, AFollowerRunsItselfWhenTheLeaderFails) { + _reader->fake_source->expand_status = Status::InternalError("the expansion failed"); + auto query = make_query(1); + SearchFlightGate gate; + gate.install(); + InvertedIndexResultBitmap leader_result; + InvertedIndexResultBitmap follower_result; + Status leader_status; + Status follower_status; + std::thread leader([&] { leader_status = run(*function_search, query.get(), &leader_result); }); + EXPECT_TRUE(gate.wait_for(1, 0)); + std::thread follower( + [&] { follower_status = run(*function_search, query.get(), &follower_result); }); + EXPECT_TRUE(gate.wait_for(1, 1)); + gate.release(); + leader.join(); + follower.join(); + + EXPECT_FALSE(leader_status.ok()); + EXPECT_FALSE(follower_status.ok()); + EXPECT_EQ((std::vector {"", ""}), _reader->fake_source->expanded); +} + +// A scoring run records the scores of its own scanner, so it runs alone. +TEST_F(SearchFlightTest, ScoringRunsRunAlone) { + auto query = make_query(1); + SearchFlightGate counter; + counter.release(); + counter.install(); + + InvertedIndexResultBitmap scored; + auto status = run(*function_search, query.get(), &scored, /*scoring=*/true); + ASSERT_TRUE(status.ok()) << status; + expect_bitmap_eq(*scored.get_data_bitmap(), {0, 2}); + EXPECT_EQ(0U, counter.leaders()); + EXPECT_EQ(0U, counter.followers()); +} + } // namespace doris diff --git a/be/test/storage/segment/inverted_index_reader_test.cpp b/be/test/storage/segment/inverted_index_reader_test.cpp index 1123bc6fb5a715..ef27c9fe55891f 100644 --- a/be/test/storage/segment/inverted_index_reader_test.cpp +++ b/be/test/storage/segment/inverted_index_reader_test.cpp @@ -24,10 +24,13 @@ #include #include #include +#include +#include #include #include #include #include +#include #include #include #include @@ -66,6 +69,54 @@ class FixedCollectionStatistics final : public CollectionStatistics { float get_or_calculate_avg_dl(const std::wstring&) override { return 3.0F; } }; +// Holds the leader of a shared index query before it computes, until the test releases it, and +// counts the queries that lead or join a run. +class IndexFlightGate { +public: + static void leader(void* opaque) noexcept { + auto& gate = *static_cast(opaque); + std::unique_lock lock(gate._mutex); + ++gate._leaders; + gate._cv.notify_all(); + gate._cv.wait(lock, [&] { return gate._released; }); + } + + static void follower(void* opaque) noexcept { + auto& gate = *static_cast(opaque); + std::lock_guard lock(gate._mutex); + ++gate._followers; + gate._cv.notify_all(); + } + + void observe(InvertedIndexReader* reader) { + reader->set_single_flight_leader_before_compute_observer_for_test(&leader, this); + reader->set_single_flight_follower_joined_observer_for_test(&follower, this); + } + + bool wait_for(size_t leaders, size_t followers) { + std::unique_lock lock(_mutex); + return _cv.wait_for(lock, std::chrono::seconds(10), + [&] { return _leaders == leaders && _followers == followers; }); + } + + void release() { + std::lock_guard lock(_mutex); + _released = true; + _cv.notify_all(); + } + + // Read only after every query has returned. + size_t leaders() const { return _leaders; } + size_t followers() const { return _followers; } + +private: + std::mutex _mutex; + std::condition_variable _cv; + size_t _leaders = 0; + size_t _followers = 0; + bool _released = false; +}; + class InvertedIndexReaderTest : public testing::Test { public: const std::string kTestDir = "./ut_dir/inverted_index_reader_test"; @@ -3072,10 +3123,12 @@ class InvertedIndexReaderTest : public testing::Test { EXPECT_EQ(run.rows, (std::vector {4})); } - void test_fulltext_independent_queries() { + // Concurrent identical queries, as from the scanners a segment is split into, each lead or + // join one shared run unless the result cache answers them, and all get the same rows. + void test_fulltext_concurrent_queries_share_runs() { TabletIndex idx_meta; const std::string prefix = write_english_index( - "test_independent_queries", &idx_meta, 0, + "test_concurrent_queries", &idx_meta, 0, {Slice("apple banana"), Slice("cherry"), Slice("apple"), Slice("banana")}, 0); auto reader = FullTextIndexReader::create_shared(&idx_meta, open_v2_file(prefix)); std::atomic flight_calls {0}; @@ -3087,8 +3140,10 @@ class InvertedIndexReaderTest : public testing::Test { reader->set_single_flight_follower_joined_observer_for_test(count_flight, &flight_calls); for (const bool cache_enabled : {false, true}) { constexpr size_t kQueries = 3; + flight_calls = 0; std::array, kQueries> rows; std::array ok {}; + std::array cache_hits {}; const auto run = [&](size_t i) { SCOPED_INIT_THREAD_CONTEXT(); OlapReaderStatistics stats; @@ -3107,6 +3162,7 @@ class InvertedIndexReaderTest : public testing::Test { InvertedIndexQueryType::MATCH_ANY_QUERY, bitmap) .ok(); rows[i] = std::vector(bitmap->begin(), bitmap->end()); + cache_hits[i] = stats.inverted_index_query_cache_hit; }; std::vector threads; for (size_t i = 0; i < kQueries; ++i) { @@ -3115,12 +3171,104 @@ class InvertedIndexReaderTest : public testing::Test { for (auto& thread : threads) { thread.join(); } - EXPECT_EQ(flight_calls.load(), 0); + int64_t answered_by_cache = 0; for (size_t i = 0; i < kQueries; ++i) { EXPECT_TRUE(ok[i]); EXPECT_EQ(rows[i], (std::vector {0, 2})); + answered_by_cache += cache_hits[i]; } + // Each query leads or joins a run exactly once, or the cache answers it. + EXPECT_EQ(flight_calls.load() + answered_by_cache, static_cast(kQueries)); + EXPECT_TRUE(cache_enabled || answered_by_cache == 0); + } + reader->set_single_flight_leader_before_compute_observer_for_test(nullptr, nullptr); + reader->set_single_flight_follower_joined_observer_for_test(nullptr, nullptr); + } + + // A follower takes the leader's rows instead of running the query itself. + void test_fulltext_identical_queries_share_one_run() { + TabletIndex idx_meta; + const std::string prefix = write_english_index( + "test_shared_text_run", &idx_meta, 0, + {Slice("apple banana"), Slice("cherry"), Slice("apple"), Slice("banana")}, 0); + auto reader = FullTextIndexReader::create_shared(&idx_meta, open_v2_file(prefix)); + IndexFlightGate gate; + gate.observe(reader.get()); + std::array, 2> bitmaps; + std::array statuses; + const auto run = [&](size_t i) { + SCOPED_INIT_THREAD_CONTEXT(); + OlapReaderStatistics stats; + RuntimeState runtime_state; + TQueryOptions query_options; + query_options.enable_inverted_index_query_cache = false; + runtime_state.set_query_options(query_options); + io::IOContext io_ctx; + auto context = std::make_shared(); + context->io_ctx = &io_ctx; + context->stats = &stats; + context->runtime_state = &runtime_state; + bitmaps[i] = std::make_shared(); + statuses[i] = reader->query(context, "1", Field::create_field("apple"), + InvertedIndexQueryType::MATCH_ANY_QUERY, bitmaps[i]); + }; + std::thread leader(run, 0); + EXPECT_TRUE(gate.wait_for(1, 0)); + std::thread follower(run, 1); + EXPECT_TRUE(gate.wait_for(1, 1)); + gate.release(); + leader.join(); + follower.join(); + for (size_t i = 0; i < 2; ++i) { + ASSERT_TRUE(statuses[i].ok()) << statuses[i]; + EXPECT_EQ(std::vector(bitmaps[i]->begin(), bitmaps[i]->end()), + (std::vector {0, 2})); } + EXPECT_EQ(bitmaps[0], bitmaps[1]); + reader->set_single_flight_leader_before_compute_observer_for_test(nullptr, nullptr); + reader->set_single_flight_follower_joined_observer_for_test(nullptr, nullptr); + } + + // BKD queries are shared the same way. + void test_bkd_identical_queries_share_one_run() { + std::vector values = {42, 100, 42, 200, 300}; + TabletIndex idx_meta; + std::string index_path_prefix; + prepare_bkd_index("test_shared_bkd_run", 0, values, &idx_meta, &index_path_prefix); + auto reader = BkdIndexReader::create_shared(&idx_meta, open_v2_file(index_path_prefix)); + IndexFlightGate gate; + gate.observe(reader.get()); + std::array, 2> bitmaps; + std::array statuses; + const auto run = [&](size_t i) { + SCOPED_INIT_THREAD_CONTEXT(); + OlapReaderStatistics stats; + RuntimeState runtime_state; + TQueryOptions query_options; + query_options.enable_inverted_index_query_cache = false; + runtime_state.set_query_options(query_options); + io::IOContext io_ctx; + auto context = std::make_shared(); + context->io_ctx = &io_ctx; + context->stats = &stats; + context->runtime_state = &runtime_state; + bitmaps[i] = std::make_shared(); + statuses[i] = reader->query(context, "0", Field::create_field(100), + InvertedIndexQueryType::GREATER_THAN_QUERY, bitmaps[i]); + }; + std::thread leader(run, 0); + EXPECT_TRUE(gate.wait_for(1, 0)); + std::thread follower(run, 1); + EXPECT_TRUE(gate.wait_for(1, 1)); + gate.release(); + leader.join(); + follower.join(); + for (size_t i = 0; i < 2; ++i) { + ASSERT_TRUE(statuses[i].ok()) << statuses[i]; + EXPECT_EQ(std::vector(bitmaps[i]->begin(), bitmaps[i]->end()), + (std::vector {3, 4})); + } + EXPECT_EQ(bitmaps[0], bitmaps[1]); reader->set_single_flight_leader_before_compute_observer_for_test(nullptr, nullptr); reader->set_single_flight_follower_joined_observer_for_test(nullptr, nullptr); } @@ -4975,8 +5123,16 @@ TEST_F(InvertedIndexReaderTest, FulltextCountOnlyFastPathWithNulls) { test_fulltext_count_only_fastpath_with_nulls(); } -TEST_F(InvertedIndexReaderTest, FulltextIndependentQueries) { - test_fulltext_independent_queries(); +TEST_F(InvertedIndexReaderTest, FulltextConcurrentQueriesShareRuns) { + test_fulltext_concurrent_queries_share_runs(); +} + +TEST_F(InvertedIndexReaderTest, FulltextIdenticalQueriesShareOneRun) { + test_fulltext_identical_queries_share_one_run(); +} + +TEST_F(InvertedIndexReaderTest, BkdIdenticalQueriesShareOneRun) { + test_bkd_identical_queries_share_one_run(); } TEST_F(InvertedIndexReaderTest, FulltextScoringBypassesTheResultCache) { diff --git a/regression-test/data/inverted_index_p0/test_split_scanners_share_index_runs.out b/regression-test/data/inverted_index_p0/test_split_scanners_share_index_runs.out new file mode 100644 index 00000000000000..328c9214f12bfd --- /dev/null +++ b/regression-test/data/inverted_index_p0/test_split_scanners_share_index_runs.out @@ -0,0 +1,43 @@ +-- This file is automatically generated. You should know what you did if you want to edit this +-- !match_any -- +4096 33550336 + +-- !match_phrase -- +4096 33554432 + +-- !match_regexp -- +8192 67112960 + +-- !keyword -- +2341 19179813 + +-- !bkd -- +1467 12022065 + +-- !search_wildcard -- +8192 67112960 + +-- !compound -- +4588 37557024 + +-- !match_any -- +4096 33550336 + +-- !match_phrase -- +4096 33554432 + +-- !match_regexp -- +8192 67112960 + +-- !keyword -- +2341 19179813 + +-- !bkd -- +1467 12022065 + +-- !search_wildcard -- +8192 67112960 + +-- !compound -- +4588 37557024 + diff --git a/regression-test/suites/inverted_index_p0/test_split_scanners_share_index_runs.groovy b/regression-test/suites/inverted_index_p0/test_split_scanners_share_index_runs.groovy new file mode 100644 index 00000000000000..e07b49ff98fe4a --- /dev/null +++ b/regression-test/suites/inverted_index_p0/test_split_scanners_share_index_runs.groovy @@ -0,0 +1,86 @@ +// 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. + +// Every scanner of a split segment evaluates its index predicates over the whole segment, and +// concurrent identical index queries share one run. Each query runs with the segment split among +// eight scanners and with it read whole; both give the same rows. +suite("test_split_scanners_share_index_runs", "p0") { + sql "DROP TABLE IF EXISTS test_split_scanners_share_index_runs" + sql """ + CREATE TABLE test_split_scanners_share_index_runs ( + id INT, + num INT, + kw VARCHAR(32), + txt TEXT, + INDEX idx_num (num) USING INVERTED, + INDEX idx_kw (kw) USING INVERTED, + INDEX idx_txt (txt) USING INVERTED PROPERTIES("parser" = "english", "support_phrase" = "true") + ) ENGINE=OLAP + DUPLICATE KEY(id) + DISTRIBUTED BY HASH(id) BUCKETS 1 + PROPERTIES ( + "replication_allocation" = "tag.location.default: 1", + "inverted_index_storage_format" = "V2", + "disable_auto_compaction" = "true" + ) + """ + // One INSERT writes the tablet's only segment. + sql """ + INSERT INTO test_split_scanners_share_index_runs + SELECT number, number % 100, concat('k', number % 7), + concat('the ', elt(number % 4 + 1, 'quick', 'lazy', 'cached', 'caching'), + ' fox ', number % 13) + FROM numbers("number" = "16384") + """ + sql "sync" + + // Eight scanners of at least 1024 rows split the 16384-row segment; without parallel scan one + // scanner reads it whole. The result cache is off, so split scanners share runs, not entries. + def split = "enable_parallel_scan=true, parallel_scan_min_rows_per_scanner=1024, " + + "parallel_scan_max_scanners_count=8, enable_inverted_index_query_cache=false" + def whole = "enable_parallel_scan=false, enable_inverted_index_query_cache=false" + for (vars in [split, whole]) { + order_qt_match_any """ + SELECT /*+ SET_VAR(${vars}) */ count(*), sum(id) FROM test_split_scanners_share_index_runs + WHERE txt MATCH_ANY 'lazy' + """ + order_qt_match_phrase """ + SELECT /*+ SET_VAR(${vars}) */ count(*), sum(id) FROM test_split_scanners_share_index_runs + WHERE txt MATCH_PHRASE 'cached fox' + """ + order_qt_match_regexp """ + SELECT /*+ SET_VAR(${vars}) */ count(*), sum(id) FROM test_split_scanners_share_index_runs + WHERE txt MATCH_REGEXP 'cach.*' + """ + order_qt_keyword """ + SELECT /*+ SET_VAR(${vars}) */ count(*), sum(id) FROM test_split_scanners_share_index_runs + WHERE kw = 'k3' + """ + order_qt_bkd """ + SELECT /*+ SET_VAR(${vars}) */ count(*), sum(id) FROM test_split_scanners_share_index_runs + WHERE num > 90 + """ + order_qt_search_wildcard """ + SELECT /*+ SET_VAR(${vars}) */ count(*), sum(id) FROM test_split_scanners_share_index_runs + WHERE search('txt:*ach*') + """ + order_qt_compound """ + SELECT /*+ SET_VAR(${vars}) */ count(*), sum(id) FROM test_split_scanners_share_index_runs + WHERE txt MATCH_ANY 'quick' OR num < 5 + """ + } +}