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/21 04:30:30 UTC
(doris) 01/01: [fix](local shuffle) Fix BUCKET_SHUFFLE destination routing with pooling scan
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 2fe2d310d4572944823bf29bf3f46e5ad0483e14
Author: 924060929 <la...@selectdb.com>
AuthorDate: Tue Apr 21 11:49:21 2026 +0800
[fix](local shuffle) Fix BUCKET_SHUFFLE destination routing with pooling scan
Root cause: filterInstancesWhichCanReceiveDataFromRemote() had a special
branch for BUCKET_SHUFFLE that returned all instances as destinations,
but BE native local exchange creates only one receiver task per BE.
This mismatch caused rows sent to non-existent receivers to be lost.
Additionally, sortDestinationInstancesByBuckets() used
assignedJoinBucketIndexes (partial bucket set) instead of
scanSource.bucketIndexToScanNodeToTablets.keySet() (full per-worker
bucket set), causing incomplete bucket-to-destination mapping.
Fix: restore baseline behavior — always use getFirstInstancePerWorker
for local shuffle destinations, and use scanSource bucket keys for
destination sorting.
Also removes the hasNonSerialExchangeInFragment() cap in
degreeOfParallelism which incorrectly limited instance count.
---
.../trees/plans/distribute/DistributePlanner.java | 23 +----
.../job/UnassignedScanBucketOlapTableJob.java | 23 +----
.../test_local_shuffle_rqg_bugs.groovy | 102 +++++++++++++++++++++
3 files changed, 105 insertions(+), 43 deletions(-)
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/distribute/DistributePlanner.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/distribute/DistributePlanner.java
index 48a7ba39413..f50a32b2913 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/distribute/DistributePlanner.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/distribute/DistributePlanner.java
@@ -31,7 +31,6 @@ import org.apache.doris.nereids.trees.plans.distribute.worker.job.AssignedJobBui
import org.apache.doris.nereids.trees.plans.distribute.worker.job.BucketScanSource;
import org.apache.doris.nereids.trees.plans.distribute.worker.job.DefaultScanSource;
import org.apache.doris.nereids.trees.plans.distribute.worker.job.LocalShuffleAssignedJob;
-import org.apache.doris.nereids.trees.plans.distribute.worker.job.LocalShuffleBucketJoinAssignedJob;
import org.apache.doris.nereids.trees.plans.distribute.worker.job.StaticAssignedJob;
import org.apache.doris.nereids.trees.plans.distribute.worker.job.UnassignedJob;
import org.apache.doris.nereids.trees.plans.distribute.worker.job.UnassignedJobBuilder;
@@ -245,13 +244,6 @@ public class DistributePlanner {
boolean useLocalShuffle = receiverPlan.getInstanceJobs().stream()
.anyMatch(LocalShuffleAssignedJob.class::isInstance);
if (useLocalShuffle) {
- // For BUCKET_SHUFFLE with local shuffle, each instance handles specific buckets.
- // The exchange must route data to each instance individually, not just the first
- // per worker. Otherwise instances 1..N-1 create ExchangeSource tasks that never
- // receive data or EOS, causing a hang.
- if (linkNode.getPartitionType() == TPartitionType.BUCKET_SHFFULE_HASH_PARTITIONED) {
- return receiverPlan.getInstanceJobs();
- }
return getFirstInstancePerWorker(receiverPlan.getInstanceJobs());
} else if (enableShareHashTableForBroadcastJoin && linkNode.isRightChildOfBroadcastHashJoin()) {
return getFirstInstancePerWorker(receiverPlan.getInstanceJobs());
@@ -264,19 +256,8 @@ public class DistributePlanner {
PipelineDistributedPlan plan, List<AssignedJob> unsorted, int bucketNum) {
AssignedJob[] instances = new AssignedJob[bucketNum];
for (AssignedJob instanceJob : unsorted) {
- // For LocalShuffleBucketJoinAssignedJob, use assignedJoinBucketIndexes which
- // gives the specific join bucket(s) this instance handles. The scanSource
- // buckets include ALL buckets on this worker (shared via pooling scan),
- // which would cause collisions when multiple instances on the same worker
- // each handle different join buckets.
- Iterable<Integer> bucketIndices;
- if (instanceJob instanceof LocalShuffleBucketJoinAssignedJob) {
- bucketIndices = ((LocalShuffleBucketJoinAssignedJob) instanceJob)
- .getAssignedJoinBucketIndexes();
- } else {
- BucketScanSource bucketScanSource = (BucketScanSource) instanceJob.getScanSource();
- bucketIndices = bucketScanSource.bucketIndexToScanNodeToTablets.keySet();
- }
+ BucketScanSource bucketScanSource = (BucketScanSource) instanceJob.getScanSource();
+ Iterable<Integer> bucketIndices = bucketScanSource.bucketIndexToScanNodeToTablets.keySet();
for (Integer bucketIndex : bucketIndices) {
if (instances[bucketIndex] != null) {
throw new IllegalStateException(
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/distribute/worker/job/UnassignedScanBucketOlapTableJob.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/distribute/worker/job/UnassignedScanBucketOlapTableJob.java
index 4b30e832229..b6319da5b9b 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/distribute/worker/job/UnassignedScanBucketOlapTableJob.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/distribute/worker/job/UnassignedScanBucketOlapTableJob.java
@@ -539,27 +539,6 @@ public class UnassignedScanBucketOlapTableJob extends AbstractUnassignedScanJob
}
int maxParallelism = (int) Math.max(tabletNum, fragment.getParallelExecNum());
- int result = Math.min(maxParallelism, colocateMaxParallelNum);
-
- // When the fragment has a non-serial exchange, the BE won't insert local exchange
- // for the exchange pipeline (distribution matches BUCKET_HASH_SHUFFLE). Instances
- // beyond per-BE bucket count are "padding" with no bucket assignment — they create
- // VDataStreamRecvrs that never receive data and hang. Cap at per-BE bucket count.
- if (useLocalShuffleToAddParallel && hasNonSerialExchangeInFragment()) {
- result = Math.min(result, maxParallel);
- }
-
- return result;
- }
-
- private boolean hasNonSerialExchangeInFragment() {
- List<ExchangeNode> exchanges = fragment.getPlanRoot()
- .collectInCurrentFragment(ExchangeNode.class::isInstance);
- for (ExchangeNode exchange : exchanges) {
- if (!exchange.isSerialNode()) {
- return true;
- }
- }
- return false;
+ return Math.min(maxParallelism, colocateMaxParallelNum);
}
}
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 8688a413c9d..7c75adcbd1a 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
@@ -1378,5 +1378,107 @@ suite("test_local_shuffle_rqg_bugs") {
assertTrue(false, "Bug 23: canUseDistinctStreamingAgg + GROUPING SETS: ${t.message}")
}
+ // ============================================================
+ // Bug 24: BUCKET_SHUFFLE join + pooling scan + local shuffle
+ // causes data loss when destination routing uses all
+ // instances instead of firstInstancePerWorker.
+ //
+ // Root cause: filterInstancesWhichCanReceiveDataFromRemote() had
+ // a special branch for BUCKET_SHUFFLE that returned all instances
+ // (40) as destinations, but BE native local exchange creates only
+ // 4 receiver tasks (one per BE). Destination mismatch causes rows
+ // sent to non-existent receivers to be lost.
+ //
+ // Trigger conditions:
+ // - BUCKET_SHUFFLE join plan (requires multi-BE + specific pptn)
+ // - ignore_storage_data_distribution=true (pooling scan)
+ // - enable_local_shuffle=true
+ // - pptn that makes scan serial (scanRanges < pptn * numBE)
+ // ============================================================
+ try {
+ logger.info("Bug 24: BUCKET_SHUFFLE + pooling scan destination routing")
+ sql "DROP TABLE IF EXISTS bug24_t1"
+ sql "DROP TABLE IF EXISTS bug24_t2"
+
+ sql """
+ CREATE TABLE bug24_t1 (
+ pk INT NOT NULL,
+ val VARCHAR(64),
+ INDEX idx_val (val) USING INVERTED
+ ) ENGINE=OLAP
+ DUPLICATE KEY(pk)
+ DISTRIBUTED BY HASH(pk) BUCKETS 10
+ PROPERTIES ("replication_num" = "1")
+ """
+
+ sql """
+ CREATE TABLE bug24_t2 (
+ pk INT NOT NULL,
+ val VARCHAR(64),
+ INDEX idx_val (val) USING INVERTED
+ ) ENGINE=OLAP
+ DUPLICATE KEY(pk)
+ DISTRIBUTED BY HASH(pk) BUCKETS 10
+ PROPERTIES ("replication_num" = "1")
+ """
+
+ // Insert 20 rows into t1, 50 into t2
+ for (int i = 1; i <= 20; i++) {
+ sql "INSERT INTO bug24_t1 VALUES (${i}, 'row_${i}')"
+ }
+ for (int i = 1; i <= 50; i++) {
+ sql "INSERT INTO bug24_t2 VALUES (${i}, 'row_${i}')"
+ }
+
+ // Baseline: no local shuffle
+ def bug24_baseline = sql """
+ SELECT /*+SET_VAR(enable_local_shuffle=false,enable_sql_cache=false)*/
+ count(*) FROM (
+ SELECT * FROM bug24_t1 AS t1
+ LEFT JOIN (SELECT * FROM bug24_t2) AS t2 ON t1.pk = t2.pk
+ ORDER BY t2.pk DESC, t1.pk DESC LIMIT 21
+ ) t
+ """
+
+ // Test with multiple pptn values to catch the specific trigger
+ for (int pptn : [1, 2, 3, 4, 5, 8, 10]) {
+ def result = sql """
+ SELECT /*+SET_VAR(
+ parallel_pipeline_task_num=${pptn},
+ ignore_storage_data_distribution=true,
+ enable_local_shuffle=true,
+ enable_local_shuffle_planner=false,
+ enable_sql_cache=false
+ )*/ count(*) FROM (
+ SELECT * FROM bug24_t1 AS t1
+ LEFT JOIN (SELECT * FROM bug24_t2) AS t2 ON t1.pk = t2.pk
+ ORDER BY t2.pk DESC, t1.pk DESC LIMIT 21
+ ) t
+ """
+ assertEquals(bug24_baseline, result,
+ "Bug 24 pptn=${pptn} planner=false: BUCKET_SHUFFLE+pooling result mismatch")
+
+ def result2 = sql """
+ SELECT /*+SET_VAR(
+ parallel_pipeline_task_num=${pptn},
+ ignore_storage_data_distribution=true,
+ enable_local_shuffle=true,
+ enable_local_shuffle_planner=true,
+ enable_sql_cache=false
+ )*/ count(*) FROM (
+ SELECT * FROM bug24_t1 AS t1
+ LEFT JOIN (SELECT * FROM bug24_t2) AS t2 ON t1.pk = t2.pk
+ ORDER BY t2.pk DESC, t1.pk DESC LIMIT 21
+ ) t
+ """
+ assertEquals(bug24_baseline, result2,
+ "Bug 24 pptn=${pptn} planner=true: BUCKET_SHUFFLE+pooling result mismatch")
+ }
+ logger.info("Bug 24: PASSED")
+ } catch (Throwable t) {
+ logger.error("Bug 24 FAILED: ${t.message}")
+ assertTrue(false, "Bug 24: BUCKET_SHUFFLE+pooling destination routing: ${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