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 + """ + } +}