This is an automated email from the ASF dual-hosted git repository.
HappenLee pushed a commit to branch opt_perf_4.1
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/opt_perf_4.1 by this push:
new 97eddc7008f [enhance] Pick fall back fix and take string into account
(#65799)
97eddc7008f is described below
commit 97eddc7008f63dfa97ee63fc02a916803708ff3d
Author: feiniaofeiafei <[email protected]>
AuthorDate: Mon Jul 20 11:32:22 2026 +0800
[enhance] Pick fall back fix and take string into account (#65799)
- Adjust the “large join” threshold by increasing the build-side
row-count limit from 400K to 1M.
- Allow aggregation to continue pushing down to the current join branch
when the opposite subtree is large or outputs String/Character types, so
that data participating in the join can be reduced earlier.
- Tighten aggregation benefit evaluation: treat Group By keys with NDV >
rowCount / 10 as having extremely low aggregation benefit; for
multi-column Group By, skip aggregation pushdown if any key falls into
this category, avoiding low-benefit pushdown caused by high cardinality.
- Propagate a “bottom small broadcast join” marker in the pushdown
context and apply stricter checks on the probe side: reject pushdown
when Group By column statistics are unknown, or when any Group By key is
estimated to reduce rows by less than 1000×.
- Extend aggregation pushdown traversal to support passing through
LogicalAggregate and LogicalIntersect nodes.
---
.../rewrite/eageraggregation/EagerAggRewriter.java | 54 ++++++++++++++++++----
.../eageraggregation/PushDownAggContext.java | 17 +++++--
.../eageraggregation/PushDownAggregation.java | 3 ++
.../eageraggregation/EagerAggRewriterTest.java | 4 +-
4 files changed, 61 insertions(+), 17 deletions(-)
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/eageraggregation/EagerAggRewriter.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/eageraggregation/EagerAggRewriter.java
index 91721adf935..c947a573153 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/eageraggregation/EagerAggRewriter.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/eageraggregation/EagerAggRewriter.java
@@ -87,10 +87,12 @@ import java.util.stream.Stream;
* ->T2(D)
*/
public class EagerAggRewriter extends DefaultPlanRewriter<PushDownAggContext> {
- public static final int BIG_JOIN_BUILD_SIZE = 400_000;
+ public static final int BIG_JOIN_BUILD_SIZE = 1_000_000;
private static final double LOWER_AGGREGATE_EFFECT_COEFFICIENT = 10000;
private static final double LOW_AGGREGATE_EFFECT_COEFFICIENT = 1000;
private static final double MEDIUM_AGGREGATE_EFFECT_COEFFICIENT = 100;
+ private static final double HIGH_AGGREGATE_EFFECT_COEFFICIENT = 10;
+ private static final double SMALL_BROADCAST_REJECT_COEFFICIENT = 1000;
private static final String JOIN_CNT = "joinCnt";
private final StatsDerive derive = new StatsDerive(false);
@@ -110,9 +112,9 @@ public class EagerAggRewriter extends
DefaultPlanRewriter<PushDownAggContext> {
}
}
ConnectContext connectContext =
context.getCascadesContext().getConnectContext();
+ boolean isSmallBroadcastBottomJoin = isSmallBroadcastJoin(join,
connectContext) && isBottomJoin(join);
if (context.isPassThroughJoinOrUnion() &&
connectContext.getSessionVariable().eagerAggregationOnBroadcastJoin
- && isSmallBroadcastJoin(join, connectContext) &&
isBottomJoin(join)
- && !outputStringType(join.right())) {
+ && isSmallBroadcastBottomJoin &&
!outputStringType(join.right())) {
Plan aggOnJoin = genAggregate(join, context);
if (aggOnJoin != join) {
return aggOnJoin;
@@ -157,15 +159,16 @@ public class EagerAggRewriter extends
DefaultPlanRewriter<PushDownAggContext> {
}
}
- boolean passThroughBigJoin = isPassThroughBigJoin(join, context);
boolean leftNeedOutputCount = needOutputCountForJoinChild(join,
toLeft, toRight,
context.needOutputCount(), rightFuncs);
boolean rightNeedOutputCount = needOutputCountForJoinChild(join,
toRight, toLeft,
context.needOutputCount(), leftFuncs);
Optional<PushDownAggContext> leftChildContext = toLeft ?
Optional.ofNullable(context.forOneBranch(leftFuncs,
- leftAliasMap, leftChildGroupByKeys, passThroughBigJoin,
leftNeedOutputCount)) : Optional.empty();
+ leftAliasMap, leftChildGroupByKeys,
isPassThroughHeavyJoin(join.right(), context),
+ leftNeedOutputCount, isSmallBroadcastBottomJoin)) :
Optional.empty();
Optional<PushDownAggContext> rightChildContext = toRight ?
Optional.ofNullable(context.forOneBranch(rightFuncs,
- rightAliasMap, rightChildGroupByKeys, passThroughBigJoin,
rightNeedOutputCount)) : Optional.empty();
+ rightAliasMap, rightChildGroupByKeys,
isPassThroughHeavyJoin(join.left(), context),
+ rightNeedOutputCount, false)) : Optional.empty();
Plan newLeft = join.left();
Plan newRight = join.right();
@@ -191,6 +194,18 @@ public class EagerAggRewriter extends
DefaultPlanRewriter<PushDownAggContext> {
return newJoin;
}
+ private boolean isPassThroughHeavyJoin(Plan joinChild, PushDownAggContext
context) {
+ if (context.isPassThroughBigJoin() ||
SessionVariable.getEagerAggregationMode() > 0) {
+ return true;
+ } else {
+ Statistics stats = joinChild.getStats();
+ if (stats == null) {
+ stats = joinChild.accept(derive, new
StatsDerive.DeriveContext());
+ }
+ return stats.getRowCount() > BIG_JOIN_BUILD_SIZE ||
outputStringType(joinChild);
+ }
+ }
+
private boolean outputStringType(Plan plan) {
return plan.getOutput().stream().anyMatch(slot -> slot.getDataType()
instanceof CharacterType);
}
@@ -434,7 +449,8 @@ public class EagerAggRewriter extends
DefaultPlanRewriter<PushDownAggContext> {
PushDownAggContext newContext = new PushDownAggContext(aggFunctions,
groupKeys, aliasMap,
context.getCascadesContext(), context.isPassThroughBigJoin(),
context.hasDecomposedAggIf, newHasCaseWhen,
- context.getBilateralState(), context.needOutputCount(),
context.isPassThroughJoinOrUnion());
+ context.getBilateralState(), context.needOutputCount(),
context.isPassThroughJoinOrUnion(),
+ context.isSmallBroadcastBottomJoin());
return newContext;
}
@@ -565,7 +581,7 @@ public class EagerAggRewriter extends
DefaultPlanRewriter<PushDownAggContext> {
PushDownAggContext contextForChild = new
PushDownAggContext(aggFunctionsForChild, groupKeysForChild,
aliasMapForChild, context.getCascadesContext(),
context.isPassThroughBigJoin(),
context.hasDecomposedAggIf, context.hasCaseWhen,
- context.getBilateralState(), context.needOutputCount(),
true);
+ context.getBilateralState(), context.needOutputCount(),
true, false);
inheritHintActionsToUnionChild(context, contextForChild,
aggFunctionsForChild);
Plan newChild = child.accept(this, contextForChild);
if (newChild != child) {
@@ -1272,8 +1288,9 @@ public class EagerAggRewriter extends
DefaultPlanRewriter<PushDownAggContext> {
List<ColumnStatistic> lower = Lists.newArrayList();
List<ColumnStatistic> medium = Lists.newArrayList();
List<ColumnStatistic> high = Lists.newArrayList();
+ List<ColumnStatistic> extremelyHigh = Lists.newArrayList();
- List<ColumnStatistic>[] cards = new List[] { lower, medium, high };
+ List<ColumnStatistic>[] cards = new List[] { lower, medium, high,
extremelyHigh };
for (NamedExpression key : context.getGroupKeys()) {
ColumnStatistic colStats =
ExpressionEstimation.INSTANCE.estimate(key, stats);
@@ -1287,6 +1304,21 @@ public class EagerAggRewriter extends
DefaultPlanRewriter<PushDownAggContext> {
cards[groupByCardinality(colStats,
stats.getRowCount())].add(colStats);
}
+ if (!extremelyHigh.isEmpty() && context.getGroupKeys().size() > 1) {
+ return false;
+ }
+
+ if (context.isSmallBroadcastBottomJoin()) {
+ for (ColumnStatistic colStats : groupKeysStats) {
+ if (colStats.isUnKnown) {
+ return false;
+ }
+ if (colStats.ndv * SMALL_BROADCAST_REJECT_COEFFICIENT >
stats.getRowCount()) {
+ return false;
+ }
+ }
+ }
+
double lowerCartesian = 1.0;
for (ColumnStatistic colStats : lower) {
lowerCartesian = lowerCartesian * colStats.ndv;
@@ -1330,7 +1362,9 @@ public class EagerAggRewriter extends
DefaultPlanRewriter<PushDownAggContext> {
// < LOW_AGGREGATE_EFFECT_COEFFICIENT
// lower(0): row_count / cardinality >= LOW_AGGREGATE_EFFECT_COEFFICIENT
private int groupByCardinality(ColumnStatistic colStats, double rowCount) {
- if (rowCount == 0 || colStats.ndv *
MEDIUM_AGGREGATE_EFFECT_COEFFICIENT > rowCount) {
+ if (rowCount == 0 || colStats.ndv * HIGH_AGGREGATE_EFFECT_COEFFICIENT
> rowCount) {
+ return 3;
+ } else if (colStats.ndv * MEDIUM_AGGREGATE_EFFECT_COEFFICIENT >
rowCount) {
return 2;
} else if (colStats.ndv * MEDIUM_AGGREGATE_EFFECT_COEFFICIENT <=
rowCount
&& colStats.ndv * LOW_AGGREGATE_EFFECT_COEFFICIENT > rowCount)
{
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/eageraggregation/PushDownAggContext.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/eageraggregation/PushDownAggContext.java
index 21805922634..1c72b271541 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/eageraggregation/PushDownAggContext.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/eageraggregation/PushDownAggContext.java
@@ -60,6 +60,7 @@ public class PushDownAggContext {
private final boolean needOutputCount;
private final boolean isPassThroughJoinOrUnion;
+ private final boolean isSmallBroadCastBottomJoin;
// Bilateral push-down plumbing.
// - bilateralState: global, shared by every context in the rewrite
invocation.
@@ -70,7 +71,7 @@ public class PushDownAggContext {
boolean passThroughBigJoin, boolean hasDecomposedAggIf, boolean
hasCaseWhen,
BilateralState bilateralState) {
this(aggFunctions, groupKeys, aliasMap, cascadesContext,
passThroughBigJoin, hasDecomposedAggIf, hasCaseWhen,
- bilateralState, false, false);
+ bilateralState, false, false, false);
}
/**
@@ -79,7 +80,8 @@ public class PushDownAggContext {
public PushDownAggContext(List<AggregateFunction> aggFunctions,
List<SlotReference> groupKeys, Map<AggregateFunction, Alias>
aliasMap, CascadesContext cascadesContext,
boolean passThroughBigJoin, boolean hasDecomposedAggIf, boolean
hasCaseWhen,
- BilateralState bilateralState, boolean needOutputCount, boolean
isPassThroughJoinOrUnion) {
+ BilateralState bilateralState, boolean needOutputCount, boolean
isPassThroughJoinOrUnion,
+ boolean isSmallBroadCastBottomJoin) {
this.groupKeys =
groupKeys.stream().distinct().collect(Collectors.toList());
this.aggFunctions = ImmutableList.copyOf(aggFunctions);
this.cascadesContext = cascadesContext;
@@ -116,6 +118,11 @@ public class PushDownAggContext {
hintAction.ifPresent(action -> bilateralState.putAction(id,
action));
}
this.isPassThroughJoinOrUnion = isPassThroughJoinOrUnion;
+ this.isSmallBroadCastBottomJoin = isSmallBroadCastBottomJoin;
+ }
+
+ public boolean isSmallBroadcastBottomJoin() {
+ return isSmallBroadCastBottomJoin;
}
public boolean isPassThroughJoinOrUnion() {
@@ -149,7 +156,7 @@ public class PushDownAggContext {
public PushDownAggContext withGroupKeys(List<SlotReference> groupKeys) {
return new PushDownAggContext(aggFunctions, groupKeys, aliasMap,
cascadesContext, passThroughBigJoin, hasDecomposedAggIf,
hasCaseWhen,
- bilateralState, needOutputCount, passThroughBigJoin);
+ bilateralState, needOutputCount, passThroughBigJoin,
isSmallBroadCastBottomJoin);
}
/**
@@ -157,13 +164,13 @@ public class PushDownAggContext {
*/
public PushDownAggContext forOneBranch(List<AggregateFunction>
branchAggFunctions,
Map<AggregateFunction, Alias> branchAliasMap, List<SlotReference>
groupKeys,
- boolean passThroughBigJoin, boolean needOutputCount) {
+ boolean passThroughBigJoin, boolean needOutputCount, boolean
isSmallBroadcastBottomJoin) {
if (branchAggFunctions.isEmpty() && groupKeys.isEmpty()) {
return null;
}
return new PushDownAggContext(branchAggFunctions, groupKeys,
branchAliasMap,
cascadesContext, passThroughBigJoin, hasDecomposedAggIf,
hasCaseWhen,
- bilateralState, needOutputCount, true);
+ bilateralState, needOutputCount, true,
isSmallBroadcastBottomJoin);
}
public BilateralState getBilateralState() {
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/eageraggregation/PushDownAggregation.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/eageraggregation/PushDownAggregation.java
index ef75f37a18f..124ef61dc8b 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/eageraggregation/PushDownAggregation.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/eageraggregation/PushDownAggregation.java
@@ -38,6 +38,7 @@ import
org.apache.doris.nereids.trees.expressions.literal.NullLiteral;
import org.apache.doris.nereids.trees.plans.Plan;
import org.apache.doris.nereids.trees.plans.logical.LogicalAggregate;
import org.apache.doris.nereids.trees.plans.logical.LogicalFilter;
+import org.apache.doris.nereids.trees.plans.logical.LogicalIntersect;
import org.apache.doris.nereids.trees.plans.logical.LogicalJoin;
import org.apache.doris.nereids.trees.plans.logical.LogicalProject;
import org.apache.doris.nereids.trees.plans.logical.LogicalRelation;
@@ -76,6 +77,8 @@ public class PushDownAggregation extends
DefaultPlanRewriter<JobContext> impleme
Min.class);
private final Set<Class> acceptNodeType = Sets.newHashSet(
+ LogicalIntersect.class,
+ LogicalAggregate.class,
LogicalUnion.class,
LogicalProject.class,
LogicalFilter.class,
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/rewrite/eageraggregation/EagerAggRewriterTest.java
b/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/rewrite/eageraggregation/EagerAggRewriterTest.java
index 6d6ade4995c..1d025906786 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/rewrite/eageraggregation/EagerAggRewriterTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/rewrite/eageraggregation/EagerAggRewriterTest.java
@@ -476,7 +476,7 @@ class EagerAggRewriterTest extends TestWithFeService
implements MemoPatternMatch
PushDownAggContext context = new PushDownAggContext(
Collections.emptyList(), Collections.emptyList(),
Collections.emptyMap(),
planChecker.getCascadesContext(),
- true, false, false, new BilateralState(), false, true);
+ true, false, false, new BilateralState(), false, true,
false);
Plan rewritten = relation.accept(new EagerAggRewriter(), context);
@@ -640,7 +640,7 @@ class EagerAggRewriterTest extends TestWithFeService
implements MemoPatternMatch
CascadesContext cascadesContext =
PlanChecker.from(connectContext).analyze("select * from t1")
.getCascadesContext();
PushDownAggContext context = new
PushDownAggContext(Collections.emptyList(), Collections.emptyList(),
- Collections.emptyMap(), cascadesContext, true, false, false,
new BilateralState(), true, false);
+ Collections.emptyMap(), cascadesContext, true, false, false,
new BilateralState(), true, false, false);
SlotReference leftCountSlot = new SlotReference("leftCnt",
BigIntType.INSTANCE, false);
SlotReference rightCountSlot = new SlotReference("rightCnt",
BigIntType.INSTANCE, false);
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
