Add support for Create Iceberg Table statement for Snowflake parser

src/ast/mod.rs

@@ -2516,6 +2516,35 @@ pub enum Statement {
     /// CREATE TABLE
     /// ```
     CreateTable(CreateTable),
+    /// ``` sql
+    /// CREATE ICEBERG TABLE
+    /// Snowflake-specific statement
+    /// <https://docs.snowflake.com/en/sql-reference/sql/create-iceberg-table>
+    /// ```
+    CreateIcebergTable {
+        or_replace: bool,
+        if_not_exists: bool,
+        /// Table name
+        #[cfg_attr(feature = "visitor", visit(with = "visit_relation"))]
+        name: ObjectName,
+        columns: Vec<ColumnDef>,
+        constraints: Vec<TableConstraint>,
+        with_options: Vec<SqlOption>,
+        comment: Option<CommentDef>,
+        cluster_by: Option<WrappedCollection<Vec<Ident>>>,
+        external_volume: Option<String>,
+        catalog: Option<String>,
+        base_location: String,
+        catalog_sync: Option<String>,
+        storage_serialization_policy: Option<StorageSerializationPolicy>,
+        data_retention_time_in_days: Option<u64>,
+        max_data_extension_time_in_days: Option<u64>,
+        change_tracking: Option<bool>,
+        copy_grants: bool,
+        with_aggregation_policy: Option<ObjectName>,
+        with_row_access_policy: Option<RowAccessPolicy>,
+        with_tags: Option<Vec<Tag>>,
+    },
     /// ```sql
     /// CREATE VIRTUAL TABLE .. USING <module_name> (<module_args>)`
     /// ```
@@ -4030,6 +4059,120 @@ impl fmt::Display for Statement {
                 }
                 Ok(())
             }
+            Statement::CreateIcebergTable {
+                or_replace,
+                if_not_exists,
+                name,
+                columns,
+                constraints,
+                with_options,
+                comment,
+                cluster_by,
+                external_volume,
+                catalog,
+                base_location,
+                catalog_sync,
+                storage_serialization_policy,
+                data_retention_time_in_days,
+                max_data_extension_time_in_days,
+                change_tracking,
+                copy_grants,
+                with_row_access_policy,
+                with_aggregation_policy,
+                with_tags,
+            } => {
+                write!(
+                    f,
+                    "CREATE {or_replace}ICEBERG TABLE {if_not_exists}{name}",
+                    if_not_exists = if *if_not_exists { "IF NOT EXISTS" } else { "" },
+                    or_replace = if *or_replace { "OR REPLACE " } else { "" }
+                )?;
+                if !columns.is_empty() || !constraints.is_empty() {
+                    write!(f, " ({}", display_comma_separated(columns))?;
+                    if !columns.is_empty() && !constraints.is_empty() {
+                        write!(f, ", ")?;
+                    }
+                    write!(f, "{})", display_comma_separated(&constraints))?;
+                }
+                if !with_options.is_empty() {
+                    write!(f, " WITH ({})", display_comma_separated(&with_options))?;
+                }
+                if let Some(comment_def) = &comment {
+                    match comment_def {
+                        CommentDef::WithEq(comment) => {
+                            write!(f, " COMMENT = '{comment}'")?;
+                        }
+                        CommentDef::WithoutEq(comment) => {
+                            write!(f, " COMMENT '{comment}'")?;
+                        }
+                        // For CommentDef::AfterColumnDefsWithoutEq will be displayed after column definition
+                        CommentDef::AfterColumnDefsWithoutEq(_) => (),
+                    }
+                }
+                if let Some(cluster_by) = cluster_by {
+                    write!(f, " CLUSTER BY {cluster_by}")?;
+                }
+                if let Some(external_volume) = external_volume {
+                    write!(f, " EXTERNAL_VOLUME = '{external_volume}'")?;
+                }
+
+                if let Some(catalog) = catalog {
+                    write!(f, " CATALOG = '{catalog}'")?;
+                }
+
+                write!(f, " BASE_LOCATION = '{base_location}'")?;
+
+                if let Some(catalog_sync) = catalog_sync {
+                    write!(f, " CATALOG_SYNC = '{catalog_sync}'")?;
+                }
+
+                if let Some(storage_serialization_policy) = storage_serialization_policy {
+                    write!(
+                        f,
+                        " STORAGE_SERIALIZATION_POLICY = {storage_serialization_policy}"
+                    )?;
+                }
+
+                if *copy_grants {
+                    write!(f, " COPY GRANTS")?;
+                }
+
+                if let Some(is_enabled) = change_tracking {
+                    write!(
+                        f,
+                        " CHANGE_TRACKING = {}",
+                        if *is_enabled { "TRUE" } else { "FALSE" }
+                    )?;
+                }
+
+                if let Some(data_retention_time_in_days) = data_retention_time_in_days {
+                    write!(
+                        f,
+                        " DATA_RETENTION_TIME_IN_DAYS = {data_retention_time_in_days}",
+                    )?;
+                }
+
+                if let Some(max_data_extension_time_in_days) = max_data_extension_time_in_days {
+                    write!(
+                        f,
+                        " MAX_DATA_EXTENSION_TIME_IN_DAYS = {max_data_extension_time_in_days}",
+                    )?;
+                }
+
+                if let Some(with_aggregation_policy) = with_aggregation_policy {
+                    write!(f, " WITH AGGREGATION POLICY {with_aggregation_policy}",)?;
+                }
+
+                if let Some(row_access_policy) = with_row_access_policy {
+                    write!(f, " {row_access_policy}",)?;
+                }
+
+                if let Some(tag) = with_tags {
+                    write!(f, " WITH TAG ({})", display_comma_separated(tag.as_slice()))?;
+                }
+
+                Ok(())
+            }
             Statement::CreateVirtualTable {
                 name,
                 if_not_exists,
@@ -8064,6 +8207,29 @@ impl fmt::Display for SessionParamValue {
     }
 }
 
+/// Snowflake StorageSerializationPolicy for Iceberg Tables
+/// ```sql
+/// [ STORAGE_SERIALIZATION_POLICY = { COMPATIBLE | OPTIMIZED } ]
+/// ```
+///
+/// <https://docs.snowflake.com/en/sql-reference/sql/create-iceberg-table>
+#[derive(Debug, Copy, Clone, PartialEq, PartialOrd, Eq, Ord, Hash)]
+#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
+#[cfg_attr(feature = "visitor", derive(Visit, VisitMut))]
+pub enum StorageSerializationPolicy {
+    Compatible,
+    Optimized,
+}
+
+impl Display for StorageSerializationPolicy {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            StorageSerializationPolicy::Compatible => write!(f, "COMPATIBLE"),
+            StorageSerializationPolicy::Optimized => write!(f, "OPTIMIZED"),
+        }
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

src/ast/spans.rs

@@ -384,6 +384,33 @@ impl Spanned for Statement {
                     .chain(to.iter().map(|i| i.span())),
             ),
             Statement::CreateTable(create_table) => create_table.span(),
+            Statement::CreateIcebergTable {
+                or_replace: _,
+                if_not_exists: _,
+                name,
+                columns,
+                constraints,
+                with_options,
+                comment: _,
+                cluster_by: _,
+                external_volume: _,
+                catalog: _,
+                base_location: _,
+                catalog_sync: _,
+                storage_serialization_policy: _,
+                data_retention_time_in_days: _,
+                max_data_extension_time_in_days: _,
+                change_tracking: _,
+                copy_grants: _,
+                with_row_access_policy: _,
+                with_aggregation_policy: _,
+                with_tags: _,
+            } => union_spans(
+                core::iter::once(name.span())
+                    .chain(columns.iter().map(|i| i.span()))
+                    .chain(constraints.iter().map(|i| i.span()))
+                    .chain(with_options.iter().map(|i| i.span())),
+            ),
             Statement::CreateVirtualTable {
                 name,
                 if_not_exists: _,