Snowflake: parse CREATE / DROP ICEBERG TABLE

Add CATALOG_TABLE_NAME and AUTO_REFRESH to CreateTable for
externally-managed Iceberg tables, accept comma- or space-separated
Iceberg options in free order, relax the BASE_LOCATION requirement for
externally-managed tables, and parse DROP ICEBERG TABLE [IF EXISTS]
<name> [PURGE].

Author: hovaesco

src/ast/ddl.rs

@@ -3073,6 +3073,12 @@ pub struct CreateTable {
     /// Snowflake "CATALOG" clause for Iceberg tables
     /// <https://docs.snowflake.com/en/sql-reference/sql/create-iceberg-table>
     pub catalog: Option<String>,
+    /// Snowflake "CATALOG_TABLE_NAME" clause for externally-managed Iceberg tables
+    /// <https://docs.snowflake.com/en/sql-reference/sql/create-iceberg-table>
+    pub catalog_table_name: Option<String>,
+    /// Snowflake "AUTO_REFRESH" clause for externally-managed Iceberg tables
+    /// <https://docs.snowflake.com/en/sql-reference/sql/create-iceberg-table>
+    pub auto_refresh: Option<bool>,
     /// Snowflake "CATALOG_SYNC" clause for Iceberg tables
     /// <https://docs.snowflake.com/en/sql-reference/sql/create-iceberg-table>
     pub catalog_sync: Option<String>,
@@ -3162,8 +3168,10 @@ impl fmt::Display for CreateTable {
             && self.like.is_none()
             && self.clone.is_none()
             && self.partition_of.is_none()
+            && !self.iceberg
         {
-            // PostgreSQL allows `CREATE TABLE t ();`, but requires empty parens
+            // PostgreSQL allows `CREATE TABLE t ();`, but requires empty parens.
+            // Externally-managed Iceberg tables legitimately have no column list.
             f.write_str(" ()")?;
         } else if let Some(CreateTableLikeKind::Parenthesized(like_in_columns_list)) = &self.like {
             write!(f, " ({like_in_columns_list})")?;
@@ -3307,6 +3315,18 @@ impl fmt::Display for CreateTable {
             write!(f, " CATALOG='{catalog}'")?;
         }
 
+        if let Some(catalog_table_name) = self.catalog_table_name.as_ref() {
+            write!(f, " CATALOG_TABLE_NAME='{catalog_table_name}'")?;
+        }
+
+        if let Some(auto_refresh) = self.auto_refresh {
+            write!(
+                f,
+                " AUTO_REFRESH={}",
+                if auto_refresh { "TRUE" } else { "FALSE" }
+            )?;
+        }
+
         if self.iceberg {
             if let Some(base_location) = self.base_location.as_ref() {
                 write!(f, " BASE_LOCATION='{base_location}'")?;

src/ast/helpers/stmt_create_table.rs

@@ -159,6 +159,10 @@ pub struct CreateTableBuilder {
     pub external_volume: Option<String>,
     /// Optional catalog name.
     pub catalog: Option<String>,
+    /// Optional externally-managed catalog table name.
+    pub catalog_table_name: Option<String>,
+    /// Optional auto-refresh flag for externally-managed Iceberg tables.
+    pub auto_refresh: Option<bool>,
     /// Optional catalog synchronization option.
     pub catalog_sync: Option<String>,
     /// Optional storage serialization policy.
@@ -236,6 +240,8 @@ impl CreateTableBuilder {
             base_location: None,
             external_volume: None,
             catalog: None,
+            catalog_table_name: None,
+            auto_refresh: None,
             catalog_sync: None,
             storage_serialization_policy: None,
             table_options: CreateTableOptions::None,
@@ -606,6 +612,8 @@ impl CreateTableBuilder {
             base_location: self.base_location,
             external_volume: self.external_volume,
             catalog: self.catalog,
+            catalog_table_name: self.catalog_table_name,
+            auto_refresh: self.auto_refresh,
             catalog_sync: self.catalog_sync,
             storage_serialization_policy: self.storage_serialization_policy,
             table_options: self.table_options,
@@ -687,6 +695,8 @@ impl From<CreateTable> for CreateTableBuilder {
             base_location: table.base_location,
             external_volume: table.external_volume,
             catalog: table.catalog,
+            catalog_table_name: table.catalog_table_name,
+            auto_refresh: table.auto_refresh,
             catalog_sync: table.catalog_sync,
             storage_serialization_policy: table.storage_serialization_policy,
             table_options: table.table_options,

src/ast/spans.rs

@@ -638,6 +638,8 @@ impl Spanned for CreateTable {
             external_volume: _,                 // todo, Snowflake specific
             base_location: _,                   // todo, Snowflake specific
             catalog: _,                         // todo, Snowflake specific
+            catalog_table_name: _,              // todo, Snowflake specific
+            auto_refresh: _,                    // todo, Snowflake specific
             catalog_sync: _,                    // todo, Snowflake specific
             storage_serialization_policy: _,
             table_options,

src/dialect/snowflake.rs

@@ -1109,6 +1109,14 @@ pub fn parse_create_table(
                     parser.expect_token(&Token::Eq)?;
                     builder.catalog = Some(parser.parse_literal_string()?);
                 }
+                Keyword::CATALOG_TABLE_NAME => {
+                    parser.expect_token(&Token::Eq)?;
+                    builder.catalog_table_name = Some(parser.parse_literal_string()?);
+                }
+                Keyword::AUTO_REFRESH => {
+                    parser.expect_token(&Token::Eq)?;
+                    builder.auto_refresh = Some(parser.parse_boolean_string()?);
+                }
                 Keyword::BASE_LOCATION => {
                     parser.expect_token(&Token::Eq)?;
                     builder.base_location = Some(parser.parse_literal_string()?);
@@ -1178,6 +1186,8 @@ pub fn parse_create_table(
                 let (columns, constraints) = parser.parse_columns()?;
                 builder = builder.columns(columns).constraints(constraints);
             }
+            // Snowflake accepts Iceberg table options either space- or comma-separated.
+            Token::Comma => {}
             Token::EOF => {
                 break;
             }
@@ -1198,7 +1208,9 @@ pub fn parse_create_table(
 
     builder = builder.table_options(table_options);
 
-    if iceberg && builder.base_location.is_none() {
+    // Managed Iceberg tables require BASE_LOCATION; externally-managed ones
+    // (identified by CATALOG_TABLE_NAME) do not.
+    if iceberg && builder.base_location.is_none() && builder.catalog_table_name.is_none() {
         return Err(ParserError::ParserError(
             "BASE_LOCATION is required for ICEBERG tables".to_string(),
         ));

src/keywords.rs

@@ -147,6 +147,7 @@ define_keywords!(
     AUTOEXTEND_SIZE,
     AUTOINCREMENT,
     AUTO_INCREMENT,
+    AUTO_REFRESH,
     AVG,
     AVG_ROW_LENGTH,
     AVRO,
@@ -213,6 +214,7 @@ define_keywords!(
     CATALOG_SYNC,
     CATALOG_SYNC_NAMESPACE_FLATTEN_DELIMITER,
     CATALOG_SYNC_NAMESPACE_MODE,
+    CATALOG_TABLE_NAME,
     CATALOG_URI,
     CATCH,
     CATEGORY,

src/parser/mod.rs

@@ -7788,7 +7788,9 @@ impl<'a> Parser<'a> {
         let persistent = dialect_of!(self is DuckDbDialect)
             && self.parse_one_of_keywords(&[Keyword::PERSISTENT]).is_some();
 
-        let object_type = if self.parse_keyword(Keyword::TABLE) {
+        let object_type = if self.parse_keyword(Keyword::TABLE)
+            || self.parse_keywords(&[Keyword::ICEBERG, Keyword::TABLE])
+        {
             ObjectType::Table
         } else if self.parse_keyword(Keyword::COLLATION) {
             ObjectType::Collation

tests/sqlparser_duckdb.rs

@@ -781,6 +781,8 @@ fn test_duckdb_union_datatype() {
             base_location: Default::default(),
             external_volume: Default::default(),
             catalog: Default::default(),
+            catalog_table_name: Default::default(),
+            auto_refresh: Default::default(),
             catalog_sync: Default::default(),
             storage_serialization_policy: Default::default(),
             table_options: CreateTableOptions::None,

tests/sqlparser_mssql.rs

@@ -2006,6 +2006,8 @@ fn parse_create_table_with_valid_options() {
                 base_location: None,
                 external_volume: None,
                 catalog: None,
+                catalog_table_name: None,
+                auto_refresh: None,
                 catalog_sync: None,
                 storage_serialization_policy: None,
                 table_options: CreateTableOptions::With(with_options),
@@ -2180,6 +2182,8 @@ fn parse_create_table_with_identity_column() {
                 base_location: None,
                 external_volume: None,
                 catalog: None,
+                catalog_table_name: None,
+                auto_refresh: None,
                 catalog_sync: None,
                 storage_serialization_policy: None,
                 table_options: CreateTableOptions::None,

tests/sqlparser_postgres.rs

@@ -6710,6 +6710,8 @@ fn parse_trigger_related_functions() {
             base_location: None,
             external_volume: None,
             catalog: None,
+            catalog_table_name: None,
+            auto_refresh: None,
             catalog_sync: None,
             storage_serialization_policy: None,
             table_options: CreateTableOptions::None,

tests/sqlparser_snowflake.rs

@@ -1051,6 +1051,120 @@ fn test_snowflake_create_iceberg_table_without_location() {
     );
 }
 
+#[test]
+fn test_snowflake_create_iceberg_table_comma_separated_options() {
+    let canonical = "CREATE ICEBERG TABLE my_table (a INT) \
+        EXTERNAL_VOLUME='volume' CATALOG='SNOWFLAKE' BASE_LOCATION='relative/path'";
+    let with_commas = "CREATE ICEBERG TABLE my_table (a INT) \
+        CATALOG='SNOWFLAKE', EXTERNAL_VOLUME='volume', BASE_LOCATION='relative/path'";
+    match snowflake().one_statement_parses_to(with_commas, canonical) {
+        Statement::CreateTable(CreateTable {
+            iceberg,
+            columns,
+            external_volume,
+            catalog,
+            base_location,
+            ..
+        }) => {
+            assert!(iceberg);
+            assert_eq!(1, columns.len());
+            assert_eq!("volume", external_volume.unwrap());
+            assert_eq!("SNOWFLAKE", catalog.unwrap());
+            assert_eq!("relative/path", base_location.unwrap());
+        }
+        _ => unreachable!(),
+    }
+}
+
+#[test]
+fn test_snowflake_create_iceberg_table_externally_managed() {
+    match snowflake().verified_stmt(
+        "CREATE OR REPLACE ICEBERG TABLE my_table \
+         CATALOG='my_catalog_integration' CATALOG_TABLE_NAME='db.tbl' AUTO_REFRESH=TRUE",
+    ) {
+        Statement::CreateTable(CreateTable {
+            or_replace,
+            iceberg,
+            columns,
+            catalog,
+            catalog_table_name,
+            auto_refresh,
+            base_location,
+            query,
+            ..
+        }) => {
+            assert!(or_replace);
+            assert!(iceberg);
+            assert!(columns.is_empty());
+            assert_eq!("my_catalog_integration", catalog.unwrap());
+            assert_eq!("db.tbl", catalog_table_name.unwrap());
+            assert_eq!(Some(true), auto_refresh);
+            assert!(base_location.is_none());
+            assert!(query.is_none());
+        }
+        _ => unreachable!(),
+    }
+}
+
+#[test]
+fn test_snowflake_create_iceberg_table_ctas() {
+    match snowflake().verified_stmt(
+        "CREATE ICEBERG TABLE my_table EXTERNAL_VOLUME='volume' CATALOG='SNOWFLAKE' \
+         BASE_LOCATION='relative/path' AS SELECT 1",
+    ) {
+        Statement::CreateTable(CreateTable {
+            iceberg,
+            columns,
+            query,
+            base_location,
+            ..
+        }) => {
+            assert!(iceberg);
+            assert!(columns.is_empty());
+            assert!(query.is_some());
+            assert_eq!("relative/path", base_location.unwrap());
+        }
+        _ => unreachable!(),
+    }
+}
+
+#[test]
+fn test_snowflake_drop_iceberg_table() {
+    match snowflake().one_statement_parses_to(
+        "DROP ICEBERG TABLE IF EXISTS my_table PURGE",
+        "DROP TABLE IF EXISTS my_table PURGE",
+    ) {
+        Statement::Drop {
+            object_type,
+            if_exists,
+            names,
+            purge,
+            ..
+        } => {
+            assert_eq!(ObjectType::Table, object_type);
+            assert!(if_exists);
+            assert!(purge);
+            assert_eq!("my_table", names[0].to_string());
+        }
+        _ => unreachable!(),
+    }
+
+    match snowflake().one_statement_parses_to("DROP ICEBERG TABLE my_table", "DROP TABLE my_table")
+    {
+        Statement::Drop {
+            object_type,
+            if_exists,
+            purge,
+            ..
+        } => {
+            assert_eq!(ObjectType::Table, object_type);
+            assert!(!if_exists);
+            assert!(!purge);
+        }
+        _ => unreachable!(),
+    }
+}
+
 #[test]
 fn test_snowflake_create_table_trailing_options() {
     // Serialization to SQL assume that in `CREATE TABLE AS` the options come before the `AS (<query>)`