[opt](inverted index) Share index runs among the scanners of a split segment

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<roaring::Roaring>(*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<std::shared_ptr<const roaring::Roaring>>;
+
+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<std::string, IndexFieldNameAndTypePair>& 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<const roaring::Roaring> 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<roaring::Roaring>(*rows),
+                                                          std::make_shared<roaring::Roaring>());
+                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<IndexQueryContext> 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<roaring::Roaring>());
+    if (flight_leader) {
+        flight_rows = std::make_shared<const roaring::Roaring>(*bitmap_result.get_data_bitmap());
+    }
 
     if (dsl_cache && cache_usable) {
         insert_dsl_result(dsl_cache, seg_prefix, dsl_sig, bitmap_result);

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,

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<roaring::Roaring>* result) {
+                    *result = std::make_shared<roaring::Roaring>();
+                    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();

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<std::shared_ptr<roaring::Roaring>>;
+
+QueryFlight& query_single_flight() {
+    static QueryFlight flight;
+    return flight;
+}
+
+} // namespace
+
+Status InvertedIndexReader::run_query_single_flight(
+        const std::string& key, std::shared_ptr<roaring::Roaring>* result,
+        const std::function<Status(std::shared_ptr<roaring::Roaring>*)>& 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<roaring::Roaring> 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<roaring::Roaring> 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);
 }

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<roaring::Roaring>* result,
+            const std::function<Status(std::shared_ptr<roaring::Roaring>*)>& compute);
+
     friend class InvertedIndexIterator;
     std::shared_ptr<IndexFileReader> _index_file_reader;
     TabletIndex _index_meta;

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<std::pair<Status, std::shared_ptr<roaring::Roaring>>>;
-
-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 <typename Compute>
-Status run_query_single_flight(
-        QueryFlight& flight, const std::string& key, std::shared_ptr<roaring::Roaring>* 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<roaring::Roaring> 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<logical::Term>()) {
@@ -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<ErrorCode::INVERTED_INDEX_NOT_SUPPORTED>(
                 "{} 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<roaring::Roaring> 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<roaring::Roaring>* shared) {
                                            RETURN_IF_ERROR(run(shared));
                                            insert_query_cache(context, cache, cache_key, *shared,