fix: Prevent SubqueryAlias to be pushed past FilterExec when unparsing

datafusion/core/tests/sql/unparser.rs

@@ -51,8 +51,8 @@ use datafusion_catalog::memory::MemorySchemaProvider;
 use datafusion_catalog::{CatalogProvider, MemoryCatalogProvider, SchemaProvider};
 use datafusion_common::Column;
 use datafusion_expr::Expr;
-use datafusion_sql::unparser::Unparser;
 use datafusion_sql::unparser::dialect::{DefaultDialect, DuckDBDialect};
+use datafusion_sql::unparser::{Unparser, plan_to_sql};
 use itertools::Itertools;
 use recursive::{set_minimum_stack_size, set_stack_allocation_size};
 
@@ -747,6 +747,31 @@ async fn optimized_duckdb_unparse_top_level_sort_over_agg_uses_select_alias() ->
     Ok(())
 }
 
+#[tokio::test]
+async fn optimized_filter_with_subquery_alias() -> Result<()> {
+    let ctx = SessionContext::new();
+    ctx.sql("create table t (a int)").await?.collect().await?;
+    let df = ctx
+        .sql(
+            "
+            select *
+            from (
+                select a
+                from t
+            ) t2
+            where a = 1
+        ",
+        )
+        .await?;
+    let plan = df.into_optimized_plan()?;
+    let sql = plan_to_sql(&plan)?.to_string();
+    assert_eq!(
+        sql,
+        "SELECT * FROM (SELECT t.a FROM t WHERE (t.a = 1)) AS t2"
+    );
+    Ok(())
+}
+
 /// The outcome of running a single roundtrip test.
 ///
 /// A successful test produces [`TestCaseResult::Success`].

datafusion/sql/src/unparser/plan.rs

@@ -1632,7 +1632,11 @@ impl Unparser<'_> {
                 // 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)
+                // Also, do not add a table alias past a Filter, as otherwise the predicates might
+                // refer to invalid tables.
+                if (unparsed_table_scan.is_none()
+                    && Self::requires_derived_subquery(plan))
+                    || matches!(plan, LogicalPlan::Filter(_))
                 {
                     // When the dialect does not support column aliases in
                     // table aliases (e.g. SQLite), inject the aliases into