fix: stop aggregate_statistics walk at the first aggregate below a Final

datafusion/physical-optimizer/src/aggregate_statistics.rs

@@ -137,12 +137,13 @@ fn take_optimizable(plan: &Arc<dyn ExecutionPlan>) -> Option<Arc<dyn ExecutionPl
     {
         let mut child = Arc::clone(agg_exec.input());
         loop {
-            if let Some(partial_agg_exec) = child.downcast_ref::<AggregateExec>()
-                && 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::<AggregateExec>() {
+                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);

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);