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/18 05:07:40 UTC
(doris) branch fe_local_shuffle updated (ab7d28db465 -> 838c92b4cb0)
This is an automated email from the ASF dual-hosted git repository.
924060929 pushed a change to branch fe_local_shuffle
in repository https://gitbox.apache.org/repos/asf/doris.git
discard ab7d28db465 [fix](local shuffle) Fix operator_id collision causing INSERT crash on multi-BE
new 838c92b4cb0 [fix](local shuffle) Fix operator_id collision causing INSERT crash on multi-BE
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 (ab7d28db465)
\
N -- N -- N refs/heads/fe_local_shuffle (838c92b4cb0)
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:
fe/fe-core/src/main/java/org/apache/doris/planner/AddLocalExchange.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
---------------------------------------------------------------------
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) Fix operator_id collision causing INSERT crash on multi-BE
Posted by hu...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
924060929 pushed a commit to branch fe_local_shuffle
in repository https://gitbox.apache.org/repos/asf/doris.git
commit 838c92b4cb0c9b312b7fc90e78791fc081007141
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 | 42 +++++++---------------
.../org/apache/doris/planner/ExchangeNode.java | 12 +++----
.../java/org/apache/doris/planner/PlanNode.java | 15 ++++++--
.../java/org/apache/doris/qe/SessionVariable.java | 4 +++
.../test_local_shuffle_fe_be_consistency.groovy | 14 +-------
.../test_local_shuffle_rqg_bugs.groovy | 42 ++++++++++++++++++++++
9 files changed, 84 insertions(+), 69 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..39ef75f7881 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,8 +93,9 @@ public class AddLocalExchange {
if (child.isSerialOperatorOnBe(context)
&& !(child instanceof LocalExchangeNode)
&& !(node instanceof LocalExchangeNode)
+ && !(node instanceof ExchangeNode)
&& !node.isSerialOperatorOnBe(context)) {
- throw new RuntimeException(
+ org.apache.logging.log4j.LogManager.getLogger(AddLocalExchange.class).warn(
"Serial " + child.getClass().getSimpleName() + "(id=" + child.getId()
+ ") feeds into non-serial " + node.getClass().getSimpleName()
+ "(id=" + node.getId() + ") without LocalExchangeNode"
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..a567b1d9d15 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
@@ -992,7 +992,18 @@ public abstract class PlanNode extends TreeNode<PlanNode> {
Pair<PlanNode, LocalExchangeType> childOutput =
child.enforceAndDeriveLocalExchange(translatorContext, this, require);
- // 3. Satisfy check: child output meets requirement → done
+ // 3. Framework-level serial child check (mirrors BE base class required_data_distribution):
+ // If child will be serial on BE but this node is not serial, the pipeline has a
+ // 1-task serial child feeding an N-task non-serial parent. Without LE, pipeline
+ // splits (AGG/JOIN) create paired pipelines with mismatched num_tasks → crash.
+ // Upgrade noRequire to requirePassthrough so LE is inserted to restore parallelism.
+ if (require instanceof LocalExchangeNode.NoRequire
+ && childOutput.first.isSerialOperatorOnBe(translatorContext.getConnectContext())
+ && !isSerialOperatorOnBe(translatorContext.getConnectContext())) {
+ require = LocalExchangeTypeRequire.requirePassthrough();
+ }
+
+ // 4. Satisfy check: child output meets requirement → done
if (require.satisfy(childOutput.second)) {
return childOutput;
}
@@ -1023,7 +1034,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)
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 d3333e74401..8688a413c9d 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
@@ -1336,5 +1336,47 @@ suite("test_local_shuffle_rqg_bugs") {
assertTrue(false, "Bug 22: AGG/SORT num_tasks propagation: ${t.message}")
}
+ // ==================== Bug 23 ====================
+ // canUseDistinctStreamingAgg + GROUPING SETS + serial scan → missing LE
+ // When enable_distinct_streaming_aggregation=true, AggregationNode's
+ // canUseDistinctStreamingAgg path set requireChild=noRequire() without
+ // checking child serial status → serial RepeatNode feeds directly into
+ // non-serial AggregationNode → shared_state mismatch on multi-BE.
+ // Fix: add isSerialOperatorOnBe check in the noRequire branch.
+ try {
+ for (def pptn : [2, 4]) {
+ def bug23_baseline = sql """
+ SELECT /*+SET_VAR(enable_local_shuffle_planner=false,
+ parallel_pipeline_task_num=${pptn},
+ ignore_storage_data_distribution=true,
+ enable_sql_cache=false,
+ disable_streaming_preaggregations=false,
+ enable_distinct_streaming_aggregation=true)*/
+ col_int_undef_signed, count(*)
+ FROM rqg_t1
+ GROUP BY GROUPING SETS ((col_int_undef_signed), (pk), ())
+ ORDER BY 1, 2
+ """
+ def bug23_result = sql """
+ SELECT /*+SET_VAR(enable_local_shuffle_planner=true,
+ parallel_pipeline_task_num=${pptn},
+ ignore_storage_data_distribution=true,
+ enable_sql_cache=false,
+ disable_streaming_preaggregations=false,
+ enable_distinct_streaming_aggregation=true)*/
+ col_int_undef_signed, count(*)
+ FROM rqg_t1
+ GROUP BY GROUPING SETS ((col_int_undef_signed), (pk), ())
+ ORDER BY 1, 2
+ """
+ assertEquals(bug23_baseline, bug23_result,
+ "Bug 23 pptn=${pptn}: GROUPING SETS + distinct streaming agg result mismatch")
+ }
+ logger.info("Bug 23: PASSED (no crash, correct results)")
+ } catch (Throwable t) {
+ logger.error("Bug 23 FAILED: ${t.message}")
+ assertTrue(false, "Bug 23: canUseDistinctStreamingAgg + GROUPING SETS: ${t.message}")
+ }
+
logger.info("=== All RQG bug reproduction tests completed ===")
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org