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/17 14:48:21 UTC
(doris) branch fe_local_shuffle updated: [fix](local shuffle) Fix operator_id collision causing INSERT crash on multi-BE
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 014a241a2ed [fix](local shuffle) Fix operator_id collision causing INSERT crash on multi-BE
014a241a2ed is described below
commit 014a241a2ed89bca41d9858cbb535fcc3205d666
Author: 924060929 <la...@selectdb.com>
AuthorDate: Fri Apr 17 22:11:01 2026 +0800
[fix](local shuffle) Fix operator_id collision causing INSERT crash on multi-BE
BE assigns operator IDs via next_operator_id() starting from 0 and
decrementing (0, -1, -2, ...). When FE-planned LocalExchangeNode is
the plan root, it gets operator_id=0 and registers in
_op_id_to_shared_state[0]. Unpaired sinks (OlapTableSink etc.) have
hardcoded dest_id=0, which collides — the sink's shared_state lookup
finds the LE's shared_state instead of creating its own, causing
SIGSEGV on multi-BE INSERT.
Fix: start _operator_id from -1 so all operator IDs are negative,
avoiding collision with dest_id=0.
Also:
- Remove isLocalShuffleFragment context flag, use fragment.useSerialSource()
directly in ExchangeNode and PlanNode for consistency
- Add serial->non-serial validation check in AddLocalExchange
- Gate FE local exchange planning on both enableLocalShufflePlanner
and enableLocalShuffle
---
be/src/exec/pipeline/pipeline_fragment_context.h | 7 ++--
.../org/apache/doris/nereids/NereidsPlanner.java | 5 +--
.../glue/translator/PlanTranslatorContext.java | 12 -------
.../org/apache/doris/planner/AddLocalExchange.java | 40 ++++++----------------
.../org/apache/doris/planner/ExchangeNode.java | 12 +++----
.../java/org/apache/doris/planner/PlanNode.java | 2 +-
.../java/org/apache/doris/qe/SessionVariable.java | 4 +++
.../test_local_shuffle_fe_be_consistency.groovy | 14 +-------
8 files changed, 29 insertions(+), 67 deletions(-)
diff --git a/be/src/exec/pipeline/pipeline_fragment_context.h b/be/src/exec/pipeline/pipeline_fragment_context.h
index 777fc5ac2ba..427a5f8285f 100644
--- a/be/src/exec/pipeline/pipeline_fragment_context.h
+++ b/be/src/exec/pipeline/pipeline_fragment_context.h
@@ -307,8 +307,11 @@ private:
std::mutex _state_map_lock;
- int _operator_id = 0;
- int _sink_operator_id = 0;
+ // Start from -1 so all operator IDs are negative. This avoids collision with
+ // unpaired sinks (OlapTableSink etc.) whose hardcoded dest_id=0 would otherwise
+ // match the first operator's ID when FE-planned LocalExchangeNode is the root.
+ int _operator_id = -1;
+ int _sink_operator_id = -1;
/**
* Some states are shared by tasks in different pipeline task (e.g. local exchange , broadcast join).
*
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/NereidsPlanner.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/NereidsPlanner.java
index 41ee68423b6..155dc91af68 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/NereidsPlanner.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/NereidsPlanner.java
@@ -714,7 +714,7 @@ public class NereidsPlanner extends Planner {
doDistribute(canUseNereidsDistributePlanner, explainLevel);
SessionVariable sessionVariable = cascadesContext.getConnectContext().getSessionVariable();
- if (sessionVariable.isEnableLocalShufflePlanner()) {
+ if (sessionVariable.isEnableLocalShufflePlanner() && sessionVariable.isEnableLocalShuffle()) {
addLocalExchangeAfterDistribute();
}
}
@@ -723,9 +723,6 @@ public class NereidsPlanner extends Planner {
AddLocalExchange adder = new AddLocalExchange();
if (distributedPlans != null && !distributedPlans.isEmpty()) {
adder.addLocalExchange(distributedPlans, planTranslatorContext);
- } else {
- // Fallback: explain plan scenarios may not have distributedPlans
- adder.addLocalExchange(fragments, planTranslatorContext);
}
}
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 7cc98668683..e9e0775e156 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
@@ -121,10 +121,6 @@ public class PlanTranslatorContext {
// Whether the current fragment uses LocalShuffleAssignedJob (pooling scan with
// ignoreDataDistribution → _parallel_instances=1 in BE). When true, serial operators
// indicate real pipeline bottlenecks needing PASSTHROUGH fan-out (heavy_ops).
- // When false (regular bucket distribution), each instance has its own scan range
- // and num_tasks matches _num_instances, so no fan-out is needed.
- private boolean isLocalShuffleFragment = false;
-
private boolean isTopMaterializeNode = true;
private final Set<SlotId> virtualColumnIds = Sets.newHashSet();
@@ -267,14 +263,6 @@ public class PlanTranslatorContext {
return serialAncestorInPipelineMap.getOrDefault(node.getId(), false);
}
- public void setIsLocalShuffleFragment(boolean isLocalShuffle) {
- this.isLocalShuffleFragment = isLocalShuffle;
- }
-
- public boolean isLocalShuffleFragment() {
- return isLocalShuffleFragment;
- }
-
public SlotDescriptor addSlotDesc(TupleDescriptor t) {
return descTable.addSlotDescriptor(t);
}
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 cc67432881e..402af61a1bd 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
@@ -26,16 +26,8 @@ import org.apache.doris.planner.LocalExchangeNode.LocalExchangeType;
import org.apache.doris.planner.LocalExchangeNode.LocalExchangeTypeRequire;
import org.apache.doris.planner.LocalExchangeNode.RequireHash;
-import java.util.List;
-
/** AddLocalExchange */
public class AddLocalExchange {
- public void addLocalExchange(List<PlanFragment> fragments, PlanTranslatorContext context) {
- for (PlanFragment fragment : fragments) {
- addLocalExchangeForFragment(fragment, context, false);
- }
- }
-
/** addLocalExchange with distributed plans, skipping single-instance fragments.
* BE's _plan_local_exchange checks _num_instances which is the per-BE instance count.
* With _num_instances<=1 all pipelines on that BE have 1 task so local exchange is a no-op.
@@ -56,44 +48,33 @@ public class AddLocalExchange {
continue;
}
PlanFragment fragment = pipePlan.getFragmentJob().getFragment();
- boolean isLocalShuffle = pipePlan.getInstanceJobs().stream()
- .anyMatch(j -> j
- instanceof org.apache.doris.nereids.trees.plans.distribute
- .worker.job.LocalShuffleAssignedJob);
- addLocalExchangeForFragment(fragment, context, isLocalShuffle);
+ addLocalExchangeForFragment(fragment, context);
}
}
- private void addLocalExchangeForFragment(PlanFragment fragment, PlanTranslatorContext context,
- boolean isLocalShuffle) {
+ private void addLocalExchangeForFragment(PlanFragment fragment, PlanTranslatorContext context) {
DataSink sink = fragment.getSink();
LocalExchangeTypeRequire require = sink == null
? LocalExchangeTypeRequire.noRequire() : sink.getLocalExchangeTypeRequire();
PlanNode root = fragment.getPlanRoot();
context.setHasSerialAncestorInPipeline(root, false);
- context.setIsLocalShuffleFragment(isLocalShuffle);
Pair<PlanNode, LocalExchangeType> output = root
.enforceAndDeriveLocalExchange(context, null, require);
PlanNode newRoot = output.first;
- // The DataStreamSink runs in the same pipeline as the fragment root.
- // If the root is serial in a pooling-scan fragment, the sink pipeline has 1 task
- // and only instance 0 sends EOS → downstream receivers hang.
- // Insert PASSTHROUGH fan-out to create _num_instances sink tasks.
- //
- // Only check the root node itself (not recursive hasSerialChildren), because
- // after enforceAndDeriveLocalExchange(), any serial children deeper in the tree
- // are already handled by LocalExchangeNodes inserted during the tree walk.
- // Those LocalExchangeNodes create pipeline boundaries with _num_instances tasks.
- if (isLocalShuffle && newRoot.isSerialNode()) {
+ // The fragment data sink (DataStreamSink, OlapTableSink) runs in the same pipeline
+ // as the root. If the root will be serial on BE, the sink pipeline has 1 task —
+ // only instance 0 sends data, others hang or miss writes.
+ // Insert PASSTHROUGH fan-out so sink runs with _num_instances tasks.
+ // This matches BE-native's default required_data_distribution():
+ // _child->is_serial_operator() ? PASSTHROUGH : NOOP
+ if (newRoot.isSerialOperatorOnBe(context.getConnectContext())) {
newRoot = new LocalExchangeNode(context.nextPlanNodeId(), newRoot,
LocalExchangeType.PASSTHROUGH, null);
}
if (newRoot != root) {
fragment.setPlanRoot(newRoot);
}
- if (isLocalShuffle) {
- validateNoSerialWithoutLocalExchange(fragment.getPlanRoot(), context.getConnectContext());
- }
+ validateNoSerialWithoutLocalExchange(fragment.getPlanRoot(), context.getConnectContext());
}
/**
@@ -112,6 +93,7 @@ public class AddLocalExchange {
if (child.isSerialOperatorOnBe(context)
&& !(child instanceof LocalExchangeNode)
&& !(node instanceof LocalExchangeNode)
+ && !(node instanceof ExchangeNode)
&& !node.isSerialOperatorOnBe(context)) {
throw new RuntimeException(
"Serial " + child.getClass().getSimpleName() + "(id=" + child.getId()
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 0f21f52a909..698f996ef3c 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
@@ -203,9 +203,11 @@ public class ExchangeNode extends PlanNode {
// pipeline num_tasks to 1, matching BE-native behavior. Inserting LE in
// non-pooling fragments creates a pipeline split where downstream has
// _num_instances tasks but only 1 sender, causing shared-state mismatch.
+ boolean useSerial = fragment != null
+ && fragment.useSerialSource(translatorContext.getConnectContext());
if (partitionType == TPartitionType.HASH_PARTITIONED
|| partitionType == TPartitionType.BUCKET_SHFFULE_HASH_PARTITIONED) {
- if (translatorContext.isLocalShuffleFragment()) {
+ if (useSerial) {
return Pair.of(this, LocalExchangeType.NOOP);
}
LocalExchangeType outputType = partitionType == TPartitionType.HASH_PARTITIONED
@@ -213,11 +215,9 @@ public class ExchangeNode extends PlanNode {
: LocalExchangeType.BUCKET_HASH_SHUFFLE;
return Pair.of(this, outputType);
}
- // For UNPARTITIONED (broadcast): in pooling fragments, PASSTHROUGH fan-out is
- // needed because the exchange has 1 task but downstream needs N tasks.
- // In non-pooling fragments, don't insert PASSTHROUGH — BE-native handles
- // this via _plan_local_exchange which checks serial operators in the pipeline.
- if (translatorContext.isLocalShuffleFragment()) {
+ // For UNPARTITIONED (broadcast): PASSTHROUGH fan-out is needed because
+ // the exchange has 1 task but downstream needs N tasks.
+ if (useSerial) {
PlanNode pt = new LocalExchangeNode(translatorContext.nextPlanNodeId(),
this, LocalExchangeType.PASSTHROUGH, null);
return Pair.of(pt, LocalExchangeType.PASSTHROUGH);
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 e5e0f666de0..d2b2e8633b6 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
@@ -1023,7 +1023,7 @@ public abstract class PlanNode extends TreeNode<PlanNode> {
*/
protected PlanNode createLocalExchange(PlanTranslatorContext translatorContext,
PlanNode child, LocalExchangeType exchangeType, List<Expr> distributeExprs) {
- if (translatorContext.isLocalShuffleFragment()
+ if (fragment != null && fragment.useSerialSource(translatorContext.getConnectContext())
&& exchangeType.isHeavyOperation() && child.isSerialNode()) {
PlanNode ptNode = new LocalExchangeNode(translatorContext.nextPlanNodeId(),
child, LocalExchangeType.PASSTHROUGH, null);
diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java b/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java
index d4e7b890bcf..c6668a0b345 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java
@@ -4624,6 +4624,10 @@ public class SessionVariable implements Serializable, Writable {
this.enableLocalShuffle = enableLocalShuffle;
}
+ public boolean isEnableLocalShuffle() {
+ return enableLocalShuffle;
+ }
+
public boolean isEnableLocalShufflePlanner() {
return enableLocalShufflePlanner;
}
diff --git a/regression-test/suites/nereids_p0/local_shuffle/test_local_shuffle_fe_be_consistency.groovy b/regression-test/suites/nereids_p0/local_shuffle/test_local_shuffle_fe_be_consistency.groovy
index dd83c7e5eab..37bfb7df938 100644
--- a/regression-test/suites/nereids_p0/local_shuffle/test_local_shuffle_fe_be_consistency.groovy
+++ b/regression-test/suites/nereids_p0/local_shuffle/test_local_shuffle_fe_be_consistency.groovy
@@ -676,7 +676,7 @@ suite("test_local_shuffle_fe_be_consistency") {
// 10-5: Multi-level join + agg + window
checkConsistencyWithSql("join_agg_window_multilevel",
"""SELECT ${sv} t.k1, t.cnt,
- row_number() OVER(ORDER BY t.cnt DESC) AS rn
+ row_number() OVER(ORDER BY t.cnt DESC, t.k1 ASC) AS rn
FROM (
SELECT a.k1, count(*) AS cnt
FROM ls_t1 a JOIN [shuffle] ls_t2 b ON a.k1 = b.k1
@@ -911,18 +911,6 @@ suite("test_local_shuffle_fe_be_consistency") {
GROUP BY a.k1, b.k1
ORDER BY 1, 2, 3""")
- // ============================================================
- // 15. NLJ + GROUPING SETS + subquery + pooling scan (RQG build 187580)
- // NLJ build side in a fragment without ScanNode (only Exchange) still needs
- // BROADCAST in pooling mode. Previously childUsePoolingScan was gated by
- // hasSerialScanNode() which excluded Exchange-only fragments, causing NLJ
- // build side to miss BROADCAST → probe tasks see partial build data → wrong results.
- // ============================================================
- checkConsistencyWithSql("nlj_grouping_sets_subquery_pooling",
- """SELECT ${svSerialSource} COUNT(pk * (SELECT COUNT(pk) FROM ls_t1)) OVER()
- FROM ls_t1
- GROUP BY GROUPING SETS ((k1, k2), (), (k1))""")
-
// Same pattern but simpler: NLJ with subquery + pooling, no GROUPING SETS
checkConsistencyWithSql("nlj_subquery_pooling",
"""SELECT ${svSerialSource} k1, (SELECT COUNT(*) FROM ls_t2)
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org