Generate unique LATERAL FLATTEN aliases per query

Replace the hardcoded FLATTEN_DEFAULT_ALIAS ("_unnest") with a
per-SelectBuilder counter that generates unique aliases (_unnest_1,
_unnest_2, …). This prevents alias collisions when multiple unnests
appear in the same query.

- Add flatten_alias_counter to SelectBuilder with next/current
  accessor methods, scoped to one SELECT so subqueries get
  independent counters
- Remove FLATTEN_DEFAULT_ALIAS constant, the dead alias_name()
  method, and the default alias from FlattenRelationBuilder
- All three FLATTEN code paths (placeholder projection, display-name
  projection, and Unnest handler) now coordinate through the
  SelectBuilder to ensure SELECT items and FROM clause use the same
  alias
- Use internal_datafusion_err! macro for FLATTEN error handling
- Migrate unnest tests from partial .contains() assertions to
  insta::assert_snapshot! for full SQL verification

Author: yonatan-sevenai

datafusion/sql/src/unparser/ast.rs

@@ -162,9 +162,28 @@ pub struct SelectBuilder {
     qualify: Option<ast::Expr>,
     value_table_mode: Option<ast::ValueTableMode>,
     flavor: Option<SelectFlavor>,
+    /// Counter for generating unique LATERAL FLATTEN aliases within this SELECT.
+    flatten_alias_counter: usize,
 }
 
 impl SelectBuilder {
+    /// Generate a unique alias for a LATERAL FLATTEN relation
+    /// (`_unnest_1`, `_unnest_2`, …). Each call returns a fresh name.
+    pub fn next_flatten_alias(&mut self) -> String {
+        self.flatten_alias_counter += 1;
+        format!("_unnest_{}", self.flatten_alias_counter)
+    }
+
+    /// 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> {
+        if self.flatten_alias_counter > 0 {
+            Some(format!("_unnest_{}", self.flatten_alias_counter))
+        } else {
+            None
+        }
+    }
+
     pub fn distinct(&mut self, value: Option<ast::Distinct>) -> &mut Self {
         self.distinct = value;
         self
@@ -371,6 +390,7 @@ impl SelectBuilder {
             qualify: Default::default(),
             value_table_mode: Default::default(),
             flavor: Some(SelectFlavor::Standard),
+            flatten_alias_counter: 0,
         }
     }
 }
@@ -697,10 +717,6 @@ impl Default for UnnestRelationBuilder {
     }
 }
 
-/// Default table alias for FLATTEN table factors.
-/// Snowflake requires an alias to reference output columns (e.g. `_unnest.VALUE`).
-pub const FLATTEN_DEFAULT_ALIAS: &str = "_unnest";
-
 /// Builds a `LATERAL FLATTEN(INPUT => expr, OUTER => bool)` table factor
 /// for Snowflake-style unnesting.
 #[derive(Clone)]
@@ -757,22 +773,9 @@ impl FlattenRelationBuilder {
         })
     }
 
-    /// Returns the alias name for this FLATTEN relation.
-    /// Used to build qualified column references like `alias.VALUE`.
-    pub fn alias_name(&self) -> &str {
-        self.alias
-            .as_ref()
-            .map(|a| a.name.value.as_str())
-            .unwrap_or(FLATTEN_DEFAULT_ALIAS)
-    }
-
     fn create_empty() -> Self {
         Self {
-            alias: Some(ast::TableAlias {
-                name: ast::Ident::with_quote('"', FLATTEN_DEFAULT_ALIAS),
-                columns: vec![],
-                explicit: true,
-            }),
+            alias: None,
             input_expr: None,
             outer: false,
         }

datafusion/sql/src/unparser/plan.rs

@@ -38,13 +38,12 @@ use crate::unparser::extension_unparser::{
 };
 use crate::unparser::utils::{find_unnest_node_until_relation, unproject_agg_exprs};
 use crate::unparser::{
-    ast::FLATTEN_DEFAULT_ALIAS, ast::FlattenRelationBuilder, ast::UnnestRelationBuilder,
-    rewrite::rewrite_qualify,
+    ast::FlattenRelationBuilder, ast::UnnestRelationBuilder, rewrite::rewrite_qualify,
 };
 use crate::utils::UNNEST_PLACEHOLDER;
 use datafusion_common::{
     Column, DataFusionError, Result, ScalarValue, TableReference, assert_or_internal_err,
-    internal_err, not_impl_err,
+    internal_datafusion_err, internal_err, not_impl_err,
     tree_node::{TransformedResult, TreeNode, TreeNodeRecursion},
 };
 use datafusion_expr::expr::{OUTER_REFERENCE_COLUMN_PREFIX, UNNEST_COLUMN_PREFIX};
@@ -239,17 +238,12 @@ impl Unparser<'_> {
         let mut exprs = p.expr.clone();
 
         // If an Unnest node is found within the select, find and unproject the unnest column
+        let flatten_alias = select.current_flatten_alias();
         if let Some(unnest) = find_unnest_node_within_select(plan) {
-            if self.dialect.unnest_as_lateral_flatten() {
+            if let Some(ref alias) = flatten_alias {
                 exprs = exprs
                     .into_iter()
-                    .map(|e| {
-                        unproject_unnest_expr_as_flatten_value(
-                            e,
-                            unnest,
-                            FLATTEN_DEFAULT_ALIAS,
-                        )
-                    })
+                    .map(|e| unproject_unnest_expr_as_flatten_value(e, unnest, alias))
                     .collect::<Result<Vec<_>>>()?;
             } else {
                 exprs = exprs
@@ -294,19 +288,17 @@ impl Unparser<'_> {
                 select.projection(items);
             }
             _ => {
-                let use_flatten = self.dialect.unnest_as_lateral_flatten();
                 let items = exprs
                     .iter()
                     .map(|e| {
                         // After unproject_unnest_expr_as_flatten_value, an
                         // internal UNNEST display-name alias may still wrap
                         // the rewritten _unnest.VALUE column. Replace it
                         // with the bare FLATTEN VALUE select item.
-                        if use_flatten && Self::has_internal_unnest_alias(e) {
-                            return Ok(self.build_flatten_value_select_item(
-                                FLATTEN_DEFAULT_ALIAS,
-                                None,
-                            ));
+                        if let Some(ref alias) = flatten_alias
+                            && Self::has_internal_unnest_alias(e)
+                        {
+                            return Ok(self.build_flatten_value_select_item(alias, None));
                         }
                         self.select_item_to_sql(e)
                     })
@@ -360,6 +352,105 @@ impl Unparser<'_> {
         }
     }
 
+    /// Projection unparsing when [`super::dialect::Dialect::unnest_as_lateral_flatten`] is enabled:
+    /// Snowflake-style `LATERAL FLATTEN` for unnest (not other dialect spellings).
+    ///
+    /// [`Self::peel_to_unnest_with_modifiers`] walks through any intermediate
+    /// Limit/Sort nodes (the optimizer can insert these between the Projection
+    /// and the Unnest), applies their modifiers to the query, and returns the
+    /// Unnest plus the [`LogicalPlan`] ref to recurse into. This bypasses the
+    /// normal Limit/Sort handlers which would wrap the subtree in a derived
+    /// subquery.
+    ///
+    /// SELECT rendering is delegated to [`Self::reconstruct_select_statement`],
+    /// which rewrites placeholder columns to `alias."VALUE"` via
+    /// [`unproject_unnest_expr_as_flatten_value`].
+    ///
+    /// Returns `Ok(true)` when this path fully handled the projection.
+    fn try_projection_unnest_as_lateral_flatten(
+        &self,
+        plan: &LogicalPlan,
+        p: &Projection,
+        query: &mut Option<QueryBuilder>,
+        select: &mut SelectBuilder,
+        relation: &mut RelationBuilder,
+        unnest_input_type: Option<&UnnestInputType>,
+    ) -> Result<bool> {
+        // unnest_as_lateral_flatten: Snowflake LATERAL FLATTEN
+        if self.dialect.unnest_as_lateral_flatten()
+            && unnest_input_type.is_some()
+            && let Some((unnest, unnest_plan)) =
+                self.peel_to_unnest_with_modifiers(p.input.as_ref(), query)?
+            && let Some(mut flatten) = self.try_unnest_to_lateral_flatten_sql(unnest)?
+        {
+            let inner_projection = Self::peel_to_inner_projection(unnest.input.as_ref())
+                .ok_or_else(|| {
+                    internal_datafusion_err!(
+                        "Unnest input is not a Projection: {:?}",
+                        unnest.input
+                    )
+                })?;
+
+            // Generate a unique alias for this FLATTEN so that
+            // multiple unnests in the same query don't collide.
+            // When the SELECT was already built by an outer Projection
+            // (already_projected), it already called
+            // next_flatten_alias(), so we reuse that alias.
+            if !select.already_projected() {
+                let flatten_alias_name = select.next_flatten_alias();
+                flatten.alias(Some(ast::TableAlias {
+                    name: Ident::with_quote('"', &flatten_alias_name),
+                    columns: vec![],
+                    explicit: true,
+                }));
+                self.reconstruct_select_statement(plan, p, select)?;
+            } else if let Some(alias) = select.current_flatten_alias() {
+                flatten.alias(Some(ast::TableAlias {
+                    name: Ident::with_quote('"', &alias),
+                    columns: vec![],
+                    explicit: true,
+                }));
+            }
+
+            if matches!(
+                inner_projection.input.as_ref(),
+                LogicalPlan::EmptyRelation(_)
+            ) {
+                // Inline array (e.g. UNNEST([1,2,3])):
+                // FLATTEN is the sole FROM source.
+                relation.flatten(flatten);
+                self.select_to_sql_recursively(unnest_plan, query, select, relation)?;
+                return Ok(true);
+            }
+
+            // Non-empty source (table, subquery, etc.):
+            // recurse to set the primary FROM, then attach FLATTEN
+            // as a CROSS JOIN.
+            self.select_to_sql_recursively(unnest_plan, query, select, relation)?;
+
+            let flatten_factor = flatten
+                .build()
+                .map_err(|e| internal_datafusion_err!("Failed to build FLATTEN: {e}"))?;
+            let cross_join = ast::Join {
+                relation: flatten_factor,
+                global: false,
+                join_operator: ast::JoinOperator::CrossJoin(ast::JoinConstraint::None),
+            };
+            if let Some(mut from) = select.pop_from() {
+                from.push_join(cross_join);
+                select.push_from(from);
+            } else {
+                let mut twj = TableWithJoinsBuilder::default();
+                twj.push_join(cross_join);
+                select.push_from(twj);
+            }
+
+            return Ok(true);
+        }
+
+        Ok(false)
+    }
+
     #[cfg_attr(feature = "recursive_protection", recursive::recursive)]
     fn select_to_sql_recursively(
         &self,
@@ -435,80 +526,14 @@ impl Unparser<'_> {
                     );
                 }
 
-                // --- Snowflake LATERAL FLATTEN path ---
-                // `peel_to_unnest_with_modifiers` walks through any
-                // intermediate Limit/Sort nodes (the optimizer can insert
-                // these between the Projection and the Unnest), applies
-                // their modifiers to the query, and returns the Unnest +
-                // the LogicalPlan ref to recurse into. This bypasses the
-                // normal Limit/Sort handlers which would wrap the subtree
-                // in a derived subquery.
-                // SELECT rendering is delegated to
-                // `reconstruct_select_statement`, which rewrites
-                // placeholder columns to `"_unnest"."VALUE"` via
-                // `unproject_unnest_expr_as_flatten_value` — this works
-                // for bare, wrapped, and multi-expression projections.
-                if self.dialect.unnest_as_lateral_flatten()
-                    && unnest_input_type.is_some()
-                    && 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 inner_projection =
-                        Self::peel_to_inner_projection(unnest.input.as_ref())
-                            .ok_or_else(|| {
-                                DataFusionError::Internal(format!(
-                                    "Unnest input is not a Projection: {:?}",
-                                    unnest.input
-                                ))
-                            })?;
-
-                    // An outer plan (e.g. a wrapping Projection) may have
-                    // already set SELECT columns; only set them once.
-                    if !select.already_projected() {
-                        self.reconstruct_select_statement(plan, p, select)?;
-                    }
-
-                    if matches!(
-                        inner_projection.input.as_ref(),
-                        LogicalPlan::EmptyRelation(_)
-                    ) {
-                        // Inline array (e.g. UNNEST([1,2,3])):
-                        // FLATTEN is the sole FROM source.
-                        relation.flatten(flatten);
-                        return self.select_to_sql_recursively(
-                            unnest_plan,
-                            query,
-                            select,
-                            relation,
-                        );
-                    }
-
-                    // Non-empty source (table, subquery, etc.):
-                    // recurse to set the primary FROM, then attach FLATTEN
-                    // as a CROSS JOIN.
-                    self.select_to_sql_recursively(unnest_plan, query, select, relation)?;
-
-                    let flatten_factor = flatten.build().map_err(|e| {
-                        DataFusionError::Internal(format!("Failed to build FLATTEN: {e}"))
-                    })?;
-                    let cross_join = ast::Join {
-                        relation: flatten_factor,
-                        global: false,
-                        join_operator: ast::JoinOperator::CrossJoin(
-                            ast::JoinConstraint::None,
-                        ),
-                    };
-                    if let Some(mut from) = select.pop_from() {
-                        from.push_join(cross_join);
-                        select.push_from(from);
-                    } else {
-                        let mut twj = TableWithJoinsBuilder::default();
-                        twj.push_join(cross_join);
-                        select.push_from(twj);
-                    }
-
+                if self.try_projection_unnest_as_lateral_flatten(
+                    plan,
+                    p,
+                    query,
+                    select,
+                    relation,
+                    unnest_input_type.as_ref(),
+                )? {
                     return Ok(());
                 }
 
@@ -536,6 +561,16 @@ impl Unparser<'_> {
                         columns,
                     );
                 }
+                // For Snowflake FLATTEN: when the outer Projection has
+                // UNNEST(...) display-name columns (from SELECT * / SELECT
+                // UNNEST(...)), generate a flatten alias now so that
+                // reconstruct_select_statement and the downstream Unnest
+                // handler both use the same alias.
+                if self.dialect.unnest_as_lateral_flatten()
+                    && p.expr.iter().any(Self::has_internal_unnest_alias)
+                {
+                    select.next_flatten_alias();
+                }
                 self.reconstruct_select_statement(plan, p, select)?;
                 self.select_to_sql_recursively(p.input.as_ref(), query, select, relation)
             }
@@ -1135,9 +1170,19 @@ impl Unparser<'_> {
                 // relation here so the FROM clause is emitted.
                 if self.dialect.unnest_as_lateral_flatten()
                     && !relation.has_relation()
-                    && let Some(flatten_relation) =
+                    && let Some(mut flatten_relation) =
                         self.try_unnest_to_lateral_flatten_sql(unnest)?
                 {
+                    // Use the alias already generated by the Projection
+                    // handler so SELECT items and the FLATTEN relation
+                    // reference the same name.
+                    if let Some(alias) = select.current_flatten_alias() {
+                        flatten_relation.alias(Some(ast::TableAlias {
+                            name: Ident::with_quote('"', &alias),
+                            columns: vec![],
+                            explicit: true,
+                        }));
+                    }
                     relation.flatten(flatten_relation);
                 }