Snowflake Unparser dialect and UNNEST support (#21593)

## Which issue does this PR close?

- Closes #21592.

## Rationale for this change

The SQL unparser needs a Snowflake dialect. Basic dialect settings
(identifier quoting, `NULLS FIRST`/`NULLS LAST`, timestamp types) are
straightforward, but `UNNEST` support required more than configuration.

Snowflake has no `UNNEST` keyword. Its equivalent, `LATERAL
FLATTEN(INPUT => expr)`, is a table function in the `FROM` clause with
output accessed via `alias."VALUE"`. This differs structurally from
standard SQL: the unparser must emit a `FROM`-clause table factor with a
`CROSS JOIN` instead of a `SELECT`-clause expression. It also must
rewrite column references to point at the FLATTEN output, and handle
several optimizer-produced plan shapes (intermediate `Limit`/`Sort`
nodes, `SubqueryAlias` wrappers, composed expressions wrapping the
unnest output, multi-expression projections). None of this can be
expressed through `CustomDialectBuilder`.

## What changes are included in this PR?

**`dialect.rs`** - New `SnowflakeDialect` with double-quote identifiers,
`NULLS FIRST`/`NULLS LAST`, no empty select lists, no column aliases in
table aliases, Snowflake timestamp types, and
`unnest_as_lateral_flatten()`. Also wired into
`CustomDialect`/`CustomDialectBuilder`.

**`ast.rs`** - New `FlattenRelationBuilder` that produces `LATERAL
FLATTEN(INPUT => expr, OUTER => bool)` table factors, parallel to the
existing `UnnestRelationBuilder`.

**`utils.rs`** - New `unproject_unnest_expr_as_flatten_value` transform
that rewrites unnest placeholder columns to `_unnest.VALUE` references.

**`plan.rs`** - Changes to `select_to_sql_recursively`:
- The `Projection` handler scans all expressions for unnest placeholders
(not just single-expression projections), then branches into the FLATTEN
path or the existing table-factor path.
- `peel_to_unnest_with_modifiers` walks through `Limit`/`Sort` nodes
between `Projection` and `Unnest`, applying their SQL modifiers to the
query builder. This handles an optimizer behavior where these nodes are
inserted between the two.
- `peel_to_inner_projection` walks through `SubqueryAlias` to find the
inner `Projection` that feeds an `Unnest`.
- `reconstruct_select_statement` gained FLATTEN-aware expression
rewriting and a `has_internal_unnest_alias` predicate to strip internal
`UNNEST(...)` display names.
- The `Unnest` handler rejects struct columns for the FLATTEN dialect
with a clear error.

## Are these changes tested?

Yes. 18 new tests covering:
- Simple inline arrays, string arrays, cross joins
- Implicit `FROM` (UNNEST in SELECT clause)
- User aliases, table aliases, literal + unnest
- Subselect source with filters and limit
- UDF result as FLATTEN input
- `Limit` between `Projection` and `Unnest`
- `Sort` between `Projection` and `Unnest`
- `Limit` + `SubqueryAlias` combined
- Composed expressions wrapping unnest output (e.g. `CAST`)
- Composed expressions with `Limit`
- Multi-expression projections
- Multi-expression projections with `Limit`
- `SubqueryAlias` between `Unnest` and inner `Projection`

## Are there any user-facing changes?

Yes. New public API surface:
- `SnowflakeDialect` struct and its constructor
- `Dialect::unnest_as_lateral_flatten()` method (default `false`)
- `CustomDialectBuilder::with_unnest_as_lateral_flatten()`
- `FlattenRelationBuilder` and `FLATTEN_DEFAULT_ALIAS` in the AST module

None of these are breaking changes, and all previous APIs should work. 
New traits have default implementations to ease migrations.

---------

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

Author: yonatan-sevenai

datafusion/sql/src/unparser/ast.rs

@@ -162,9 +162,52 @@ 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,
+    /// 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>,
 }
 
+/// Prefix used for auto-generated LATERAL FLATTEN table aliases.
+const FLATTEN_ALIAS_PREFIX: &str = "_unnest";
+
 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!("{FLATTEN_ALIAS_PREFIX}_{}", self.flatten_alias_counter)
+    }
+
+    /// Register a table alias as pointing to a LATERAL FLATTEN relation.
+    pub fn add_flatten_table_alias(&mut self, alias: String) {
+        self.flatten_table_aliases.push(alias);
+    }
+
+    /// Returns true if no FLATTEN table aliases have been registered.
+    pub fn flatten_table_aliases_empty(&self) -> bool {
+        self.flatten_table_aliases.is_empty()
+    }
+
+    /// Returns true if the given table alias refers to a FLATTEN relation.
+    pub fn is_flatten_table_alias(&self, alias: &str) -> bool {
+        self.flatten_table_aliases.iter().any(|a| a == alias)
+    }
+
+    /// 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!(
+                "{FLATTEN_ALIAS_PREFIX}_{}",
+                self.flatten_alias_counter
+            ))
+        } else {
+            None
+        }
+    }
+
     pub fn distinct(&mut self, value: Option<ast::Distinct>) -> &mut Self {
         self.distinct = value;
         self
@@ -371,6 +414,8 @@ impl SelectBuilder {
             qualify: Default::default(),
             value_table_mode: Default::default(),
             flavor: Some(SelectFlavor::Standard),
+            flatten_alias_counter: 0,
+            flatten_table_aliases: Vec::new(),
         }
     }
 }
@@ -432,11 +477,11 @@ pub struct RelationBuilder {
 }
 
 #[derive(Clone)]
-#[expect(clippy::large_enum_variant)]
 enum TableFactorBuilder {
     Table(TableRelationBuilder),
     Derived(DerivedRelationBuilder),
     Unnest(UnnestRelationBuilder),
+    Flatten(FlattenRelationBuilder),
     Empty,
 }
 
@@ -458,6 +503,11 @@ impl RelationBuilder {
         self
     }
 
+    pub fn flatten(&mut self, value: FlattenRelationBuilder) -> &mut Self {
+        self.relation = Some(TableFactorBuilder::Flatten(value));
+        self
+    }
+
     pub fn empty(&mut self) -> &mut Self {
         self.relation = Some(TableFactorBuilder::Empty);
         self
@@ -474,6 +524,9 @@ impl RelationBuilder {
             Some(TableFactorBuilder::Unnest(ref mut rel_builder)) => {
                 rel_builder.alias = value;
             }
+            Some(TableFactorBuilder::Flatten(ref mut rel_builder)) => {
+                rel_builder.alias = value;
+            }
             Some(TableFactorBuilder::Empty) => (),
             None => (),
         }
@@ -484,6 +537,7 @@ impl RelationBuilder {
             Some(TableFactorBuilder::Table(ref value)) => Some(value.build()?),
             Some(TableFactorBuilder::Derived(ref value)) => Some(value.build()?),
             Some(TableFactorBuilder::Unnest(ref value)) => Some(value.build()?),
+            Some(TableFactorBuilder::Flatten(ref value)) => Some(value.build()?),
             Some(TableFactorBuilder::Empty) => None,
             None => return Err(Into::into(UninitializedFieldError::from("relation"))),
         })
@@ -688,6 +742,77 @@ impl Default for UnnestRelationBuilder {
     }
 }
 
+/// Builds a `LATERAL FLATTEN(INPUT => expr, OUTER => bool)` table factor
+/// for Snowflake-style unnesting.
+#[derive(Clone)]
+pub struct FlattenRelationBuilder {
+    pub alias: Option<ast::TableAlias>,
+    /// The input expression to flatten (e.g. a column reference).
+    pub input_expr: Option<ast::Expr>,
+    /// Whether to preserve rows for NULL/empty inputs (Snowflake `OUTER` param).
+    pub outer: bool,
+}
+
+impl FlattenRelationBuilder {
+    pub fn alias(&mut self, value: Option<ast::TableAlias>) -> &mut Self {
+        self.alias = value;
+        self
+    }
+
+    pub fn input_expr(&mut self, value: ast::Expr) -> &mut Self {
+        self.input_expr = Some(value);
+        self
+    }
+
+    pub fn outer(&mut self, value: bool) -> &mut Self {
+        self.outer = value;
+        self
+    }
+
+    pub fn build(&self) -> Result<ast::TableFactor, BuilderError> {
+        let input = self.input_expr.clone().ok_or_else(|| {
+            BuilderError::from(UninitializedFieldError::from("input_expr"))
+        })?;
+
+        let mut args = vec![ast::FunctionArg::Named {
+            name: ast::Ident::new("INPUT"),
+            arg: ast::FunctionArgExpr::Expr(input),
+            operator: ast::FunctionArgOperator::RightArrow,
+        }];
+
+        if self.outer {
+            args.push(ast::FunctionArg::Named {
+                name: ast::Ident::new("OUTER"),
+                arg: ast::FunctionArgExpr::Expr(ast::Expr::Value(
+                    ast::Value::Boolean(true).into(),
+                )),
+                operator: ast::FunctionArgOperator::RightArrow,
+            });
+        }
+
+        Ok(ast::TableFactor::Function {
+            lateral: true,
+            name: ast::ObjectName::from(vec![ast::Ident::new("FLATTEN")]),
+            args,
+            alias: self.alias.clone(),
+        })
+    }
+
+    fn create_empty() -> Self {
+        Self {
+            alias: None,
+            input_expr: None,
+            outer: false,
+        }
+    }
+}
+
+impl Default for FlattenRelationBuilder {
+    fn default() -> Self {
+        Self::create_empty()
+    }
+}
+
 /// Runtime error when a `build()` method is called and one or more required fields
 /// do not have a value.
 #[derive(Debug, Clone)]

datafusion/sql/src/unparser/dialect.rs

@@ -211,6 +211,15 @@ pub trait Dialect: Send + Sync {
         false
     }
 
+    /// Unparse the unnest plan as `LATERAL FLATTEN(INPUT => expr, ...)`.
+    ///
+    /// Snowflake uses FLATTEN as a table function instead of the SQL-standard UNNEST.
+    /// When this returns `true`, the unparser emits
+    /// `LATERAL FLATTEN(INPUT => <col>, OUTER => <bool>)` in the FROM clause.
+    fn unnest_as_lateral_flatten(&self) -> bool {
+        false
+    }
+
     /// Allows the dialect to override column alias unparsing if the dialect has specific rules.
     /// Returns None if the default unparsing should be used, or Some(String) if there is
     /// a custom implementation for the alias.
@@ -718,6 +727,59 @@ impl BigQueryDialect {
     }
 }
 
+/// Dialect for Snowflake SQL.
+///
+/// Key differences from the default dialect:
+/// - Uses double-quote identifier quoting
+/// - Supports `NULLS FIRST`/`NULLS LAST` in `ORDER BY`
+/// - Does not support empty select lists (`SELECT FROM t`)
+/// - Does not support column aliases in table alias definitions
+///   (Snowflake accepts the syntax but silently ignores the renames in join contexts)
+/// - Unparses `UNNEST` plans as `LATERAL FLATTEN(INPUT => expr, ...)`
+pub struct SnowflakeDialect {}
+
+#[expect(clippy::new_without_default)]
+impl SnowflakeDialect {
+    #[must_use]
+    pub fn new() -> Self {
+        Self {}
+    }
+}
+
+impl Dialect for SnowflakeDialect {
+    fn identifier_quote_style(&self, _: &str) -> Option<char> {
+        Some('"')
+    }
+
+    fn supports_nulls_first_in_sort(&self) -> bool {
+        true
+    }
+
+    fn supports_empty_select_list(&self) -> bool {
+        false
+    }
+
+    fn supports_column_alias_in_table_alias(&self) -> bool {
+        false
+    }
+
+    fn timestamp_cast_dtype(
+        &self,
+        _time_unit: &TimeUnit,
+        tz: &Option<Arc<str>>,
+    ) -> ast::DataType {
+        if tz.is_some() {
+            ast::DataType::Timestamp(None, TimezoneInfo::WithTimeZone)
+        } else {
+            ast::DataType::Timestamp(None, TimezoneInfo::None)
+        }
+    }
+
+    fn unnest_as_lateral_flatten(&self) -> bool {
+        true
+    }
+}
+
 pub struct CustomDialect {
     identifier_quote_style: Option<char>,
     supports_nulls_first_in_sort: bool,
@@ -740,6 +802,7 @@ pub struct CustomDialect {
     window_func_support_window_frame: bool,
     full_qualified_col: bool,
     unnest_as_table_factor: bool,
+    unnest_as_lateral_flatten: bool,
 }
 
 impl Default for CustomDialect {
@@ -769,6 +832,7 @@ impl Default for CustomDialect {
             window_func_support_window_frame: true,
             full_qualified_col: false,
             unnest_as_table_factor: false,
+            unnest_as_lateral_flatten: false,
         }
     }
 }
@@ -883,6 +947,10 @@ impl Dialect for CustomDialect {
     fn unnest_as_table_factor(&self) -> bool {
         self.unnest_as_table_factor
     }
+
+    fn unnest_as_lateral_flatten(&self) -> bool {
+        self.unnest_as_lateral_flatten
+    }
 }
 
 /// `CustomDialectBuilder` to build `CustomDialect` using builder pattern
@@ -921,6 +989,7 @@ pub struct CustomDialectBuilder {
     window_func_support_window_frame: bool,
     full_qualified_col: bool,
     unnest_as_table_factor: bool,
+    unnest_as_lateral_flatten: bool,
 }
 
 impl Default for CustomDialectBuilder {
@@ -956,6 +1025,7 @@ impl CustomDialectBuilder {
             window_func_support_window_frame: true,
             full_qualified_col: false,
             unnest_as_table_factor: false,
+            unnest_as_lateral_flatten: false,
         }
     }
 
@@ -983,6 +1053,7 @@ impl CustomDialectBuilder {
             window_func_support_window_frame: self.window_func_support_window_frame,
             full_qualified_col: self.full_qualified_col,
             unnest_as_table_factor: self.unnest_as_table_factor,
+            unnest_as_lateral_flatten: self.unnest_as_lateral_flatten,
         }
     }
 
@@ -1129,4 +1200,12 @@ impl CustomDialectBuilder {
         self.unnest_as_table_factor = unnest_as_table_factor;
         self
     }
+
+    pub fn with_unnest_as_lateral_flatten(
+        mut self,
+        unnest_as_lateral_flatten: bool,
+    ) -> Self {
+        self.unnest_as_lateral_flatten = unnest_as_lateral_flatten;
+        self
+    }
 }