Quoting fix

Author: yonatan-sevenai

datafusion/sql/src/unparser/ast.rs

@@ -769,7 +769,7 @@ impl FlattenRelationBuilder {
     fn create_empty() -> Self {
         Self {
             alias: Some(ast::TableAlias {
-                name: ast::Ident::new(FLATTEN_DEFAULT_ALIAS),
+                name: ast::Ident::with_quote('"', FLATTEN_DEFAULT_ALIAS),
                 columns: vec![],
                 explicit: true,
             }),

datafusion/sql/src/unparser/plan.rs

@@ -421,25 +421,16 @@ impl Unparser<'_> {
                 }
 
                 // Projection can be top-level plan for unnest relation.
-                // The projection generated by the `RecursiveUnnestRewriter` from a UNNEST relation will have
-                // only one expression, which is the placeholder column generated by the rewriter.
-                //
-                // Two cases:
-                //  - "bare": the expression IS the placeholder (+ aliases):
-                //      `__unnest_placeholder(...) AS item`
-                //  - "wrapped": the placeholder is inside a function call:
-                //      `json_as_text(__unnest_placeholder(...), 'type') AS target_type`
+                // The projection generated by the `RecursiveUnnestRewriter`
+                // will have exactly one expression referencing the unnest
+                // placeholder column. The placeholder may be at the top level
+                // ("bare": `__unnest_placeholder(...) AS item`) or wrapped
+                // inside a function call ("wrapped":
+                // `json_as_text(__unnest_placeholder(...), 'type') AS t`).
                 let (unnest_input_type, placeholder_is_bare) = if p.expr.len() == 1 {
-                    if let Some(t) =
-                        Self::check_unnest_placeholder_with_outer_ref(&p.expr[0])
-                    {
-                        (Some(t), true)
-                    } else if let Some(t) =
-                        Self::find_unnest_placeholder_in_expr(&p.expr[0])
-                    {
-                        (Some(t), false)
-                    } else {
-                        (None, false)
+                    match Self::find_unnest_placeholder(&p.expr[0]) {
+                        Some((t, is_bare)) => (Some(t), is_bare),
+                        None => (None, false),
                     }
                 } else {
                     (None, false)
@@ -451,16 +442,19 @@ impl Unparser<'_> {
                 } else {
                     None
                 };
+                // Snowflake LATERAL FLATTEN path.
+                // `peel_to_unnest_with_modifiers` walks through any
+                // intermediate Limit/Sort nodes, applies their modifiers to
+                // the query, and returns the Unnest + the LogicalPlan ref to
+                // recurse into (bypassing the normal Limit/Sort handlers that
+                // would create a derived subquery).
                 if self.dialect.unnest_as_lateral_flatten()
                     && unnest_input_type.is_some()
-                    && let Some(unnest) = Self::peel_to_unnest(p.input.as_ref())
-                    && let Some(flatten_relation) =
+                    && let Some((unnest, unnest_plan)) =
+                        self.peel_to_unnest_with_modifiers(p.input.as_ref(), query)?
+                    && let Some(flatten) =
                         self.try_unnest_to_lateral_flatten_sql(unnest)?
                 {
-                    let alias_name = flatten_relation.alias_name().to_string();
-
-                    // Check if the Unnest source is a real query (subselect)
-                    // vs an inline array (EmptyRelation).
                     let inner_projection =
                         Self::peel_to_inner_projection(unnest.input.as_ref())
                             .ok_or_else(|| {
@@ -470,35 +464,30 @@ impl Unparser<'_> {
                                 ))
                             })?;
 
-                    // Apply any intermediate Limit/Sort modifiers and get
-                    // a reference to the Unnest LogicalPlan node for recursion.
-                    // This bypasses the Limit/Sort handlers (which would create
-                    // unwanted derived subqueries when already_projected is set).
-                    let unnest_plan = self.apply_transparent_and_find_unnest_plan(
-                        p.input.as_ref(),
-                        query,
-                    )?;
+                    // Set the SELECT projection.
+                    if !select.already_projected() {
+                        if placeholder_is_bare {
+                            let value_expr = self.build_flatten_value_select_item(
+                                flatten.alias_name(),
+                                user_alias.as_deref(),
+                            );
+                            select.projection(vec![value_expr]);
+                        } else {
+                            // Composed expression wrapping the placeholder:
+                            // reconstruct_select_statement rewrites the
+                            // placeholder column to `_unnest."VALUE"` via
+                            // unproject_unnest_expr_as_flatten_value.
+                            self.reconstruct_select_statement(plan, p, select)?;
+                        }
+                    }
 
                     if matches!(
                         inner_projection.input.as_ref(),
                         LogicalPlan::EmptyRelation(_)
                     ) {
                         // Inline array case (e.g. UNNEST([1,2,3])):
                         // FLATTEN is the sole FROM source.
-                        relation.flatten(flatten_relation);
-
-                        if !select.already_projected() {
-                            if placeholder_is_bare {
-                                let value_expr = self.build_flatten_value_select_item(
-                                    &alias_name,
-                                    user_alias.as_deref(),
-                                );
-                                select.projection(vec![value_expr]);
-                            } else {
-                                self.reconstruct_select_statement(plan, p, select)?;
-                            }
-                        }
-
+                        relation.flatten(flatten);
                         return self.select_to_sql_recursively(
                             unnest_plan,
                             query,
@@ -508,41 +497,9 @@ impl Unparser<'_> {
                     }
 
                     // Non-empty source (table, subquery, etc.):
-                    // The FLATTEN references columns/expressions from the source.
-                    // Convert the inner Projection's expression to SQL for the
-                    // FLATTEN INPUT, then store the FLATTEN to be added as a
-                    // CROSS JOIN after the source relation is processed.
-                    let first_expr = inner_projection.expr.first().ok_or_else(|| {
-                        DataFusionError::Internal(
-                            "Unnest inner projection has no expressions".to_string(),
-                        )
-                    })?;
-                    // Strip the __unnest_placeholder alias to get the raw expression
-                    let raw_expr = match first_expr {
-                        Expr::Alias(Alias { expr, .. }) => expr.as_ref(),
-                        other => other,
-                    };
-                    let input_sql = self.expr_to_sql(raw_expr)?;
-
-                    let mut flatten = FlattenRelationBuilder::default();
-                    flatten.input_expr(input_sql);
-                    flatten.outer(unnest.options.preserve_nulls);
-
-                    if !select.already_projected() {
-                        if placeholder_is_bare {
-                            let value_expr = self.build_flatten_value_select_item(
-                                &alias_name,
-                                user_alias.as_deref(),
-                            );
-                            select.projection(vec![value_expr]);
-                        } else {
-                            self.reconstruct_select_statement(plan, p, select)?;
-                        }
-                    }
-
-                    // Recurse into the Unnest → inner source to set the primary
-                    // relation (table scan, subquery, etc.), then add FLATTEN
-                    // as a CROSS JOIN.
+                    // Recurse into the Unnest → inner source to set the
+                    // primary FROM relation, then add the FLATTEN as a
+                    // CROSS JOIN.
                     self.select_to_sql_recursively(unnest_plan, query, select, relation)?;
 
                     let flatten_factor = flatten.build().map_err(|e| {
@@ -559,26 +516,23 @@ impl Unparser<'_> {
                         from.push_join(cross_join);
                         select.push_from(from);
                     } else {
-                        // No existing FROM — create one with just the FLATTEN
                         let mut twj = TableWithJoinsBuilder::default();
                         twj.push_join(cross_join);
                         select.push_from(twj);
                     }
 
                     return Ok(());
                 }
+                // Standard UNNEST table factor path (BigQuery, etc.).
                 if self.dialect.unnest_as_table_factor()
                     && unnest_input_type.is_some()
                     && user_alias.is_none() // Skip if user alias present — fall through to reconstruct_select_statement which preserves aliases
-                    && let Some(unnest) = Self::peel_to_unnest(p.input.as_ref())
+                    && let Some((unnest, unnest_plan)) =
+                        self.peel_to_unnest_with_modifiers(p.input.as_ref(), query)?
                     && let Some(unnest_relation) =
                         self.try_unnest_to_table_factor_sql(unnest)?
                 {
                     relation.unnest(unnest_relation);
-                    let unnest_plan = self.apply_transparent_and_find_unnest_plan(
-                        p.input.as_ref(),
-                        query,
-                    )?;
                     return self.select_to_sql_recursively(
                         unnest_plan,
                         query,
@@ -1262,59 +1216,40 @@ impl Unparser<'_> {
         }
     }
 
-    /// Walk through "transparent" plan nodes (Limit, Sort) to find an Unnest.
-    ///
-    /// The DataFusion optimizer may insert Limit or Sort between the outer
-    /// Projection and the Unnest node. These nodes modify result quantity or
-    /// ordering but do not change the plan shape for unnest detection. The
-    /// normal recursion in [`Self::select_to_sql_recursively`] handles their
-    /// SQL rendering (LIMIT/OFFSET/ORDER BY); this helper only needs to
-    /// locate the Unnest so the FLATTEN / table-factor code path can fire.
-    fn peel_to_unnest(plan: &LogicalPlan) -> Option<&Unnest> {
-        match plan {
-            LogicalPlan::Unnest(unnest) => Some(unnest),
-            LogicalPlan::Limit(limit) => Self::peel_to_unnest(limit.input.as_ref()),
-            LogicalPlan::Sort(sort) => Self::peel_to_unnest(sort.input.as_ref()),
-            _ => None,
-        }
-    }
-
-    /// Walk through "transparent" plan nodes (SubqueryAlias, Limit, Sort) to
-    /// find the inner Projection that feeds an Unnest node.
+    /// Walk through transparent nodes (SubqueryAlias) to find the inner
+    /// Projection that feeds an Unnest node.
     ///
-    /// The inner Projection is created by the `RecursiveUnnestRewriter` and
-    /// contains the array expression that the Unnest operates on. A
-    /// `SubqueryAlias` (e.g. from a virtual/passthrough table) may wrap the
-    /// Projection; Limit/Sort are also handled for robustness.
+    /// The inner Projection is created atomically by the
+    /// `RecursiveUnnestRewriter` and contains the array expression that the
+    /// Unnest operates on. A `SubqueryAlias` (e.g. from a virtual/passthrough
+    /// table) may wrap the Projection.
     fn peel_to_inner_projection(plan: &LogicalPlan) -> Option<&Projection> {
         match plan {
             LogicalPlan::Projection(p) => Some(p),
             LogicalPlan::SubqueryAlias(alias) => {
                 Self::peel_to_inner_projection(alias.input.as_ref())
             }
-            LogicalPlan::Limit(limit) => {
-                Self::peel_to_inner_projection(limit.input.as_ref())
-            }
-            LogicalPlan::Sort(sort) => {
-                Self::peel_to_inner_projection(sort.input.as_ref())
-            }
             _ => None,
         }
     }
 
     /// Walk through transparent nodes (Limit, Sort) between the outer
     /// Projection and the Unnest, applying their SQL modifiers (LIMIT,
-    /// OFFSET, ORDER BY) to the query builder. Returns a reference to the
-    /// Unnest `LogicalPlan` node so the caller can recurse into it directly,
-    /// bypassing the intermediate handlers that would otherwise create
-    /// unwanted derived subqueries.
-    fn apply_transparent_and_find_unnest_plan<'a>(
+    /// OFFSET, ORDER BY) to the query builder. Returns the `Unnest` node
+    /// and a reference to the enclosing `LogicalPlan` for recursion, or
+    /// `Ok(None)` if no Unnest is found.
+    ///
+    /// By processing Limit/Sort inline and then recursing into the Unnest
+    /// plan directly, we bypass the normal Limit/Sort handlers which would
+    /// create unwanted derived subqueries (since `already_projected` is
+    /// set at the point this is called).
+    fn peel_to_unnest_with_modifiers<'a>(
         &self,
         plan: &'a LogicalPlan,
         query: &mut Option<QueryBuilder>,
-    ) -> Result<&'a LogicalPlan> {
+    ) -> Result<Option<(&'a Unnest, &'a LogicalPlan)>> {
         match plan {
-            LogicalPlan::Unnest(_) => Ok(plan),
+            LogicalPlan::Unnest(unnest) => Ok(Some((unnest, plan))),
             LogicalPlan::Limit(limit) => {
                 if let Some(fetch) = &limit.fetch
                     && let Some(q) = query.as_mut()
@@ -1329,7 +1264,7 @@ impl Unparser<'_> {
                         value: self.expr_to_sql(skip)?,
                     }));
                 }
-                self.apply_transparent_and_find_unnest_plan(limit.input.as_ref(), query)
+                self.peel_to_unnest_with_modifiers(limit.input.as_ref(), query)
             }
             LogicalPlan::Sort(sort) => {
                 let Some(query_ref) = query.as_mut() else {
@@ -1344,32 +1279,49 @@ impl Unparser<'_> {
                     ))));
                 }
                 query_ref.order_by(self.sorts_to_sql(&sort.expr)?);
-                self.apply_transparent_and_find_unnest_plan(sort.input.as_ref(), query)
-            }
-            other => {
-                internal_err!("Unexpected node between Projection and Unnest: {other:?}")
+                self.peel_to_unnest_with_modifiers(sort.input.as_ref(), query)
             }
+            _ => Ok(None),
         }
     }
 
-    /// Try to find the placeholder column name generated by `RecursiveUnnestRewriter`.
+    /// Search an expression for an unnest placeholder column reference.
     ///
-    /// - If the column is a placeholder column match the pattern `Expr::Alias(Expr::Column("__unnest_placeholder(...)"))`,
-    ///   it means it is a scalar column, return [UnnestInputType::Scalar].
-    /// - If the column is a placeholder column match the pattern `Expr::Alias(Expr::Column("__unnest_placeholder(outer_ref(...)))")`,
-    ///   it means it is an outer reference column, return [UnnestInputType::OuterReference].
-    /// - If the column is not a placeholder column, return [None].
+    /// Returns both the [`UnnestInputType`] and whether the placeholder is
+    /// "bare" (the expression IS the placeholder, modulo aliases) or
+    /// "wrapped" (the placeholder is inside a function call or other
+    /// transformation).
     ///
-    /// `outer_ref` is the display result of [Expr::OuterReferenceColumn]
-    fn check_unnest_placeholder_with_outer_ref(expr: &Expr) -> Option<UnnestInputType> {
-        // Peel through all Alias layers to find the inner Column.
-        // The expression may have multiple aliases, e.g.:
-        //   Alias("items", Alias("UNNEST(...)", Column("__unnest_placeholder(...)")))
+    /// Examples:
+    /// - `Alias("item", Column("__unnest_placeholder(...)"))` → `Some((Scalar, true))`
+    /// - `Alias("t", Cast(Column("__unnest_placeholder(...)"), Int64))` → `Some((Scalar, false))`
+    /// - `Column("unrelated")` → `None`
+    fn find_unnest_placeholder(expr: &Expr) -> Option<(UnnestInputType, bool)> {
+        // Fast path: check if the expression IS the placeholder (peel aliases).
         let mut inner = expr;
         while let Expr::Alias(Alias { expr, .. }) = inner {
             inner = expr.as_ref();
         }
-        if let Expr::Column(Column { name, .. }) = inner
+        if let Some(t) = Self::classify_placeholder_column(inner) {
+            return Some((t, true));
+        }
+        // Slow path: walk the full expression tree.
+        let mut result = None;
+        let _ = expr.apply(|e| {
+            if let Some(t) = Self::classify_placeholder_column(e) {
+                result = Some(t);
+                return Ok(TreeNodeRecursion::Stop);
+            }
+            Ok(TreeNodeRecursion::Continue)
+        });
+        result.map(|t| (t, false))
+    }
+
+    /// If `expr` is a `Column` whose name starts with `__unnest_placeholder`,
+    /// classify it as [`UnnestInputType::OuterReference`] or
+    /// [`UnnestInputType::Scalar`].
+    fn classify_placeholder_column(expr: &Expr) -> Option<UnnestInputType> {
+        if let Expr::Column(Column { name, .. }) = expr
             && let Some(prefix) = name.strip_prefix(UNNEST_PLACEHOLDER)
         {
             if prefix.starts_with(&format!("({OUTER_REFERENCE_COLUMN_PREFIX}(")) {
@@ -1380,39 +1332,6 @@ impl Unparser<'_> {
         None
     }
 
-    /// Recursively search the expression tree for an unnest placeholder column.
-    ///
-    /// Unlike [`Self::check_unnest_placeholder_with_outer_ref`] which only
-    /// matches when the placeholder IS the expression (modulo aliases), this
-    /// function finds placeholders buried inside function calls or other
-    /// transformations. For example:
-    ///
-    /// ```text
-    /// Alias("target_type",
-    ///   json_as_text(
-    ///     Column("__unnest_placeholder(...)"),  ← found here
-    ///     Literal("type")))
-    /// ```
-    fn find_unnest_placeholder_in_expr(expr: &Expr) -> Option<UnnestInputType> {
-        let mut result = None;
-        let _ = expr.apply(|e| {
-            if let Expr::Column(Column { name, .. }) = e
-                && let Some(prefix) = name.strip_prefix(UNNEST_PLACEHOLDER)
-            {
-                result = if prefix
-                    .starts_with(&format!("({OUTER_REFERENCE_COLUMN_PREFIX}("))
-                {
-                    Some(UnnestInputType::OuterReference)
-                } else {
-                    Some(UnnestInputType::Scalar)
-                };
-                return Ok(TreeNodeRecursion::Stop);
-            }
-            Ok(TreeNodeRecursion::Continue)
-        });
-        result
-    }
-
     /// Extract the outermost user-provided alias from an unnest expression.
     /// Returns `None` if the outermost alias is DataFusion's internal display
     /// name (e.g. `UNNEST(make_array(...))`), or if there is no alias at all.
@@ -1463,7 +1382,7 @@ impl Unparser<'_> {
         user_alias: Option<&str>,
     ) -> ast::SelectItem {
         let compound = ast::Expr::CompoundIdentifier(vec![
-            self.new_ident_without_quote_style(flatten_alias.to_string()),
+            self.new_ident_quoted_if_needs(flatten_alias.to_string()),
             Ident::with_quote('"', "VALUE"),
         ]);
         match user_alias {