[doris] branch compaction_opt updated: [bugfix](runtime_filter) cherry-pick 14362 heap use after free (#14363)

[dataroaring]

This is an automated email from the ASF dual-hosted git repository.

dataroaring pushed a commit to branch compaction_opt
in repository https://gitbox.apache.org/repos/asf/doris.git


The following commit(s) were added to refs/heads/compaction_opt by this push:
     new 27c5fd84a7 [bugfix](runtime_filter) cherry-pick 14362 heap use after 
free (#14363)
27c5fd84a7 is described below

commit 27c5fd84a738cad92aabffc779ae1acafa8bae51
Author: yixiutt <102007456+yixi...@users.noreply.github.com>
AuthorDate: Thu Nov 17 16:17:29 2022 +0800

    [bugfix](runtime_filter) cherry-pick 14362 heap use after free (#14363)
---
 be/src/runtime/fragment_mgr.cpp     | 15 +++++++++++++++
 be/src/runtime/runtime_filter_mgr.h |  4 +++-
 2 files changed, 18 insertions(+), 1 deletion(-)

diff --git a/be/src/runtime/fragment_mgr.cpp b/be/src/runtime/fragment_mgr.cpp
index 58b3b1d95b..cb88fb9481 100644
--- a/be/src/runtime/fragment_mgr.cpp
+++ b/be/src/runtime/fragment_mgr.cpp
@@ -1027,6 +1027,21 @@ Status FragmentMgr::merge_filter(const 
PMergeFilterRequest* request,
     UniqueId queryid = request->query_id();
     std::shared_ptr<RuntimeFilterMergeControllerEntity> filter_controller;
     RETURN_IF_ERROR(_runtimefilter_controller.acquire(queryid, 
&filter_controller));
+    auto fragment_instance_id = filter_controller->instance_id();
+    TUniqueId tfragment_instance_id = fragment_instance_id.to_thrift();
+    std::shared_ptr<FragmentExecState> fragment_state;
+    {
+        std::unique_lock<std::mutex> lock(_lock);
+        auto iter = _fragment_map.find(tfragment_instance_id);
+        if (iter == _fragment_map.end()) {
+            VLOG_CRITICAL << "unknown fragment-id:" << fragment_instance_id;
+            return Status::InvalidArgument("fragment-id: {}", 
fragment_instance_id.to_string());
+        }
+
+        // hold reference to fragment_state, or else runtime_state can be 
destroyed
+        // when filter_controller->merge is still in progress
+        fragment_state = iter->second;
+    }
     RETURN_IF_ERROR(filter_controller->merge(request, attach_data));
     return Status::OK();
 }
diff --git a/be/src/runtime/runtime_filter_mgr.h 
b/be/src/runtime/runtime_filter_mgr.h
index 63960f0182..1228bcbdc4 100644
--- a/be/src/runtime/runtime_filter_mgr.h
+++ b/be/src/runtime/runtime_filter_mgr.h
@@ -122,7 +122,9 @@ public:
     Status merge(const PMergeFilterRequest* request,
                  butil::IOBufAsZeroCopyInputStream* attach_data);
 
-    UniqueId query_id() { return _query_id; }
+    UniqueId query_id() const { return _query_id; }
+
+    UniqueId instance_id() const { return _fragment_instance_id; }
 
     struct RuntimeFilterCntlVal {
         int64_t create_time;


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@doris.apache.org
For additional commands, e-mail: commits-h...@doris.apache.org