Snowflake: Add support for CREATE DYNAMIC TABLE (#1960)

Author: yoavcloud

src/ast/ddl.rs

@@ -33,10 +33,11 @@ use crate::ast::{
     display_comma_separated, display_separated, ArgMode, CommentDef, CreateFunctionBody,
     CreateFunctionUsing, CreateTableLikeKind, CreateTableOptions, DataType, Expr, FileFormat,
     FunctionBehavior, FunctionCalledOnNull, FunctionDeterminismSpecifier, FunctionParallel,
-    HiveDistributionStyle, HiveFormat, HiveIOFormat, HiveRowFormat, Ident, MySQLColumnPosition,
-    ObjectName, OnCommit, OneOrManyWithParens, OperateFunctionArg, OrderByExpr, ProjectionSelect,
-    Query, RowAccessPolicy, SequenceOptions, Spanned, SqlOption, StorageSerializationPolicy, Tag,
-    Value, ValueWithSpan, WrappedCollection,
+    HiveDistributionStyle, HiveFormat, HiveIOFormat, HiveRowFormat, Ident, InitializeKind,
+    MySQLColumnPosition, ObjectName, OnCommit, OneOrManyWithParens, OperateFunctionArg,
+    OrderByExpr, ProjectionSelect, Query, RefreshModeKind, RowAccessPolicy, SequenceOptions,
+    Spanned, SqlOption, StorageSerializationPolicy, TableVersion, Tag, Value, ValueWithSpan,
+    WrappedCollection,
 };
 use crate::display_utils::{DisplayCommaSeparated, Indent, NewLine, SpaceOrNewline};
 use crate::keywords::Keyword;
@@ -2428,6 +2429,7 @@ pub struct CreateTable {
     pub or_replace: bool,
     pub temporary: bool,
     pub external: bool,
+    pub dynamic: bool,
     pub global: Option<bool>,
     pub if_not_exists: bool,
     pub transient: bool,
@@ -2448,6 +2450,7 @@ pub struct CreateTable {
     pub without_rowid: bool,
     pub like: Option<CreateTableLikeKind>,
     pub clone: Option<ObjectName>,
+    pub version: Option<TableVersion>,
     // For Hive dialect, the table comment is after the column definitions without `=`,
     // so the `comment` field is optional and different than the comment field in the general options list.
     // [Hive](https://cwiki.apache.org/confluence/display/Hive/LanguageManual+DDL#LanguageManualDDL-CreateTable)
@@ -2525,6 +2528,21 @@ pub struct CreateTable {
     /// Snowflake "STORAGE_SERIALIZATION_POLICY" clause for Iceberg tables
     /// <https://docs.snowflake.com/en/sql-reference/sql/create-iceberg-table>
     pub storage_serialization_policy: Option<StorageSerializationPolicy>,
+    /// Snowflake "TARGET_LAG" clause for dybamic tables
+    /// <https://docs.snowflake.com/en/sql-reference/sql/create-dynamic-table>
+    pub target_lag: Option<String>,
+    /// Snowflake "WAREHOUSE" clause for dybamic tables
+    /// <https://docs.snowflake.com/en/sql-reference/sql/create-dynamic-table>
+    pub warehouse: Option<Ident>,
+    /// Snowflake "REFRESH_MODE" clause for dybamic tables
+    /// <https://docs.snowflake.com/en/sql-reference/sql/create-dynamic-table>
+    pub refresh_mode: Option<RefreshModeKind>,
+    /// Snowflake "INITIALIZE" clause for dybamic tables
+    /// <https://docs.snowflake.com/en/sql-reference/sql/create-dynamic-table>
+    pub initialize: Option<InitializeKind>,
+    /// Snowflake "REQUIRE USER" clause for dybamic tables
+    /// <https://docs.snowflake.com/en/sql-reference/sql/create-dynamic-table>
+    pub require_user: bool,
 }
 
 impl fmt::Display for CreateTable {
@@ -2538,7 +2556,7 @@ impl fmt::Display for CreateTable {
         //   `CREATE TABLE t (a INT) AS SELECT a from t2`
         write!(
             f,
-            "CREATE {or_replace}{external}{global}{temporary}{transient}{volatile}{iceberg}TABLE {if_not_exists}{name}",
+            "CREATE {or_replace}{external}{global}{temporary}{transient}{volatile}{dynamic}{iceberg}TABLE {if_not_exists}{name}",
             or_replace = if self.or_replace { "OR REPLACE " } else { "" },
             external = if self.external { "EXTERNAL " } else { "" },
             global = self.global
@@ -2556,6 +2574,7 @@ impl fmt::Display for CreateTable {
             volatile = if self.volatile { "VOLATILE " } else { "" },
             // Only for Snowflake
             iceberg = if self.iceberg { "ICEBERG " } else { "" },
+            dynamic = if self.dynamic { "DYNAMIC " } else { "" },
             name = self.name,
         )?;
         if let Some(on_cluster) = &self.on_cluster {
@@ -2598,6 +2617,10 @@ impl fmt::Display for CreateTable {
             write!(f, " CLONE {c}")?;
         }
 
+        if let Some(version) = &self.version {
+            write!(f, " {version}")?;
+        }
+
         match &self.hive_distribution {
             HiveDistributionStyle::PARTITIONED { columns } => {
                 write!(f, " PARTITIONED BY ({})", display_comma_separated(columns))?;
@@ -2700,27 +2723,27 @@ impl fmt::Display for CreateTable {
             write!(f, " {options}")?;
         }
         if let Some(external_volume) = self.external_volume.as_ref() {
-            write!(f, " EXTERNAL_VOLUME = '{external_volume}'")?;
+            write!(f, " EXTERNAL_VOLUME='{external_volume}'")?;
         }
 
         if let Some(catalog) = self.catalog.as_ref() {
-            write!(f, " CATALOG = '{catalog}'")?;
+            write!(f, " CATALOG='{catalog}'")?;
         }
 
         if self.iceberg {
             if let Some(base_location) = self.base_location.as_ref() {
-                write!(f, " BASE_LOCATION = '{base_location}'")?;
+                write!(f, " BASE_LOCATION='{base_location}'")?;
             }
         }
 
         if let Some(catalog_sync) = self.catalog_sync.as_ref() {
-            write!(f, " CATALOG_SYNC = '{catalog_sync}'")?;
+            write!(f, " CATALOG_SYNC='{catalog_sync}'")?;
         }
 
         if let Some(storage_serialization_policy) = self.storage_serialization_policy.as_ref() {
             write!(
                 f,
-                " STORAGE_SERIALIZATION_POLICY = {storage_serialization_policy}"
+                " STORAGE_SERIALIZATION_POLICY={storage_serialization_policy}"
             )?;
         }
 
@@ -2774,6 +2797,26 @@ impl fmt::Display for CreateTable {
             write!(f, " WITH TAG ({})", display_comma_separated(tag.as_slice()))?;
         }
 
+        if let Some(target_lag) = &self.target_lag {
+            write!(f, " TARGET_LAG='{target_lag}'")?;
+        }
+
+        if let Some(warehouse) = &self.warehouse {
+            write!(f, " WAREHOUSE={warehouse}")?;
+        }
+
+        if let Some(refresh_mode) = &self.refresh_mode {
+            write!(f, " REFRESH_MODE={refresh_mode}")?;
+        }
+
+        if let Some(initialize) = &self.initialize {
+            write!(f, " INITIALIZE={initialize}")?;
+        }
+
+        if self.require_user {
+            write!(f, " REQUIRE USER")?;
+        }
+
         if self.on_commit.is_some() {
             let on_commit = match self.on_commit {
                 Some(OnCommit::DeleteRows) => "ON COMMIT DELETE ROWS",

src/ast/helpers/stmt_create_table.rs

@@ -26,9 +26,9 @@ use sqlparser_derive::{Visit, VisitMut};
 
 use crate::ast::{
     ClusteredBy, ColumnDef, CommentDef, CreateTable, CreateTableLikeKind, CreateTableOptions, Expr,
-    FileFormat, HiveDistributionStyle, HiveFormat, Ident, ObjectName, OnCommit,
-    OneOrManyWithParens, Query, RowAccessPolicy, Statement, StorageSerializationPolicy,
-    TableConstraint, Tag, WrappedCollection,
+    FileFormat, HiveDistributionStyle, HiveFormat, Ident, InitializeKind, ObjectName, OnCommit,
+    OneOrManyWithParens, Query, RefreshModeKind, RowAccessPolicy, Statement,
+    StorageSerializationPolicy, TableConstraint, TableVersion, Tag, WrappedCollection,
 };
 
 use crate::parser::ParserError;
@@ -72,6 +72,7 @@ pub struct CreateTableBuilder {
     pub transient: bool,
     pub volatile: bool,
     pub iceberg: bool,
+    pub dynamic: bool,
     pub name: ObjectName,
     pub columns: Vec<ColumnDef>,
     pub constraints: Vec<TableConstraint>,
@@ -83,6 +84,7 @@ pub struct CreateTableBuilder {
     pub without_rowid: bool,
     pub like: Option<CreateTableLikeKind>,
     pub clone: Option<ObjectName>,
+    pub version: Option<TableVersion>,
     pub comment: Option<CommentDef>,
     pub on_commit: Option<OnCommit>,
     pub on_cluster: Option<Ident>,
@@ -108,6 +110,11 @@ pub struct CreateTableBuilder {
     pub catalog_sync: Option<String>,
     pub storage_serialization_policy: Option<StorageSerializationPolicy>,
     pub table_options: CreateTableOptions,
+    pub target_lag: Option<String>,
+    pub warehouse: Option<Ident>,
+    pub refresh_mode: Option<RefreshModeKind>,
+    pub initialize: Option<InitializeKind>,
+    pub require_user: bool,
 }
 
 impl CreateTableBuilder {
@@ -121,6 +128,7 @@ impl CreateTableBuilder {
             transient: false,
             volatile: false,
             iceberg: false,
+            dynamic: false,
             name,
             columns: vec![],
             constraints: vec![],
@@ -132,6 +140,7 @@ impl CreateTableBuilder {
             without_rowid: false,
             like: None,
             clone: None,
+            version: None,
             comment: None,
             on_commit: None,
             on_cluster: None,
@@ -157,6 +166,11 @@ impl CreateTableBuilder {
             catalog_sync: None,
             storage_serialization_policy: None,
             table_options: CreateTableOptions::None,
+            target_lag: None,
+            warehouse: None,
+            refresh_mode: None,
+            initialize: None,
+            require_user: false,
         }
     }
     pub fn or_replace(mut self, or_replace: bool) -> Self {
@@ -199,6 +213,11 @@ impl CreateTableBuilder {
         self
     }
 
+    pub fn dynamic(mut self, dynamic: bool) -> Self {
+        self.dynamic = dynamic;
+        self
+    }
+
     pub fn columns(mut self, columns: Vec<ColumnDef>) -> Self {
         self.columns = columns;
         self
@@ -248,6 +267,11 @@ impl CreateTableBuilder {
         self
     }
 
+    pub fn version(mut self, version: Option<TableVersion>) -> Self {
+        self.version = version;
+        self
+    }
+
     pub fn comment_after_column_def(mut self, comment: Option<CommentDef>) -> Self {
         self.comment = comment;
         self
@@ -382,24 +406,29 @@ impl CreateTableBuilder {
         self
     }
 
-    /// Returns true if the statement has exactly one source of info on the schema of the new table.
-    /// This is Snowflake-specific, some dialects allow more than one source.
-    pub(crate) fn validate_schema_info(&self) -> bool {
-        let mut sources = 0;
-        if !self.columns.is_empty() {
-            sources += 1;
-        }
-        if self.query.is_some() {
-            sources += 1;
-        }
-        if self.like.is_some() {
-            sources += 1;
-        }
-        if self.clone.is_some() {
-            sources += 1;
-        }
+    pub fn target_lag(mut self, target_lag: Option<String>) -> Self {
+        self.target_lag = target_lag;
+        self
+    }
+
+    pub fn warehouse(mut self, warehouse: Option<Ident>) -> Self {
+        self.warehouse = warehouse;
+        self
+    }
 
-        sources == 1
+    pub fn refresh_mode(mut self, refresh_mode: Option<RefreshModeKind>) -> Self {
+        self.refresh_mode = refresh_mode;
+        self
+    }
+
+    pub fn initialize(mut self, initialize: Option<InitializeKind>) -> Self {
+        self.initialize = initialize;
+        self
+    }
+
+    pub fn require_user(mut self, require_user: bool) -> Self {
+        self.require_user = require_user;
+        self
     }
 
     pub fn build(self) -> Statement {
@@ -412,6 +441,7 @@ impl CreateTableBuilder {
             transient: self.transient,
             volatile: self.volatile,
             iceberg: self.iceberg,
+            dynamic: self.dynamic,
             name: self.name,
             columns: self.columns,
             constraints: self.constraints,
@@ -423,6 +453,7 @@ impl CreateTableBuilder {
             without_rowid: self.without_rowid,
             like: self.like,
             clone: self.clone,
+            version: self.version,
             comment: self.comment,
             on_commit: self.on_commit,
             on_cluster: self.on_cluster,
@@ -448,6 +479,11 @@ impl CreateTableBuilder {
             catalog_sync: self.catalog_sync,
             storage_serialization_policy: self.storage_serialization_policy,
             table_options: self.table_options,
+            target_lag: self.target_lag,
+            warehouse: self.warehouse,
+            refresh_mode: self.refresh_mode,
+            initialize: self.initialize,
+            require_user: self.require_user,
         })
     }
 }
@@ -468,6 +504,7 @@ impl TryFrom<Statement> for CreateTableBuilder {
                 transient,
                 volatile,
                 iceberg,
+                dynamic,
                 name,
                 columns,
                 constraints,
@@ -479,6 +516,7 @@ impl TryFrom<Statement> for CreateTableBuilder {
                 without_rowid,
                 like,
                 clone,
+                version,
                 comment,
                 on_commit,
                 on_cluster,
@@ -504,13 +542,19 @@ impl TryFrom<Statement> for CreateTableBuilder {
                 catalog_sync,
                 storage_serialization_policy,
                 table_options,
+                target_lag,
+                warehouse,
+                refresh_mode,
+                initialize,
+                require_user,
             }) => Ok(Self {
                 or_replace,
                 temporary,
                 external,
                 global,
                 if_not_exists,
                 transient,
+                dynamic,
                 name,
                 columns,
                 constraints,
@@ -522,6 +566,7 @@ impl TryFrom<Statement> for CreateTableBuilder {
                 without_rowid,
                 like,
                 clone,
+                version,
                 comment,
                 on_commit,
                 on_cluster,
@@ -549,6 +594,11 @@ impl TryFrom<Statement> for CreateTableBuilder {
                 catalog_sync,
                 storage_serialization_policy,
                 table_options,
+                target_lag,
+                warehouse,
+                refresh_mode,
+                initialize,
+                require_user,
             }),
             _ => Err(ParserError::ParserError(format!(
                 "Expected create table statement, but received: {stmt}"