diff --git a/datafusion/physical-optimizer/src/aggregate_statistics.rs b/datafusion/physical-optimizer/src/aggregate_statistics.rs index b9763ffe2895f..b229371fd064d 100644 --- a/datafusion/physical-optimizer/src/aggregate_statistics.rs +++ b/datafusion/physical-optimizer/src/aggregate_statistics.rs @@ -137,12 +137,13 @@ fn take_optimizable(plan: &Arc) -> Option() - && partial_agg_exec.mode().input_mode() == AggregateInputMode::Raw - && partial_agg_exec.group_expr().is_empty() - && partial_agg_exec.filter_expr().iter().all(|e| e.is_none()) - { - return Some(child); + // Stop at the first aggregate: only this `Final`'s own `Partial` is usable. + if let Some(partial_agg_exec) = child.downcast_ref::() { + let usable = partial_agg_exec.mode().input_mode() + == AggregateInputMode::Raw + && partial_agg_exec.group_expr().is_empty() + && partial_agg_exec.filter_expr().iter().all(|e| e.is_none()); + return usable.then(|| Arc::clone(&child)); } if let [childrens_child] = child.children().as_slice() { child = Arc::clone(childrens_child); diff --git a/datafusion/sqllogictest/test_files/aggregate.slt b/datafusion/sqllogictest/test_files/aggregate.slt index 1e12ee4194a49..242aedf1db1ed 100644 --- a/datafusion/sqllogictest/test_files/aggregate.slt +++ b/datafusion/sqllogictest/test_files/aggregate.slt @@ -8853,6 +8853,46 @@ select distinct count() from empty; statement ok DROP TABLE empty; +# rule `aggregate_statistics` should not use the statistics of an aggregate +# inside a subquery to replace an outer aggregate with a FILTER +# https://github.com/apache/datafusion/issues/27220 + +statement ok +CREATE TABLE t AS VALUES (1), (2), (3), (4), (5); + +query I +SELECT count(*) FILTER (WHERE c > 0) AS n FROM (SELECT count(*) AS c FROM t); +---- +1 + +query I +SELECT count(1) FILTER (WHERE c > 0) FROM (SELECT count(1) AS c FROM t); +---- +1 + +query I +SELECT count(*) AS n FROM (SELECT count(*) AS c FROM t); +---- +1 + +query TT +EXPLAIN SELECT count(*) FILTER (WHERE c > 0) AS n FROM (SELECT count(*) AS c FROM t); +---- +logical_plan +01)Projection: count(Int64(1)) FILTER (WHERE c > Int64(0)) AS count(*) FILTER (WHERE c > Int64(0)) AS n +02)--Aggregate: groupBy=[[]], aggr=[[count(Int64(1)) FILTER (WHERE c > Int64(0))]] +03)----Projection: count(Int64(1)) AS count(*) AS c +04)------Aggregate: groupBy=[[]], aggr=[[count(Int64(1))]] +05)--------TableScan: t projection=[] +physical_plan +01)ProjectionExec: expr=[count(Int64(1)) FILTER (WHERE c > Int64(0))@0 as n] +02)--AggregateExec: mode=Single, gby=[], aggr=[count(Int64(1)) FILTER (WHERE c > Int64(0))] +03)----ProjectionExec: expr=[5 as c] +04)------PlaceholderRowExec + +statement ok +DROP TABLE t; + statement ok CREATE TABLE t(col0 INTEGER) as VALUES(2);