add query handle to extend RuntimeFilterMergeControllerEntity lifecircuit

BiteTheDDDDt · BiteTheDDDDt · commit a43c10f302e9 · 2025-11-20T17:53:28.000+08:00
diff --git a/be/src/olap/id_manager.h b/be/src/olap/id_manager.h
@@ -195,13 +195,13 @@ class IdFileMap {
         return it->second;
     }
 
-    int64_t get_delayed_expired_timestamp() { return delayed_expired_timestamp; }
+    int64_t get_delayed_expired_timestamp() const { return delayed_expired_timestamp; }
 
     void set_external_scan_params(QueryContext* query_ctx, int max_file_scanners) {
         std::call_once(once_flag_for_external, [&] {
             DCHECK(query_ctx != nullptr);
             _query_global = query_ctx->get_query_globals();
-            _query_options = query_ctx->get_query_options();
+            _query_options = query_ctx->query_options();
             _file_scan_range_params_map = query_ctx->file_scan_range_params_map;
             _max_file_scanners = max_file_scanners;
         });
diff --git a/be/src/runtime/fragment_mgr.cpp b/be/src/runtime/fragment_mgr.cpp
@@ -69,13 +69,13 @@
 #include "runtime/frontend_info.h"
 #include "runtime/primitive_type.h"
 #include "runtime/query_context.h"
+#include "runtime/query_handle.h"
 #include "runtime/runtime_query_statistics_mgr.h"
 #include "runtime/runtime_state.h"
 #include "runtime/stream_load/new_load_stream_mgr.h"
 #include "runtime/stream_load/stream_load_context.h"
 #include "runtime/stream_load/stream_load_executor.h"
 #include "runtime/thread_context.h"
-#include "runtime/types.h"
 #include "runtime/workload_group/workload_group.h"
 #include "runtime/workload_group/workload_group_manager.h"
 #include "runtime_filter/runtime_filter_consumer.h"
@@ -728,6 +728,10 @@ Status FragmentMgr::_get_or_create_query_ctx(const TPipelineFragmentParams& para
                         query_ctx = QueryContext::create(query_id, _exec_env, params.query_options,
                                                          params.coord, params.is_nereids,
                                                          params.current_connect_fe, query_source);
+                        auto query_handle =
+                                std::make_shared<QueryHandle>(query_id, parent.query_options, query_ctx);
+                        _query_handle_map.insert(query_id, query_handle);
+                        query_ctx->set_query_handle(query_handle);
                         SCOPED_SWITCH_THREAD_MEM_TRACKER_LIMITER(query_ctx->query_mem_tracker());
                         RETURN_IF_ERROR(DescriptorTbl::create(
                                 &(query_ctx->obj_pool), params.desc_tbl, &(query_ctx->desc_tbl)));
@@ -755,9 +759,8 @@ Status FragmentMgr::_get_or_create_query_ctx(const TPipelineFragmentParams& para
                             if (info.__isset.runtime_filter_params) {
                                 auto handler =
                                         std::make_shared<RuntimeFilterMergeControllerEntity>();
-                                RETURN_IF_ERROR(
-                                        handler->init(query_ctx, info.runtime_filter_params));
-                                query_ctx->set_merge_controller_handler(handler);
+                                RETURN_IF_ERROR(handler->init(info.runtime_filter_params));
+                                query_handle->set_merge_controller_handler(handler);
 
                                 query_ctx->runtime_filter_mgr()->set_runtime_filter_params(
                                         info.runtime_filter_params);
@@ -787,6 +790,15 @@ std::string FragmentMgr::dump_pipeline_tasks(int64_t duration) {
         timespec now;
         clock_gettime(CLOCK_MONOTONIC, &now);
 
+        _query_handle_map.apply(
+                [&](phmap::flat_hash_map<TUniqueId, std::shared_ptr<QueryHandle>>& map) -> Status {
+                    for (auto& it : map) {
+                        fmt::format_to(debug_string_buffer, "QueryHandle: {}\n",
+                                       it.second->debug_string());
+                    }
+                    return Status::OK();
+                });
+
         _pipeline_map.apply([&](phmap::flat_hash_map<
                                     std::pair<TUniqueId, int>,
                                     std::shared_ptr<pipeline::PipelineFragmentContext>>& map)
@@ -804,13 +816,6 @@ std::string FragmentMgr::dump_pipeline_tasks(int64_t duration) {
                             debug_string_buffer, "QueryId: {}, global_runtime_filter_mgr: {}\n",
                             print_id(it.first.first),
                             it.second->get_query_ctx()->runtime_filter_mgr()->debug_string());
-
-                    if (it.second->get_query_ctx()->get_merge_controller_handler()) {
-                        fmt::format_to(debug_string_buffer, "{}\n",
-                                       it.second->get_query_ctx()
-                                               ->get_merge_controller_handler()
-                                               ->debug_string());
-                    }
                 }
 
                 auto timeout_second = it.second->timeout_second();
@@ -914,6 +919,7 @@ void FragmentMgr::cancel_query(const TUniqueId query_id, const Status reason) {
     }
     query_ctx->cancel(reason);
     remove_query_context(query_id);
+    _query_handle_map.erase(query_id);
     LOG(INFO) << "Query " << print_id(query_id)
               << " is cancelled and removed. Reason: " << reason.to_string();
 }
@@ -1310,8 +1316,8 @@ Status FragmentMgr::send_filter_size(const PSendFilterSizeRequest* request) {
         return Status::EndOfFile("inject FragmentMgr::send_filter_size.return_eof");
     }
 
-    if (auto q_ctx = get_query_ctx(query_id)) {
-        return q_ctx->get_merge_controller_handler()->send_filter_size(q_ctx, request);
+    if (auto q_handle = _query_handle_map.find(query_id); q_handle != nullptr) {
+        return q_handle->get_merge_controller_handler()->send_filter_size(q_handle, request);
     } else {
         return Status::EndOfFile(
                 "Send filter size failed: Query context (query-id: {}) not found, maybe "
@@ -1347,12 +1353,11 @@ Status FragmentMgr::merge_filter(const PMergeFilterRequest* request,
     TUniqueId query_id;
     query_id.__set_hi(queryid.hi);
     query_id.__set_lo(queryid.lo);
-    if (auto q_ctx = get_query_ctx(query_id)) {
-        SCOPED_ATTACH_TASK(q_ctx.get());
-        if (!q_ctx->get_merge_controller_handler()) {
+    if (auto q_handle = _query_handle_map.find(query_id); q_handle != nullptr) {
+        if (!q_handle->get_merge_controller_handler()) {
             return Status::InternalError("Merge filter failed: Merge controller handler is null");
         }
-        return q_ctx->get_merge_controller_handler()->merge(q_ctx, request, attach_data);
+        return q_handle->get_merge_controller_handler()->merge(q_handle, request, attach_data);
     } else {
         return Status::EndOfFile(
                 "Merge filter size failed: Query context (query-id: {}) already finished",
diff --git a/be/src/runtime/fragment_mgr.h b/be/src/runtime/fragment_mgr.h
@@ -53,6 +53,7 @@ namespace pipeline {
 class PipelineFragmentContext;
 } // namespace pipeline
 class QueryContext;
+class QueryHandle;
 class ExecEnv;
 class ThreadPool;
 class PExecPlanFragmentStartRequest;
@@ -209,6 +210,7 @@ class FragmentMgr : public RestMonitorIface {
 
     // query id -> QueryContext
     ConcurrentContextMap<TUniqueId, std::weak_ptr<QueryContext>, QueryContext> _query_ctx_map;
+    ConcurrentContextMap<TUniqueId, std::shared_ptr<QueryHandle>, QueryHandle> _query_handle_map;
 
     CountDownLatch _stop_background_threads_latch;
     std::shared_ptr<Thread> _cancel_thread;
diff --git a/be/src/runtime/query_context.cpp b/be/src/runtime/query_context.cpp
@@ -23,7 +23,6 @@
 #include <gen_cpp/Types_types.h>
 #include <glog/logging.h>
 
-#include <algorithm>
 #include <exception>
 #include <memory>
 #include <mutex>
@@ -32,7 +31,6 @@
 
 #include "common/logging.h"
 #include "common/status.h"
-#include "olap/olap_common.h"
 #include "pipeline/dependency.h"
 #include "pipeline/pipeline_fragment_context.h"
 #include "runtime/exec_env.h"
@@ -43,10 +41,8 @@
 #include "runtime/thread_context.h"
 #include "runtime/workload_group/workload_group_manager.h"
 #include "runtime/workload_management/query_task_controller.h"
-#include "runtime_filter/runtime_filter_definitions.h"
 #include "util/mem_info.h"
 #include "util/uid_util.h"
-#include "vec/spill/spill_stream_manager.h"
 
 namespace doris {
 
@@ -106,8 +102,6 @@ QueryContext::QueryContext(TUniqueId query_id, ExecEnv* exec_env,
     _memory_sufficient_dependency =
             pipeline::Dependency::create_unique(-1, -1, "MemorySufficientDependency", true);
 
-    _runtime_filter_mgr = std::make_unique<RuntimeFilterMgr>(true);
-
     _timeout_second = query_options.execution_timeout;
 
     bool is_query_type_valid = query_options.query_type == TQueryType::SELECT ||
@@ -231,10 +225,6 @@ QueryContext::~QueryContext() {
     _runtime_predicates.clear();
     file_scan_range_params_map.clear();
     obj_pool.clear();
-    if (_merge_controller_handler) {
-        _merge_controller_handler->release_undone_filters(this);
-    }
-    _merge_controller_handler.reset();
 
     DorisMetrics::instance()->query_ctx_cnt->increment(-1);
     ExecEnv::GetInstance()->fragment_mgr()->remove_query_context(this->_query_id);
diff --git a/be/src/runtime/query_context.h b/be/src/runtime/query_context.h
@@ -22,7 +22,6 @@
 #include <gen_cpp/Types_types.h>
 #include <glog/logging.h>
 
-#include <atomic>
 #include <cstdint>
 #include <memory>
 #include <mutex>
@@ -34,16 +33,16 @@
 #include "common/object_pool.h"
 #include "runtime/exec_env.h"
 #include "runtime/memory/mem_tracker_limiter.h"
+#include "runtime/query_handle.h"
 #include "runtime/runtime_predicate.h"
 #include "runtime/workload_management/resource_context.h"
-#include "runtime_filter/runtime_filter_mgr.h"
-#include "util/hash_util.hpp"
-#include "util/threadpool.h"
 #include "vec/exec/scan/scanner_scheduler.h"
 #include "workload_group/workload_group.h"
 
 namespace doris {
 
+class QueryHandle;
+
 namespace pipeline {
 class PipelineFragmentContext;
 class PipelineTask;
@@ -177,12 +176,6 @@ class QueryContext : public std::enable_shared_from_this<QueryContext> {
         return _query_options.__isset.fe_process_uuid ? _query_options.fe_process_uuid : 0;
     }
 
-    bool ignore_runtime_filter_error() const {
-        return _query_options.__isset.ignore_runtime_filter_error
-                       ? _query_options.ignore_runtime_filter_error
-                       : false;
-    }
-
     bool enable_force_spill() const {
         return _query_options.__isset.enable_force_spill && _query_options.enable_force_spill;
     }
@@ -214,14 +207,6 @@ class QueryContext : public std::enable_shared_from_this<QueryContext> {
 
     doris::pipeline::TaskScheduler* get_pipe_exec_scheduler();
 
-    void set_merge_controller_handler(
-            std::shared_ptr<RuntimeFilterMergeControllerEntity>& handler) {
-        _merge_controller_handler = handler;
-    }
-    std::shared_ptr<RuntimeFilterMergeControllerEntity> get_merge_controller_handler() const {
-        return _merge_controller_handler;
-    }
-
     bool is_nereids() const { return _is_nereids; }
 
     WorkloadGroupPtr workload_group() const { return _resource_ctx->workload_group(); }
@@ -299,6 +284,10 @@ class QueryContext : public std::enable_shared_from_this<QueryContext> {
     void set_first_error_msg(std::string error_msg);
     std::string get_first_error_msg();
 
+    void set_query_handle(std::shared_ptr<QueryHandle> query_handle) {
+        _query_handle = query_handle;
+    }
+
 private:
     friend class QueryTaskController;
 
@@ -330,10 +319,6 @@ class QueryContext : public std::enable_shared_from_this<QueryContext> {
     // This dependency indicates if memory is sufficient to execute.
     std::unique_ptr<pipeline::Dependency> _memory_sufficient_dependency;
 
-    // This shared ptr is never used. It is just a reference to hold the object.
-    // There is a weak ptr in runtime filter manager to reference this object.
-    std::shared_ptr<RuntimeFilterMergeControllerEntity> _merge_controller_handler;
-
     std::map<int, std::weak_ptr<pipeline::PipelineFragmentContext>> _fragment_id_to_pipeline_ctx;
     std::mutex _pipeline_map_write_lock;
 
@@ -377,6 +362,8 @@ class QueryContext : public std::enable_shared_from_this<QueryContext> {
     std::string _load_error_url;
     std::string _first_error_msg;
 
+    std::shared_ptr<QueryHandle> _query_handle;
+
 public:
     // when fragment of pipeline is closed, it will register its profile to this map by using add_fragment_profile
     void add_fragment_profile(
@@ -392,8 +379,6 @@ class QueryContext : public std::enable_shared_from_this<QueryContext> {
 
     timespec get_query_arrival_timestamp() const { return this->_query_arrival_timestamp; }
     QuerySource get_query_source() const { return this->_query_source; }
-
-    const TQueryOptions get_query_options() const { return _query_options; }
 };
 
 } // namespace doris
diff --git a/be/src/runtime/query_handle.h b/be/src/runtime/query_handle.h
@@ -0,0 +1,82 @@
+// 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.
+
+#pragma once
+
+#include <gen_cpp/PaloInternalService_types.h>
+#include <gen_cpp/Types_types.h>
+
+#include <memory>
+#include <string>
+#include <utility>
+
+#include "runtime/query_context.h"
+#include "runtime_filter/runtime_filter_mgr.h"
+
+namespace doris {
+
+class QueryHandle;
+
+// The role of QueryHandle is similar to QueryContext, but its lifecycle is longer than QueryContext.
+// QueryHandle will exist until the entire query ends, rather than being released as the current BE task ends like QueryContext.
+// It is mainly used to store runtime states that need coordination between BEs, such as the MergeControllerHandler of RuntimeFilter.
+// This way, even if the QueryContext of one BE has been released, other BEs can still access these coordination states through QueryHandle to ensure the correctness and consistency of the query.
+// QueryContext hold shared_ptr of QueryHandle, and QueryHandle hold weak_ptr of QueryContext to avoid circular references.
+class QueryHandle {
+public:
+    QueryHandle(TUniqueId query_id, TQueryOptions query_options,
+                std::weak_ptr<QueryContext> query_ctx)
+            : _query_id(std::move(query_id)),
+              _query_options(std::move(query_options)),
+              _query_ctx(std::move(query_ctx)) {}
+    ~QueryHandle() = default;
+
+    std::string debug_string() const {
+        return fmt::format(
+                "QueryHandle(query_id={}): {}", print_id(_query_id),
+                _merge_controller_handler ? _merge_controller_handler->debug_string() : "null");
+    }
+
+    void set_merge_controller_handler(
+            std::shared_ptr<RuntimeFilterMergeControllerEntity>& handler) {
+        _merge_controller_handler = handler;
+    }
+    std::shared_ptr<RuntimeFilterMergeControllerEntity> get_merge_controller_handler() const {
+        return _merge_controller_handler;
+    }
+
+    const TQueryOptions& query_options() const { return _query_options; }
+
+    std::weak_ptr<QueryContext> weak_query_ctx() const { return _query_ctx; }
+
+    int execution_timeout() const {
+        return _query_options.__isset.execution_timeout ? _query_options.execution_timeout
+                                                        : _query_options.query_timeout;
+    }
+
+    TUniqueId query_id() const { return _query_id; }
+
+private:
+    TUniqueId _query_id;
+    TQueryOptions _query_options;
+    std::weak_ptr<QueryContext> _query_ctx;
+    // This shared ptr is never used. It is just a reference to hold the object.
+    // There is a weak ptr in runtime filter manager to reference this object.
+    std::shared_ptr<RuntimeFilterMergeControllerEntity> _merge_controller_handler;
+};
+
+} // namespace doris
diff --git a/be/src/runtime_filter/runtime_filter_merger.h b/be/src/runtime_filter/runtime_filter_merger.h
@@ -35,9 +35,9 @@ class RuntimeFilterMerger : public RuntimeFilter {
         READY // Collecting all products(_received_producer_num == _expected_producer_num) will transfer to this state, and filter is already available
     };
 
-    static Status create(const QueryContext* query_ctx, const TRuntimeFilterDesc* desc,
+    static Status create(const TRuntimeFilterDesc* desc,
                          std::shared_ptr<RuntimeFilterMerger>* res) {
-        *res = std::shared_ptr<RuntimeFilterMerger>(new RuntimeFilterMerger(query_ctx, desc));
+        *res = std::shared_ptr<RuntimeFilterMerger>(new RuntimeFilterMerger(desc));
         vectorized::VExprContextSPtr build_ctx;
         RETURN_IF_ERROR(vectorized::VExpr::create_expr_tree(desc->src_expr, build_ctx));
         (*res)->_wrapper = std::make_shared<RuntimeFilterWrapper>(
@@ -104,7 +104,7 @@ class RuntimeFilterMerger : public RuntimeFilter {
     }
 
 private:
-    RuntimeFilterMerger(const QueryContext* query_ctx, const TRuntimeFilterDesc* desc)
+    RuntimeFilterMerger(const TRuntimeFilterDesc* desc)
             : RuntimeFilter(desc), _rf_state(State::WAITING_FOR_PRODUCT) {}
 
     std::atomic<State> _rf_state;
diff --git a/be/src/runtime_filter/runtime_filter_mgr.cpp b/be/src/runtime_filter/runtime_filter_mgr.cpp
diff --git a/be/src/runtime_filter/runtime_filter_mgr.h b/be/src/runtime_filter/runtime_filter_mgr.h
diff --git a/be/src/runtime_filter/runtime_filter_producer.h b/be/src/runtime_filter/runtime_filter_producer.h
