fix(snowflake): accept column list after CREATE DYNAMIC TABLE options

src/parser/mod.rs

@@ -8725,7 +8725,7 @@ impl<'a> Parser<'a> {
         };
 
         // parse optional column list (schema)
-        let (mut columns, constraints, projections) = self.parse_columns()?;
+        let (mut columns, mut constraints, mut projections) = self.parse_columns()?;
 
         // PostgreSQL: partition bound for `PARTITION OF parent`
         let partition_bound = if partition_of.is_some() {
@@ -9147,6 +9147,18 @@ impl<'a> Parser<'a> {
                         }
                     }
                     // Value was a paren group; skip preserving since we don't have AST for it.
+                } else if matches!(self.peek_token().token, Token::Word(_)) {
+                    // Bare-word option value (TARGET_LAG = DOWNSTREAM,
+                    // REFRESH_MODE = AUTO, INITIALIZE = ON_CREATE, WAREHOUSE = wh).
+                    // Parse just the identifier — NOT a full expression — so a
+                    // trailing column-definition-list `(` (Snowflake allows
+                    // the column list *after* the options header) is not mistaken
+                    // for a function call on the value word.
+                    let ident = self.parse_identifier(false)?;
+                    table_options.push(SqlOption {
+                        name: ObjectName(vec![Ident::new(key)]),
+                        value: Expr::Identifier(ident),
+                    });
                 } else if let Ok(value) = self.parse_expr() {
                     table_options.push(SqlOption {
                         name: ObjectName(vec![Ident::new(key)]),
@@ -9184,6 +9196,20 @@ impl<'a> Parser<'a> {
                 continue;
             }
 
+            // Snowflake dynamic / iceberg tables: the column-definition list may
+            // appear *after* the options header rather than before it, e.g.
+            // `TARGET_LAG = ... WAREHOUSE = ... ( "col" TYPE, ... ) AS SELECT ...`.
+            // The up-front parse_columns() saw the first option keyword (not `(`),
+            // so `columns` is still empty here. Pick the list up now.
+            // https://docs.snowflake.com/en/sql-reference/sql/create-dynamic-table
+            if columns.is_empty() && constraints.is_empty() && self.peek_token_is(&Token::LParen) {
+                let (cols, cons, projs) = self.parse_columns()?;
+                columns = cols;
+                constraints = cons;
+                projections = projs;
+                continue;
+            }
+
             break;
         }
 

tests/sqlparser_snowflake.rs

@@ -3463,6 +3463,82 @@ fn test_snowflake_create_dynamic_table_full_options() {
     }
 }
 
+#[test]
+fn test_snowflake_create_dynamic_table_column_list_after_options() {
+    // Snowflake allows the column-definition list *after* the options header
+    // (TARGET_LAG / WAREHOUSE / REFRESH_MODE / INITIALIZE), then AS SELECT. The
+    // column list must be captured even though the up-front column parse saw
+    // the first option keyword instead of `(`.
+    // https://docs.snowflake.com/en/sql-reference/sql/create-dynamic-table
+    let sql = "CREATE OR REPLACE DYNAMIC TABLE db.sch.t \
+        TARGET_LAG = '100000 Days' WAREHOUSE = COMPUTE_WH \
+        REFRESH_MODE = AUTO INITIALIZE = ON_CREATE \
+        ( \
+            \"SYSTEM_CREATE_DATE\" TIMESTAMP_NTZ(9), \
+            \"TS_TZ\" TIMESTAMP_TZ(9), \
+            \"ID\" VARCHAR(256), \
+            \"AMOUNT\" NUMBER(38,2), \
+            \"DT\" DATE, \
+            \"IS_OK\" BOOLEAN \
+        ) \
+        AS SELECT a, b, c, d, e, f FROM src";
+    match snowflake().parse_sql_statements(sql).unwrap().remove(0) {
+        Statement::CreateTable {
+            dynamic,
+            columns,
+            query,
+            table_options,
+            ..
+        } => {
+            assert!(dynamic);
+            assert!(query.is_some());
+            let names: Vec<String> = columns.iter().map(|c| c.name.to_string()).collect();
+            assert_eq!(
+                names,
+                vec![
+                    "\"SYSTEM_CREATE_DATE\"",
+                    "\"TS_TZ\"",
+                    "\"ID\"",
+                    "\"AMOUNT\"",
+                    "\"DT\"",
+                    "\"IS_OK\"",
+                ]
+            );
+            let keys: Vec<String> = table_options
+                .iter()
+                .map(|o| o.name.to_string().to_uppercase())
+                .collect();
+            for k in ["TARGET_LAG", "WAREHOUSE", "REFRESH_MODE", "INITIALIZE"] {
+                assert!(keys.iter().any(|x| x == k), "missing {k} in {keys:?}");
+            }
+        }
+        other => panic!("expected CreateTable, got {other:?}"),
+    }
+
+    // Column-list-only variant (no AS SELECT). The tokenizer abuts the closing
+    // quote against the type (`"ID"VARCHAR`), which must still parse.
+    let sql_no_as = "CREATE OR REPLACE DYNAMIC TABLE t \
+        TARGET_LAG = '1 minute' WAREHOUSE = wh \
+        (\"ID\"VARCHAR(256), \"AMOUNT\"NUMBER(38,2), \"IS_OK\"BOOLEAN)";
+    match snowflake()
+        .parse_sql_statements(sql_no_as)
+        .unwrap()
+        .remove(0)
+    {
+        Statement::CreateTable {
+            dynamic,
+            columns,
+            query,
+            ..
+        } => {
+            assert!(dynamic);
+            assert!(query.is_none());
+            assert_eq!(columns.len(), 3);
+        }
+        other => panic!("expected CreateTable, got {other:?}"),
+    }
+}
+
 #[test]
 fn test_snowflake_create_hybrid_table() {
     let sql = "CREATE HYBRID TABLE prefs (customer_id INT NOT NULL, pref_key VARCHAR(64) NOT NULL, PRIMARY KEY (customer_id, pref_key))";