(doris) branch fe_local_shuffle updated: [fix](local shuffle) Propagate num_tasks from LOCAL_EXCHANGE to ancestor pipelines

[hu...@apache.org]

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

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


The following commit(s) were added to refs/heads/fe_local_shuffle by this push:
     new b03722b95e9 [fix](local shuffle) Propagate num_tasks from LOCAL_EXCHANGE to ancestor pipelines
b03722b95e9 is described below

commit b03722b95e98b9d84943e2818cca4c4d710999de
Author: 924060929 <la...@selectdb.com>
AuthorDate: Wed Apr 15 19:59:35 2026 +0800

    [fix](local shuffle) Propagate num_tasks from LOCAL_EXCHANGE to ancestor pipelines
    
    When FE inserts LOCAL_EXCHANGE_NODE below a pipeline-splitting operator
    (AGG, SORT, JOIN), the LOCAL_EXCHANGE handler restores its immediate
    pipeline to _num_instances tasks. However, ancestor pipelines created
    earlier by the splitting operators still carry the reduced num_tasks
    inherited from a serial operator (e.g., pooling scan or UNPARTITIONED
    Exchange).
    
    This causes a num_tasks mismatch: e.g., AggSink pipeline has
    _num_instances tasks but AggSource pipeline has only 1 task. Instance
    1+ creates AggSink tasks but no AggSource task, leaving source_deps
    uninitialized and causing DCHECK failure in set_ready_to_read().
    
    Fix: add _propagate_local_exchange_num_tasks() after _build_pipelines
    to walk the DAG and raise ancestor pipelines to _num_instances, stopping
    at serial source operators. Only runs when FE-planned LOCAL_EXCHANGE
    nodes exist (no impact on BE-native path).
---
 be/src/exec/pipeline/pipeline_fragment_context.cpp | 50 ++++++++++++++++++++++
 be/src/exec/pipeline/pipeline_fragment_context.h   |  4 ++
 2 files changed, 54 insertions(+)

diff --git a/be/src/exec/pipeline/pipeline_fragment_context.cpp b/be/src/exec/pipeline/pipeline_fragment_context.cpp
index abfe3048fff..358873e6323 100644
--- a/be/src/exec/pipeline/pipeline_fragment_context.cpp
+++ b/be/src/exec/pipeline/pipeline_fragment_context.cpp
@@ -261,6 +261,10 @@ Status PipelineFragmentContext::_build_and_prepare_full_pipeline(ThreadPool* thr
         RETURN_IF_ERROR(_build_pipelines(_runtime_state->obj_pool(), *_query_ctx->desc_tbl,
                                          &_root_op, root_pipeline));
 
+        // Propagate _num_instances from LOCAL_EXCHANGE pipelines to ancestor pipelines
+        // that inherited reduced num_tasks from a serial operator.
+        _propagate_local_exchange_num_tasks();
+
         // Create deferred local exchangers now that all pipelines have final num_tasks.
         RETURN_IF_ERROR(_create_deferred_local_exchangers());
 
@@ -729,6 +733,52 @@ Status PipelineFragmentContext::_create_deferred_local_exchangers() {
     return Status::OK();
 }
 
+void PipelineFragmentContext::_propagate_local_exchange_num_tasks() {
+    if (_deferred_exchangers.empty()) {
+        return;
+    }
+    // When FE inserts LOCAL_EXCHANGE_NODE below a pipeline-splitting operator (AGG, SORT,
+    // JOIN), LOCAL_EXCHANGE_NODE processing restores its immediate pipeline to
+    // _num_instances.  However, ancestor pipelines created earlier by the splitting
+    // operators still carry the reduced num_tasks inherited from a serial operator.
+    //
+    // Walk the DAG: for each downstream pipeline whose num_tasks < _num_instances, if any
+    // upstream dependency already has _num_instances tasks, raise the downstream too —
+    // unless its source operator is serial (it legitimately has 1 task).  Iterate until
+    // stable.
+    std::unordered_map<PipelineId, PipelinePtr> id_to_pipe;
+    for (auto& p : _pipelines) {
+        id_to_pipe[p->id()] = p;
+    }
+
+    bool changed = true;
+    while (changed) {
+        changed = false;
+        for (const auto& [downstream_id, upstream_ids] : _dag) {
+            auto dit = id_to_pipe.find(downstream_id);
+            if (dit == id_to_pipe.end()) {
+                continue;
+            }
+            auto& downstream = dit->second;
+            if (downstream->num_tasks() >= _num_instances) {
+                continue;
+            }
+            if (!downstream->operators().empty() &&
+                downstream->operators().front()->is_serial_operator()) {
+                continue;
+            }
+            for (auto upstream_id : upstream_ids) {
+                auto uit = id_to_pipe.find(upstream_id);
+                if (uit != id_to_pipe.end() && uit->second->num_tasks() >= _num_instances) {
+                    downstream->set_num_tasks(_num_instances);
+                    changed = true;
+                    break;
+                }
+            }
+        }
+    }
+}
+
 Status PipelineFragmentContext::_create_tree_helper(
         ObjectPool* pool, const std::vector<TPlanNode>& tnodes, const DescriptorTbl& descs,
         OperatorPtr parent, int* node_idx, OperatorPtr* root, PipelinePtr& cur_pipe, int child_idx,
diff --git a/be/src/exec/pipeline/pipeline_fragment_context.h b/be/src/exec/pipeline/pipeline_fragment_context.h
index 97bfbef524d..777fc5ac2ba 100644
--- a/be/src/exec/pipeline/pipeline_fragment_context.h
+++ b/be/src/exec/pipeline/pipeline_fragment_context.h
@@ -345,6 +345,10 @@ private:
     };
     std::vector<DeferredExchangerInfo> _deferred_exchangers;
     Status _create_deferred_local_exchangers();
+    // After _build_pipelines, propagate _num_instances from FE-planned LOCAL_EXCHANGE
+    // pipelines upward through the DAG to ancestor pipelines that inherited reduced
+    // num_tasks from a serial operator.
+    void _propagate_local_exchange_num_tasks();
 
     //Here are two types of runtime states:
     //    - _runtime state is at the Fragment level.


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