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/15 03:23:41 UTC
(doris) branch fe_local_shuffle updated (491003d2c9a -> 97d05678ebb)
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 491003d2c9a [fix](local shuffle) Fix NLJ childUsePoolingScan: use useSerialSource without hasSerialScanNode gate
new 97d05678ebb [fix](local shuffle) Fix NLJ childUsePoolingScan: use useSerialSource without hasSerialScanNode gate
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 (491003d2c9a)
\
N -- N -- N refs/heads/fe_local_shuffle (97d05678ebb)
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/ExchangeNode.java | 38 +++++++---------------
1 file changed, 12 insertions(+), 26 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) Fix NLJ childUsePoolingScan: use useSerialSource without hasSerialScanNode gate
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 97d05678ebb4399c98b06fe9abc6e5b0992edf5e
Author: 924060929 <la...@selectdb.com>
AuthorDate: Tue Apr 14 23:19:12 2026 +0800
[fix](local shuffle) Fix NLJ childUsePoolingScan: use useSerialSource without hasSerialScanNode gate
The hasSerialScanNode() gate excluded fragments with serial Exchanges but no
ScanNode (e.g., receive fragments with NLJ). In pooling mode these fragments
still need BROADCAST on the build side — without it, probe tasks only see
partial build data, causing row count mismatch in GROUPING SETS + subquery
queries on 4+ BE clusters.
Reverts the overly strict check introduced in the enforceRequire refactoring.
---
.../org/apache/doris/planner/ExchangeNode.java | 38 +++++++---------------
.../apache/doris/planner/NestedLoopJoinNode.java | 8 ++---
.../planner/LocalShuffleNodeCoverageTest.java | 2 --
.../test_local_shuffle_fe_be_consistency.groovy | 18 ++++++++++
4 files changed, 33 insertions(+), 33 deletions(-)
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 cc084c0f4b5..ae27e4a4a03 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
@@ -157,37 +157,15 @@ public class ExchangeNode extends PlanNode {
*
* So FRAGMENT 0 should not use serial source.
*/
- /**
- * Whether this Exchange will be serial on BE. Matches toThrift condition:
- * (staticSerial || hasSerialScanNode) && useSerialSource.
- * Without useSerialSource gate, non-pooling UNPARTITIONED exchanges would be treated
- * as serial, causing incorrect PASS_TO_ONE insertion in broadcast joins.
- */
@Override
public boolean isSerialOperator() {
- ConnectContext ctx = ConnectContext.get();
- if (ctx == null || fragment == null) {
- return false;
- }
- if (!fragment.useSerialSource(ctx)) {
- return false;
- }
- return isStaticSerialOperator() || fragment.hasSerialScanNode();
- }
-
- /** Static serial check without fragment context. Used by hasSerialChildren()
- * to avoid infinite recursion: isSerialOperator → useSerialSource → hasSerialChildren
- * → isSerialOperator. */
- private boolean isStaticSerialOperator() {
- ConnectContext ctx = ConnectContext.get();
- return (ctx != null && ctx.getSessionVariable().isUseSerialExchange()
+ return (ConnectContext.get() != null && ConnectContext.get().getSessionVariable().isUseSerialExchange()
|| partitionType == TPartitionType.UNPARTITIONED) && mergeInfo == null;
}
@Override
public boolean hasSerialChildren() {
- // Use static check to avoid recursion through useSerialSource → hasSerialScanNode
- return isStaticSerialOperator();
+ return isSerialOperator();
}
@Override
@@ -198,8 +176,16 @@ public class ExchangeNode extends PlanNode {
@Override
public Pair<PlanNode, LocalExchangeType> enforceAndDeriveLocalExchange(PlanTranslatorContext translatorContext,
PlanNode parent, LocalExchangeTypeRequire parentRequire) {
- // isSerialOperator() now matches toThrift: (staticSerial || hasSerialScanNode) && useSerialSource
- boolean willBeSerialOnBe = isSerialOperator();
+ // Must match the BE serial condition in toThrift(): isSerialOperator() && useSerialSource().
+ // 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 = (isSerialOperator() || fragment.hasSerialScanNode())
+ && 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
diff --git a/fe/fe-core/src/main/java/org/apache/doris/planner/NestedLoopJoinNode.java b/fe/fe-core/src/main/java/org/apache/doris/planner/NestedLoopJoinNode.java
index d9395e7c9a4..49915ad5851 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/planner/NestedLoopJoinNode.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/planner/NestedLoopJoinNode.java
@@ -179,11 +179,9 @@ public class NestedLoopJoinNode extends JoinNodeBase {
public Pair<PlanNode, LocalExchangeType> enforceAndDeriveLocalExchange(PlanTranslatorContext translatorContext,
PlanNode parent, LocalExchangeTypeRequire parentRequire) {
- // Pooling scan: useSerialSource() AND actually has a pooling scan node in the fragment.
- // Without hasSerialScanNode(), useSerialSource() can be true for fragments with only
- // serial Exchanges (use_serial_exchange=true) but no pooling scan, causing false positives.
- boolean childUsePoolingScan = fragment.useSerialSource(translatorContext.getConnectContext())
- && fragment.hasSerialScanNode();
+ // Pooling mode: the fragment uses serial source (pooling scan or serial exchange).
+ // NLJ build side needs BROADCAST in pooling mode so all probe tasks see full build data.
+ boolean childUsePoolingScan = fragment.useSerialSource(translatorContext.getConnectContext());
LocalExchangeTypeRequire probeSideRequire;
LocalExchangeTypeRequire buildSideRequire;
diff --git a/fe/fe-core/src/test/java/org/apache/doris/planner/LocalShuffleNodeCoverageTest.java b/fe/fe-core/src/test/java/org/apache/doris/planner/LocalShuffleNodeCoverageTest.java
index 23f935091a9..8241f6d4c2a 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/planner/LocalShuffleNodeCoverageTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/planner/LocalShuffleNodeCoverageTest.java
@@ -310,8 +310,6 @@ public class LocalShuffleNodeCoverageTest {
Lists.newArrayList(new TupleId(NEXT_ID.getAndIncrement())), JoinOperator.INNER_JOIN, false);
serialSourceJoin.fragment = Mockito.mock(PlanFragment.class);
Mockito.when(serialSourceJoin.fragment.useSerialSource(Mockito.any())).thenReturn(true);
- // childUsePoolingScan requires BOTH useSerialSource AND hasSerialScanNode
- Mockito.when(serialSourceJoin.fragment.hasSerialScanNode()).thenReturn(true);
Pair<PlanNode, LocalExchangeType> serialOutput = serialSourceJoin.enforceAndDeriveLocalExchange(
ctx, null, LocalExchangeTypeRequire.noRequire());
Assertions.assertEquals(LocalExchangeType.ADAPTIVE_PASSTHROUGH, serialOutput.second);
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 37b5b26ad72..dd83c7e5eab 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
@@ -911,6 +911,24 @@ 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)
+ FROM ls_t1
+ ORDER BY k1""")
+
// ================================================================
// Phase 2: Fetch all profiles and compare results
// By now, all queries have been executed and profiles are being
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org