[bugfix](fragment mgr) heap used after free in fragment manager when query is cancelled (#23817)

Co-authored-by: yiguolei <yiguolei@gmail.com>

be/src/runtime/fragment_mgr.cpp

@@ -617,7 +617,13 @@ Status FragmentMgr::exec_plan_fragment(const TExecPlanFragmentParams& params,
                                    params.query_options.enable_pipeline_engine;
     RETURN_IF_ERROR(
             _get_query_ctx(params, params.params.query_id, pipeline_engine_enabled, query_ctx));
-    query_ctx->fragment_ids.push_back(fragment_instance_id);
+    {
+        // Need lock here, because it will modify fragment ids and std::vector may resize and reallocate
+        // memory, but query_is_canncelled will traverse the vector, it will core.
+        // query_is_cancelled is called in allocator, we has to avoid dead lock.
+        std::lock_guard<std::mutex> lock(_lock);
+        query_ctx->fragment_ids.push_back(fragment_instance_id);
+    }
 
     auto fragment_executor = std::make_shared<PlanFragmentExecutor>(
             _exec_env, query_ctx, params.params.fragment_instance_id, -1, params.backend_num,

be/src/runtime/fragment_mgr.h

@@ -169,6 +169,11 @@ private:
     // This is input params
     ExecEnv* _exec_env;
 
+    // The lock should only be used to protect the structures in fragment manager. Has to be
+    // used in a very small scope because it may dead lock. For example, if the _lock is used
+    // in prepare stage, the call path is  prepare --> expr prepare --> may call allocator
+    // when allocate failed, allocator may call query_is_cancelled, query is callced will also
+    // call _lock, so that there is dead lock.
     std::mutex _lock;
 
     std::condition_variable _cv;