This is an automated email from the ASF dual-hosted git repository.
Mryange pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new e1df149d617 [fix](local shuffle) Prevent local exchange under serial
parent pipeline (#67490)
e1df149d617 is described below
commit e1df149d617d16066fec9c97d770c14b3ac045b2
Author: Mryange <[email protected]>
AuthorDate: Tue Sep 8 14:55:15 2026 +0800
[fix](local shuffle) Prevent local exchange under serial parent pipeline
(#67490)
When FE local shuffle planning inserted a local exchange into a pipeline
whose parent pipeline was serial, it could increase only the downstream
pipeline task count while the paired sink and source operators remained
one-to-one. This produced inconsistent pipeline concurrency and could
fail during execution. Root cause: FE did not propagate the serial state
of a parent pipeline across pipeline boundaries. This change tracks that
state and skips local exchange insertion when the parent pipeline is
serial, matching the existing BE planning constraint.
### Release note
None
### Check List (For Author)
- Test <!-- At least one of them must be included. -->
- [ ] Regression test
- [ ] Unit Test
- [ ] Manual test (add detailed scripts or steps below)
- [ ] No need to test or manual test. Explain why:
- [ ] This is a refactor/code format and no logic has been changed.
- [ ] Previous test can cover this change.
- [ ] No code files have been changed.
- [ ] Other reason <!-- Add your reason? -->
- Behavior changed:
- [ ] No.
- [ ] Yes. <!-- Explain the behavior change -->
- Does this need documentation?
- [ ] No.
- [ ] Yes. <!-- Add document PR link here. eg:
https://github.com/apache/doris-website/pull/1214 -->
### Check List (For Reviewer who merge this PR)
- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label <!-- Add branch pick label that this PR
should merge into -->
---
.../glue/translator/PlanTranslatorContext.java | 14 +++++++++
.../org/apache/doris/planner/AddLocalExchange.java | 1 +
.../java/org/apache/doris/planner/PlanNode.java | 16 ++++++++--
.../test_local_shuffle_rqg_bugs.groovy | 35 ++++++++++++++++++++++
4 files changed, 63 insertions(+), 3 deletions(-)
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PlanTranslatorContext.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PlanTranslatorContext.java
index 0a09cd0670b..476579d2f12 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PlanTranslatorContext.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PlanTranslatorContext.java
@@ -133,6 +133,12 @@ public class PlanTranslatorContext {
// root and across pipeline boundaries (see shouldResetSerialFlagForChild).
private final Map<PlanNodeId, Boolean> serialAncestorInPipelineMap =
Maps.newHashMap();
+ // Per-node "does this pipeline have a serial parent pipeline" flag.
Mirrors BE's
+ // Pipeline::num_tasks_of_parent() gate in _add_local_exchange: a pipeline
whose parent
+ // has one task must not be split by another local exchange, because that
would raise only
+ // the new source side to N tasks while the paired sink/source operators
stay one-to-one.
+ private final Map<PlanNodeId, Boolean> serialParentPipelineMap =
Maps.newHashMap();
+
// Per-node "is there a downstream operator that depends on hash
distribution for
// correctness, with HASH/NOOP path connecting it to me" flag. Mirrors
BE's
// _followed_by_shuffled_operator propagation in
pipeline_fragment_context.cpp.
@@ -292,6 +298,14 @@ public class PlanTranslatorContext {
return serialAncestorInPipelineMap.getOrDefault(node.getId(), false);
}
+ public void setHasSerialParentPipeline(PlanNode node, boolean value) {
+ serialParentPipelineMap.put(node.getId(), value);
+ }
+
+ public boolean hasSerialParentPipeline(PlanNode node) {
+ return serialParentPipelineMap.getOrDefault(node.getId(), false);
+ }
+
public void setHasShuffleForCorrectnessAncestor(PlanNode node, boolean
value) {
shuffledAncestorMap.put(node.getId(), value);
}
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 1adea56c13e..b5cf3f49766 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
@@ -197,6 +197,7 @@ public class AddLocalExchange {
? LocalExchangeTypeRequire.noRequire() :
sink.getLocalExchangeTypeRequire();
PlanNode root = fragment.getPlanRoot();
context.setHasSerialAncestorInPipeline(root, false);
+ context.setHasSerialParentPipeline(root, false);
Pair<PlanNode, LocalExchangeType> output = root
.enforceAndDeriveLocalExchange(context, null, require);
PlanNode newRoot = output.first;
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 a635ca26730..b592c15f627 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
@@ -50,6 +50,7 @@ import org.apache.doris.thrift.TPushAggOp;
import com.google.common.base.Joiner;
import com.google.common.base.Preconditions;
+import com.google.common.base.Suppliers;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import org.apache.commons.collections4.CollectionUtils;
@@ -65,6 +66,7 @@ import java.util.Map.Entry;
import java.util.Set;
import java.util.function.Consumer;
import java.util.function.Predicate;
+import java.util.function.Supplier;
import java.util.stream.Collectors;
/**
@@ -1069,7 +1071,12 @@ public abstract class PlanNode extends
TreeNode<PlanNode> {
// node's sink, e.g. Exchange is in AGG_Sink pipeline).
// For non-splitting operators (shouldReset=false, e.g. streaming AGG):
// Inherit parent's serial flag + this node's own.
- boolean inheritedSerial = shouldResetSerialFlagForChild(childIndex)
+ boolean startsNewPipeline = shouldResetSerialFlagForChild(childIndex);
+ Supplier<Boolean> currentNodeSerialOnBe = Suppliers.memoize(
+ () ->
isSerialOperatorOnBe(translatorContext.getConnectContext()));
+ boolean currentPipelineSerial =
translatorContext.hasSerialAncestorInPipeline(this)
+ || currentNodeSerialOnBe.get();
+ boolean inheritedSerial = startsNewPipeline
? false : translatorContext.hasSerialAncestorInPipeline(this);
// Use isSerialOperatorOnBe (= isSerialNode &&
fragment.useSerialSource) instead of the
// raw isSerialNode(). BE's OperatorBase reads the Thrift
`is_serial_operator` flag —
@@ -1078,8 +1085,10 @@ public abstract class PlanNode extends
TreeNode<PlanNode> {
// Using isSerialNode here would set the child's serial-ancestor flag
wider than BE's
// view and over-skip required LocalExchanges downstream.
boolean childHasSerialAncestor = inheritedSerial
- || isSerialOperatorOnBe(translatorContext.getConnectContext());
+ || currentNodeSerialOnBe.get();
translatorContext.setHasSerialAncestorInPipeline(child,
childHasSerialAncestor);
+ translatorContext.setHasSerialParentPipeline(child, startsNewPipeline
+ ? currentPipelineSerial :
translatorContext.hasSerialParentPipeline(this));
// 1b. Propagate shuffle-for-correctness-ancestor flag to child.
// Mirrors BE's _followed_by_shuffled_operator: a downstream operator
needs hash
@@ -1140,7 +1149,8 @@ public abstract class PlanNode extends TreeNode<PlanNode>
{
// Use isSerialOperatorOnBe (not isSerialNode) because BE's
Pipeline::need_to_local_exchange
// checks op->is_serial_operator() which reads the Thrift flag set
from isSerialOperatorOnBe;
// when fragment.useSerialSource is false, BE treats this node as
non-serial.
- if (translatorContext.hasSerialAncestorInPipeline(this)
+ if (translatorContext.hasSerialParentPipeline(this)
+ || translatorContext.hasSerialAncestorInPipeline(this)
||
isSerialOperatorOnBe(translatorContext.getConnectContext())) {
return childOutput;
}
diff --git
a/regression-test/suites/nereids_p0/local_shuffle/test_local_shuffle_rqg_bugs.groovy
b/regression-test/suites/nereids_p0/local_shuffle/test_local_shuffle_rqg_bugs.groovy
index 4401f0362d3..f53ab6204f5 100644
---
a/regression-test/suites/nereids_p0/local_shuffle/test_local_shuffle_rqg_bugs.groovy
+++
b/regression-test/suites/nereids_p0/local_shuffle/test_local_shuffle_rqg_bugs.groovy
@@ -1663,5 +1663,40 @@ suite("test_local_shuffle_rqg_bugs") {
assertTrue(false, "Bug 26: ${t.message}")
}
+ // Bug 27: do not insert a local exchange in a pipeline whose parent
pipeline is serial.
+ // Otherwise the local exchange raises the lower AggSink pipeline to N
tasks while its
+ // paired AggSource remains at one task, leaving task 1+ without a source
dependency.
+ def bug27Query = { planner -> """
+ SELECT /*+SET_VAR(enable_local_shuffle_planner=${planner},
+ enable_local_exchange_before_agg=false,
+ enable_local_exchange_before_streaming_agg=true,
+ enable_broadcast_join_force_passthrough=true,
+ enable_share_hash_table_for_broadcast_join=false,
+ parallel_pipeline_task_num=3,
+ enable_sql_cache=false)*/
+ COUNT(*),
+ COUNT(DISTINCT CAST(f.pk AS STRING)),
+ MIN(CAST(f.pk AS STRING)),
+ MAX(CAST(f.pk AS STRING)),
+ COUNT(DISTINCT CAST(d.pk AS STRING)),
+ MIN(CAST(d.pk AS STRING)),
+ MAX(CAST(d.pk AS STRING)),
+ COUNT(DISTINCT CAST(f.col_int_undef_signed AS STRING)),
+ MIN(CAST(f.col_int_undef_signed AS STRING)),
+ MAX(CAST(f.col_int_undef_signed AS STRING))
+ FROM rqg_t1 f
+ INNER JOIN (
+ SELECT * FROM rqg_t2 d
+ WHERE d.col_int_undef_signed = 1 AND COALESCE(d.pk, -1) >= 10
+ ) d ON d.col_int_undef_signed = f.col_int_undef_signed
+ AND f.col_int_undef_signed = f.col_int_undef_signed2
+ AND f.pk = d.pk
+ """ }
+
+ def bug27BeResult = sql bug27Query(false)
+ for (int i = 0; i < 20; i++) {
+ assertEquals(bug27BeResult, sql(bug27Query(true)), "Bug 27 run ${i}")
+ }
+
logger.info("=== All RQG bug reproduction tests completed ===")
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
