(doris) branch master updated: [pipelineX](fix) Fix coredump if prepare failed (#30250)

[gabriellee]

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

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


The following commit(s) were added to refs/heads/master by this push:
     new bbc99119e0d [pipelineX](fix) Fix coredump if prepare failed (#30250)
bbc99119e0d is described below

commit bbc99119e0dcacc3894751dba1de388cdfa6bb3d
Author: Gabriel <gabrielleeb...@gmail.com>
AuthorDate: Tue Jan 23 14:19:17 2024 +0800

    [pipelineX](fix) Fix coredump if prepare failed (#30250)
---
 be/src/pipeline/pipeline_x/operator.cpp        |  6 +++++-
 be/src/pipeline/pipeline_x/operator.h          |  8 ++++++--
 be/src/pipeline/pipeline_x/pipeline_x_task.cpp | 14 ++------------
 3 files changed, 13 insertions(+), 15 deletions(-)

diff --git a/be/src/pipeline/pipeline_x/operator.cpp 
b/be/src/pipeline/pipeline_x/operator.cpp
index a59f4ced6de..21453dfbc3c 100644
--- a/be/src/pipeline/pipeline_x/operator.cpp
+++ b/be/src/pipeline/pipeline_x/operator.cpp
@@ -151,7 +151,11 @@ Status OperatorXBase::close(RuntimeState* state) {
     if (_child_x && !is_source()) {
         RETURN_IF_ERROR(_child_x->close(state));
     }
-    return state->get_local_state(operator_id())->close(state);
+    auto result = state->get_local_state_result(operator_id());
+    if (!result) {
+        return result.error();
+    }
+    return result.value()->close(state);
 }
 
 void PipelineXLocalStateBase::clear_origin_block() {
diff --git a/be/src/pipeline/pipeline_x/operator.h 
b/be/src/pipeline/pipeline_x/operator.h
index 98feb60ac22..e40c7849c09 100644
--- a/be/src/pipeline/pipeline_x/operator.h
+++ b/be/src/pipeline/pipeline_x/operator.h
@@ -537,8 +537,12 @@ public:
 
     [[nodiscard]] bool is_source() const override { return false; }
 
-    virtual Status close(RuntimeState* state, Status exec_status) {
-        return state->get_sink_local_state(operator_id())->close(state, 
exec_status);
+    Status close(RuntimeState* state, Status exec_status) {
+        auto result = state->get_sink_local_state_result(operator_id());
+        if (!result) {
+            return result.error();
+        }
+        return result.value()->close(state, exec_status);
     }
 
     [[nodiscard]] RuntimeProfile* get_runtime_profile() const override {
diff --git a/be/src/pipeline/pipeline_x/pipeline_x_task.cpp 
b/be/src/pipeline/pipeline_x/pipeline_x_task.cpp
index c44570eb7c6..fece30296c9 100644
--- a/be/src/pipeline/pipeline_x/pipeline_x_task.cpp
+++ b/be/src/pipeline/pipeline_x/pipeline_x_task.cpp
@@ -126,11 +126,7 @@ Status PipelineXTask::_extract_dependencies() {
         }
     }
     {
-        auto result = 
_state->get_sink_local_state_result(_sink->operator_id());
-        if (!result) {
-            return result.error();
-        }
-        auto* local_state = result.value();
+        auto* local_state = _state->get_sink_local_state(_sink->operator_id());
         auto* dep = local_state->dependency();
         DCHECK(dep != nullptr);
         _write_dependencies = dep;
@@ -139,13 +135,7 @@ Status PipelineXTask::_extract_dependencies() {
             _finish_dependencies.push_back(fin_dep);
         }
     }
-    {
-        auto result = _state->get_local_state_result(_source->operator_id());
-        if (!result) {
-            return result.error();
-        }
-        _filter_dependency = result.value()->filterdependency();
-    }
+    { _filter_dependency = 
_state->get_local_state(_source->operator_id())->filterdependency(); }
     return Status::OK();
 }
 


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