update

BiteTheDDDDt · BiteTheDDDDt · commit cb4fc020dc2b · 2025-11-24T20:52:26.000+08:00
diff --git a/be/src/runtime/coordinator_context.cpp b/be/src/runtime/coordinator_context.cpp
@@ -15,56 +15,59 @@
 // specific language governing permissions and limitations
 // under the License.
 
-#include "runtime/query_handle.h"
+#include "runtime/coordinator_context.h"
 
 #include "runtime/query_context.h"
 
 namespace doris {
 
-QueryHandle::QueryHandle(TUniqueId query_id, TQueryOptions query_options,
-                         std::weak_ptr<QueryContext> query_ctx)
+CoordinatorContext::CoordinatorContext(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)) {
     CHECK(_query_ctx.lock() != nullptr);
     _resource_ctx = _query_ctx.lock()->_resource_ctx;
 }
 
-QueryHandle::~QueryHandle() = default;
+CoordinatorContext::~CoordinatorContext() {
+    SCOPED_ATTACH_TASK(_resource_ctx);
+    _merge_controller_handler.reset();
+}
 
-std::string QueryHandle::debug_string() const {
+std::string CoordinatorContext::debug_string() const {
     return fmt::format(
-            "QueryHandle(query_id={}): {}", print_id(_query_id),
+            "CoordinatorContext(query_id={}): {}", print_id(_query_id),
             _merge_controller_handler ? _merge_controller_handler->debug_string() : "null");
 }
 
-void QueryHandle::set_merge_controller_handler(
+void CoordinatorContext::set_merge_controller_handler(
         std::shared_ptr<RuntimeFilterMergeControllerEntity>& handler) {
     _merge_controller_handler = handler;
 }
-std::shared_ptr<RuntimeFilterMergeControllerEntity> QueryHandle::get_merge_controller_handler()
-        const {
+std::shared_ptr<RuntimeFilterMergeControllerEntity>
+CoordinatorContext::get_merge_controller_handler() const {
     return _merge_controller_handler;
 }
 
-const TQueryOptions& QueryHandle::query_options() const {
+const TQueryOptions& CoordinatorContext::query_options() const {
     return _query_options;
 }
 
-std::weak_ptr<QueryContext> QueryHandle::weak_query_ctx() const {
+std::weak_ptr<QueryContext> CoordinatorContext::weak_query_ctx() const {
     return _query_ctx;
 }
 
-int QueryHandle::execution_timeout() const {
+int CoordinatorContext::execution_timeout() const {
     return _query_options.__isset.execution_timeout ? _query_options.execution_timeout
                                                     : _query_options.query_timeout;
 }
 
-TUniqueId QueryHandle::query_id() const {
+TUniqueId CoordinatorContext::query_id() const {
     return _query_id;
 }
 
-std::shared_ptr<ResourceContext> QueryHandle::resource_ctx() const {
+std::shared_ptr<ResourceContext> CoordinatorContext::resource_ctx() const {
     return _resource_ctx;
 }
 
diff --git a/be/src/runtime/coordinator_context.h b/be/src/runtime/coordinator_context.h
@@ -22,25 +22,25 @@
 
 #include <memory>
 #include <string>
-#include <utility>
 
 #include "runtime/query_context.h"
 #include "runtime_filter/runtime_filter_mgr.h"
 
 namespace doris {
 
-class QueryHandle;
+class CoordinatorContext;
+class ResourceContext;
 
-// 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.
+// The role of CoordinatorContext is similar to QueryContext, but its lifecycle is longer than QueryContext.
+// CoordinatorContext 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 {
+// This way, even if the QueryContext of one BE has been released, other BEs can still access these coordination states through CoordinatorContext to ensure the correctness and consistency of the query.
+// QueryContext hold shared_ptr of CoordinatorContext, and CoordinatorContext hold weak_ptr of QueryContext to avoid circular references.
+class CoordinatorContext {
 public:
-    QueryHandle(TUniqueId query_id, TQueryOptions query_options,
-                std::weak_ptr<QueryContext> query_ctx);
-    ~QueryHandle();
+    CoordinatorContext(TUniqueId query_id, TQueryOptions query_options,
+                       std::weak_ptr<QueryContext> query_ctx);
+    ~CoordinatorContext();
 
     std::string debug_string() const;
 
diff --git a/be/src/runtime/fragment_mgr.cpp b/be/src/runtime/fragment_mgr.cpp
@@ -64,12 +64,12 @@
 #include "io/fs/stream_load_pipe.h"
 #include "pipeline/pipeline_fragment_context.h"
 #include "runtime/client_cache.h"
+#include "runtime/coordinator_context.h"
 #include "runtime/descriptors.h"
 #include "runtime/exec_env.h"
 #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"
@@ -728,10 +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>(
+                        auto coordinator_context = std::make_shared<CoordinatorContext>(
                                 query_id, parent.query_options, query_ctx);
-                        _query_handle_map.insert(query_id, query_handle);
-                        query_ctx->set_query_handle(query_handle);
+                        _coordinator_context_map.insert(query_id, coordinator_context);
+                        query_ctx->set_coordinator_context(coordinator_context);
                         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)));
@@ -760,7 +760,7 @@ Status FragmentMgr::_get_or_create_query_ctx(const TPipelineFragmentParams& para
                                 auto handler =
                                         std::make_shared<RuntimeFilterMergeControllerEntity>();
                                 RETURN_IF_ERROR(handler->init(info.runtime_filter_params));
-                                query_handle->set_merge_controller_handler(handler);
+                                coordinator_context->set_merge_controller_handler(handler);
 
                                 query_ctx->runtime_filter_mgr()->set_runtime_filter_params(
                                         info.runtime_filter_params);
@@ -790,10 +790,11 @@ 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 {
+        _coordinator_context_map.apply(
+                [&](phmap::flat_hash_map<TUniqueId, std::shared_ptr<CoordinatorContext>>& map)
+                        -> Status {
                     for (auto& it : map) {
-                        fmt::format_to(debug_string_buffer, "QueryHandle: {}\n",
+                        fmt::format_to(debug_string_buffer, "CoordinatorContext: {}\n",
                                        it.second->debug_string());
                     }
                     return Status::OK();
@@ -920,7 +921,7 @@ void FragmentMgr::cancel_query(const TUniqueId query_id, const Status reason) {
     SCOPED_ATTACH_TASK(query_ctx->resource_ctx());
     query_ctx->cancel(reason);
     remove_query_context(query_id);
-    _query_handle_map.erase(query_id);
+    _coordinator_context_map.erase(query_id);
     LOG(INFO) << "Query " << print_id(query_id)
               << " is cancelled and removed. Reason: " << reason.to_string();
 }
@@ -1317,7 +1318,7 @@ Status FragmentMgr::send_filter_size(const PSendFilterSizeRequest* request) {
         return Status::EndOfFile("inject FragmentMgr::send_filter_size.return_eof");
     }
 
-    if (auto q_handle = _query_handle_map.find(query_id); q_handle != nullptr) {
+    if (auto q_handle = _coordinator_context_map.find(query_id); q_handle != nullptr) {
         return q_handle->get_merge_controller_handler()->send_filter_size(q_handle, request);
     } else {
         return Status::EndOfFile(
@@ -1354,7 +1355,7 @@ 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_handle = _query_handle_map.find(query_id); q_handle != nullptr) {
+    if (auto q_handle = _coordinator_context_map.find(query_id); q_handle != nullptr) {
         SCOPED_ATTACH_TASK(q_handle->resource_ctx());
         if (!q_handle->get_merge_controller_handler()) {
             return Status::InternalError("Merge filter failed: Merge controller handler is null");
diff --git a/be/src/runtime/fragment_mgr.h b/be/src/runtime/fragment_mgr.h
@@ -53,7 +53,7 @@ namespace pipeline {
 class PipelineFragmentContext;
 } // namespace pipeline
 class QueryContext;
-class QueryHandle;
+class CoordinatorContext;
 class ExecEnv;
 class ThreadPool;
 class PExecPlanFragmentStartRequest;
@@ -210,7 +210,8 @@ 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;
+    ConcurrentContextMap<TUniqueId, std::shared_ptr<CoordinatorContext>, CoordinatorContext>
+            _coordinator_context_map;
 
     CountDownLatch _stop_background_threads_latch;
     std::shared_ptr<Thread> _cancel_thread;
diff --git a/be/src/runtime/query_context.h b/be/src/runtime/query_context.h
@@ -31,17 +31,17 @@
 #include "common/config.h"
 #include "common/factory_creator.h"
 #include "common/object_pool.h"
+#include "runtime/coordinator_context.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 "vec/exec/scan/scanner_scheduler.h"
 #include "workload_group/workload_group.h"
 
 namespace doris {
 
-class QueryHandle;
+class CoordinatorContext;
 
 namespace pipeline {
 class PipelineFragmentContext;
@@ -284,13 +284,13 @@ 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;
+    void set_coordinator_context(std::shared_ptr<CoordinatorContext> coordinator_context) {
+        _coordinator_context = coordinator_context;
     }
 
 private:
     friend class QueryTaskController;
-    friend class QueryHandle;
+    friend class CoordinatorContext;
 
     int _timeout_second;
     TUniqueId _query_id;
@@ -363,7 +363,7 @@ 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;
+    std::shared_ptr<CoordinatorContext> _coordinator_context;
 
 public:
     // when fragment of pipeline is closed, it will register its profile to this map by using add_fragment_profile
diff --git a/be/src/runtime_filter/runtime_filter_mgr.cpp b/be/src/runtime_filter/runtime_filter_mgr.cpp
@@ -32,10 +32,10 @@
 #include "common/config.h"
 #include "common/logging.h"
 #include "common/status.h"
+#include "runtime/coordinator_context.h"
 #include "runtime/exec_env.h"
 #include "runtime/memory/mem_tracker.h"
 #include "runtime/query_context.h"
-#include "runtime/query_handle.h"
 #include "runtime/runtime_state.h"
 #include "runtime/thread_context.h"
 #include "runtime_filter/runtime_filter_consumer.h"
@@ -216,7 +216,8 @@ Status RuntimeFilterMergeControllerEntity::init(const TRuntimeFilterParams& runt
 }
 
 Status RuntimeFilterMergeControllerEntity::send_filter_size(
-        std::shared_ptr<QueryHandle> query_handle, const PSendFilterSizeRequest* request) {
+        std::shared_ptr<CoordinatorContext> coordinator_context,
+        const PSendFilterSizeRequest* request) {
     SCOPED_CONSUME_MEM_TRACKER(_mem_tracker);
 
     auto filter_id = request->filter_id();
@@ -236,9 +237,9 @@ Status RuntimeFilterMergeControllerEntity::send_filter_size(
     Status st = Status::OK();
     // After all runtime filters' size are collected, we should send response to all producers.
     if (cnt_val.merger->add_rf_size(request->filter_size())) {
-        auto ctx = query_handle->query_options().ignore_runtime_filter_error
+        auto ctx = coordinator_context->query_options().ignore_runtime_filter_error
                            ? std::weak_ptr<QueryContext> {}
-                           : query_handle->weak_query_ctx();
+                           : coordinator_context->weak_query_ctx();
         for (auto addr : cnt_val.source_addrs) {
             std::shared_ptr<PBackendService_Stub> stub(
                     ExecEnv::GetInstance()->brpc_internal_client_cache()->get_client(addr));
@@ -255,10 +256,10 @@ Status RuntimeFilterMergeControllerEntity::send_filter_size(
                                   DummyBrpcCallback<PSyncFilterSizeResponse>::create_shared(), ctx);
 
             auto* pquery_id = closure->request_->mutable_query_id();
-            pquery_id->set_hi(query_handle->query_id().hi);
-            pquery_id->set_lo(query_handle->query_id().lo);
+            pquery_id->set_hi(coordinator_context->query_id().hi);
+            pquery_id->set_lo(coordinator_context->query_id().lo);
             closure->cntl_->set_timeout_ms(
-                    get_execution_rpc_timeout_ms(query_handle->execution_timeout()));
+                    get_execution_rpc_timeout_ms(coordinator_context->execution_timeout()));
             if (config::execution_ignore_eovercrowded) {
                 closure->cntl_->ignore_eovercrowded();
             }
@@ -302,9 +303,9 @@ std::string RuntimeFilterMgr::debug_string() {
 }
 
 // merge data
-Status RuntimeFilterMergeControllerEntity::merge(std::shared_ptr<QueryHandle> query_handle,
-                                                 const PMergeFilterRequest* request,
-                                                 butil::IOBufAsZeroCopyInputStream* attach_data) {
+Status RuntimeFilterMergeControllerEntity::merge(
+        std::shared_ptr<CoordinatorContext> coordinator_context, const PMergeFilterRequest* request,
+        butil::IOBufAsZeroCopyInputStream* attach_data) {
     SCOPED_CONSUME_MEM_TRACKER(_mem_tracker);
     int64_t merge_time = 0;
     auto filter_id = request->filter_id();
@@ -327,7 +328,7 @@ Status RuntimeFilterMergeControllerEntity::merge(std::shared_ptr<QueryHandle> qu
             return Status::OK();
         }
         std::shared_ptr<RuntimeFilterProducer> tmp_filter;
-        RETURN_IF_ERROR(RuntimeFilterProducer::create(query_handle->query_options(),
+        RETURN_IF_ERROR(RuntimeFilterProducer::create(coordinator_context->query_options(),
                                                       &cnt_val.runtime_filter_desc, &tmp_filter));
 
         RETURN_IF_ERROR(tmp_filter->assign(*request, attach_data));
@@ -340,11 +341,11 @@ Status RuntimeFilterMergeControllerEntity::merge(std::shared_ptr<QueryHandle> qu
 
     if (is_ready) {
         return _send_rf_to_target(cnt_val,
-                                  query_handle->query_options().ignore_runtime_filter_error
+                                  coordinator_context->query_options().ignore_runtime_filter_error
                                           ? std::weak_ptr<QueryContext> {}
-                                          : query_handle->weak_query_ctx(),
+                                          : coordinator_context->weak_query_ctx(),
                                   merge_time, request->query_id(),
-                                  query_handle->execution_timeout());
+                                  coordinator_context->execution_timeout());
     }
     return Status::OK();
 }
diff --git a/be/src/runtime_filter/runtime_filter_mgr.h b/be/src/runtime_filter/runtime_filter_mgr.h
@@ -55,7 +55,7 @@ class RuntimeFilterWrapper;
 class QueryContext;
 class ExecEnv;
 class RuntimeProfile;
-class QueryHandle;
+class CoordinatorContext;
 
 struct LocalMergeContext {
     std::mutex mtx;
@@ -142,10 +142,11 @@ class RuntimeFilterMergeControllerEntity {
     Status init(const TRuntimeFilterParams& runtime_filter_params);
 
     // handle merge rpc
-    Status merge(std::shared_ptr<QueryHandle> query_handle, const PMergeFilterRequest* request,
+    Status merge(std::shared_ptr<CoordinatorContext> coordinator_context,
+                 const PMergeFilterRequest* request,
                  butil::IOBufAsZeroCopyInputStream* attach_data);
 
-    Status send_filter_size(std::shared_ptr<QueryHandle> query_handle,
+    Status send_filter_size(std::shared_ptr<CoordinatorContext> coordinator_context,
                             const PSendFilterSizeRequest* request);
 
     std::string debug_string();
diff --git a/be/test/runtime_filter/runtime_filter_test_utils.h b/be/test/runtime_filter/runtime_filter_test_utils.h
@@ -38,9 +38,9 @@ class RuntimeFilterTest : public testing::Test {
         _query_ctx =
                 QueryContext::create(TUniqueId(), ExecEnv::GetInstance(), _query_options,
                                      fe_address, true, fe_address, QuerySource::INTERNAL_FRONTEND);
-        _query_handle =
-                std::make_shared<QueryHandle>(_query_ctx->query_id(), _query_options, _query_ctx);
-        _query_ctx->set_query_handle(_query_handle);
+        _coordinator_context = std::make_shared<CoordinatorContext>(_query_ctx->query_id(),
+                                                                    _query_options, _query_ctx);
+        _query_ctx->set_coordinator_context(_coordinator_context);
         _query_ctx->runtime_filter_mgr()->set_runtime_filter_params(
                 TRuntimeFilterParamsBuilder().build());
 
@@ -61,7 +61,7 @@ class RuntimeFilterTest : public testing::Test {
 protected:
     RuntimeProfile _profile = RuntimeProfile("");
     std::shared_ptr<QueryContext> _query_ctx;
-    std::shared_ptr<QueryHandle> _query_handle;
+    std::shared_ptr<CoordinatorContext> _coordinator_context;
     TQueryOptions _query_options;
     const std::string LOCALHOST = BackendOptions::get_localhost();
     const int DUMMY_PORT = config::brpc_port;
