fix: Snowflake LATERAL FLATTEN handles SubqueryAlias between Unnest and Projection

When a table is accessed through a passthrough/virtual table mapping,
DataFusion inserts a SubqueryAlias node between Unnest and its inner
Projection. The FLATTEN rendering code assumed a direct Projection child
and failed with "Unnest input is not a Projection: SubqueryAlias(...)".

Peel through SubqueryAlias in three code paths that inspect unnest.input:
try_unnest_to_lateral_flatten_sql, the inline-vs-table source check, and
the general unnest recursion. Also fix a pre-existing collapsible_if
clippy warning in check_unnest_placeholder_with_outer_ref.

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

Author: yonatan-sevenai

datafusion/sql/src/unparser/plan.rs

@@ -442,6 +442,14 @@ impl Unparser<'_> {
                     // vs an inline array (EmptyRelation).
                     let inner_projection = match unnest.input.as_ref() {
                         LogicalPlan::Projection(proj) => proj,
+                        LogicalPlan::SubqueryAlias(alias) => match alias.input.as_ref() {
+                            LogicalPlan::Projection(proj) => proj,
+                            other => {
+                                return internal_err!(
+                                    "Unnest input (through SubqueryAlias) is not a Projection: {other:?}"
+                                );
+                            }
+                        },
                         other => {
                             return internal_err!(
                                 "Unnest input is not a Projection: {other:?}"
@@ -1186,6 +1194,11 @@ impl Unparser<'_> {
                 if let LogicalPlan::Projection(p) = unnest.input.as_ref() {
                     // continue with projection input
                     self.select_to_sql_recursively(&p.input, query, select, relation)
+                } else if let LogicalPlan::SubqueryAlias(alias) = unnest.input.as_ref()
+                    && let LogicalPlan::Projection(p) = alias.input.as_ref()
+                {
+                    // SubqueryAlias wraps the Projection (e.g. passthrough tables)
+                    self.select_to_sql_recursively(&p.input, query, select, relation)
                 } else {
                     internal_err!("Unnest input is not a Projection: {unnest:?}")
                 }
@@ -1250,13 +1263,13 @@ impl Unparser<'_> {
             }
             _ => return None,
         };
-        if let Expr::Column(Column { name, .. }) = inner {
-            if let Some(prefix) = name.strip_prefix(UNNEST_PLACEHOLDER) {
-                if prefix.starts_with(&format!("({OUTER_REFERENCE_COLUMN_PREFIX}(")) {
-                    return Some(UnnestInputType::OuterReference);
-                }
-                return Some(UnnestInputType::Scalar);
+        if let Expr::Column(Column { name, .. }) = inner
+            && let Some(prefix) = name.strip_prefix(UNNEST_PLACEHOLDER)
+        {
+            if prefix.starts_with(&format!("({OUTER_REFERENCE_COLUMN_PREFIX}(")) {
+                return Some(UnnestInputType::OuterReference);
             }
+            return Some(UnnestInputType::Scalar);
         }
         None
     }
@@ -1306,8 +1319,16 @@ impl Unparser<'_> {
         &self,
         unnest: &Unnest,
     ) -> Result<Option<FlattenRelationBuilder>> {
-        let LogicalPlan::Projection(projection) = unnest.input.as_ref() else {
-            return Ok(None);
+        let projection = match unnest.input.as_ref() {
+            LogicalPlan::Projection(p) => p,
+            LogicalPlan::SubqueryAlias(alias) => {
+                if let LogicalPlan::Projection(p) = alias.input.as_ref() {
+                    p
+                } else {
+                    return Ok(None);
+                }
+            }
+            _ => return Ok(None),
         };
 
         // For now, handle the simple case of a single expression to flatten.

datafusion/sql/tests/cases/plan_to_sql.rs

@@ -3183,3 +3183,36 @@ fn snowflake_flatten_unnest_udf_result() -> Result<(), DataFusionError> {
     insta::assert_snapshot!(actual, @r#"SELECT _unnest."VALUE" FROM "j1" CROSS JOIN LATERAL FLATTEN(INPUT => json_get_array("j1"."j1_string")) AS _unnest LIMIT 5"#);
     Ok(())
 }
+
+#[test]
+fn snowflake_unnest_through_subquery_alias() -> Result<(), DataFusionError> {
+    // Build: Projection → Unnest → SubqueryAlias → Projection → TableScan
+    // This simulates the plan produced when a virtual/passthrough table
+    // wraps the source in a SubqueryAlias, which sits between the Unnest
+    // and its inner Projection.
+
+    let schema = Schema::new(vec![Field::new(
+        "items",
+        DataType::List(Arc::new(Field::new_list_field(DataType::Utf8, true))),
+        true,
+    )]);
+
+    let plan = table_scan(Some("source"), &schema, None)?
+        .project(vec![col("items").alias("__unnest_placeholder(items)")])?
+        .alias("t")? // SubqueryAlias — this is what breaks
+        .unnest_column("__unnest_placeholder(items)")?
+        .project(vec![col("__unnest_placeholder(items)").alias("item")])?
+        .build()?;
+
+    let snowflake = SnowflakeDialect::new();
+    let unparser = Unparser::new(&snowflake);
+    let result = unparser.plan_to_sql(&plan)?;
+    let sql_str = result.to_string();
+
+    // Should contain LATERAL FLATTEN, not error
+    assert!(
+        sql_str.contains("LATERAL FLATTEN"),
+        "Expected LATERAL FLATTEN in SQL, got: {sql_str}"
+    );
+    Ok(())
+}