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 3cd360be0db [fix](fe) Fix serial AnalyticEval crash and DataStreamSink
hang in local shuffle
3cd360be0db is described below
commit 3cd360be0db90c770c22842c6a592e419d11eca1
Author: 924060929 <[email protected]>
AuthorDate: Thu Apr 9 11:25:22 2026 +0800
[fix](fe) Fix serial AnalyticEval crash and DataStreamSink hang in local
shuffle
### What problem does this PR solve?
Issue Number: close #xxx
Problem Summary:
Two bugs when serial AnalyticEval (OVER() with no PARTITION BY) runs in
fragments with LocalShuffleAssignedJob (multiple instances on one BE):
1. ANALYTIC_EVAL_SINK crash: Exchange.enforceAndDeriveLocalExchange() wraps
itself with PASSTHROUGH LocalExchange. This restores AnalyticSink
pipeline
to _num_instances tasks while serial AnalyticSource stays at 1 task.
For instance_idx > 0, source_deps is empty → DCHECK crash.
Fix: AnalyticEvalNode strips any LocalExchange inserted by its child when
partitionExprs is empty (serial mode).
2. DataStreamSink hang: After fixing the crash, serial AnalyticSource
reduces
all downstream pipeline tasks to 1 via add_pipeline() inheritance. Only
instance 0 runs DataStreamSink, sending 1 EOS, but receiver expects
_num_instances EOSes → hang.
Fix: AddLocalExchange checks hasSerialChildren() in addition to
isSerialOperator() to insert PASSTHROUGH above the fragment root,
restoring DataStreamSink to _num_instances tasks.
### Release note
None
### Check List (For Author)
- Test: Manual test on 3-BE local cluster with rqg_t20 table
- Verified crash query returns 41 rows correctly with pptn=0,1,4
- Verified all shuffle/broadcast/union/bucket combos pass
- Behavior changed: No
- Does this need documentation: No
Co-authored-by: Copilot <[email protected]>
---
.../org/apache/doris/planner/AddLocalExchange.java | 10 ++++++----
.../org/apache/doris/planner/AnalyticEvalNode.java | 20 ++++++++++++++++++--
2 files changed, 24 insertions(+), 6 deletions(-)
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/planner/AddLocalExchange.java
b/fe/fe-core/src/main/java/org/apache/doris/planner/AddLocalExchange.java
index 120125ff008..40b79a03471 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/planner/AddLocalExchange.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/planner/AddLocalExchange.java
@@ -71,14 +71,16 @@ public class AddLocalExchange {
.enforceAndDeriveLocalExchange(context, null, require);
PlanNode newRoot = output.first;
// Mirror BE OperatorBase base class required_data_distribution():
- // when child (fragment root) is serial AND the fragment uses pooling
scan
+ // when any operator in the fragment plan is serial AND the fragment
uses pooling scan
// (LocalShuffleAssignedJob → ignoreDataDistribution →
_parallel_instances=1),
- // the DataStreamSink pipeline has num_tasks=1 so only instance 0
creates a
- // task and sends EOS. Downstream receivers expect _num_instances
EOSes → hang.
+ // serial operators reduce pipeline num_tasks to 1, and this
propagates via
+ // add_pipeline() to all downstream pipelines (including the
DataStreamSink pipeline).
+ // Only instance 0 creates tasks and sends EOS. Downstream receivers
expect
+ // _num_instances EOSes → hang.
// Insert PASSTHROUGH fan-out to create _num_instances sink tasks.
// Non-pooling fragments (regular bucket distribution) have
_num_instances ==
// bucket_count and every instance gets a scan range, so all sink
tasks run.
- if (isLocalShuffle && newRoot.isSerialOperator()) {
+ if (isLocalShuffle && (newRoot.isSerialOperator() ||
newRoot.hasSerialChildren())) {
newRoot = new LocalExchangeNode(context.nextPlanNodeId(), newRoot,
LocalExchangeType.PASSTHROUGH, null);
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/planner/AnalyticEvalNode.java
b/fe/fe-core/src/main/java/org/apache/doris/planner/AnalyticEvalNode.java
index 816ad1f37de..e7a859dc824 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/planner/AnalyticEvalNode.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/planner/AnalyticEvalNode.java
@@ -172,8 +172,24 @@ public class AnalyticEvalNode extends PlanNode {
LocalExchangeTypeRequire requireChild;
LocalExchangeType outputType = null;
if (partitionExprs.isEmpty()) {
- requireChild = LocalExchangeTypeRequire.requirePassthrough();
- outputType = LocalExchangeType.PASSTHROUGH;
+ // Serial AnalyticEval (OVER() with no PARTITION BY):
+ // Must NOT have any LocalExchange between AnalyticEval and its
child.
+ // On BE, AnalyticSink and AnalyticSource share state
(source_deps/sink_deps).
+ // A LocalExchange below would restore the AnalyticSink pipeline
to _num_instances
+ // tasks while the serial AnalyticSource pipeline stays at 1 task.
For instance_idx > 0,
+ // the AnalyticSource task is never created, leaving source_deps
empty →
+ // crash when AnalyticSink calls set_ready_to_read(0).
+ //
+ // We call enforceChild with noRequire to traverse children, then
strip any
+ // LocalExchange the child inserted (e.g., Exchange wrapping
itself with PASSTHROUGH).
+ Pair<PlanNode, LocalExchangeType> enforceResult
+ = enforceChild(translatorContext,
LocalExchangeTypeRequire.noRequire(), children.get(0));
+ PlanNode newChild = enforceResult.first;
+ if (newChild instanceof LocalExchangeNode) {
+ newChild = newChild.getChild(0);
+ }
+ children = Lists.newArrayList(newChild);
+ return Pair.of(this, LocalExchangeType.PASSTHROUGH);
} else if (orderByElements.isEmpty()) {
if (AddLocalExchange.isColocated(this)) {
requireChild = LocalExchangeTypeRequire.requireHash();
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
