fix: avoid invalid qualifiers in unparsed subqueries

datafusion/sql/src/unparser/ast.rs

@@ -167,6 +167,8 @@ pub struct SelectBuilder {
     /// Table aliases that correspond to LATERAL FLATTEN relations.
     /// Column references into these aliases must use `VALUE` as the column name.
     flatten_table_aliases: Vec<String>,
+    /// Depth of explicitly named subqueries currently being rendered.
+    subquery_alias_depth: usize,
 }
 
 /// Prefix used for auto-generated LATERAL FLATTEN table aliases.
@@ -195,6 +197,19 @@ impl SelectBuilder {
         self.flatten_table_aliases.iter().any(|a| a == alias)
     }
 
+    pub fn enter_subquery_alias(&mut self) {
+        self.subquery_alias_depth += 1;
+    }
+
+    pub fn exit_subquery_alias(&mut self) {
+        debug_assert!(self.subquery_alias_depth > 0);
+        self.subquery_alias_depth -= 1;
+    }
+
+    pub fn inside_subquery_alias(&self) -> bool {
+        self.subquery_alias_depth > 0
+    }
+
     /// Returns the most recently generated flatten alias, or `None` if
     /// `next_flatten_alias` has not been called yet.
     pub fn current_flatten_alias(&self) -> Option<String> {
@@ -419,6 +434,7 @@ impl SelectBuilder {
             flavor: Some(SelectFlavor::Standard),
             flatten_alias_counter: 0,
             flatten_table_aliases: Vec::new(),
+            subquery_alias_depth: 0,
         }
     }
 }

datafusion/sql/src/unparser/plan.rs

@@ -54,7 +54,7 @@ use datafusion_expr::{
     TableScan, Unnest, UserDefinedLogicalNode, Window, expr::Alias,
 };
 use sqlparser::ast::{self, Ident, OrderByKind, SetExpr, TableAliasColumnDef};
-use std::{sync::Arc, vec};
+use std::{collections::HashSet, sync::Arc, vec};
 
 /// Convert a DataFusion [`LogicalPlan`] to [`ast::Statement`]
 ///
@@ -451,6 +451,24 @@ impl Unparser<'_> {
         }
     }
 
+    /// Return the alias recursion would assign when `plan` must become a
+    /// derived relation below an already rendered projection.
+    fn derived_input_alias(plan: &LogicalPlan) -> Option<&'static str> {
+        match plan {
+            LogicalPlan::Projection(_) => Some("derived_projection"),
+            LogicalPlan::Limit(_) => Some("derived_limit"),
+            LogicalPlan::Sort(_) => Some("derived_sort"),
+            LogicalPlan::Distinct(_) => Some("derived_distinct"),
+            LogicalPlan::Filter(filter) => {
+                Self::derived_input_alias(filter.input.as_ref())
+            }
+            LogicalPlan::Repartition(repartition) => {
+                Self::derived_input_alias(repartition.input.as_ref())
+            }
+            _ => None,
+        }
+    }
+
     fn contains_aggregate_before_relation(plan: &LogicalPlan) -> bool {
         match plan {
             LogicalPlan::Aggregate(_) => true,
@@ -829,6 +847,61 @@ impl Unparser<'_> {
                         columns,
                     );
                 }
+
+                let qualified_projection = p.expr.iter().try_fold(false, |found, expr| {
+                    if found {
+                        Ok(true)
+                    } else {
+                        expr.exists(|expr| {
+                            Ok(matches!(expr, Expr::Column(column) if column.relation.is_some()))
+                        })
+                    }
+                })?;
+                let mut input_names = HashSet::new();
+                let unique_input_names = p
+                    .input
+                    .schema()
+                    .fields()
+                    .iter()
+                    .all(|field| input_names.insert(field.name()));
+                if let Some(input_alias) = Self::derived_input_alias(p.input.as_ref())
+                    && qualified_projection
+                    && unique_input_names
+                    && find_unnest_node_within_select(plan).is_none()
+                    && !select.inside_subquery_alias()
+                {
+                    // The input is about to enter a new SQL scope. Preserve that
+                    // boundary explicitly and make the outer expressions resolve
+                    // against the relation that will actually be visible there.
+                    let requires_alias = self.dialect.requires_derived_table_alias();
+                    let alias = requires_alias
+                        .then(|| self.new_table_alias(input_alias.to_string(), vec![]));
+                    self.derive(p.input.as_ref(), relation, alias, false)?;
+
+                    let items = p
+                        .expr
+                        .iter()
+                        .cloned()
+                        .map(|expr| {
+                            if requires_alias {
+                                let mut alias_rewriter = TableAliasRewriter {
+                                    table_schema: p.input.schema().as_ref(),
+                                    alias_name: TableReference::bare(input_alias),
+                                    rewrite_unqualified: false,
+                                };
+                                expr.rewrite(&mut alias_rewriter).data()
+                            } else {
+                                Self::strip_column_qualifiers_for_schema(
+                                    expr,
+                                    p.input.schema().as_ref(),
+                                )
+                            }
+                        })
+                        .map(|expr| self.select_item_to_sql(&expr?))
+                        .collect::<Result<Vec<_>>>()?;
+                    select.projection(items);
+                    return Ok(());
+                }
                 // For Snowflake FLATTEN: when the outer Projection has
                 // UNNEST(...) display-name columns (from SELECT * / SELECT
                 // UNNEST(...)), generate a flatten alias now so that
@@ -1537,7 +1610,8 @@ impl Unparser<'_> {
                     )]);
                 }
                 let plan = unparsed_table_scan.unwrap_or_else(|| plan.clone());
-                if !columns.is_empty()
+                select.enter_subquery_alias();
+                let recursive_result = if !columns.is_empty()
                     && !self.dialect.supports_column_alias_in_table_alias()
                 {
                     // Instead of specifying column aliases as part of the outer table, inject them directly into the inner projection
@@ -1558,10 +1632,12 @@ impl Unparser<'_> {
                         query,
                         select,
                         relation,
-                    )?;
+                    )
                 } else {
-                    self.select_to_sql_recursively(&plan, query, select, relation)?;
-                }
+                    self.select_to_sql_recursively(&plan, query, select, relation)
+                };
+                select.exit_subquery_alias();
+                recursive_result?;
 
                 relation.alias(Some(
                     self.new_table_alias(plan_alias.alias.table().to_string(), columns),

datafusion/sql/tests/cases/plan_to_sql.rs

@@ -392,13 +392,54 @@ fn roundtrip_statement_with_dialect_4() -> Result<(), DataFusionError> {
     Ok(())
 }
 
+#[test]
+fn roundtrip_rebases_derived_projection_references() -> Result<(), DataFusionError> {
+    roundtrip_statement_with_dialect_helper!(
+        sql: "select j1_id from (select ta.j1_id as j1_id from j1 ta);",
+        parser_dialect: GenericDialect {},
+        unparser_dialect: UnparserDefaultDialect {},
+        expected: @"SELECT j1_id FROM (SELECT ta.j1_id FROM j1 AS ta)",
+    );
+    roundtrip_statement_with_dialect_helper!(
+        sql: "select j1_id from (select ta.j1_id as j1_id from j1 ta);",
+        parser_dialect: MySqlDialect {},
+        unparser_dialect: UnparserMySqlDialect {},
+        expected: @"SELECT `derived_projection`.`j1_id` FROM (SELECT `ta`.`j1_id` FROM `j1` AS `ta`) AS `derived_projection`",
+    );
+    roundtrip_statement_with_dialect_helper!(
+        sql: "select j1_id from (select ta.j1_id as j1_id from j1 ta) where j1_id > 1;",
+        parser_dialect: GenericDialect {},
+        unparser_dialect: UnparserDefaultDialect {},
+        expected: @"SELECT j1_id FROM (SELECT ta.j1_id FROM j1 AS ta WHERE (ta.j1_id > 1))",
+    );
+    roundtrip_statement_with_dialect_helper!(
+        sql: "select j1_id from (select ta.j1_id as j1_id from j1 ta) where j1_id > 1;",
+        parser_dialect: MySqlDialect {},
+        unparser_dialect: UnparserMySqlDialect {},
+        expected: @"SELECT `derived_projection`.`j1_id` FROM (SELECT `ta`.`j1_id` FROM `j1` AS `ta` WHERE (`ta`.`j1_id` > 1)) AS `derived_projection`",
+    );
+    roundtrip_statement_with_dialect_helper!(
+        sql: "select j1_id from (select distinct ta.j1_id as j1_id from j1 ta);",
+        parser_dialect: GenericDialect {},
+        unparser_dialect: UnparserDefaultDialect {},
+        expected: @"SELECT j1_id FROM (SELECT DISTINCT ta.j1_id FROM j1 AS ta)",
+    );
+    roundtrip_statement_with_dialect_helper!(
+        sql: "select j1_id from (select distinct ta.j1_id as j1_id from j1 ta);",
+        parser_dialect: MySqlDialect {},
+        unparser_dialect: UnparserMySqlDialect {},
+        expected: @"SELECT `derived_distinct`.`j1_id` FROM (SELECT DISTINCT `ta`.`j1_id` FROM `j1` AS `ta`) AS `derived_distinct`",
+    );
+    Ok(())
+}
+
 #[test]
 fn roundtrip_statement_with_dialect_5() -> Result<(), DataFusionError> {
     roundtrip_statement_with_dialect_helper!(
         sql: "select j1_id from (select j1_id from j1 limit 10);",
         parser_dialect: MySqlDialect {},
         unparser_dialect: UnparserMySqlDialect {},
-        expected: @"SELECT `j1`.`j1_id` FROM (SELECT `j1`.`j1_id` FROM `j1` LIMIT 10) AS `derived_limit`",
+        expected: @"SELECT `derived_limit`.`j1_id` FROM (SELECT `j1`.`j1_id` FROM `j1` LIMIT 10) AS `derived_limit`",
     );
     Ok(())
 }
@@ -1614,7 +1655,7 @@ fn test_table_scan_pushdown() -> Result<()> {
         plan_to_sql(&query_from_table_scan_with_two_projections)?;
     assert_snapshot!(
         query_from_table_scan_with_two_projections,
-        @"SELECT t1.id, t1.age FROM (SELECT t1.id, t1.age FROM t1)"
+        @"SELECT id, age FROM (SELECT t1.id, t1.age FROM t1)"
     );
 
     let table_scan_with_filter = table_scan_with_filters(
@@ -1791,7 +1832,7 @@ fn test_sort_with_scalar_fn_and_push_down_fetch() -> Result<()> {
     let sql = plan_to_sql(&plan)?;
     assert_snapshot!(
         sql,
-        @"SELECT t1.search_phrase FROM (SELECT t1.search_phrase, t1.event_time FROM t1 WHERE (t1.search_phrase <> '') ORDER BY substr(t1.event_time, 1, 5) ASC NULLS FIRST LIMIT 10)"
+        @"SELECT search_phrase FROM (SELECT t1.search_phrase, t1.event_time FROM t1 WHERE (t1.search_phrase <> '') ORDER BY substr(t1.event_time, 1, 5) ASC NULLS FIRST LIMIT 10)"
     );
     Ok(())
 }