From 78fd46f9261be26271faa062c023a7b994d52196 Mon Sep 17 00:00:00 2001 From: blinding-pixels <281499151+blinding-pixels@users.noreply.github.com> Date: Sun, 30 Aug 2026 22:56:11 -0400 Subject: [PATCH 1/2] fix: avoid invalid qualifiers in unparsed subqueries --- datafusion/sql/src/unparser/ast.rs | 16 +++++ datafusion/sql/src/unparser/plan.rs | 86 +++++++++++++++++++++-- datafusion/sql/tests/cases/plan_to_sql.rs | 47 ++++++++++++- 3 files changed, 141 insertions(+), 8 deletions(-) diff --git a/datafusion/sql/src/unparser/ast.rs b/datafusion/sql/src/unparser/ast.rs index 7418d0b5b7605..c4cc1de28c141 100644 --- a/datafusion/sql/src/unparser/ast.rs +++ b/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, + /// 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 { @@ -419,6 +434,7 @@ impl SelectBuilder { flavor: Some(SelectFlavor::Standard), flatten_alias_counter: 0, flatten_table_aliases: Vec::new(), + subquery_alias_depth: 0, } } } diff --git a/datafusion/sql/src/unparser/plan.rs b/datafusion/sql/src/unparser/plan.rs index 9922509a0e609..2cc2acf17f417 100644 --- a/datafusion/sql/src/unparser/plan.rs +++ b/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::>>()?; + 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), diff --git a/datafusion/sql/tests/cases/plan_to_sql.rs b/datafusion/sql/tests/cases/plan_to_sql.rs index d6c31570bf1b0..3ebb41744f12d 100644 --- a/datafusion/sql/tests/cases/plan_to_sql.rs +++ b/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(()) } From 5422e973d45040d50dfca5e1cd2175075abd4298 Mon Sep 17 00:00:00 2001 From: blinding-pixels <281499151+blinding-pixels@users.noreply.github.com> Date: Mon, 31 Aug 2026 04:15:21 -0400 Subject: [PATCH 2/2] fix: rebase outer derived-table clauses --- datafusion/sql/src/unparser/ast.rs | 6 +- datafusion/sql/src/unparser/plan.rs | 140 +++++++++++++++------- datafusion/sql/tests/cases/plan_to_sql.rs | 28 +++++ 3 files changed, 129 insertions(+), 45 deletions(-) diff --git a/datafusion/sql/src/unparser/ast.rs b/datafusion/sql/src/unparser/ast.rs index c4cc1de28c141..c335d3ee4c16d 100644 --- a/datafusion/sql/src/unparser/ast.rs +++ b/datafusion/sql/src/unparser/ast.rs @@ -197,16 +197,16 @@ impl SelectBuilder { self.flatten_table_aliases.iter().any(|a| a == alias) } - pub fn enter_subquery_alias(&mut self) { + pub(super) fn enter_subquery_alias(&mut self) { self.subquery_alias_depth += 1; } - pub fn exit_subquery_alias(&mut self) { + pub(super) 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 { + pub(super) fn inside_subquery_alias(&self) -> bool { self.subquery_alias_depth > 0 } diff --git a/datafusion/sql/src/unparser/plan.rs b/datafusion/sql/src/unparser/plan.rs index 2cc2acf17f417..8a19e92435ca3 100644 --- a/datafusion/sql/src/unparser/plan.rs +++ b/datafusion/sql/src/unparser/plan.rs @@ -164,6 +164,12 @@ impl<'a> UnparserAggScope<'a> { } } +#[derive(Clone, Copy)] +struct DerivedInputScope<'a> { + alias: &'static str, + schema: &'a DFSchema, +} + impl Unparser<'_> { pub fn plan_to_sql(&self, plan: &LogicalPlan) -> Result { let mut plan = normalize_union_schema(plan)?; @@ -469,6 +475,74 @@ impl Unparser<'_> { } } + fn derived_input_scope<'a>( + plan: &'a LogicalPlan, + select: &SelectBuilder, + ) -> Option> { + if select.inside_subquery_alias() { + return None; + } + + match plan { + LogicalPlan::Projection(projection) => { + let alias = Self::derived_input_alias(projection.input.as_ref())?; + let qualified_projection = projection.expr.iter().any(|expr| { + expr.column_refs() + .iter() + .any(|column| column.relation.is_some()) + }); + let mut input_names = HashSet::new(); + let unique_input_names = projection + .input + .schema() + .fields() + .iter() + .all(|field| input_names.insert(field.name())); + + (qualified_projection + && unique_input_names + && find_unnest_node_within_select(plan).is_none()) + .then_some(DerivedInputScope { + alias, + schema: projection.input.schema().as_ref(), + }) + } + LogicalPlan::Filter(filter) => { + Self::derived_input_scope(filter.input.as_ref(), select) + } + LogicalPlan::Limit(limit) => { + Self::derived_input_scope(limit.input.as_ref(), select) + } + LogicalPlan::Sort(sort) => { + Self::derived_input_scope(sort.input.as_ref(), select) + } + LogicalPlan::Repartition(repartition) => { + Self::derived_input_scope(repartition.input.as_ref(), select) + } + _ => None, + } + } + + fn rebase_derived_input_expr( + &self, + expr: Expr, + scope: Option>, + ) -> Result { + let Some(scope) = scope else { + return Ok(expr); + }; + if self.dialect.requires_derived_table_alias() { + let mut alias_rewriter = TableAliasRewriter { + table_schema: scope.schema, + alias_name: TableReference::bare(scope.alias), + rewrite_unqualified: false, + }; + expr.rewrite(&mut alias_rewriter).data() + } else { + Self::strip_column_qualifiers_for_schema(expr, scope.schema) + } + } + fn contains_aggregate_before_relation(plan: &LogicalPlan) -> bool { match plan { LogicalPlan::Aggregate(_) => true, @@ -848,55 +922,20 @@ impl Unparser<'_> { ); } - 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() - { + if let Some(scope) = Self::derived_input_scope(plan, select) { // 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![])); + .then(|| self.new_table_alias(scope.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.rebase_derived_input_expr(expr, Some(scope))) .map(|expr| self.select_item_to_sql(&expr?)) .collect::>>()?; select.projection(items); @@ -1145,6 +1184,7 @@ impl Unparser<'_> { self.select_to_sql_recursively(cur, query, select, relation) } LogicalPlan::Filter(filter) => { + let derived_input_scope = Self::derived_input_scope(plan, select); let window = find_window_nodes_within_select( plan, None, @@ -1161,15 +1201,23 @@ impl Unparser<'_> { unprojected = UnparserAggScope::new(agg).prepare(unprojected, None)?; } + unprojected = + self.rebase_derived_input_expr(unprojected, derived_input_scope)?; let filter_expr = self.expr_to_sql(&unprojected)?; select.qualify(Some(filter_expr)); } else if let Some(agg) = agg { - let unprojected = UnparserAggScope::new(agg) + let mut unprojected = UnparserAggScope::new(agg) .prepare(filter.predicate.clone(), None)?; + unprojected = + self.rebase_derived_input_expr(unprojected, derived_input_scope)?; let filter_expr = self.expr_to_sql(&unprojected)?; select.having(Some(filter_expr)); } else { - let filter_expr = self.expr_to_sql(&filter.predicate)?; + let predicate = self.rebase_derived_input_expr( + filter.predicate.clone(), + derived_input_scope, + )?; + let filter_expr = self.expr_to_sql(&predicate)?; select.selection(Some(filter_expr)); } @@ -1245,17 +1293,25 @@ impl Unparser<'_> { )))); }; + let derived_input_scope = Self::derived_input_scope(plan, select); let agg = find_agg_node_within_select(plan, select.already_projected()); // unproject sort expressions let sort_exprs: Vec = sort .expr .iter() .map(|sort_expr| { - Self::unproject_sort_expr_in_scope( + let sort_expr = Self::unproject_sort_expr_in_scope( sort_expr.clone(), agg, sort.input.as_ref(), - ) + )?; + Ok(SortExpr { + expr: self.rebase_derived_input_expr( + sort_expr.expr, + derived_input_scope, + )?, + ..sort_expr + }) }) .collect::>>()?; diff --git a/datafusion/sql/tests/cases/plan_to_sql.rs b/datafusion/sql/tests/cases/plan_to_sql.rs index 3ebb41744f12d..fedd252eb77d6 100644 --- a/datafusion/sql/tests/cases/plan_to_sql.rs +++ b/datafusion/sql/tests/cases/plan_to_sql.rs @@ -406,6 +406,18 @@ fn roundtrip_rebases_derived_projection_references() -> Result<(), DataFusionErr 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) order by j1_id;", + parser_dialect: GenericDialect {}, + unparser_dialect: UnparserDefaultDialect {}, + expected: @"SELECT j1_id FROM (SELECT ta.j1_id FROM j1 AS ta) ORDER BY j1_id ASC NULLS LAST", + ); + roundtrip_statement_with_dialect_helper!( + sql: "select j1_id from (select ta.j1_id as j1_id from j1 ta) order by j1_id;", + parser_dialect: MySqlDialect {}, + unparser_dialect: UnparserMySqlDialect {}, + expected: @"SELECT `derived_projection`.`j1_id` FROM (SELECT `ta`.`j1_id` FROM `j1` AS `ta`) AS `derived_projection` ORDER BY `derived_projection`.`j1_id` ASC", + ); 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 {}, @@ -430,6 +442,22 @@ fn roundtrip_rebases_derived_projection_references() -> Result<(), DataFusionErr unparser_dialect: UnparserMySqlDialect {}, expected: @"SELECT `derived_distinct`.`j1_id` FROM (SELECT DISTINCT `ta`.`j1_id` FROM `j1` AS `ta`) AS `derived_distinct`", ); + + let statement = Parser::new(&GenericDialect {}) + .try_with_sql("select j1_id from (select ta.j1_id as j1_id from j1 ta)")? + .parse_statement()?; + let context = MockContextProvider { + state: MockSessionState::default(), + }; + let plan = SqlToRel::new(&context).sql_statement_to_plan(statement)?; + let plan = LogicalPlanBuilder::from(plan) + .filter(col("ta.j1_id").gt(lit(1)))? + .build()?; + let unparser = Unparser::new(&UnparserMySqlDialect {}); + assert_snapshot!( + unparser.plan_to_sql(&plan)?, + @"SELECT `derived_projection`.`j1_id` FROM (SELECT `ta`.`j1_id` FROM `j1` AS `ta`) AS `derived_projection` WHERE (`derived_projection`.`j1_id` > 1)" + ); Ok(()) }