Add support for Create Iceberg Table statement for Snowflake parser (apache#1664)

Author: Vedin

src/ast/dml.rs

@@ -117,6 +117,7 @@ pub struct CreateTable {
     pub if_not_exists: bool,
     pub transient: bool,
     pub volatile: bool,
+    pub iceberg: bool,
     /// Table name
     #[cfg_attr(feature = "visitor", visit(with = "visit_relation"))]
     pub name: ObjectName,
@@ -192,6 +193,21 @@ pub struct CreateTable {
     /// Snowflake "WITH TAG" clause
     /// <https://docs.snowflake.com/en/sql-reference/sql/create-table>
     pub with_tags: Option<Vec<Tag>>,
+    /// Snowflake "EXTERNAL_VOLUME" clause for Iceberg tables
+    /// <https://docs.snowflake.com/en/sql-reference/sql/create-iceberg-table>
+    pub external_volume: Option<String>,
+    /// Snowflake "BASE_LOCATION" clause for Iceberg tables
+    /// <https://docs.snowflake.com/en/sql-reference/sql/create-iceberg-table>
+    pub base_location: Option<String>,
+    /// Snowflake "CATALOG" clause for Iceberg tables
+    /// <https://docs.snowflake.com/en/sql-reference/sql/create-iceberg-table>
+    pub catalog: Option<String>,
+    /// Snowflake "CATALOG_SYNC" clause for Iceberg tables
+    /// <https://docs.snowflake.com/en/sql-reference/sql/create-iceberg-table>
+    pub catalog_sync: Option<String>,
+    /// 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>,
 }
 
 impl Display for CreateTable {
@@ -205,7 +221,7 @@ impl Display for CreateTable {
         //   `CREATE TABLE t (a INT) AS SELECT a from t2`
         write!(
             f,
-            "CREATE {or_replace}{external}{global}{temporary}{transient}{volatile}TABLE {if_not_exists}{name}",
+            "CREATE {or_replace}{external}{global}{temporary}{transient}{volatile}{iceberg}TABLE {if_not_exists}{name}",
             or_replace = if self.or_replace { "OR REPLACE " } else { "" },
             external = if self.external { "EXTERNAL " } else { "" },
             global = self.global
@@ -221,6 +237,8 @@ impl Display for CreateTable {
             temporary = if self.temporary { "TEMPORARY " } else { "" },
             transient = if self.transient { "TRANSIENT " } else { "" },
             volatile = if self.volatile { "VOLATILE " } else { "" },
+            // Only for Snowflake
+            iceberg = if self.iceberg { "ICEBERG " } else { "" },
             name = self.name,
         )?;
         if let Some(on_cluster) = &self.on_cluster {
@@ -382,6 +400,31 @@ impl Display for CreateTable {
             )?;
         }
 
+        if let Some(external_volume) = self.external_volume.as_ref() {
+            write!(f, " EXTERNAL_VOLUME = '{external_volume}'")?;
+        }
+
+        if let Some(catalog) = self.catalog.as_ref() {
+            write!(f, " CATALOG = '{catalog}'")?;
+        }
+
+        if self.iceberg {
+            if let Some(base_location) = self.base_location.as_ref() {
+                write!(f, " BASE_LOCATION = '{base_location}'")?;
+            }
+        }
+
+        if let Some(catalog_sync) = self.catalog_sync.as_ref() {
+            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}"
+            )?;
+        }
+
         if self.copy_grants {
             write!(f, " COPY GRANTS")?;
         }

src/ast/helpers/stmt_create_table.rs

@@ -28,7 +28,7 @@ use super::super::dml::CreateTable;
 use crate::ast::{
     ClusteredBy, ColumnDef, CommentDef, Expr, FileFormat, HiveDistributionStyle, HiveFormat, Ident,
     ObjectName, OnCommit, OneOrManyWithParens, Query, RowAccessPolicy, SqlOption, Statement,
-    TableConstraint, TableEngine, Tag, WrappedCollection,
+    StorageSerializationPolicy, TableConstraint, TableEngine, Tag, WrappedCollection,
 };
 use crate::parser::ParserError;
 
@@ -71,6 +71,7 @@ pub struct CreateTableBuilder {
     pub if_not_exists: bool,
     pub transient: bool,
     pub volatile: bool,
+    pub iceberg: bool,
     pub name: ObjectName,
     pub columns: Vec<ColumnDef>,
     pub constraints: Vec<TableConstraint>,
@@ -107,6 +108,11 @@ pub struct CreateTableBuilder {
     pub with_aggregation_policy: Option<ObjectName>,
     pub with_row_access_policy: Option<RowAccessPolicy>,
     pub with_tags: Option<Vec<Tag>>,
+    pub base_location: Option<String>,
+    pub external_volume: Option<String>,
+    pub catalog: Option<String>,
+    pub catalog_sync: Option<String>,
+    pub storage_serialization_policy: Option<StorageSerializationPolicy>,
 }
 
 impl CreateTableBuilder {
@@ -119,6 +125,7 @@ impl CreateTableBuilder {
             if_not_exists: false,
             transient: false,
             volatile: false,
+            iceberg: false,
             name,
             columns: vec![],
             constraints: vec![],
@@ -155,6 +162,11 @@ impl CreateTableBuilder {
             with_aggregation_policy: None,
             with_row_access_policy: None,
             with_tags: None,
+            base_location: None,
+            external_volume: None,
+            catalog: None,
+            catalog_sync: None,
+            storage_serialization_policy: None,
         }
     }
     pub fn or_replace(mut self, or_replace: bool) -> Self {
@@ -192,6 +204,11 @@ impl CreateTableBuilder {
         self
     }
 
+    pub fn iceberg(mut self, iceberg: bool) -> Self {
+        self.iceberg = iceberg;
+        self
+    }
+
     pub fn columns(mut self, columns: Vec<ColumnDef>) -> Self {
         self.columns = columns;
         self
@@ -371,6 +388,34 @@ impl CreateTableBuilder {
         self
     }
 
+    pub fn base_location(mut self, base_location: Option<String>) -> Self {
+        self.base_location = base_location;
+        self
+    }
+
+    pub fn external_volume(mut self, external_volume: Option<String>) -> Self {
+        self.external_volume = external_volume;
+        self
+    }
+
+    pub fn catalog(mut self, catalog: Option<String>) -> Self {
+        self.catalog = catalog;
+        self
+    }
+
+    pub fn catalog_sync(mut self, catalog_sync: Option<String>) -> Self {
+        self.catalog_sync = catalog_sync;
+        self
+    }
+
+    pub fn storage_serialization_policy(
+        mut self,
+        storage_serialization_policy: Option<StorageSerializationPolicy>,
+    ) -> Self {
+        self.storage_serialization_policy = storage_serialization_policy;
+        self
+    }
+
     pub fn build(self) -> Statement {
         Statement::CreateTable(CreateTable {
             or_replace: self.or_replace,
@@ -380,6 +425,7 @@ impl CreateTableBuilder {
             if_not_exists: self.if_not_exists,
             transient: self.transient,
             volatile: self.volatile,
+            iceberg: self.iceberg,
             name: self.name,
             columns: self.columns,
             constraints: self.constraints,
@@ -416,6 +462,11 @@ impl CreateTableBuilder {
             with_aggregation_policy: self.with_aggregation_policy,
             with_row_access_policy: self.with_row_access_policy,
             with_tags: self.with_tags,
+            base_location: self.base_location,
+            external_volume: self.external_volume,
+            catalog: self.catalog,
+            catalog_sync: self.catalog_sync,
+            storage_serialization_policy: self.storage_serialization_policy,
         })
     }
 }
@@ -435,6 +486,7 @@ impl TryFrom<Statement> for CreateTableBuilder {
                 if_not_exists,
                 transient,
                 volatile,
+                iceberg,
                 name,
                 columns,
                 constraints,
@@ -471,6 +523,11 @@ impl TryFrom<Statement> for CreateTableBuilder {
                 with_aggregation_policy,
                 with_row_access_policy,
                 with_tags,
+                base_location,
+                external_volume,
+                catalog,
+                catalog_sync,
+                storage_serialization_policy,
             }) => Ok(Self {
                 or_replace,
                 temporary,
@@ -505,6 +562,7 @@ impl TryFrom<Statement> for CreateTableBuilder {
                 clustered_by,
                 options,
                 strict,
+                iceberg,
                 copy_grants,
                 enable_schema_evolution,
                 change_tracking,
@@ -515,6 +573,11 @@ impl TryFrom<Statement> for CreateTableBuilder {
                 with_row_access_policy,
                 with_tags,
                 volatile,
+                base_location,
+                external_volume,
+                catalog,
+                catalog_sync,
+                storage_serialization_policy,
             }),
             _ => Err(ParserError::ParserError(format!(
                 "Expected create table statement, but received: {stmt}"

src/ast/spans.rs

@@ -532,6 +532,7 @@ impl Spanned for CreateTable {
             if_not_exists: _, // bool
             transient: _,     // bool
             volatile: _,      // bool
+            iceberg: _,       // bool, Snowflake specific
             name,
             columns,
             constraints,
@@ -568,6 +569,11 @@ impl Spanned for CreateTable {
             with_aggregation_policy: _,         // todo, Snowflake specific
             with_row_access_policy: _,          // todo, Snowflake specific
             with_tags: _,                       // todo, Snowflake specific
+            external_volume: _,                 // todo, Snowflake specific
+            base_location: _,                   // todo, Snowflake specific
+            catalog: _,                         // todo, Snowflake specific
+            catalog_sync: _,                    // todo, Snowflake specific
+            storage_serialization_policy: _,    // todo, Snowflake specific
         } = self;
 
         union_spans(

src/dialect/snowflake.rs

@@ -37,6 +37,7 @@ use alloc::string::String;
 use alloc::vec::Vec;
 #[cfg(not(feature = "std"))]
 use alloc::{format, vec};
+use sqlparser::ast::StorageSerializationPolicy;
 
 use super::keywords::RESERVED_FOR_IDENTIFIER;
 
@@ -130,16 +131,19 @@ impl Dialect for SnowflakeDialect {
             let mut temporary = false;
             let mut volatile = false;
             let mut transient = false;
+            let mut iceberg = false;
 
             match parser.parse_one_of_keywords(&[
                 Keyword::TEMP,
                 Keyword::TEMPORARY,
                 Keyword::VOLATILE,
                 Keyword::TRANSIENT,
+                Keyword::ICEBERG,
             ]) {
                 Some(Keyword::TEMP | Keyword::TEMPORARY) => temporary = true,
                 Some(Keyword::VOLATILE) => volatile = true,
                 Some(Keyword::TRANSIENT) => transient = true,
+                Some(Keyword::ICEBERG) => iceberg = true,
                 _ => {}
             }
 
@@ -148,7 +152,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, parser,
+                    or_replace, global, temporary, volatile, transient, iceberg, parser,
                 ));
             } else {
                 // need to go back with the cursor
@@ -255,7 +259,7 @@ impl Dialect for SnowflakeDialect {
     fn is_select_item_alias(&self, explicit: bool, kw: &Keyword, parser: &mut Parser) -> bool {
         explicit
             || match kw {
-            // The following keywords can be considered an alias as long as 
+            // The following keywords can be considered an alias as long as
             // they are not followed by other tokens that may change their meaning
             // e.g. `SELECT * EXCEPT (col1) FROM tbl`
             Keyword::EXCEPT
@@ -277,8 +281,8 @@ impl Dialect for SnowflakeDialect {
                 false
             }
 
-            // Reserved keywords by the Snowflake dialect, which seem to be less strictive 
-            // than what is listed in `keywords::RESERVED_FOR_COLUMN_ALIAS`. The following 
+            // Reserved keywords by the Snowflake dialect, which seem to be less strictive
+            // than what is listed in `keywords::RESERVED_FOR_COLUMN_ALIAS`. The following
             // keywords were tested with the this statement: `SELECT 1 <KW>`.
             Keyword::FROM
             | Keyword::GROUP
@@ -325,12 +329,14 @@ fn parse_file_staging_command(kw: Keyword, parser: &mut Parser) -> Result<Statem
 
 /// 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>
 pub fn parse_create_table(
     or_replace: bool,
     global: Option<bool>,
     temporary: bool,
     volatile: bool,
     transient: bool,
+    iceberg: bool,
     parser: &mut Parser,
 ) -> Result<Statement, ParserError> {
     let if_not_exists = parser.parse_keywords(&[Keyword::IF, Keyword::NOT, Keyword::EXISTS]);
@@ -342,6 +348,7 @@ pub fn parse_create_table(
         .temporary(temporary)
         .transient(transient)
         .volatile(volatile)
+        .iceberg(iceberg)
         .global(global)
         .hive_formats(Some(Default::default()));
 
@@ -468,6 +475,28 @@ pub fn parse_create_table(
                     let on_commit = Some(parser.parse_create_table_on_commit()?);
                     builder = builder.on_commit(on_commit);
                 }
+                Keyword::EXTERNAL_VOLUME => {
+                    parser.expect_token(&Token::Eq)?;
+                    builder.external_volume = Some(parser.parse_literal_string()?);
+                }
+                Keyword::CATALOG => {
+                    parser.expect_token(&Token::Eq)?;
+                    builder.catalog = Some(parser.parse_literal_string()?);
+                }
+                Keyword::BASE_LOCATION => {
+                    parser.expect_token(&Token::Eq)?;
+                    builder.base_location = Some(parser.parse_literal_string()?);
+                }
+                Keyword::CATALOG_SYNC => {
+                    parser.expect_token(&Token::Eq)?;
+                    builder.catalog_sync = Some(parser.parse_literal_string()?);
+                }
+                Keyword::STORAGE_SERIALIZATION_POLICY => {
+                    parser.expect_token(&Token::Eq)?;
+
+                    builder.storage_serialization_policy =
+                        Some(parse_storage_serialization_policy(parser)?);
+                }
                 _ => {
                     return parser.expected("end of statement", next_token);
                 }
@@ -502,9 +531,29 @@ pub fn parse_create_table(
         }
     }
 
+    if iceberg && builder.base_location.is_none() {
+        return Err(ParserError::ParserError(
+            "BASE_LOCATION is required for ICEBERG tables".to_string(),
+        ));
+    }
+
     Ok(builder.build())
 }
 
+pub fn parse_storage_serialization_policy(
+    parser: &mut Parser,
+) -> Result<StorageSerializationPolicy, ParserError> {
+    let next_token = parser.next_token();
+    match &next_token.token {
+        Token::Word(w) => match w.keyword {
+            Keyword::COMPATIBLE => Ok(StorageSerializationPolicy::Compatible),
+            Keyword::OPTIMIZED => Ok(StorageSerializationPolicy::Optimized),
+            _ => parser.expected("storage_serialization_policy", next_token),
+        },
+        _ => parser.expected("storage_serialization_policy", next_token),
+    }
+}
+
 pub fn parse_create_stage(
     or_replace: bool,
     temporary: bool,