This is an automated email from the ASF dual-hosted git repository.
englefly pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 15e3a17cb26 [improvement](repeat) Optimize shuffle key selection for
Repeat decomposition (#66532)
15e3a17cb26 is described below
commit 15e3a17cb26999537e62a62aa771534707b9afd0
Author: feiniaofeiafei <[email protected]>
AuthorDate: Fri Aug 7 11:30:14 2026 +0800
[improvement](repeat) Optimize shuffle key selection for Repeat
decomposition (#66532)
## What problem does this PR solve?
Optimize shuffle-key selection for Repeat decomposition:
- Allow high-NDV columns to be selected when hot-value statistics are
unavailable.
- Use a stricter NDV threshold (`instanceNum * 128`) to ensure
sufficient distribution.
- Preserve the existing conservative behavior for other skew-detection
scenarios.
---
.../rules/rewrite/DecomposeRepeatWithPreAggregation.java | 13 ++++++++++---
.../org/apache/doris/statistics/util/StatisticsUtil.java | 7 ++++---
.../rewrite/DecomposeRepeatWithPreAggregationTest.java | 2 +-
3 files changed, 15 insertions(+), 7 deletions(-)
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/DecomposeRepeatWithPreAggregation.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/DecomposeRepeatWithPreAggregation.java
index e0f81b2cf36..4c7f2a23e4a 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/DecomposeRepeatWithPreAggregation.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/DecomposeRepeatWithPreAggregation.java
@@ -97,6 +97,7 @@ public class DecomposeRepeatWithPreAggregation extends
DefaultPlanRewriter<Disti
private static final Set<Class<? extends AggregateFunction>>
SUPPORT_AGG_FUNCTIONS =
ImmutableSet.of(Sum.class, Sum0.class, Min.class, Max.class,
AnyValue.class, Count.class);
private static final int DECOMPOSE_REPEAT_THRESHOLD = 3;
+ private static final int BALANCE_MULTIPLIER = 128;
@Override
public Plan rewriteRoot(Plan plan, JobContext jobContext) {
@@ -550,11 +551,10 @@ public class DecomposeRepeatWithPreAggregation extends
DefaultPlanRewriter<Disti
}
for (Expression candidate : candidates) {
ColumnStatistic columnStatistic =
inputStats.findColumnStatistics(candidate);
- if (columnStatistic == null || columnStatistic.isUnKnown() ||
columnStatistic.hotValues == null) {
+ if (columnStatistic == null || columnStatistic.isUnKnown()) {
continue;
}
- if (StatisticsUtil.isBalanced(columnStatistic, totalInstanceNum,
- ShuffleKeyPruneUtils.shuffleKeyHotValueThreshold,
inputStats.getRowCount())) {
+ if (isBalanced(columnStatistic, totalInstanceNum,
inputStats.getRowCount())) {
return Optional.of(candidate);
}
}
@@ -627,4 +627,11 @@ public class DecomposeRepeatWithPreAggregation extends
DefaultPlanRewriter<Disti
new SlotReference(groupingId.getName(),
groupingId.getDataType(), false), remainingGroupingIdValues,
child);
}
+
+ private static boolean isBalanced(ColumnStatistic columnStatistic, int
instanceNum, double rowCount) {
+ double ndv = columnStatistic.ndv;
+ return ndv > instanceNum * BALANCE_MULTIPLIER
+ && !StatisticsUtil.hasSignificantHotValues(columnStatistic,
+ ShuffleKeyPruneUtils.shuffleKeyHotValueThreshold, rowCount,
false);
+ }
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/statistics/util/StatisticsUtil.java
b/fe/fe-core/src/main/java/org/apache/doris/statistics/util/StatisticsUtil.java
index 2cdb451ce4d..e3b67a4dd51 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/statistics/util/StatisticsUtil.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/statistics/util/StatisticsUtil.java
@@ -1078,10 +1078,11 @@ public class StatisticsUtil {
* Used by shuffle key prune and skew detection rules.
* Returns false when hotValues is null (not collected) or empty
(collected but no hot values).
*/
- public static boolean hasSignificantHotValues(ColumnStatistic
columnStatistic, double minRatio, double rowCount) {
+ public static boolean hasSignificantHotValues(ColumnStatistic
columnStatistic, double minRatio, double rowCount,
+ boolean strictWhenHotValueUnknow) {
Map<Literal, Float> hotValues = columnStatistic.getHotValues();
if (hotValues == null) {
- return true;
+ return strictWhenHotValueUnknow;
}
return columnStatistic.numNulls / rowCount > minRatio
|| hotValues.values().stream().anyMatch(ratio -> ratio >=
minRatio);
@@ -1091,6 +1092,6 @@ public class StatisticsUtil {
double rowCount) {
double ndv = columnStatistic.ndv;
return ndv > instanceNum *
AggregateUtils.NDV_INSTANCE_BALANCE_MULTIPLIER
- && !hasSignificantHotValues(columnStatistic, minRatio,
rowCount);
+ && !hasSignificantHotValues(columnStatistic, minRatio,
rowCount, true);
}
}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/rewrite/DecomposeRepeatWithPreAggregationTest.java
b/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/rewrite/DecomposeRepeatWithPreAggregationTest.java
index 66434072777..888a375fb90 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/rewrite/DecomposeRepeatWithPreAggregationTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/rewrite/DecomposeRepeatWithPreAggregationTest.java
@@ -681,7 +681,7 @@ public class DecomposeRepeatWithPreAggregationTest extends
TestWithFeService imp
Optional<Expression> chosen2 = (Optional<Expression>) method.invoke(
rule, groupingSets, -1, candidates, stats, 50);
Assertions.assertTrue(chosen2.isPresent());
- Assertions.assertEquals(b, chosen2.get());
+ Assertions.assertEquals(c, chosen2.get());
// inputStats null -> chooseByNdv returns empty for every group ->
empty
@SuppressWarnings("unchecked")
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
