You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/10/23 05:54:10 UTC
(datafusion) 01/01: fix: keep a subquery correlated when an aggregate sits above a pulled up scalar aggregate (#26544)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-26544-53e121026b5a966d508437af75ca697e56a419c2
in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit fcc4ca8856074d3e07b5ddbaa9c15c962c3afdf1
Author: Oleks V <co...@users.noreply.github.com>
AuthorDate: Fri Oct 23 05:53:45 2026 +0000
fix: keep a subquery correlated when an aggregate sits above a pulled up scalar aggregate (#26544)
## Which issue does this PR close?
- Part of #26508.
## Rationale for this change
An aggregate without `GROUP BY` returns one row even when its input is
empty. When such an aggregate is in a correlated subquery and another
aggregate sits above it, DataFusion returns wrong results for outer rows
that match no inner rows:
```sql
SELECT x.a,
(SELECT count(*) FROM (SELECT max(y.b) FROM (VALUES (1, 10)) AS y(a, b) WHERE y.a = x.a)) AS v1,
(SELECT max(n) FROM (SELECT count(*) AS n FROM (VALUES (1, 10)) AS y(a, b) WHERE y.a = x.a)) AS v2
FROM (VALUES (1), (2)) AS x(a)
```
For `a = 2` this returns `v1 = 0, v2 = NULL` instead of `v1 = 1, v2 =
0`, which DuckDB, SQLite and PostgreSQL return. The same shape gives
wrong results under `IN` and `LEFT JOIN LATERAL` too.
When the correlated filter is pulled up, the inner scalar aggregate
becomes a grouped aggregate, which returns no row for the outer rows it
does not match. The aggregate above then runs over an empty input. The
count bug handling only fixes the aggregate the join reads, which here
is the outer one, so the inner row is never restored. The existing
`TODO` next to `pulled_up_scalar_agg` described this case.
## What changes are included in this PR?
`PullUpCorrelatedExpr` stops the pull up when it reaches an `Aggregate`
after a scalar aggregate has already been pulled up. The subquery stays
correlated, the same way a window over an aggregate with count bug
handling does today. The rewrite stops at that `Aggregate`, so the nodes
above it are not rebuilt over the pulled up plan. Before, the rebuild
could fail with a schema error. Such a query now fails with `Physical
plan does not support logical expression ScalarSubquery` instead of
returning wrong rows.
A projection over the scalar aggregate (`SELECT n + 1 FROM (SELECT
count(*) AS n ...)`) and an aggregate over a grouped aggregate are still
pulled up.
## What is the testing strategy for this PR?
New `sqllogictest` cases at the end of `subquery.slt` cover the two
queries from the issue, the `IN`, `LATERAL` and `EXISTS` forms, a
projection, a filter, a `GROUP BY`-only aggregate and a grouped
aggregate between the aggregates, a `max` over `max` query that is now
blocked, and the two shapes that must still be pulled up. All
`sqllogictest` files, the `datafusion-optimizer` unit tests and the
`core_integration` `sql::` tests (which include the TPC-DS planning
tests) pass locally.
## Are there any user-facing changes?
Queries with this shape now return a "not implemented" error instead of
wrong results. Some queries of this shape returned correct rows before
and now fail too, for example `max` over `max`. Telling them apart needs
the outer aggregate evaluated on the inner one's empty-input row. No API
changes.
---
datafusion/optimizer/src/decorrelate.rs | 13 ++++-
datafusion/sqllogictest/test_files/subquery.slt | 73 +++++++++++++++++++++++++
2 files changed, 84 insertions(+), 2 deletions(-)
diff --git a/datafusion/optimizer/src/decorrelate.rs b/datafusion/optimizer/src/decorrelate.rs
index e6f235da4d..b81f5cb447 100644
--- a/datafusion/optimizer/src/decorrelate.rs
+++ b/datafusion/optimizer/src/decorrelate.rs
@@ -83,6 +83,9 @@ pub struct PullUpCorrelatedExpr {
pub pull_up_having_expr: Option<Expr>,
/// whether we have converted a scalar aggregation into a group aggregation. When unnesting
/// lateral joins, we need to produce a left outer join in such cases.
+ ///
+ /// An `Aggregate` above stops the pull up when this is set. `f_up` visits
+ /// the aggregate that sets it before any node above.
pub pulled_up_scalar_agg: bool,
/// Every correlated conjunct that a `Filter` of the subquery applies,
/// before `remove_duplicated_filter` drops the ones that the `IN`
@@ -556,6 +559,14 @@ impl TreeNodeRewriter for PullUpCorrelatedExpr {
LogicalPlan::Aggregate(aggregate)
if self.in_predicate_opt.is_some() || !self.join_filters.is_empty() =>
{
+ // A pulled up scalar aggregate below gives no row for unmatched
+ // outer rows, so this aggregate would see an empty input. Stop
+ // the rewrite here, since rebuilding the nodes above can fail
+ // with a schema error. See https://github.com/apache/datafusion/issues/26508
+ if self.pulled_up_scalar_agg {
+ self.can_pull_up = false;
+ return Ok(Transformed::new(plan, false, TreeNodeRecursion::Stop));
+ }
// If the aggregation is from a distinct it will not change the result for
// exists/in subqueries so we can still pull up all predicates.
let is_distinct = aggregate.aggr_expr.is_empty();
@@ -647,8 +658,6 @@ impl TreeNodeRewriter for PullUpCorrelatedExpr {
}
}
if aggregate.group_expr.is_empty() {
- // TODO: how do we handle the case where we have pulled multiple aggregations? For example,
- // a group agg with a scalar agg as child.
self.pulled_up_scalar_agg = true;
}
let new_plan = LogicalPlanBuilder::from((*aggregate.input).clone())
diff --git a/datafusion/sqllogictest/test_files/subquery.slt b/datafusion/sqllogictest/test_files/subquery.slt
index a4474c4b03..45ec0345f7 100644
--- a/datafusion/sqllogictest/test_files/subquery.slt
+++ b/datafusion/sqllogictest/test_files/subquery.slt
@@ -4003,3 +4003,76 @@ DROP TABLE unnest_outer;
statement ok
DROP TABLE unnest_inner;
+
+# An aggregate over a correlated scalar aggregate stays correlated.
+# https://github.com/apache/datafusion/issues/26508
+statement ok
+CREATE TABLE nested_agg_outer(k INT) AS VALUES (1), (2);
+
+statement ok
+CREATE TABLE nested_agg_inner(id INT, v INT) AS VALUES (1, 10);
+
+# Errors rather than returning wrong rows. Expected `(1, 1), (2, 1)`.
+query error Physical plan does not support logical expression ScalarSubquery
+SELECT o.k, (SELECT count(*) FROM (SELECT max(i.v) FROM nested_agg_inner AS i WHERE i.id = o.k)) FROM nested_agg_outer AS o ORDER BY o.k;
+
+# Errors rather than returning wrong rows. Expected `(1, 1), (2, 0)`.
+query error Physical plan does not support logical expression ScalarSubquery
+SELECT o.k, (SELECT max(n) FROM (SELECT count(*) AS n FROM nested_agg_inner AS i WHERE i.id = o.k)) FROM nested_agg_outer AS o ORDER BY o.k;
+
+# IN. Errors rather than returning wrong rows. Expected `2`.
+query error Physical plan does not support logical expression InSubquery
+SELECT o.k FROM nested_agg_outer AS o WHERE 0 IN (SELECT max(n) FROM (SELECT count(*) AS n FROM nested_agg_inner AS i WHERE i.id = o.k)) ORDER BY o.k;
+
+# LATERAL. Errors rather than returning wrong rows. Expected `(1, 1), (2, 0)`.
+query error Physical plan does not support logical expression OuterReferenceColumn
+SELECT o.k, s.m FROM nested_agg_outer AS o LEFT JOIN LATERAL (SELECT max(n) AS m FROM (SELECT count(*) AS n FROM nested_agg_inner AS i WHERE i.id = o.k)) AS s ON true ORDER BY o.k;
+
+# EXISTS. Errors rather than returning wrong rows. Expected `1, 2`.
+query error Physical plan does not support logical expression Exists
+SELECT o.k FROM nested_agg_outer AS o WHERE EXISTS (SELECT max(n) FROM (SELECT count(*) AS n FROM nested_agg_inner AS i WHERE i.id = o.k)) ORDER BY o.k;
+
+# A projection between the aggregates. Errors rather than returning wrong rows. Expected `(1, 2), (2, 1)`.
+query error Physical plan does not support logical expression ScalarSubquery
+SELECT o.k, (SELECT max(n2) FROM (SELECT n + 1 AS n2 FROM (SELECT count(*) AS n FROM nested_agg_inner AS i WHERE i.id = o.k))) FROM nested_agg_outer AS o ORDER BY o.k;
+
+# A filter between the aggregates. Errors rather than returning wrong rows.
+# Expected `(1, 1), (2, 1)`.
+query error Physical plan does not support logical expression ScalarSubquery
+SELECT o.k, (SELECT count(*) FROM (SELECT count(*) AS n FROM nested_agg_inner AS i WHERE i.id = o.k) WHERE n >= 0) FROM nested_agg_outer AS o ORDER BY o.k;
+
+# A GROUP BY-only aggregate over the scalar aggregate.
+# Errors rather than returning wrong rows. Expected `(1, 1), (2, 1)`.
+query error Physical plan does not support logical expression ScalarSubquery
+SELECT o.k, (SELECT count(*) FROM (SELECT n FROM (SELECT count(*) AS n FROM nested_agg_inner AS i WHERE i.id = o.k) GROUP BY n)) FROM nested_agg_outer AS o ORDER BY o.k;
+
+# A grouped aggregate over the scalar aggregate.
+# Errors rather than returning wrong rows. Expected `(1, 1), (2, 1)`.
+query error Physical plan does not support logical expression ScalarSubquery
+SELECT o.k, (SELECT max(c) FROM (SELECT n, count(*) AS c FROM (SELECT count(*) AS n FROM nested_agg_inner AS i WHERE i.id = o.k) GROUP BY n)) FROM nested_agg_outer AS o ORDER BY o.k;
+
+# Returned the right rows before, but is blocked too, since telling it apart
+# needs the outer aggregate evaluated on the inner one's empty input row.
+# Expected `(1, 10), (2, NULL)`.
+query error Physical plan does not support logical expression ScalarSubquery
+SELECT o.k, (SELECT max(m) FROM (SELECT max(i.v) AS m FROM nested_agg_inner AS i WHERE i.id = o.k)) FROM nested_agg_outer AS o ORDER BY o.k;
+
+# A projection over the scalar aggregate is still pulled up.
+query II
+SELECT o.k, (SELECT n + 1 FROM (SELECT count(*) AS n FROM nested_agg_inner AS i WHERE i.id = o.k)) FROM nested_agg_outer AS o ORDER BY o.k;
+----
+1 2
+2 1
+
+# An aggregate over a grouped aggregate is still pulled up.
+query II
+SELECT o.k, (SELECT max(c) FROM (SELECT i.id, count(*) AS c FROM nested_agg_inner AS i WHERE i.id = o.k GROUP BY i.id)) FROM nested_agg_outer AS o ORDER BY o.k;
+----
+1 1
+2 NULL
+
+statement ok
+DROP TABLE nested_agg_outer;
+
+statement ok
+DROP TABLE nested_agg_inner;
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org