fix: preserve subquery structure when unparsing SubqueryAlias over Ag… (#21099)

When the SQL unparser encountered a SubqueryAlias node whose direct
child was an Aggregate (or other clause-building plan like Window, Sort,
Limit, Union), it would flatten the subquery into a simple table alias,
losing the aggregate entirely.

For example, a plan representing:
SELECT j1.col FROM j1 JOIN (SELECT max(id) AS m FROM j2) AS b ON j1.id =
b.m

would unparse to:
  SELECT j1.col FROM j1 INNER JOIN j2 AS b ON j1.id = b.m

dropping the MAX aggregate and the subquery.

Root cause: the SubqueryAlias handler in select_to_sql_recursively would
call subquery_alias_inner_query_and_columns (which only unwraps
Projection children) and unparse_table_scan_pushdown (which only handles
TableScan/SubqueryAlias/Projection). When both returned nothing useful
for an Aggregate child, the code recursed directly into the Aggregate,
merging its GROUP BY into the outer SELECT instead of wrapping it in a
derived subquery.

The fix adds an early check: if the SubqueryAlias's direct child is a
plan type that builds its own SELECT clauses (Aggregate, Window, Sort,
Limit, Union), emit it as a derived subquery via self.derive() with the
alias always attached, rather than falling through to the recursive path
that would flatten it.

## Which issue does this PR close?
- Closes #21098 

## Rationale for this change

The SQL unparser silently drops subquery structure when a SubqueryAlias
node directly wraps an Aggregate (or Window, Sort, Limit, Union). For
example, a plan representing
```sql 
SELECT ... FROM j1 JOIN (SELECT max(id) FROM j2) AS b ...
``` 
unparses to 
```sql
SELECT ... FROM j1 JOIN j2 AS b ...
```
losing the aggregate entirely. This produces semantically incorrect SQL.

## What changes are included in this PR?

In the SubqueryAlias handler within select_to_sql_recursively
(`datafusion/sql/src/unparser/plan.rs`):
- Added an early check: if the SubqueryAlias's direct child is a plan
type that builds its own SELECT clauses (Aggregate, Window, Sort, Limit,
Union) and cannot be reduced to a table scan, emit it as a derived
subquery (SELECT ...) AS alias via self.derive() instead of recursing
into the child and flattening it.
- Added a helper requires_derived_subquery() that identifies plan types
requiring this treatment.

## Are these changes tested?

Yes. A new test test_unparse_manual_join_with_subquery_aggregate is
added that constructs a SubqueryAlias > Aggregate plan (without an
intermediate Projection) and asserts the unparsed SQL preserves the
MAX() aggregate function call. This test fails without the fix. All
current unparser tests succeed without modification

## Are there any user-facing changes?

No.

---------

Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: yonatan-sevenai

datafusion/sql/src/unparser/plan.rs

@@ -828,6 +828,50 @@ impl Unparser<'_> {
                     Some(plan_alias.alias.clone()),
                     select.already_projected(),
                 )?;
+
+                // If the (possibly rewritten) inner plan builds its own
+                // SELECT clauses (e.g. Aggregate adds GROUP BY, Window adds
+                // OVER, etc.) and unparse_table_scan_pushdown couldn't reduce it,
+                // we must emit a derived subquery: (SELECT ...) AS alias.
+                // Without this, the recursive handler would merge those clauses
+                // into the outer SELECT, losing the subquery structure entirely.
+                if unparsed_table_scan.is_none() && Self::requires_derived_subquery(plan)
+                {
+                    // When the dialect does not support column aliases in
+                    // table aliases (e.g. SQLite), inject the aliases into
+                    // the inner projection before wrapping as a derived
+                    // subquery.
+                    if !columns.is_empty()
+                        && !self.dialect.supports_column_alias_in_table_alias()
+                    {
+                        let Ok(rewritten_plan) =
+                            inject_column_aliases_into_subquery(plan.clone(), columns)
+                        else {
+                            return internal_err!(
+                                "Failed to transform SubqueryAlias plan"
+                            );
+                        };
+                        return self.derive(
+                            &rewritten_plan,
+                            relation,
+                            Some(self.new_table_alias(
+                                plan_alias.alias.table().to_string(),
+                                vec![],
+                            )),
+                            false,
+                        );
+                    }
+                    return self.derive(
+                        plan,
+                        relation,
+                        Some(self.new_table_alias(
+                            plan_alias.alias.table().to_string(),
+                            columns,
+                        )),
+                        false,
+                    );
+                }
+
                 // if the child plan is a TableScan with pushdown operations, we don't need to
                 // create an additional subquery for it
                 if !select.already_projected() && unparsed_table_scan.is_none() {
@@ -1060,6 +1104,22 @@ impl Unparser<'_> {
         scan.projection.is_some() || !scan.filters.is_empty() || scan.fetch.is_some()
     }
 
+    /// Returns true if a plan, when used as the direct child of a SubqueryAlias,
+    /// must be emitted as a derived subquery `(SELECT ...) AS alias`.
+    ///
+    /// Plans like Aggregate or Window build their own SELECT clauses (GROUP BY,
+    /// window functions).
+    fn requires_derived_subquery(plan: &LogicalPlan) -> bool {
+        matches!(
+            plan,
+            LogicalPlan::Aggregate(_)
+                | LogicalPlan::Window(_)
+                | LogicalPlan::Sort(_)
+                | LogicalPlan::Limit(_)
+                | LogicalPlan::Union(_)
+        )
+    }
+
     /// Try to unparse a table scan with pushdown operations into a new subquery plan.
     /// If the table scan is without any pushdown operations, return None.
     fn unparse_table_scan_pushdown(

datafusion/sql/tests/cases/plan_to_sql.rs

@@ -23,7 +23,7 @@ use datafusion_common::{
 };
 use datafusion_expr::expr::{WindowFunction, WindowFunctionParams};
 use datafusion_expr::test::function_stub::{
-    count_udaf, max_udaf, min_udaf, sum, sum_udaf,
+    count_udaf, max, max_udaf, min_udaf, sum, sum_udaf,
 };
 use datafusion_expr::{
     EmptyRelation, Expr, Extension, LogicalPlan, LogicalPlanBuilder, Union,
@@ -2904,3 +2904,72 @@ fn test_json_access_3() {
         @r#"SELECT (j1.j1_string : 'field.inner1[''inner2'']') FROM j1"#
     );
 }
+
+/// Roundtrip test for a subquery aggregate with column aliases.
+/// Ensures that `subquery_alias_inner_query_and_columns` unwrapping
+/// a Projection -> Aggregate still triggers the derived-subquery path.
+#[test]
+fn roundtrip_subquery_aggregate_with_column_alias() -> Result<(), DataFusionError> {
+    roundtrip_statement_with_dialect_helper!(
+        sql: "SELECT id FROM (SELECT max(j1_id) FROM j1) AS c(id)",
+        parser_dialect: GenericDialect {},
+        unparser_dialect: UnparserDefaultDialect {},
+        expected: @"SELECT c.id FROM (SELECT max(j1.j1_id) FROM j1) AS c (id)",
+    );
+    Ok(())
+}
+
+/// Test that unparsing a manually constructed join with a subquery aggregate
+/// preserves the MAX aggregate function.
+///
+/// Builds the equivalent of:
+///   SELECT j1.j1_string FROM j1
+///     JOIN (SELECT max(j2_id) AS max_id FROM j2) AS b
+///     ON j1.j1_id = b.max_id
+#[test]
+fn test_unparse_manual_join_with_subquery_aggregate() -> Result<()> {
+    let context = MockContextProvider {
+        state: MockSessionState::default(),
+    };
+    let j1_schema = context
+        .get_table_source(TableReference::bare("j1"))?
+        .schema();
+    let j2_schema = context
+        .get_table_source(TableReference::bare("j2"))?
+        .schema();
+
+    // Build the right side: SELECT max(j2_id) AS max_id FROM j2
+    let right_scan = table_scan(Some("j2"), &j2_schema, None)?.build()?;
+    let right_agg = LogicalPlanBuilder::from(right_scan)
+        .aggregate(
+            vec![] as Vec<Expr>,
+            vec![max(col("j2.j2_id")).alias("max_id")],
+        )?
+        .build()?;
+    let right_subquery = subquery_alias(right_agg, "b")?;
+
+    // Build the full plan: SELECT j1.j1_string FROM j1 JOIN (...) AS b ON j1.j1_id = b.max_id
+    let left_scan = table_scan(Some("j1"), &j1_schema, None)?.build()?;
+    let plan = LogicalPlanBuilder::from(left_scan)
+        .join(
+            right_subquery,
+            datafusion_expr::JoinType::Inner,
+            (
+                vec![Column::from_qualified_name("j1.j1_id")],
+                vec![Column::from_qualified_name("b.max_id")],
+            ),
+            None,
+        )?
+        .project(vec![col("j1.j1_string")])?
+        .build()?;
+
+    let unparser = Unparser::default();
+    let sql = unparser.plan_to_sql(&plan)?.to_string();
+    let sql_upper = sql.to_uppercase();
+    assert!(
+        sql_upper.contains("MAX("),
+        "Unparsed SQL should preserve the MAX aggregate function call, got: {sql}"
+    );
+
+    Ok(())
+}