[Chore](query) add _query_ctx_map_delay_delete (#59262)

This pull request introduces a delayed deletion mechanism for
`QueryContext` objects in the fragment manager to ensure that runtime
filter merging works correctly after a query reaches its end-of-stream
(EOS). The main changes involve adding a new map to retain query
contexts temporarily, modifying the lifecycle management of query
contexts, and introducing utility methods to support this behavior.

**Query Context Lifecycle Management:**

* Added a new member `_query_ctx_map_delay_delete` to `FragmentMgr` to
keep `QueryContext` objects alive for a short period after query EOS,
ensuring that the runtime filter coordinator can complete its work.
* Modified `FragmentMgr::remove_query_context` to erase entries from
`_query_ctx_map_delay_delete` as part of the cleanup process.
* Updated the `QueryContext` destructor to call `remove_query_context`
only if the fragment manager is available, preventing issues during unit
tests.

**Runtime Filter Merge Coordination:**

* In `FragmentMgr::_get_or_create_query_ctx`, now inserts the
`QueryContext` into `_query_ctx_map_delay_delete` if the runtime filter
merge handler is not empty, ensuring the context is retained for
merging.
* Added an `empty()` method to `RuntimeFilterMergeControllerEntity` to
check if the filter map is empty, which is used to determine if delayed
deletion is needed.

Author: BiteTheDDDDt

be/src/runtime/fragment_mgr.cpp

@@ -670,6 +670,7 @@ void FragmentMgr::remove_pipeline_context(std::pair<TUniqueId, int> key) {
 }
 
 void FragmentMgr::remove_query_context(const TUniqueId& key) {
+    _query_ctx_map_delay_delete.erase(key);
 #ifndef BE_TEST
     _query_ctx_map.erase(key);
 #endif
@@ -768,6 +769,9 @@ Status FragmentMgr::_get_or_create_query_ctx(const TPipelineFragmentParams& para
 
                                 query_ctx->runtime_filter_mgr()->set_runtime_filter_params(
                                         info.runtime_filter_params);
+                                if (!handler->empty()) {
+                                    _query_ctx_map_delay_delete.insert(query_id, query_ctx);
+                                }
                             }
                             if (info.__isset.topn_filter_descs) {
                                 query_ctx->init_runtime_predicates(info.topn_filter_descs);

be/src/runtime/fragment_mgr.h

@@ -209,6 +209,9 @@ class FragmentMgr : public RestMonitorIface {
 
     // query id -> QueryContext
     ConcurrentContextMap<TUniqueId, std::weak_ptr<QueryContext>, QueryContext> _query_ctx_map;
+    // keep query ctx do not delete immediately to make rf coordinator merge filter work well after query eos
+    ConcurrentContextMap<TUniqueId, std::shared_ptr<QueryContext>, QueryContext>
+            _query_ctx_map_delay_delete;
 
     CountDownLatch _stop_background_threads_latch;
     std::shared_ptr<Thread> _cancel_thread;

be/src/runtime/query_context.cpp

@@ -231,13 +231,13 @@ 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);
+    // fragment_mgr is nullptr in unittest
+    if (ExecEnv::GetInstance()->fragment_mgr()) {
+        ExecEnv::GetInstance()->fragment_mgr()->remove_query_context(this->_query_id);
+    }
     // the only one msg shows query's end. any other msg should append to it if need.
     LOG_INFO("Query {} deconstructed, mem_tracker: {}", print_id(this->_query_id), mem_tracker_msg);
 }

be/src/runtime_filter/runtime_filter_mgr.cpp

@@ -433,30 +433,6 @@ Status RuntimeFilterMergeControllerEntity::_send_rf_to_target(GlobalMergeContext
     return st;
 }
 
-void RuntimeFilterMergeControllerEntity::release_undone_filters(QueryContext* query_ctx) {
-    std::unique_lock<std::shared_mutex> guard(_filter_map_mutex);
-    for (auto& [filter_id, ctx] : _filter_map) {
-        if (!ctx.done && !ctx.targetv2_info.empty()) {
-            {
-                std::lock_guard<std::mutex> l(ctx.mtx);
-                ctx.merger->set_wrapper_state_and_ready_to_apply(
-                        RuntimeFilterWrapper::State::DISABLED,
-                        "rf coordinator's query context released before runtime filter is ready to "
-                        "apply");
-            }
-            auto st = _send_rf_to_target(ctx, std::weak_ptr<QueryContext> {}, 0,
-                                         UniqueId(query_ctx->query_id()).to_proto(),
-                                         query_ctx->execution_timeout());
-            if (!st.ok()) {
-                LOG(WARNING)
-                        << "Failed to send runtime filter to target before query done. filter_id:"
-                        << filter_id << " " << ctx.merger->debug_string() << " reason:" << st;
-            }
-        }
-    }
-    _filter_map.clear();
-}
-
 std::string RuntimeFilterMergeControllerEntity::debug_string() {
     std::string result = "RuntimeFilterMergeControllerEntity Info:\n";
     std::shared_lock<std::shared_mutex> guard(_filter_map_mutex);

be/src/runtime_filter/runtime_filter_mgr.h

@@ -150,7 +150,10 @@ class RuntimeFilterMergeControllerEntity {
 
     std::string debug_string();
 
-    void release_undone_filters(QueryContext* query_ctx);
+    bool empty() {
+        std::shared_lock<std::shared_mutex> read_lock(_filter_map_mutex);
+        return _filter_map.empty();
+    }
 
 private:
     Status _init_with_desc(std::shared_ptr<QueryContext> query_ctx,