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 10:11:19 UTC
(doris) branch fe_local_shuffle updated (04449643620 -> fe3b023a356)
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 04449643620 [fix](local shuffle) Match BE toThrift serial condition in FE enforceRequire
new fe3b023a356 [fix](local shuffle) Match BE toThrift serial condition in FE enforceRequire
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 (04449643620)
\
N -- N -- N refs/heads/fe_local_shuffle (fe3b023a356)
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:
.../org/apache/doris/planner/AggregationNode.java | 7 ++-----
.../org/apache/doris/planner/AnalyticEvalNode.java | 4 +---
.../org/apache/doris/planner/ExchangeNode.java | 24 ++++++++++------------
.../org/apache/doris/planner/HashJoinNode.java | 18 ++++------------
.../java/org/apache/doris/planner/PlanNode.java | 9 +++++++-
5 files changed, 26 insertions(+), 36 deletions(-)
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
(doris) 01/01: [fix](local shuffle) Match BE toThrift serial condition in FE enforceRequire
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 fe3b023a35689da05544dcdeeff84551999df45c
Author: 924060929 <la...@selectdb.com>
AuthorDate: Thu Apr 16 15:37:45 2026 +0800
[fix](local shuffle) Match BE toThrift serial condition in FE enforceRequire
FE's enforceRequire checked `children.get(i).isSerialOperator()` to detect
serial children, but BE's toThrift sets `is_serial_operator = (isSerialOperator()
|| hasSerialScanNode()) && useSerialSource()`. When a fragment has a pooling scan
(hasSerialScanNode=true) but the Exchange child is non-merging UNPARTITIONED
(isSerialOperator=false due to mergeInfo!=null), FE missed the serial condition.
This caused FE to skip LOCAL_EXCHANGE insertion for:
- HashJoin BROADCAST build side: build pipeline had 1 task, probe had N tasks,
instance 1+ probes had no hash table → data loss
- Streaming AGG child: same mismatch pattern
- Analytic eval child: same pattern
Fix: add `|| fragment.hasSerialScanNode()` to serial child checks in
HashJoinNode, AggregationNode, and AnalyticEvalNode, matching the BE's
toThrift condition.
Also restore _deferred_exchangers.empty() early return in BE propagation
to avoid interfering with BE-native _plan_local_exchange.
---
.../org/apache/doris/planner/AggregationNode.java | 8 ++------
.../org/apache/doris/planner/AnalyticEvalNode.java | 4 +---
.../org/apache/doris/planner/ExchangeNode.java | 24 ++++++++++------------
.../org/apache/doris/planner/HashJoinNode.java | 11 ++++------
.../java/org/apache/doris/planner/PlanNode.java | 9 +++++++-
5 files changed, 26 insertions(+), 30 deletions(-)
diff --git a/fe/fe-core/src/main/java/org/apache/doris/planner/AggregationNode.java b/fe/fe-core/src/main/java/org/apache/doris/planner/AggregationNode.java
index 13e126031ce..d184b6a92e5 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/planner/AggregationNode.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/planner/AggregationNode.java
@@ -297,11 +297,8 @@ public class AggregationNode extends PlanNode {
if (children.get(0) instanceof HashJoinNode
&& sessionVariable.enableStreamingAggHashJoinForcePassthrough) {
requireChild = LocalExchangeTypeRequire.requirePassthrough();
- } else if (fragment != null && fragment.useSerialSource(connectContext)
- && children.get(0).isSerialOperator()) {
+ } else if (children.get(0).willBeSerialOnBe(connectContext)) {
// BE: _child->is_serial_operator() ? PASSTHROUGH : NOOP
- // useSerialSource gate needed: ScanNode.isSerialOperator() can be true in
- // non-pooling mode (scanRanges < pptn) but isn't serial on BE without useSerialSource.
requireChild = LocalExchangeTypeRequire.requirePassthrough();
} else {
requireChild = LocalExchangeTypeRequire.noRequire();
@@ -314,8 +311,7 @@ public class AggregationNode extends PlanNode {
requireChild = LocalExchangeTypeRequire.noRequire();
} else {
// Serialize agg, no group key: BE → _child->is_serial_operator() ? PT : NOOP
- if (fragment != null && fragment.useSerialSource(connectContext)
- && children.get(0).isSerialOperator()) {
+ if (children.get(0).willBeSerialOnBe(connectContext)) {
requireChild = LocalExchangeTypeRequire.requirePassthrough();
} else {
requireChild = LocalExchangeTypeRequire.noRequire();
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 e93ebd9f017..320ebf964ea 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
@@ -201,10 +201,8 @@ public class AnalyticEvalNode extends PlanNode {
requireChild = LocalExchangeTypeRequire.noRequire();
outputType = LocalExchangeType.NOOP;
}
- } else if (fragment.useSerialSource(translatorContext.getConnectContext())
- && children.get(0).isSerialOperator()) {
+ } else if (children.get(0).willBeSerialOnBe(translatorContext.getConnectContext())) {
// BE base class: _child->is_serial_operator() ? PASSTHROUGH : NOOP
- // useSerialSource gate: ScanNode.isSerialOperator() can be true in non-pooling mode
requireChild = LocalExchangeTypeRequire.requirePassthrough();
outputType = LocalExchangeType.PASSTHROUGH;
} else {
diff --git a/fe/fe-core/src/main/java/org/apache/doris/planner/ExchangeNode.java b/fe/fe-core/src/main/java/org/apache/doris/planner/ExchangeNode.java
index 8f6e8a7994a..691a7bb110a 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/planner/ExchangeNode.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/planner/ExchangeNode.java
@@ -102,12 +102,8 @@ public class ExchangeNode extends PlanNode {
@Override
protected void toThrift(TPlanNode msg) {
- // BE local shuffle only supports two modes: 1 instance or all instances receiving data.
- // hasSerialScanNode() ensures Exchange is serial when sibling scan is pooling,
- // avoiding a middle-ground where some but not all instances receive data
- // (EOS count mismatch → hang).
- msg.setIsSerialOperator((isSerialOperator() || fragment.hasSerialScanNode())
- && fragment.useSerialSource(ConnectContext.get()));
+ // Reuse willBeSerialOnBe() — single source of truth for BE serial condition.
+ msg.setIsSerialOperator(willBeSerialOnBe(ConnectContext.get()));
msg.node_type = TPlanNodeType.EXCHANGE_NODE;
msg.exchange_node = new TExchangeNode();
for (TupleId tid : tupleIds) {
@@ -163,6 +159,13 @@ public class ExchangeNode extends PlanNode {
|| partitionType == TPartitionType.UNPARTITIONED) && mergeInfo != null;
}
+ @Override
+ public boolean willBeSerialOnBe(ConnectContext context) {
+ return fragment != null
+ && (isSerialOperator() || fragment.hasSerialScanNode())
+ && fragment.useSerialSource(context);
+ }
+
@Override
public boolean hasSerialChildren() {
return isSerialOperator();
@@ -180,13 +183,8 @@ public class ExchangeNode extends PlanNode {
// Only insert PASSTHROUGH when the exchange will actually be serial on the BE.
// Without useSerialSource() check, we'd insert PASSTHROUGH in non-pooling fragments
// where the exchange has N tasks, corrupting broadcast join data distribution
- // (PASSTHROUGH round-robin splits complete-dataset sinks into 1/N subsets per source).
- // Must match toThrift: include hasSerialScanNode() — when sibling scan is pooling,
- // Exchange becomes serial on BE even if Exchange itself isn't inherently serial.
- boolean willBeSerialOnBe = fragment != null
- && (isSerialOperator() || fragment.hasSerialScanNode())
- && fragment.useSerialSource(ConnectContext.get());
- if (willBeSerialOnBe) {
+ // Use the unified willBeSerialOnBe() which matches toThrift serial condition.
+ if (willBeSerialOnBe(ConnectContext.get())) {
// If there is already a serial ancestor in the same pipeline (e.g., serial NLJ
// for RIGHT_OUTER/FULL_OUTER join), don't insert any local exchange. The serial
// ancestor already constrains the pipeline to 1 task. Inserting a PASSTHROUGH LE
diff --git a/fe/fe-core/src/main/java/org/apache/doris/planner/HashJoinNode.java b/fe/fe-core/src/main/java/org/apache/doris/planner/HashJoinNode.java
index daff478885c..15b25d1b55f 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/planner/HashJoinNode.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/planner/HashJoinNode.java
@@ -307,13 +307,10 @@ public class HashJoinNode extends JoinNodeBase {
outputType = LocalExchangeType.NOOP;
} else if (distrMode == DistributionMode.BROADCAST) {
// BE: _child->is_serial_operator() ? PASSTHROUGH/PASS_TO_ONE : NOOP
- // useSerialSource gate: ScanNode.isSerialOperator() can be true in non-pooling
- // (scanRanges < pptn) but isn't serial on BE. ExchangeNode.isSerialOperator()
- // already includes useSerialSource, so the gate is harmless for Exchange children.
- boolean useSerialSource = fragment != null
- && fragment.useSerialSource(translatorContext.getConnectContext());
- boolean probeChildSerial = useSerialSource && children.get(0).isSerialOperator();
- boolean buildChildSerial = useSerialSource && children.get(1).isSerialOperator();
+ boolean probeChildSerial = children.get(0).willBeSerialOnBe(
+ translatorContext.getConnectContext());
+ boolean buildChildSerial = children.get(1).willBeSerialOnBe(
+ translatorContext.getConnectContext());
probeSideRequire = probeChildSerial
? LocalExchangeTypeRequire.requirePassthrough()
: LocalExchangeTypeRequire.noRequire();
diff --git a/fe/fe-core/src/main/java/org/apache/doris/planner/PlanNode.java b/fe/fe-core/src/main/java/org/apache/doris/planner/PlanNode.java
index 0335b298dda..3fe34492a0a 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/planner/PlanNode.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/planner/PlanNode.java
@@ -472,7 +472,7 @@ public abstract class PlanNode extends TreeNode<PlanNode> {
TPlanNode msg = new TPlanNode();
msg.node_id = id.asInt();
msg.setNereidsId(nereidsId);
- msg.setIsSerialOperator(isSerialOperator() && fragment.useSerialSource(ConnectContext.get()));
+ msg.setIsSerialOperator(willBeSerialOnBe(ConnectContext.get()));
msg.num_children = children.size();
msg.limit = limit;
for (TupleId tid : tupleIds) {
@@ -834,6 +834,13 @@ public abstract class PlanNode extends TreeNode<PlanNode> {
return false;
}
+ // Whether this node will have is_serial_operator=true on BE.
+ // Must match the condition in toThrift()/treeToThriftHelper().
+ // Subclasses (ExchangeNode) override to add hasSerialScanNode().
+ public boolean willBeSerialOnBe(ConnectContext context) {
+ return fragment != null && isSerialOperator() && fragment.useSerialSource(context);
+ }
+
public boolean hasSerialChildren() {
if (children.isEmpty()) {
return isSerialOperator();
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org