This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.1 by this push:
new 2267b487300 branch-4.1: [fix](test) stabilize the flaky
shuffle_left_join regression test #65769 (#65839)
2267b487300 is described below
commit 2267b487300afec461aa712f38631cfa46ddb8a0
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Sat Aug 29 18:04:44 2026 +0800
branch-4.1: [fix](test) stabilize the flaky shuffle_left_join regression
test #65769 (#65839)
Cherry-picked from #65769
---------
Co-authored-by: 924060929 <[email protected]>
---
.../properties/ChildrenPropertiesRegulator.java | 9 ++++++++-
.../java/org/apache/doris/qe/SessionVariable.java | 19 +++++++++++++++++++
.../distribute/shuffle_left_join.groovy | 18 ++++++++++++++++--
3 files changed, 43 insertions(+), 3 deletions(-)
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/properties/ChildrenPropertiesRegulator.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/properties/ChildrenPropertiesRegulator.java
index c15c43a58a2..43aa2d92830 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/properties/ChildrenPropertiesRegulator.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/properties/ChildrenPropertiesRegulator.java
@@ -304,7 +304,14 @@ public class ChildrenPropertiesRegulator extends
PlanVisitor<List<List<PhysicalP
int bucketNum =
candidate.getTable().getDefaultDistributionInfo().getBucketNum();
int totalBucketNum = prunedPartNum * bucketNum;
ConnectContext connectContext = ConnectContext.get();
- return totalBucketNum < connectContext.getTotalInstanceNum() *
0.8;
+ // <= 0 disables the downgrade entirely, so a test or a tuning
session can keep
+ // bucket shuffle (the anchored side needs no re-shuffle)
regardless of how many
+ // instances the cluster has.
+ double downgradeRatio =
connectContext.getSessionVariable().getBucketShuffleDowngradeRatio();
+ if (downgradeRatio <= 0) {
+ return false;
+ }
+ return totalBucketNum < connectContext.getTotalInstanceNum() *
downgradeRatio;
}
}
}
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 f6076510641..6425c3fc35c 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
@@ -374,6 +374,8 @@ public class SessionVariable implements Serializable,
Writable {
public static final String FORCE_TO_LOCAL_SHUFFLE =
"force_to_local_shuffle";
+ public static final String BUCKET_SHUFFLE_DOWNGRADE_RATIO =
"bucket_shuffle_downgrade_ratio";
+
public static final String ENABLE_LOCAL_MERGE_SORT =
"enable_local_merge_sort";
public static final String ENABLE_SHARED_EXCHANGE_SINK_BUFFER =
"enable_shared_exchange_sink_buffer";
@@ -1680,6 +1682,15 @@ public class SessionVariable implements Serializable,
Writable {
"Whether to force to local shuffle on pipelineX
engine."})
private boolean forceToLocalShuffle = false;
+ @VariableMgr.VarAttr(
+ name = BUCKET_SHUFFLE_DOWNGRADE_RATIO, fuzzy = false, varType =
VariableAnnotation.EXPERIMENTAL,
+ description = {"当一侧基表总桶数小于总实例数的该倍数时, 放弃bucket shuffle
join降级为shuffle join。"
+ + "小于等于0时永不降级。默认0.8保持原有行为",
+ "Downgrade bucket shuffle join to shuffle join when the
base table side's total"
+ + " bucket count is less than total instance count times
this ratio. Values <= 0"
+ + " never downgrade. Default 0.8 keeps the original
behavior."}, needForward = true)
+ private double bucketShuffleDowngradeRatio = 0.8;
+
@VariableMgr.VarAttr(name = ENABLE_LOCAL_MERGE_SORT)
private boolean enableLocalMergeSort = true;
@@ -6542,6 +6553,14 @@ public class SessionVariable implements Serializable,
Writable {
this.forceToLocalShuffle = forceToLocalShuffle;
}
+ public double getBucketShuffleDowngradeRatio() {
+ return bucketShuffleDowngradeRatio;
+ }
+
+ public void setBucketShuffleDowngradeRatio(double
bucketShuffleDowngradeRatio) {
+ this.bucketShuffleDowngradeRatio = bucketShuffleDowngradeRatio;
+ }
+
public boolean isFetchAllFeForSystemTable() {
return fetchAllFeForSystemTable;
}
diff --git
a/regression-test/suites/nereids_syntax_p0/distribute/shuffle_left_join.groovy
b/regression-test/suites/nereids_syntax_p0/distribute/shuffle_left_join.groovy
index 4ee14eba481..356bf9e5fd4 100644
---
a/regression-test/suites/nereids_syntax_p0/distribute/shuffle_left_join.groovy
+++
b/regression-test/suites/nereids_syntax_p0/distribute/shuffle_left_join.groovy
@@ -18,6 +18,17 @@ import java.util.stream.Collectors
// under the License.
suite("shuffle_left_join") {
+ // The point of this suite is the left-to-right bucket shuffle: with the
nereids distribute
+ // planner on, the aggregated left side is shuffled onto the right table's
storage buckets, so
+ // the join becomes a bucket shuffle join that saves one exchange versus a
plain partitioned
+ // shuffle. The planner chooses between the two candidates by cost, which
depends on the scan
+ // row count and on the bucket-shuffle downgrade gate. Both are pinned
below so the asserted
+ // plan is stable:
+ // - `analyze ... with sync` fixes the row count (otherwise it is
reported asynchronously
+ // after the insert, and the plan flips depending on whether the
report has landed yet);
+ // - `bucket_shuffle_downgrade_ratio=0` disables the downgrade that
turns bucket shuffle back
+ // into a partitioned shuffle when the bucket count is small relative
to the instance count,
+ // which otherwise makes the plan depend on the number of backends.
multi_sql """
drop table if exists test_shuffle_left;
@@ -35,11 +46,14 @@ suite("shuffle_left_join") {
sync;
+ analyze table test_shuffle_left with sync;
+
set enable_nereids_distribute_planner=false;
set enable_pipeline_x_engine=true;
set disable_join_reorder=true;
set enable_local_shuffle=false;
set force_to_local_shuffle=false;
+ set bucket_shuffle_downgrade_ratio=0;
"""
def extractFragment = { String sqlStr, String containsString,
Closure<Integer> checkExchangeNum ->
@@ -95,8 +109,8 @@ suite("shuffle_left_join") {
.collect(Collectors.joining("\n"))
logger.info("Variables:\n${variableString}")
- extractFragment(sqlStr, "INNER JOIN(PARTITIONED)") { exchangeNum ->
- assertTrue(exchangeNum == 2)
+ extractFragment(sqlStr, "INNER JOIN(BUCKET_SHUFFLE)") { exchangeNum ->
+ assertTrue(exchangeNum == 1)
}
explain {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
