You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@doris.apache.org by hu...@apache.org on 2026/04/16 03:32:49 UTC
(doris) branch fe_local_shuffle updated (821d1a90d7c -> 523b38af81e)
This is an automated email from the ASF dual-hosted git repository.
huajianlan pushed a change to branch fe_local_shuffle
in repository https://gitbox.apache.org/repos/asf/doris.git
discard 821d1a90d7c Revert "[fix](fe) Fix BUCKET_SHUFFLE data loss in serial exchange with FE-planned local shuffle"
new 523b38af81e Revert "[fix](fe) Fix BUCKET_SHUFFLE data loss in serial exchange with FE-planned local shuffle"
This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
branch are not in the new version. This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:
* -- * -- B -- O -- O -- O (821d1a90d7c)
\
N -- N -- N refs/heads/fe_local_shuffle (523b38af81e)
You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.
Any revisions marked "omit" are not gone; other references still
refer to them. Any revisions marked "discard" are gone forever.
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
be/src/exec/pipeline/pipeline_fragment_context.cpp | 6 ++++++
1 file changed, 6 insertions(+)
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
(doris) 01/01: Revert "[fix](fe) Fix BUCKET_SHUFFLE data loss in serial exchange with FE-planned local shuffle"
Posted by 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
commit 523b38af81e8a994755025ebc7d155d702ecb71f
Author: 924060929 <la...@selectdb.com>
AuthorDate: Wed Apr 15 20:28:28 2026 +0800
Revert "[fix](fe) Fix BUCKET_SHUFFLE data loss in serial exchange with FE-planned local shuffle"
This reverts commit 782dc35888d6d744a6fba1b5f271e21245515070.
---
be/src/exec/pipeline/pipeline_fragment_context.cpp | 79 +++++++++++++++++++---
.../trees/plans/distribute/DistributePlanner.java | 48 +++----------
2 files changed, 79 insertions(+), 48 deletions(-)
diff --git a/be/src/exec/pipeline/pipeline_fragment_context.cpp b/be/src/exec/pipeline/pipeline_fragment_context.cpp
index 358873e6323..527670652a1 100644
--- a/be/src/exec/pipeline/pipeline_fragment_context.cpp
+++ b/be/src/exec/pipeline/pipeline_fragment_context.cpp
@@ -734,23 +734,25 @@ Status PipelineFragmentContext::_create_deferred_local_exchangers() {
}
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.
+ // Fix num_tasks mismatches between paired pipelines created by pipeline-splitting
+ // operators (AGG, SORT, JOIN). These pairs share state via inject_shared_state and
+ // must have consistent num_tasks; otherwise instance 1+ tasks access null shared_state.
//
- // 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.
+ // Pass 1 (upward): when FE inserts LOCAL_EXCHANGE_NODE below a splitting operator,
+ // the LE pipeline gets _num_instances but ancestor pipelines still have the reduced
+ // count. Raise them to _num_instances (skip serial source operators).
+ //
+ // Pass 2 (downward): when a serial Exchange feeds a splitting operator (e.g., scalar
+ // merge-finalize AGG), the sink pipeline has 1 task but the source pipeline may have
+ // been raised (by LE or by inheriting _num_instances from add_pipeline). Lower it
+ // to match, unless it contains LocalExchangeSource (deliberately raised by LE).
std::unordered_map<PipelineId, PipelinePtr> id_to_pipe;
for (auto& p : _pipelines) {
id_to_pipe[p->id()] = p;
}
+ // Pass 1 (upward): raise downstream pipelines to _num_instances when any upstream
+ // dependency was already raised by LOCAL_EXCHANGE.
bool changed = true;
while (changed) {
changed = false;
@@ -777,6 +779,61 @@ void PipelineFragmentContext::_propagate_local_exchange_num_tasks() {
}
}
}
+
+ // Pass 2 (downward): when a pipeline-splitting operator (AGG, SORT, JOIN) creates
+ // paired pipelines (source + sink), they share state via inject_shared_state and must
+ // have consistent num_tasks. If the sink pipeline has a serial source (Exchange) with
+ // num_tasks < _num_instances, but the source pipeline was raised (by inheriting from
+ // an LE-fixed parent during add_pipeline), lower the source pipeline to match.
+ // Skip pipelines containing LocalExchangeSource — they were deliberately raised.
+ //
+ // Example: serial Exchange → AGG(merge finalize) → LOCAL_EXCHANGE → NLJ
+ // P_sink = [Exchange, AGGSink] num_tasks=1 (serial)
+ // P_source = [AGGSource] num_tasks=_num_instances (inherited)
+ // → lower P_source to 1; the LE above fans out the 1 result to N NLJ tasks.
+ 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() <= 1) {
+ continue;
+ }
+ // Don't lower pipelines that contain LocalExchangeSource — they were
+ // deliberately raised to _num_instances by LOCAL_EXCHANGE processing.
+ bool has_le_source = false;
+ for (auto& op : downstream->operators()) {
+ if (dynamic_cast<LocalExchangeSourceOperatorX*>(op.get())) {
+ has_le_source = true;
+ break;
+ }
+ }
+ if (has_le_source) {
+ 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() < downstream->num_tasks() &&
+ !uit->second->operators().empty() &&
+ uit->second->operators().front()->is_serial_operator()) {
+ LOG(INFO) << "propagate pass2: lower pipeline " << downstream_id
+ << " (" << downstream->debug_string() << ") from "
+ << downstream->num_tasks() << " to "
+ << uit->second->num_tasks()
+ << " to match serial upstream " << upstream_id
+ << " (" << uit->second->debug_string() << ")";
+ downstream->set_num_tasks(uit->second->num_tasks());
+ changed = true;
+ break;
+ }
+ }
+ }
+ }
}
Status PipelineFragmentContext::_create_tree_helper(
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/distribute/DistributePlanner.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/distribute/DistributePlanner.java
index cedaff46f57..48a7ba39413 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/distribute/DistributePlanner.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/distribute/DistributePlanner.java
@@ -60,13 +60,11 @@ import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import java.util.Arrays;
-import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Objects;
import java.util.Optional;
-import java.util.Set;
import java.util.function.Consumer;
/** DistributePlanner */
@@ -247,21 +245,12 @@ public class DistributePlanner {
boolean useLocalShuffle = receiverPlan.getInstanceJobs().stream()
.anyMatch(LocalShuffleAssignedJob.class::isInstance);
if (useLocalShuffle) {
+ // For BUCKET_SHUFFLE with local shuffle, each instance handles specific buckets.
+ // The exchange must route data to each instance individually, not just the first
+ // per worker. Otherwise instances 1..N-1 create ExchangeSource tasks that never
+ // receive data or EOS, causing a hang.
if (linkNode.getPartitionType() == TPartitionType.BUCKET_SHFFULE_HASH_PARTITIONED) {
- // When Exchange is serial on BE (1 task per worker), only instance 0
- // creates VDataStreamRecvr. Data sent to other instances is silently
- // dropped (EOF). Route all data to the first instance per worker;
- // the FE-planned local exchange (BUCKET_HASH_SHUFFLE) redistributes
- // locally to the correct join build tasks.
- // When Exchange is non-serial, each instance has its own Exchange task,
- // so route specific bucket data to each instance directly.
- PlanFragment receiverFragment = receiverPlan.getFragmentJob().getFragment();
- boolean exchangeWillBeSerial = (linkNode.isSerialOperator()
- || receiverFragment.hasSerialScanNode())
- && receiverFragment.useSerialSource(ConnectContext.get());
- if (!exchangeWillBeSerial) {
- return receiverPlan.getInstanceJobs();
- }
+ return receiverPlan.getInstanceJobs();
}
return getFirstInstancePerWorker(receiverPlan.getInstanceJobs());
} else if (enableShareHashTableForBroadcastJoin && linkNode.isRightChildOfBroadcastHashJoin()) {
@@ -273,20 +262,15 @@ public class DistributePlanner {
private List<AssignedJob> sortDestinationInstancesByBuckets(
PipelineDistributedPlan plan, List<AssignedJob> unsorted, int bucketNum) {
- // Detect first-instance-per-worker mode: if each worker appears at most once,
- // we must use scan-level buckets (all buckets on that worker) instead of
- // join-level buckets (only this instance's assigned subset).
- boolean firstPerWorkerOnly = isFirstPerWorkerOnly(unsorted);
-
AssignedJob[] instances = new AssignedJob[bucketNum];
for (AssignedJob instanceJob : unsorted) {
- // When multiple instances per worker (non-serial exchange), use
- // assignedJoinBucketIndexes to avoid collisions — each instance handles
- // only its specific join buckets.
- // When first-per-worker (serial exchange), use scan-level buckets so
- // all buckets on the worker map to this single receiver instance.
+ // For LocalShuffleBucketJoinAssignedJob, use assignedJoinBucketIndexes which
+ // gives the specific join bucket(s) this instance handles. The scanSource
+ // buckets include ALL buckets on this worker (shared via pooling scan),
+ // which would cause collisions when multiple instances on the same worker
+ // each handle different join buckets.
Iterable<Integer> bucketIndices;
- if (instanceJob instanceof LocalShuffleBucketJoinAssignedJob && !firstPerWorkerOnly) {
+ if (instanceJob instanceof LocalShuffleBucketJoinAssignedJob) {
bucketIndices = ((LocalShuffleBucketJoinAssignedJob) instanceJob)
.getAssignedJoinBucketIndexes();
} else {
@@ -317,16 +301,6 @@ public class DistributePlanner {
return Arrays.asList(instances);
}
- private boolean isFirstPerWorkerOnly(List<AssignedJob> instances) {
- Set<DistributedPlanWorker> seen = new HashSet<>();
- for (AssignedJob instance : instances) {
- if (!seen.add(instance.getAssignedWorker())) {
- return false;
- }
- }
- return true;
- }
-
private List<AssignedJob> getFirstInstancePerWorker(List<AssignedJob> instances) {
Map<DistributedPlanWorker, AssignedJob> firstInstancePerWorker = Maps.newLinkedHashMap();
for (AssignedJob instance : instances) {
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org