This is an automated email from the ASF dual-hosted git repository.
morrySnow 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 b357fc249cc [fix](eager-agg)Forbid pushdown when the child aggregate
has no GROUP BY keys (#65703)
b357fc249cc is described below
commit b357fc249ccd8849cc49ec581ac5f9cd523bbaf2
Author: minghong <[email protected]>
AuthorDate: Fri Jul 17 14:22:49 2026 +0800
[fix](eager-agg)Forbid pushdown when the child aggregate has no GROUP BY
keys (#65703)
### What problem does this PR solve?
Related PR: #63690
Problem Summary:
do not push down a scalar agg to join's child
```
t1 (a, b)
(1,10)
(2,20);
SELECT t1.a, SUM(t1.b), t2.c
FROM t1, t2
WHERE t1.a = 999
GROUP BY t1.a, t2.c
```
plan
```
agg(sum(t1.b) groupkey=[t2.c])
->nlj
->filter(a = 999)
->t1
-> t2
```
without eager-agg, the output of nlj is empty, becase all rows of t1 are
filtered.
if we push down agg uppon filter, the agg is scalar agg, since t1.a is
eliminated by constant propagation.
the plan becomes
```
agg1(sum(x) groupkey=[t2.c])
->nlj
-> agg2(sum(t1.b) as x, groupkey=[])
->filter(a = 999)
->t1
-> t2
```
agg2 output at least one row, and hence nlj output is not empty(t2 is not
empty). and the final result is not empty now.
---
.../rewrite/eageraggregation/EagerAggRewriter.java | 37 ++++++
.../query_p0/eager_agg/scalar_agg_pushdown.out | 41 +++++++
.../query_p0/eager_agg/scalar_agg_pushdown.groovy | 126 +++++++++++++++++++++
3 files changed, 204 insertions(+)
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 748a17f99b3..088de2e0f06 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
@@ -776,6 +776,43 @@ public class EagerAggRewriter extends
DefaultPlanRewriter<PushDownAggContext> {
if (isPushDisabledByVariable(context)) {
return child;
}
+ // Forbid creating a scalar aggregate (aggregate without GROUP BY
keys) during
+ // push-down. A scalar aggregate emits 1 row even when its input is
empty
+ // (SQL standard: SELECT COUNT(*) FROM empty_table returns 1 row with
0).
+ // When placed below a join, that 1 row joins with rows from the other
side,
+ // producing phantom rows that do not exist in the original query.
+ //
+ // This guard is placed at the aggregate-creation boundary rather than
upstream
+ // (e.g. visitLogicalJoin or createContextFromProject) because
context.groupKeys
+ // can change during intermediate rewrites. Checking
groupKeys.isEmpty() early
+ // would either fire on a stale empty state (missing a later filter
that adds
+ // keys) or fail to fire because a constant key has not yet been
resolved to
+ // empty input slots. Example plan that reaches genAggregate with
groupKeys=[]:
+ //
+ // Aggregate(group=[k], sum(z))
+ // CrossJoin
+ // Project(1 AS k, A.v + B.v AS z)
+ // InnerJoin(A.id = B.id)
+ // Scan A
+ // Scan B
+ // Scan R
+ //
+ // Walk:
+ // 1. visitLogicalJoin(crossJoin): parentContext.groupKeys=[k]
+ // → fillGroupByKeys puts k in leftChildGroupByKeys → forOneBranch
succeeds
+ // 2. visitLogicalProject: createContextFromProject maps k through "1
AS k"
+ // → literal 1 has no input slots → newContext.groupKeys=[]
+ // 3. visitLogicalJoin(innerJoin): sum(A.v+B.v) spans both sides
+ // → toLeft=false, toRight=false → falls back to genAggregate with
groupKeys=[]
+ // → without this guard, creates scalar agg below crossJoin →
phantom rows
+ //
+ // If the guard were at visitLogicalJoin step 1, groupKeys=[k]
(non-empty) would
+ // pass. If at createContextFromProject step 2, it would block before
a downstream
+ // filter had the chance to add keys. Only genAggregate sees the final
state.
+ // See regression test: scalar_agg_pushdown.groovy
+ if (context.getGroupKeys().isEmpty()) {
+ return child;
+ }
if (checkStats(child, context) || isPushEnabledByVariable(context)) {
List<NamedExpression> aggOutputExpressions = new ArrayList<>();
for (AggregateFunction func : context.getAggFunctions()) {
diff --git a/regression-test/data/query_p0/eager_agg/scalar_agg_pushdown.out
b/regression-test/data/query_p0/eager_agg/scalar_agg_pushdown.out
new file mode 100644
index 00000000000..8fa4341a049
--- /dev/null
+++ b/regression-test/data/query_p0/eager_agg/scalar_agg_pushdown.out
@@ -0,0 +1,41 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !scalar_agg_no_pushdown --
+
+-- !const_group_key_shape --
+PhysicalResultSink
+--hashAgg[GLOBAL]
+----hashAgg[LOCAL]
+------NestedLoopJoin[CROSS_JOIN]
+--------NestedLoopJoin[INNER_JOIN]
+----------filter((ta.a = 999))
+------------PhysicalOlapScan[t1(ta)]
+----------filter((tb.a = 999))
+------------PhysicalOlapScan[t1(tb)]
+--------PhysicalStorageLayerAggregate[t2]
+
+Hint log:
+Used: leading(ta tb )
+UnUsed:
+SyntaxError:
+
+-- !const_group_key_exe --
+
+-- !pushdown_preserved_shape --
+PhysicalResultSink
+--hashAgg[GLOBAL]
+----hashAgg[LOCAL]
+------NestedLoopJoin[CROSS_JOIN]
+--------hashJoin[INNER_JOIN] hashCondition=((ta.a = tb.a)) otherCondition=()
+----------hashAgg[GLOBAL]
+------------PhysicalOlapScan[t1(ta)]
+----------PhysicalOlapScan[t1(tb)]
+--------PhysicalStorageLayerAggregate[t2]
+
+Hint log:
+Used: leading(ta broadcast tb )
+UnUsed:
+SyntaxError:
+
+-- !pushdown_preserved_exe --
+1 60
+
diff --git
a/regression-test/suites/query_p0/eager_agg/scalar_agg_pushdown.groovy
b/regression-test/suites/query_p0/eager_agg/scalar_agg_pushdown.groovy
new file mode 100644
index 00000000000..64c502184fe
--- /dev/null
+++ b/regression-test/suites/query_p0/eager_agg/scalar_agg_pushdown.groovy
@@ -0,0 +1,126 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+suite("scalar_agg_pushdown") {
+ sql "set eager_aggregation_mode=1;"
+ sql "set eager_aggregation_on_join=true;"
+ sql "set disable_nereids_rules='SALT_JOIN';"
+ sql "set runtime_filter_mode=OFF;"
+ sql 'set ignore_shape_nodes="PhysicalProject, PhysicalDistribute";'
+
+ // Test: scalar aggregation (no GROUP BY) must not be pushed below a join,
+ // because scalar agg on empty input returns 1 NULL row (SQL standard),
+ // which would be cross-joined with the other side to produce phantom rows.
+ // Regression test for: scalar agg pushdown producing phantom rows when
+ // constant propagation replaces GROUP BY key with a literal that doesn't
+ // exist in the table.
+
+ sql """drop table if exists t1;"""
+ sql """create table t1 (a int, b int) distributed by hash(a)
properties("replication_num" = "1");"""
+ sql """insert into t1 values (1,10),(2,20);"""
+
+ sql """drop table if exists t2;"""
+ sql """create table t2 (c int) distributed by hash(c)
properties("replication_num" = "1");"""
+ sql """insert into t2 values (1),(2);"""
+
+ // WHERE t1.a = 999 matches no rows in t1.
+ // GROUP BY includes t1.a, t2.c (cross-table GROUP BY).
+ // Without the fix, eager aggregation pushes a scalar partial aggregate
+ // below the join, which on empty input returns 1 NULL row.
+ // That NULL row cross-joins with t2's 2 rows, producing 2 phantom rows.
+ // With the fix, pushdown is forbidden for scalar child aggregates.
+ qt_scalar_agg_no_pushdown """
+ SELECT t1.a, SUM(t1.b), t2.c
+ FROM t1, t2
+ WHERE t1.a = 999
+ GROUP BY t1.a, t2.c
+ ORDER BY t2.c;
+ """
+
+ // Test: reviewer's example — constant group key k survives
ConstantPropagation
+ // (retained when it is the only GROUP BY key, to keep
group-by-with-literal
+ // semantics), then goes through a Project where createContextFromProject
+ // resolves k to empty input slots. sum(z) = sum(ta.b + tb.b) spans both
+ // inner-join children, so visitLogicalJoin falls back to genAggregate with
+ // groupKeys=[]. Plan after subquery unnesting (t1 self-joined as ta/tb,
+ // t2 as the cross side):
+ //
+ // Aggregate(group=[k], sum(z))
+ // CrossJoin
+ // Project(1 AS k, ta.b + tb.b AS z)
+ // InnerJoin(ta.a = tb.a)
+ // Scan t1 ta
+ // Scan t1 tb
+ // Scan t2
+ //
+ // WHERE ta.a = 999 empties the inner join. Without the guard at
+ // genAggregate(), a scalar aggregate is created above the inner join;
+ // its 0->1 row semantic produces a phantom row that cross-joins with
+ // t2's rows, yielding (1, NULL) instead of an empty result.
+
+ // no scalar aggregate is allowed between the inner join and the cross join
+ qt_const_group_key_shape """
+ explain shape plan
+ SELECT k, SUM(z)
+ FROM (
+ SELECT /*+leading(ta broadcast tb)*/ 1 AS k, ta.b + tb.b AS z
+ FROM t1 ta JOIN t1 tb ON ta.a = tb.a
+ WHERE ta.a = 999
+ ) sub, t2
+ GROUP BY k;
+ """
+
+ order_qt_const_group_key_exe """
+ SELECT k, SUM(z)
+ FROM (
+ SELECT /*+leading(ta broadcast tb)*/ 1 AS k, ta.b + tb.b AS z
+ FROM t1 ta JOIN t1 tb ON ta.a = tb.a
+ WHERE ta.a = 999
+ ) sub, t2
+ GROUP BY k;
+ """
+
+ // Test: valid pushdown must be preserved. Same shape as above, but
+ // z = ta.b comes from one side only. The context reaching the Project
+ // still ends up with groupKeys=[] (k is a constant), yet at the inner
+ // join fillGroupByKeys adds the join key ta.a, so the aggregate finally
+ // pushed onto scan(ta) is a grouped one: agg(group=[a], sum(b)).
+ // This is why the empty-group-keys guard must sit at genAggregate (the
+ // aggregate-creation boundary) instead of firing early when an
+ // intermediate context has no group keys.
+
+ // expect an aggregate pushed below the inner join, on top of scan(ta)
+ qt_pushdown_preserved_shape """
+ explain shape plan
+ SELECT k, SUM(z)
+ FROM (
+ SELECT /*+leading(ta broadcast tb)*/ 1 AS k, ta.b AS z
+ FROM t1 ta JOIN t1 tb ON ta.a = tb.a
+ ) sub, t2
+ GROUP BY k;
+ """
+
+ // (10 + 20) * 2 rows of t2 = 60
+ order_qt_pushdown_preserved_exe """
+ SELECT k, SUM(z)
+ FROM (
+ SELECT /*+leading(ta broadcast tb)*/ 1 AS k, ta.b AS z
+ FROM t1 ta JOIN t1 tb ON ta.a = tb.a
+ ) sub, t2
+ GROUP BY k;
+ """
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
