Snowflake: CREATE DYNAMIC TABLE

Author: yoavcloud

src/ast/dml.rs

@@ -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<ObjectName>,
     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}"

src/ast/helpers/stmt_create_table.rs

@@ -10465,6 +10465,48 @@ impl fmt::Display for CreateUser {
     }
 }
 
+/// Specifies the refresh mode for the dynamic table.
+///
+/// [Snowflake](https://docs.snowflake.com/en/sql-reference/sql/create-dynamic-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 RefreshModeKind {
+    Auto,
+    Full,
+    Incremental,
+}
+
+impl fmt::Display for RefreshModeKind {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        match self {
+            RefreshModeKind::Auto => write!(f, "AUTO"),
+            RefreshModeKind::Full => write!(f, "FULL"),
+            RefreshModeKind::Incremental => write!(f, "INCREMENTAL"),
+        }
+    }
+}
+
+/// Specifies the behavior of the initial refresh of the dynamic table.
+///
+/// [Snowflake](https://docs.snowflake.com/en/sql-reference/sql/create-dynamic-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 InitializeKind {
+    OnCreate,
+    OnSchedule,
+}
+
+impl fmt::Display for InitializeKind {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        match self {
+            InitializeKind::OnCreate => write!(f, "ON_CREATE"),
+            InitializeKind::OnSchedule => write!(f, "ON_SCHEDULE"),
+        }
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use crate::tokenizer::Location;

src/ast/mod.rs

@@ -579,6 +579,7 @@ impl Spanned for CreateTable {
             temporary: _,     // bool
             external: _,      // bool
             global: _,        // bool
+            dynamic: _,       // bool
             if_not_exists: _, // bool
             transient: _,     // bool
             volatile: _,      // bool
@@ -619,6 +620,12 @@ impl Spanned for CreateTable {
             catalog_sync: _,                    // todo, Snowflake specific
             storage_serialization_policy: _,
             table_options,
+            target_lag: _,
+            warehouse: _,
+            version: _,
+            refresh_mode: _,
+            initialize: _,
+            require_user: _,
         } = self;
 
         union_spans(

src/ast/spans.rs

@@ -235,6 +235,8 @@ impl Dialect for SnowflakeDialect {
                 _ => None,
             };
 
+            let dynamic = parser.parse_keyword(Keyword::DYNAMIC);
+
             let mut temporary = false;
             let mut volatile = false;
             let mut transient = false;
@@ -259,7 +261,7 @@ impl Dialect for SnowflakeDialect {
                 return Some(parse_create_stage(or_replace, temporary, parser));
             } else if parser.parse_keyword(Keyword::TABLE) {
                 return Some(parse_create_table(
-                    or_replace, global, temporary, volatile, transient, iceberg, parser,
+                    or_replace, global, temporary, volatile, transient, iceberg, dynamic, parser,
                 ));
             } else if parser.parse_keyword(Keyword::DATABASE) {
                 return Some(parse_create_database(or_replace, transient, parser));
@@ -614,13 +616,15 @@ fn parse_alter_session(parser: &mut Parser, set: bool) -> Result<Statement, Pars
 /// Parse snowflake create table statement.
 /// <https://docs.snowflake.com/en/sql-reference/sql/create-table>
 /// <https://docs.snowflake.com/en/sql-reference/sql/create-iceberg-table>
+#[allow(clippy::too_many_arguments)]
 pub fn parse_create_table(
     or_replace: bool,
     global: Option<bool>,
     temporary: bool,
     volatile: bool,
     transient: bool,
     iceberg: bool,
+    dynamic: bool,
     parser: &mut Parser,
 ) -> Result<Statement, ParserError> {
     let if_not_exists = parser.parse_keywords(&[Keyword::IF, Keyword::NOT, Keyword::EXISTS]);
@@ -634,6 +638,7 @@ pub fn parse_create_table(
         .volatile(volatile)
         .iceberg(iceberg)
         .global(global)
+        .dynamic(dynamic)
         .hive_formats(Some(Default::default()));
 
     // Snowflake does not enforce order of the parameters in the statement. The parser needs to
@@ -767,6 +772,49 @@ pub fn parse_create_table(
                 Keyword::IF if parser.parse_keywords(&[Keyword::NOT, Keyword::EXISTS]) => {
                     builder = builder.if_not_exists(true);
                 }
+                Keyword::TARGET_LAG => {
+                    parser.expect_token(&Token::Eq)?;
+                    let target_lag = parser.parse_literal_string()?;
+                    builder = builder.target_lag(Some(target_lag));
+                }
+                Keyword::WAREHOUSE => {
+                    parser.expect_token(&Token::Eq)?;
+                    let warehouse = parser.parse_identifier()?;
+                    builder = builder.warehouse(Some(warehouse));
+                }
+                Keyword::AT | Keyword::BEFORE => {
+                    parser.prev_token();
+                    let version = parser.maybe_parse_table_version()?;
+                    builder = builder.version(version);
+                }
+                Keyword::REFRESH_MODE => {
+                    parser.expect_token(&Token::Eq)?;
+                    let refresh_mode = match parser.parse_one_of_keywords(&[
+                        Keyword::AUTO,
+                        Keyword::FULL,
+                        Keyword::INCREMENTAL,
+                    ]) {
+                        Some(Keyword::AUTO) => Some(RefreshModeKind::Auto),
+                        Some(Keyword::FULL) => Some(RefreshModeKind::Full),
+                        Some(Keyword::INCREMENTAL) => Some(RefreshModeKind::Incremental),
+                        _ => return parser.expected("AUTO, FULL or INCREMENTAL", next_token),
+                    };
+                    builder = builder.refresh_mode(refresh_mode);
+                }
+                Keyword::INITIALIZE => {
+                    parser.expect_token(&Token::Eq)?;
+                    let initialize = match parser
+                        .parse_one_of_keywords(&[Keyword::ON_CREATE, Keyword::ON_SCHEDULE])
+                    {
+                        Some(Keyword::ON_CREATE) => Some(InitializeKind::OnCreate),
+                        Some(Keyword::ON_SCHEDULE) => Some(InitializeKind::OnSchedule),
+                        _ => return parser.expected("ON_CREATE or ON_SCHEDULE", next_token),
+                    };
+                    builder = builder.initialize(initialize);
+                }
+                Keyword::REQUIRE if parser.parse_keyword(Keyword::USER) => {
+                    builder = builder.require_user(true);
+                }
                 _ => {
                     return parser.expected("end of statement", next_token);
                 }
@@ -777,21 +825,9 @@ pub fn parse_create_table(
                 builder = builder.columns(columns).constraints(constraints);
             }
             Token::EOF => {
-                if !builder.validate_schema_info() {
-                    return Err(ParserError::ParserError(
-                        "unexpected end of input".to_string(),
-                    ));
-                }
-
                 break;
             }
             Token::SemiColon => {
-                if !builder.validate_schema_info() {
-                    return Err(ParserError::ParserError(
-                        "unexpected end of input".to_string(),
-                    ));
-                }
-
                 parser.prev_token();
                 break;
             }