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 b0475298d70 [fix](fe) Skip PASSTHROUGH LE wrapping for Exchange under
serial ancestor
b0475298d70 is described below
commit b0475298d7033f9c74e09c4bc3cf2d8f055313d8
Author: 924060929 <[email protected]>
AuthorDate: Thu Apr 9 18:32:35 2026 +0800
[fix](fe) Skip PASSTHROUGH LE wrapping for Exchange under serial ancestor
### What problem does this PR solve?
Issue Number: close #xxx
Problem Summary:
When a serial NLJ (RIGHT_OUTER/FULL_OUTER) has an UNPARTITIONED Exchange
child, ExchangeNode.enforceAndDeriveLocalExchange() wraps the Exchange
with a PASSTHROUGH LocalExchange. On BE, the NLJ_PROBE (serial) sets
pipeline num_tasks=1, then the LE handler set_num_tasks(_num_instances)
overrides it to N, triggering DCHECK crash: serial operator in pipeline
with num_tasks > 1.
Fix: Check hasSerialAncestorInPipeline before wrapping. When a serial
ancestor already constrains the pipeline to 1 task, inserting a
PASSTHROUGH LE would create a pipeline boundary where the BE LE handler
resets num_tasks, conflicting with the serial constraint. Return NOOP
instead to keep the Exchange directly in the serial pipeline.
### Release note
None
### Check List (For Author)
- Test: Regression test (Bug 18) + manual test with various pptn values
- Behavior changed: No
- Does this need documentation: No
Co-authored-by: Copilot <[email protected]>
---
.../org/apache/doris/planner/ExchangeNode.java | 9 ++++
.../test_local_shuffle_rqg_bugs.groovy | 53 ++++++++++++++++++++++
2 files changed, 62 insertions(+)
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 60d42500d5e..4f03834ab9b 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
@@ -189,6 +189,15 @@ public class ExchangeNode extends PlanNode {
&& fragment != null
&& fragment.useSerialSource(ConnectContext.get());
if (willBeSerialOnBe) {
+ // 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
+ // would create a pipeline boundary where the BE
LOCAL_EXCHANGE_NODE handler calls
+ // set_num_tasks(_num_instances), overriding the serial constraint
and causing a
+ // DCHECK crash (serial operator in pipeline with num_tasks > 1).
+ if (translatorContext.hasSerialAncestorInPipeline(this)) {
+ return Pair.of(this, LocalExchangeType.NOOP);
+ }
// Serial exchange → 1 task. Must fan out to N tasks for
downstream operators.
// For HASH/BUCKET exchanges: return NOOP and let parent insert
the appropriate
// redistribution (HASH_SHUFFLE or BUCKET_HASH_SHUFFLE).
PASSTHROUGH round-robin
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 f838e5f61de..f2d2dd71072 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
@@ -988,5 +988,58 @@ suite("test_local_shuffle_rqg_bugs") {
assertTrue(false, "Bug 16+17: Serial AnalyticEval crash/hang:
${t.message}")
}
+ // Bug 18: DCHECK crash in Pipeline::set_num_tasks when PASSTHROUGH LE is
inserted
+ // between serial NLJ and its child Exchange.
+ // Root cause: ExchangeNode.enforceAndDeriveLocalExchange wraps
UNPARTITIONED serial
+ // Exchange with PASSTHROUGH LE. On BE, NLJ_PROBE (serial) sets pipeline
num_tasks=1,
+ // then the LE handler's set_num_tasks(_num_instances) overrides it to N,
triggering
+ // DCHECK (serial operator in pipeline with num_tasks > 1).
+ // Fix: skip PASSTHROUGH wrapping when hasSerialAncestorInPipeline is
true.
+ // Query: LEFT JOIN with always-true self-ref condition (table.pk =
table.pk) creates
+ // RIGHT_OUTER NLJ (serial). With pptn>1 and ignore_data_distribution,
the fragment
+ // gets N instances but NLJ forces 1 task.
+ try {
+ logger.info("Bug 18: Testing serial NLJ with PASSTHROUGH LE crash")
+ // Use existing rqg_t1 table (10 rows, 10 buckets)
+ def bug18_baseline = sql """
+ SELECT /*+SET_VAR(enable_local_shuffle_planner=false,
+ enable_sql_cache=false)*/
+ table1.col_varchar_10__undef_signed AS field1
+ FROM rqg_t1 AS table1
+ LEFT JOIN rqg_t1 AS table2
+ ON table2.pk = table2.pk
+ WHERE table1.pk BETWEEN 2 AND 11
+ GROUP BY field1
+ ORDER BY 1
+ """
+
+ // Test with various pptn values — crash requires pptn > 1
+ for (int ppt : [4, 7]) {
+ def bug18_result = sql """
+ SELECT /*+SET_VAR(use_serial_exchange=false,
+ parallel_pipeline_task_num=${ppt},
+ enable_local_shuffle_planner=true,
+ ignore_storage_data_distribution=true,
+ enable_sql_cache=false,
+
enable_share_hash_table_for_broadcast_join=false,
+ enable_broadcast_join_force_passthrough=true,
+ enable_parallel_scan=false)*/
+ table1.col_varchar_10__undef_signed AS field1
+ FROM rqg_t1 AS table1
+ LEFT JOIN rqg_t1 AS table2
+ ON table2.pk = table2.pk
+ WHERE table1.pk BETWEEN 2 AND 11
+ GROUP BY field1
+ ORDER BY 1
+ """
+ assertEquals(bug18_baseline, bug18_result,
+ "Bug 18 pptn=${ppt}: result mismatch with serial NLJ + local
exchange")
+ }
+ logger.info("Bug 18: PASSED (no crash, correct results)")
+ } catch (Throwable t) {
+ logger.error("Bug 18 FAILED: ${t.message}")
+ assertTrue(false, "Bug 18: Serial NLJ PASSTHROUGH LE crash:
${t.message}")
+ }
+
logger.info("=== All RQG bug reproduction tests completed ===")
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
