fix: Unparser drops column alias needed by ORDER BY when flattening Projection

In `rewrite_plan_for_sort_on_non_projected_fields`, when the outer
Projection excludes a Sort column that was defined as an alias in the
inner Projection (e.g. `Z AS c`), the rewrite replaced the inner
Projection's expressions with only the outer Projection's mapped
expressions, dropping the alias definition. This left ORDER BY
referencing a non-existent column.

The fix inlines the underlying physical expression into the Sort when
an alias is dropped (e.g. `ORDER BY c` → `ORDER BY t."Z"`).

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

Author: yonatan-sevenai

datafusion/sql/src/unparser/rewrite.rs

@@ -254,6 +254,46 @@ pub(super) fn rewrite_plan_for_sort_on_non_projected_fields(
             .map(|e| map.get(e).unwrap_or(e).clone())
             .collect::<Vec<_>>();
 
+        // Build a reverse map from alias name → underlying expression for
+        // sort columns that are dropped from the outer projection.  When
+        // the inner Projection is trimmed to `new_exprs` below, any alias
+        // that only existed in the inner Projection disappears, so ORDER BY
+        // references to it become dangling.  We inline the physical
+        // expression instead (e.g. ORDER BY "c" → ORDER BY "Z").
+        let projected_aliases: HashSet<&str> = new_exprs
+            .iter()
+            .filter_map(|e| match e {
+                Expr::Alias(alias) => Some(alias.name.as_str()),
+                _ => None,
+            })
+            .collect();
+
+        let alias_to_underlying: HashMap<String, Expr> = inner_p
+            .expr
+            .iter()
+            .filter_map(|e| match e {
+                Expr::Alias(alias) if !projected_aliases.contains(alias.name.as_str()) => {
+                    Some((alias.name.clone(), (*alias.expr).clone()))
+                }
+                _ => None,
+            })
+            .collect();
+
+        if !alias_to_underlying.is_empty() {
+            for sort_expr in &mut sort.expr {
+                let mut expr = sort_expr.expr.clone();
+                while let Expr::Alias(alias) = expr {
+                    expr = *alias.expr;
+                }
+                if let Expr::Column(ref col) = expr {
+                    let name = col.name();
+                    if let Some(underlying) = alias_to_underlying.get(name) {
+                        sort_expr.expr = underlying.clone();
+                    }
+                }
+            }
+        }
+
         inner_p.expr.clone_from(&new_exprs);
         sort.input = Arc::new(LogicalPlan::Projection(inner_p));
 

datafusion/sql/tests/cases/plan_to_sql.rs

@@ -2973,3 +2973,57 @@ fn test_unparse_manual_join_with_subquery_aggregate() -> Result<()> {
 
     Ok(())
 }
+
+/// Regression test: when the outer Projection excludes a Sort column that was
+/// defined as an alias in the inner Projection (through a SubqueryAlias), the
+/// Unparser must either preserve the subquery boundary or inline the physical
+/// column name into the ORDER BY clause.
+///
+/// See: https://github.com/apache/datafusion/issues/XXXX
+#[test]
+fn test_sort_on_aliased_column_dropped_by_outer_projection() -> Result<()> {
+    let schema = Schema::new(vec![
+        Field::new("X", DataType::Utf8, true),
+        Field::new("Y", DataType::Utf8, true),
+        Field::new("Z", DataType::Utf8, true),
+    ]);
+
+    // Build:
+    //   Projection: [a, b]                         -- outer: excludes sort column "c"
+    //     Sort: [c DESC, fetch=1]                   -- references alias "c"
+    //       Projection: [X AS a, Y AS b, Z AS c]    -- defines alias "c"
+    //         SubqueryAlias: t
+    //           TableScan: phys_table [X, Y, Z]
+    let plan = table_scan(Some("phys_table"), &schema, None)?
+        .alias("t")?
+        .project(vec![
+            Expr::Column(Column::new(Some(TableReference::bare("t")), "X"))
+                .alias("a"),
+            Expr::Column(Column::new(Some(TableReference::bare("t")), "Y"))
+                .alias("b"),
+            Expr::Column(Column::new(Some(TableReference::bare("t")), "Z"))
+                .alias("c"),
+        ])?
+        .sort_with_limit(
+            vec![Expr::Column(Column::new_unqualified("c")).sort(false, true)],
+            Some(1),
+        )?
+        .project(vec![
+            Expr::Column(Column::new_unqualified("a")),
+            Expr::Column(Column::new_unqualified("b")),
+        ])?
+        .build()?;
+
+    let unparser = Unparser::default();
+    let sql = unparser.plan_to_sql(&plan)?;
+
+    // The sort column "c" (aliased from "Z" in the inner Projection) must be
+    // inlined to the physical column in ORDER BY, since the outer Projection
+    // excludes it from the SELECT list.
+    assert_snapshot!(
+        sql,
+        @r#"SELECT t."X" AS a, t."Y" AS b FROM phys_table AS t ORDER BY t."Z" DESC NULLS FIRST LIMIT 1"#
+    );
+
+    Ok(())
+}