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/09/03 06:47:20 UTC
(datafusion) branch main updated: fix: avoid invalid qualifiers in unparsed subqueries (#24808)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion.git
The following commit(s) were added to refs/heads/main by this push:
new a6f1dce70c fix: avoid invalid qualifiers in unparsed subqueries (#24808)
a6f1dce70c is described below
commit a6f1dce70c6cb1faac48ec51f85f6d6e53829a4a
Author: blinding-pixels <bl...@protonmail.com>
AuthorDate: Thu Sep 3 06:38:48 2026 +0000
fix: avoid invalid qualifiers in unparsed subqueries (#24808)
## Which issue does this PR close?
- Closes #13027.
## Rationale for this change
I noticed that this issue had an earlier PR from almost two years ago.
That implementation was closed because it used broad identifier cleanup,
and maintainers were concerned it could introduce subtle bugs.
Since then, DataFusion has added much of the machinery needed for the
approaches recommended in that review. This solution uses that newer
machinery and combines both approaches. It removes the qualifier when
the derived table does not need an alias, and rewrites the reference to
the derived alias when the dialect requires one.
## What changes are included in this PR?
This PR detects when a projection enters a new derived-table scope and
prevents its outer expressions from referring to an inner table alias
that is no longer visible.
For dialects that do not require a derived-table alias, the invalid
inner qualifier is removed. For dialects such as MySQL that require an
alias, the outer reference is rewritten to the generated derived-table
alias. Explicitly named subquery scopes remain unchanged.
## What is the testing strategy for this PR?
The SQL unparser round-trip tests cover the issue's original query with
both the generic and MySQL dialects. They also cover filtered and
distinct derived inputs, and update existing limit and nested-projection
cases to assert valid outer references.
The required formatting, Clippy, and extended workspace test suite all
pass, including all 505 SQL logic test files.
## Are there any user-facing changes?
Yes. SQL produced by the unparser no longer contains qualifiers that
refer to tables outside their visible scope. Dialects that require
derived-table aliases now qualify the outer reference with the generated
alias. There are no public API changes.
---------
Co-authored-by: blinding-pixels <28...@users.noreply.github.com>
---
datafusion/sql/src/unparser/ast.rs | 16 +++
datafusion/sql/src/unparser/plan.rs | 160 ++++++++++++++++++++++++++++--
datafusion/sql/tests/cases/plan_to_sql.rs | 87 +++++++++++++++-
3 files changed, 251 insertions(+), 12 deletions(-)
diff --git a/datafusion/sql/src/unparser/ast.rs b/datafusion/sql/src/unparser/ast.rs
index 7418d0b5b7..c335d3ee4c 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<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(super) fn enter_subquery_alias(&mut self) {
+ self.subquery_alias_depth += 1;
+ }
+
+ pub(super) fn exit_subquery_alias(&mut self) {
+ debug_assert!(self.subquery_alias_depth > 0);
+ self.subquery_alias_depth -= 1;
+ }
+
+ pub(super) 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,
}
}
}
diff --git a/datafusion/sql/src/unparser/plan.rs b/datafusion/sql/src/unparser/plan.rs
index 30320acbb2..1a4a094e96 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`]
///
@@ -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<ast::Statement> {
let mut plan = normalize_union_schema(plan)?;
@@ -451,6 +457,102 @@ 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 derived_input_scope<'a>(
+ plan: &'a LogicalPlan,
+ select: &SelectBuilder,
+ ) -> Option<DerivedInputScope<'a>> {
+ if select.inside_subquery_alias() {
+ return None;
+ }
+
+ if select.already_projected()
+ && find_unnest_node_within_select(plan).is_none()
+ && let Some(alias) = Self::derived_input_alias(plan)
+ {
+ return Some(DerivedInputScope {
+ alias,
+ schema: plan.schema().as_ref(),
+ });
+ }
+
+ 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<DerivedInputScope<'_>>,
+ ) -> Result<Expr> {
+ 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,
@@ -829,6 +931,26 @@ impl Unparser<'_> {
columns,
);
}
+
+ 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(scope.alias.to_string(), vec![]));
+ self.derive(p.input.as_ref(), relation, alias, false)?;
+
+ let items = p
+ .expr
+ .iter()
+ .cloned()
+ .map(|expr| self.rebase_derived_input_expr(expr, Some(scope)))
+ .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
@@ -1072,6 +1194,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,
@@ -1088,15 +1211,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));
}
@@ -1172,17 +1303,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<SortExpr> = 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::<Result<Vec<_>>>()?;
@@ -1537,7 +1676,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 +1698,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 d6c31570bf..32dfade056 100644
--- a/datafusion/sql/tests/cases/plan_to_sql.rs
+++ b/datafusion/sql/tests/cases/plan_to_sql.rs
@@ -392,13 +392,94 @@ 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) 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 {},
+ 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`",
+ );
+
+ 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)"
+ );
+
+ let schema = Schema::new(vec![Field::new("j1_id", DataType::Int32, false)]);
+ let plan = table_scan(Some("j1"), &schema, None)?
+ .alias("ta")?
+ .project(vec![col("ta.j1_id")])?
+ .filter(col("ta.j1_id").gt(lit(0)))?
+ .project(vec![lit(1)])?
+ .build()?;
+ assert_snapshot!(
+ unparser.plan_to_sql(&plan)?,
+ @"SELECT 1 FROM (SELECT `ta`.`j1_id` FROM `j1` AS `ta`) AS `derived_projection` WHERE (`derived_projection`.`j1_id` > 0)"
+ );
+ 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 +1695,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 +1872,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(())
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org