You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@doris.apache.org by yi...@apache.org on 2025/10/11 07:29:39 UTC
(doris) branch branch-4.0 updated: branch-4.0: [fix](shuffle) EOF iff all channels done #56731 (#56740)
This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.0 by this push:
new f53f62e8d71 branch-4.0: [fix](shuffle) EOF iff all channels done #56731 (#56740)
f53f62e8d71 is described below
commit f53f62e8d7180e61400b8d72bfa1672f0ad9f5d7
Author: github-actions[bot] <41...@users.noreply.github.com>
AuthorDate: Sat Oct 11 15:29:31 2025 +0800
branch-4.0: [fix](shuffle) EOF iff all channels done #56731 (#56740)
Cherry-picked from #56731
Co-authored-by: Gabriel <li...@selectdb.com>
---
be/src/pipeline/exec/exchange_sink_operator.cpp | 15 ++++++++++++++-
be/src/pipeline/pipeline.cpp | 4 ++--
be/src/pipeline/pipeline.h | 2 +-
be/src/pipeline/pipeline_fragment_context.cpp | 2 +-
be/src/pipeline/pipeline_task.cpp | 7 ++++---
be/src/pipeline/pipeline_task.h | 6 +++++-
6 files changed, 27 insertions(+), 9 deletions(-)
diff --git a/be/src/pipeline/exec/exchange_sink_operator.cpp b/be/src/pipeline/exec/exchange_sink_operator.cpp
index d3b8152bd6b..c32860811af 100644
--- a/be/src/pipeline/exec/exchange_sink_operator.cpp
+++ b/be/src/pipeline/exec/exchange_sink_operator.cpp
@@ -514,7 +514,20 @@ Status ExchangeSinkOperatorX::sink(RuntimeState* state, vectorized::Block* block
for (auto& channel : local_state.channels) {
COUNTER_UPDATE(local_state.memory_used_counter(), -channel->mem_usage());
Status st = channel->close(state);
- if (!st.ok() && final_st.ok()) {
+ /**
+ * Consider this case below:
+ *
+ * +--- Channel0 (Running)
+ * |
+ * ExchangeSink ---+--- Channel1 (EOF)
+ * |
+ * +--- Channel2 (Running)
+ *
+ * Channel1 is EOF now and return `END_OF_FILE` here. However, Channel0 and Channel2
+ * still need new data. If ExchangeSink returns EOF, downstream tasks will no longer receive
+ * blocks including EOS signal. So we must ensure to return EOF iff all channels are EOF.
+ */
+ if (!st.ok() && !st.is<ErrorCode::END_OF_FILE>() && final_st.ok()) {
final_st = st;
}
}
diff --git a/be/src/pipeline/pipeline.cpp b/be/src/pipeline/pipeline.cpp
index 49d86bf4eb9..774561bbe37 100644
--- a/be/src/pipeline/pipeline.cpp
+++ b/be/src/pipeline/pipeline.cpp
@@ -111,7 +111,7 @@ Status Pipeline::set_sink(DataSinkOperatorPtr& sink) {
return Status::OK();
}
-void Pipeline::make_all_runnable() {
+void Pipeline::make_all_runnable(PipelineId wake_by) {
DBUG_EXECUTE_IF("Pipeline::make_all_runnable.sleep", {
auto pipeline_id = DebugPoints::instance()->get_debug_param_or_default<int32_t>(
"Pipeline::make_all_runnable.sleep", "pipeline_id", -1);
@@ -124,7 +124,7 @@ void Pipeline::make_all_runnable() {
if (_sink->count_down_destination()) {
for (auto* task : _tasks) {
if (task) {
- task->set_wake_up_early();
+ task->set_wake_up_early(wake_by);
}
}
for (auto* task : _tasks) {
diff --git a/be/src/pipeline/pipeline.h b/be/src/pipeline/pipeline.h
index 7fff24cf8d9..91ac4dd3cd6 100644
--- a/be/src/pipeline/pipeline.h
+++ b/be/src/pipeline/pipeline.h
@@ -104,7 +104,7 @@ public:
_tasks[i] = task;
}
- void make_all_runnable();
+ void make_all_runnable(PipelineId wake_by);
void set_num_tasks(int num_tasks) {
_num_tasks = num_tasks;
diff --git a/be/src/pipeline/pipeline_fragment_context.cpp b/be/src/pipeline/pipeline_fragment_context.cpp
index ec048d0db51..bd0ce44d78d 100644
--- a/be/src/pipeline/pipeline_fragment_context.cpp
+++ b/be/src/pipeline/pipeline_fragment_context.cpp
@@ -1786,7 +1786,7 @@ void PipelineFragmentContext::decrement_running_task(PipelineId pipeline_id) {
if (_pip_id_to_pipeline[pipeline_id]->close_task()) {
if (_dag.contains(pipeline_id)) {
for (auto dep : _dag[pipeline_id]) {
- _pip_id_to_pipeline[dep]->make_all_runnable();
+ _pip_id_to_pipeline[dep]->make_all_runnable(pipeline_id);
}
}
}
diff --git a/be/src/pipeline/pipeline_task.cpp b/be/src/pipeline/pipeline_task.cpp
index e069cb840d1..5a86dd2626a 100644
--- a/be/src/pipeline/pipeline_task.cpp
+++ b/be/src/pipeline/pipeline_task.cpp
@@ -723,10 +723,11 @@ std::string PipelineTask::debug_string() {
fmt::format_to(debug_string_buffer,
"PipelineTask[id = {}, open = {}, eos = {}, state = {}, dry run = "
- "{}, _wake_up_early = {}, time elapsed since last state changing = {}s, spilling"
- " = {}, is running = {}]",
+ "{}, _wake_up_early = {}, _wake_up_by = {}, time elapsed since last state "
+ "changing = {}s, spilling = {}, is running = {}]",
_index, _opened, _eos, _to_string(_exec_state), _dry_run, _wake_up_early.load(),
- _state_change_watcher.elapsed_time() / NANOS_PER_SEC, _spilling, is_running());
+ _wake_by, _state_change_watcher.elapsed_time() / NANOS_PER_SEC, _spilling,
+ is_running());
std::unique_lock<std::mutex> lc(_dependency_lock);
auto* cur_blocked_dep = _blocked_dep;
auto fragment = _fragment_context.lock();
diff --git a/be/src/pipeline/pipeline_task.h b/be/src/pipeline/pipeline_task.h
index 8c11cb87b61..08b57e0c630 100644
--- a/be/src/pipeline/pipeline_task.h
+++ b/be/src/pipeline/pipeline_task.h
@@ -127,7 +127,10 @@ public:
int task_id() const { return _index; };
bool is_finalized() const { return _exec_state == State::FINALIZED; }
- void set_wake_up_early() { _wake_up_early = true; }
+ void set_wake_up_early(PipelineId wake_by = -1) {
+ _wake_up_early = true;
+ _wake_by = wake_by;
+ }
// Execution phase should be terminated. This is called if this task is canceled or waken up early.
void terminate();
@@ -309,6 +312,7 @@ private:
MonotonicStopWatch _state_change_watcher;
std::atomic<bool> _spilling = false;
const std::string _pipeline_name;
+ int _wake_by = -1;
};
using PipelineTaskSPtr = std::shared_ptr<PipelineTask>;
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org