update

apache · Jan 27, 2025 · bb15b66 · bb15b66
1 parent d6e500d
commit bb15b66
Show file tree

Hide file tree

Showing 3 changed files with 15 additions and 6 deletions.
diff --git a/be/src/pipeline/exec/multi_cast_data_stream_sink.cpp b/be/src/pipeline/exec/multi_cast_data_stream_sink.cpp
@@ -50,10 +50,6 @@ Status MultiCastDataStreamSinkOperatorX::sink(RuntimeState* state, vectorized::B
     if (in_block->rows() > 0 || eos) {
         COUNTER_UPDATE(local_state.rows_input_counter(), (int64_t)in_block->rows());
         auto st = local_state._shared_state->multi_cast_data_streamer->push(state, in_block, eos);
-        // TODO: improvement: if sink returned END_OF_FILE, pipeline task can be finished
-        if (st.template is<ErrorCode::END_OF_FILE>()) {
-            return Status::OK();
-        }
         return st;
     }
     return Status::OK();

diff --git a/be/src/pipeline/pipeline.cpp b/be/src/pipeline/pipeline.cpp
@@ -122,7 +122,20 @@ void Pipeline::make_all_runnable() {
 
     for (auto* task : _tasks) {
         if (task) {
-            task->make_runnable_if_all_downstream_finished();
+            task->wake_up_early_if_all_downstream_finished();
+        }
+    }
+
+    if (_sink->count_down_destination()) {
+        for (auto* task : _tasks) {
+            if (task) {
+                task->set_wake_up_early();
+            }
+        }
+        for (auto* task : _tasks) {
+            if (task) {
+                task->clear_blocking_state();
+            }
         }
     }
 }

diff --git a/be/src/pipeline/pipeline_task.h b/be/src/pipeline/pipeline_task.h
@@ -233,7 +233,7 @@ class PipelineTask {
 
     bool wake_up_early() const { return _wake_up_early; }
 
-    void make_runnable_if_all_downstream_finished() {
+    void wake_up_early_if_all_downstream_finished() {
         if (!_sink_shared_state) {
             return;
         }