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 f2aeb639446 branch-4.1:[fix](eager-agg) Handle duplicate aggregate
functions pushed through projects #66531 (#66741)
f2aeb639446 is described below
commit f2aeb6394463a5742600b248f3fd04b4b2837deb
Author: feiniaofeiafei <[email protected]>
AuthorDate: Fri Aug 14 09:26:11 2026 +0800
branch-4.1:[fix](eager-agg) Handle duplicate aggregate functions pushed
through projects #66531 (#66741)
picked from #66531
---
.../rewrite/eageraggregation/EagerAggRewriter.java | 78 +++++++++++++---------
.../nereids_p0/eager_agg/bilateral_eager_agg.out | 4 ++
.../data/nereids_p0/eager_agg/eager_agg.out | 8 ++-
.../eager_agg/bilateral_eager_agg.groovy | 74 ++++++++++++++++++++
4 files changed, 130 insertions(+), 34 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 3a022f10c01..bc1b8ee133f 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
@@ -61,11 +61,13 @@ import org.apache.doris.qe.SessionVariable;
import org.apache.doris.statistics.ColumnStatistic;
import org.apache.doris.statistics.Statistics;
+import com.google.common.collect.ImmutableList;
import com.google.common.collect.Lists;
import java.util.ArrayList;
+import java.util.HashMap;
import java.util.HashSet;
-import java.util.IdentityHashMap;
+import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Optional;
@@ -135,8 +137,8 @@ public class EagerAggRewriter extends
DefaultPlanRewriter<PushDownAggContext> {
// construct left and right aggFuncs and aliasMap
List<AggregateFunction> leftFuncs = new ArrayList<>();
List<AggregateFunction> rightFuncs = new ArrayList<>();
- Map<AggregateFunction, Alias> leftAliasMap = new IdentityHashMap<>();
- Map<AggregateFunction, Alias> rightAliasMap = new IdentityHashMap<>();
+ Map<AggregateFunction, Alias> leftAliasMap = new HashMap<>();
+ Map<AggregateFunction, Alias> rightAliasMap = new HashMap<>();
for (AggregateFunction f : context.getAggFunctions()) {
Set<Slot> inputs = f.getInputSlots();
Alias a = context.getAliasMap().get(f);
@@ -198,7 +200,7 @@ public class EagerAggRewriter extends
DefaultPlanRewriter<PushDownAggContext> {
}
private boolean isPassThroughHeavyJoin(Plan joinChild, PushDownAggContext
context) {
- if (context.isPassThroughHeavyJoin() ||
SessionVariable.getEagerAggregationMode() > 0) {
+ if (context.isPassThroughHeavyJoin()) {
return true;
} else {
Statistics stats = joinChild.getStats();
@@ -405,7 +407,7 @@ public class EagerAggRewriter extends
DefaultPlanRewriter<PushDownAggContext> {
private PushDownAggContext createContextFromProject(
LogicalProject<? extends Plan> project,
- PushDownAggContext context) {
+ PushDownAggContext context, Map<ExprId, ExprId>
projectToChildExprIdMap) {
/*
* context: sum(a) groupBy(y+z as x, l)
* proj: b+c as a, u+v as y, m+n as l
@@ -419,35 +421,38 @@ public class EagerAggRewriter extends
DefaultPlanRewriter<PushDownAggContext> {
.stream().map(slot -> (SlotReference)
slot).collect(Collectors.toList()));
}
- List<AggregateFunction> aggFunctions = new ArrayList<>();
- Map<AggregateFunction, Alias> aliasMap = new IdentityHashMap<>();
+ Set<AggregateFunction> aggFunctions = new LinkedHashSet<>();
+ Map<AggregateFunction, Alias> aliasMap = new HashMap<>();
+ boolean newContainsNullToNonNull = context.hasCaseWhen;
for (AggregateFunction aggFunc : context.getAggFunctions()) {
AggregateFunction newAggFunc = (AggregateFunction)
project.pushDownExpressionPastProject(aggFunc);
Alias alias = context.getAliasMap().get(aggFunc);
- aliasMap.put(newAggFunc, (Alias) alias.withChildren(newAggFunc));
+ Alias aliasForChild;
+ if (aliasMap.containsKey(newAggFunc)) {
+ aliasForChild = aliasMap.get(newAggFunc);
+ } else {
+ aliasForChild = (Alias) alias.withChildren(newAggFunc);
+ aliasMap.put(newAggFunc, aliasForChild);
+ }
+ projectToChildExprIdMap.put(alias.getExprId(),
aliasForChild.getExprId());
aggFunctions.add(newAggFunc);
- }
- // After pushing expressions past the project, the agg functions may
now
- // contain If/CaseWhen that were hidden behind slot references before.
- // e.g. count(#slot) where #slot = if(cond, a, b) in the project.
- // We must re-check and update hasCaseWhen accordingly.
- boolean newHasCaseWhen = context.hasCaseWhen;
- if (!newHasCaseWhen) {
- for (AggregateFunction aggFunc : aggFunctions) {
- if (aggFunc.children().stream().anyMatch(
- arg -> arg.anyMatch(e ->
-
NullToNonNullFunction.canConvertNullToNonNull((Expression) e)))) {
- newHasCaseWhen = true;
- break;
- }
+ // After pushing expressions past the project, the agg functions
may now
+ // contain NullToNonNull expressions that were hidden behind slot
references before.
+ // e.g. count(#slot) where #slot = coalesce(a, 0) in the project.
+ // We must re-check and update containsNullToNonNull accordingly.
+ if (!newContainsNullToNonNull
+ && newAggFunc.children().stream().anyMatch(
+ arg -> arg.anyMatch(e ->
+
NullToNonNullFunction.canConvertNullToNonNull((Expression) e)))) {
+ newContainsNullToNonNull = true;
}
}
- PushDownAggContext newContext = new PushDownAggContext(aggFunctions,
groupKeys, aliasMap,
+
+ return new PushDownAggContext(ImmutableList.copyOf(aggFunctions),
groupKeys, aliasMap,
context.getCascadesContext(), context.isPassThroughHeavyJoin(),
- context.hasDecomposedAggIf, newHasCaseWhen,
+ context.hasDecomposedAggIf, newContainsNullToNonNull,
context.getBilateralState(), context.needOutputCount(),
context.isPassThroughJoinOrUnion(),
context.isSmallBroadcastBottomJoin());
- return newContext;
}
private boolean canPushThroughProject(LogicalProject<? extends Plan>
project, PushDownAggContext context) {
@@ -557,7 +562,7 @@ public class EagerAggRewriter extends
DefaultPlanRewriter<PushDownAggContext> {
Plan child = union.children().get(idx);
final int childIdx = idx;
List<AggregateFunction> aggFunctionsForChild = new ArrayList<>();
- IdentityHashMap<AggregateFunction, Alias> aliasMapForChild = new
IdentityHashMap<>();
+ Map<AggregateFunction, Alias> aliasMapForChild = new HashMap<>();
for (AggregateFunction func : context.getAggFunctions()) {
AggregateFunction newFunc = (AggregateFunction)
union.pushDownExpressionPastSetOperator(func, childIdx);
aggFunctionsForChild.add(newFunc);
@@ -686,7 +691,8 @@ public class EagerAggRewriter extends
DefaultPlanRewriter<PushDownAggContext> {
if (!canPushThroughProject(project, context)) {
return genAggregate(project, context);
}
- PushDownAggContext newContext = createContextFromProject(project,
context);
+ Map<ExprId, ExprId> projectToChildExprIdMap = new HashMap<>();
+ PushDownAggContext newContext = createContextFromProject(project,
context, projectToChildExprIdMap);
if (newContext.aggFuncAndGroupKeyAllEmpty()) {
return project;
}
@@ -714,9 +720,19 @@ public class EagerAggRewriter extends
DefaultPlanRewriter<PushDownAggContext> {
BilateralState state = context.getBilateralState();
for (AggregateFunction aggFunc : context.getAggFunctions()) {
Alias alias = context.getAliasMap().get(aggFunc);
- NamedExpression namedExpression =
state.getPushedAggFuncSlot(alias.getExprId());
- newProjections.add(namedExpression.toSlot());
+ ExprId childExprId =
projectToChildExprIdMap.get(alias.getExprId());
+ NamedExpression namedExpression =
state.getPushedAggFuncSlot(childExprId);
+ NamedExpression output;
+ if (namedExpression.getExprId().equals(alias.getExprId())) {
+ output = namedExpression.toSlot();
+ } else {
+ output = (Alias)
alias.withChildren(namedExpression.toSlot());
+ state.registerAggFuncOutput(alias.getExprId(),
output.toSlot(),
+ state.isAggFuncActuallyPushed(childExprId));
+ }
+ newProjections.add(output);
}
+
for (SlotReference slot : context.getGroupKeys()) {
boolean valid = false;
for (NamedExpression ne : project.getProjects()) {
@@ -1302,9 +1318,7 @@ public class EagerAggRewriter extends
DefaultPlanRewriter<PushDownAggContext> {
}
if (mode > 0) {
- // when mode=1, any join is regarded as big join in order to
- // push down aggregation through at least one join
- return context.isPassThroughHeavyJoin();
+ return true;
}
if (!context.isPassThroughHeavyJoin() && !context.hasDecomposedAggIf) {
diff --git a/regression-test/data/nereids_p0/eager_agg/bilateral_eager_agg.out
b/regression-test/data/nereids_p0/eager_agg/bilateral_eager_agg.out
index a386ad329a1..1ff0e1a9dc8 100644
--- a/regression-test/data/nereids_p0/eager_agg/bilateral_eager_agg.out
+++ b/regression-test/data/nereids_p0/eager_agg/bilateral_eager_agg.out
@@ -335,3 +335,7 @@
2000-06-03 true
2020-01-01 false
+-- !union_2_same_agg_func --
+1 10 10
+2 20 20
+
diff --git a/regression-test/data/nereids_p0/eager_agg/eager_agg.out
b/regression-test/data/nereids_p0/eager_agg/eager_agg.out
index 71acc3997b4..71f0fa7f701 100644
--- a/regression-test/data/nereids_p0/eager_agg/eager_agg.out
+++ b/regression-test/data/nereids_p0/eager_agg/eager_agg.out
@@ -307,8 +307,12 @@ PhysicalResultSink
------PhysicalUnion
--------hashJoin[INNER_JOIN] hashCondition=((dt.d_date_sk =
ss.ss_sold_date_sk)) otherCondition=()
----------PhysicalOlapScan[store_sales]
-----------PhysicalOlapScan[date_dim]
---------PhysicalOlapScan[date_dim]
+----------hashAgg[GLOBAL]
+------------hashAgg[LOCAL]
+--------------PhysicalOlapScan[date_dim]
+--------hashAgg[GLOBAL]
+----------hashAgg[LOCAL]
+------------PhysicalOlapScan[date_dim]
Hint log:
Used:
diff --git
a/regression-test/suites/nereids_p0/eager_agg/bilateral_eager_agg.groovy
b/regression-test/suites/nereids_p0/eager_agg/bilateral_eager_agg.groovy
index 76c1cc5f1d7..d59fc4e1896 100644
--- a/regression-test/suites/nereids_p0/eager_agg/bilateral_eager_agg.groovy
+++ b/regression-test/suites/nereids_p0/eager_agg/bilateral_eager_agg.groovy
@@ -950,4 +950,78 @@ suite("bilateral_eager_agg") {
WHERE l.filter_date = '2018-01-08'
GROUP BY group_flag;
"""
+
+ multi_sql """
+ DROP TABLE IF EXISTS src_a;
+ DROP TABLE IF EXISTS src_b;
+ DROP TABLE IF EXISTS src_c;
+
+ CREATE TABLE src_a (
+ k BIGINT NOT NULL,
+ v BIGINT NOT NULL
+ )
+ DUPLICATE KEY(k)
+ DISTRIBUTED BY HASH(k) BUCKETS 1
+ PROPERTIES (
+ "replication_num" = "1"
+ );
+
+ CREATE TABLE src_b (
+ k BIGINT NOT NULL,
+ join_id BIGINT NOT NULL
+ )
+ DUPLICATE KEY(k, join_id)
+ DISTRIBUTED BY HASH(k) BUCKETS 1
+ PROPERTIES (
+ "replication_num" = "1"
+ );
+
+ CREATE TABLE src_c (
+ join_id BIGINT NOT NULL
+ )
+ DUPLICATE KEY(join_id)
+ DISTRIBUTED BY HASH(join_id) BUCKETS 1
+ PROPERTIES (
+ "replication_num" = "1"
+ );
+
+ INSERT INTO src_a VALUES
+ (1, 10),
+ (2, 20);
+
+ INSERT INTO src_b VALUES
+ (1, 101),
+ (2, 102);
+
+ INSERT INTO src_c VALUES
+ (101),
+ (102);
+
+ SET disable_join_reorder = true;
+ SET eager_aggregation_mode = 1;
+ SET fe_debug = true;
+ """
+
+ order_qt_union_2_same_agg_func """
+ SELECT
+ u.k,
+ SUM(u.x) AS sum_x,
+ SUM(u.y) AS sum_y
+ FROM (
+ SELECT
+ a.k,
+ a.v AS x,
+ a.v AS y
+ FROM src_a a
+ UNION ALL
+ SELECT
+ b.k,
+ CAST(0 AS BIGINT) AS x,
+ CAST(0 AS BIGINT) AS y
+ FROM src_b b
+ INNER JOIN src_c c
+ ON b.join_id = c.join_id
+ ) u
+ GROUP BY u.k;
+ """
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
