Snowflake: support for extended column options in CREATE TABLE (#1454)

src/ast/ddl.rs

@@ -31,7 +31,7 @@ use sqlparser_derive::{Visit, VisitMut};
 use crate::ast::value::escape_single_quote_string;
 use crate::ast::{
     display_comma_separated, display_separated, DataType, Expr, Ident, MySQLColumnPosition,
-    ObjectName, OrderByExpr, ProjectionSelect, SequenceOptions, SqlOption, Value,
+    ObjectName, OrderByExpr, ProjectionSelect, SequenceOptions, SqlOption, Tag, Value,
 };
 use crate::keywords::Keyword;
 use crate::tokenizer::Token;
@@ -1096,17 +1096,221 @@ impl fmt::Display for ColumnOptionDef {
     }
 }
 
+/// Identity is a column option for defining an identity or autoincrement column in a `CREATE TABLE` statement.
+/// Syntax
+/// ```sql
+/// { IDENTITY | AUTOINCREMENT } [ (seed , increment) | START num INCREMENT num ] [ ORDER | NOORDER ]
+/// ```
+/// [MS SQL Server]: https://learn.microsoft.com/en-us/sql/t-sql/statements/create-table-transact-sql-identity-property
+/// [Snowflake]: https://docs.snowflake.com/en/sql-reference/sql/create-table
+#[derive(Debug, Clone, PartialEq, PartialOrd, Eq, Ord, Hash)]
+#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
+#[cfg_attr(feature = "visitor", derive(Visit, VisitMut))]
+pub enum IdentityPropertyKind {
+    /// An identity property declared via the `AUTOINCREMENT` key word
+    /// Example:
+    /// ```sql
+    ///  AUTOINCREMENT(100, 1) NOORDER
+    ///  AUTOINCREMENT START 100 INCREMENT 1 ORDER
+    /// ```
+    /// [Snowflake]: https://docs.snowflake.com/en/sql-reference/sql/create-table
+    Autoincrement(IdentityProperty),
+    /// An identity property declared via the `IDENTITY` key word
+    /// Example, [MS SQL Server] or [Snowflake]:
+    /// ```sql
+    ///  IDENTITY(100, 1)
+    /// ```
+    /// [Snowflake]
+    /// ```sql
+    ///  IDENTITY(100, 1) ORDER
+    ///  IDENTITY START 100 INCREMENT 1 NOORDER
+    /// ```
+    /// [MS SQL Server]: https://learn.microsoft.com/en-us/sql/t-sql/statements/create-table-transact-sql-identity-property
+    /// [Snowflake]: https://docs.snowflake.com/en/sql-reference/sql/create-table
+    Identity(IdentityProperty),
+}
+
+impl fmt::Display for IdentityPropertyKind {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        let (command, property) = match self {
+            IdentityPropertyKind::Identity(property) => ("IDENTITY", property),
+            IdentityPropertyKind::Autoincrement(property) => ("AUTOINCREMENT", property),
+        };
+        write!(f, "{command}")?;
+        if let Some(parameters) = &property.parameters {
+            write!(f, "{parameters}")?;
+        }
+        if let Some(order) = &property.order {
+            write!(f, "{order}")?;
+        }
+        Ok(())
+    }
+}
+
 #[derive(Debug, Clone, PartialEq, PartialOrd, Eq, Ord, Hash)]
 #[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
 #[cfg_attr(feature = "visitor", derive(Visit, VisitMut))]
 pub struct IdentityProperty {
+    pub parameters: Option<IdentityPropertyFormatKind>,
+    pub order: Option<IdentityPropertyOrder>,
+}
+
+/// A format of parameters of identity column.
+///
+/// It is [Snowflake] specific.
+/// Syntax
+/// ```sql
+/// (seed , increment) | START num INCREMENT num
+/// ```
+/// [MS SQL Server] uses one way of representing these parameters.
+/// Syntax
+/// ```sql
+/// (seed , increment)
+/// ```
+/// [MS SQL Server]: https://learn.microsoft.com/en-us/sql/t-sql/statements/create-table-transact-sql-identity-property
+/// [Snowflake]: https://docs.snowflake.com/en/sql-reference/sql/create-table
+#[derive(Debug, Clone, PartialEq, PartialOrd, Eq, Ord, Hash)]
+#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
+#[cfg_attr(feature = "visitor", derive(Visit, VisitMut))]
+pub enum IdentityPropertyFormatKind {
+    /// A parameters of identity column declared like parameters of function call
+    /// Example:
+    /// ```sql
+    ///  (100, 1)
+    /// ```
+    /// [MS SQL Server]: https://learn.microsoft.com/en-us/sql/t-sql/statements/create-table-transact-sql-identity-property
+    /// [Snowflake]: https://docs.snowflake.com/en/sql-reference/sql/create-table
+    FunctionCall(IdentityParameters),
+    /// A parameters of identity column declared with keywords `START` and `INCREMENT`
+    /// Example:
+    /// ```sql
+    ///  START 100 INCREMENT 1
+    /// ```
+    /// [Snowflake]: https://docs.snowflake.com/en/sql-reference/sql/create-table
+    StartAndIncrement(IdentityParameters),
+}
+
+impl fmt::Display for IdentityPropertyFormatKind {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        match self {
+            IdentityPropertyFormatKind::FunctionCall(parameters) => {
+                write!(f, "({}, {})", parameters.seed, parameters.increment)
+            }
+            IdentityPropertyFormatKind::StartAndIncrement(parameters) => {
+                write!(
+                    f,
+                    " START {} INCREMENT {}",
+                    parameters.seed, parameters.increment
+                )
+            }
+        }
+    }
+}
+#[derive(Debug, Clone, PartialEq, PartialOrd, Eq, Ord, Hash)]
+#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
+#[cfg_attr(feature = "visitor", derive(Visit, VisitMut))]
+pub struct IdentityParameters {
     pub seed: Expr,
     pub increment: Expr,
 }
 
-impl fmt::Display for IdentityProperty {
+/// The identity column option specifies how values are generated for the auto-incremented column, either in increasing or decreasing order.
+/// Syntax
+/// ```sql
+/// ORDER | NOORDER
+/// ```
+/// [Snowflake]: https://docs.snowflake.com/en/sql-reference/sql/create-table
+#[derive(Debug, Clone, PartialEq, PartialOrd, Eq, Ord, Hash)]
+#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
+#[cfg_attr(feature = "visitor", derive(Visit, VisitMut))]
+pub enum IdentityPropertyOrder {
+    Order,
+    NoOrder,
+}
+
+impl fmt::Display for IdentityPropertyOrder {
     fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
-        write!(f, "{}, {}", self.seed, self.increment)
+        match self {
+            IdentityPropertyOrder::Order => write!(f, " ORDER"),
+            IdentityPropertyOrder::NoOrder => write!(f, " NOORDER"),
+        }
+    }
+}
+
+/// Column policy that identify a security policy of access to a column.
+/// Syntax
+/// ```sql
+/// [ WITH ] MASKING POLICY <policy_name> [ USING ( <col_name> , <cond_col1> , ... ) ]
+/// [ WITH ] PROJECTION POLICY <policy_name>
+/// ```
+/// [Snowflake]: https://docs.snowflake.com/en/sql-reference/sql/create-table
+#[derive(Debug, Clone, PartialEq, PartialOrd, Eq, Ord, Hash)]
+#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
+#[cfg_attr(feature = "visitor", derive(Visit, VisitMut))]
+pub enum ColumnPolicy {
+    MaskingPolicy(ColumnPolicyProperty),
+    ProjectionPolicy(ColumnPolicyProperty),
+}
+
+impl fmt::Display for ColumnPolicy {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        let (command, property) = match self {
+            ColumnPolicy::MaskingPolicy(property) => ("MASKING POLICY", property),
+            ColumnPolicy::ProjectionPolicy(property) => ("PROJECTION POLICY", property),
+        };
+        if property.with {
+            write!(f, "WITH ")?;
+        }
+        write!(f, "{command} {}", property.policy_name)?;
+        if let Some(using_columns) = &property.using_columns {
+            write!(f, " USING ({})", display_comma_separated(using_columns))?;
+        }
+        Ok(())
+    }
+}
+
+#[derive(Debug, Clone, PartialEq, PartialOrd, Eq, Ord, Hash)]
+#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
+#[cfg_attr(feature = "visitor", derive(Visit, VisitMut))]
+pub struct ColumnPolicyProperty {
+    /// This flag indicates that the column policy option is declared using the `WITH` prefix.
+    /// Example
+    /// ```sql
+    /// WITH PROJECTION POLICY sample_policy
+    /// ```
+    /// [Snowflake]: https://docs.snowflake.com/en/sql-reference/sql/create-table
+    pub with: bool,
+    pub policy_name: Ident,
+    pub using_columns: Option<Vec<Ident>>,
+}
+
+/// Tags option of column
+/// Syntax
+/// ```sql
+/// [ WITH ] TAG ( <tag_name> = '<tag_value>' [ , <tag_name> = '<tag_value>' , ... ] )
+/// ```
+/// [Snowflake]: https://docs.snowflake.com/en/sql-reference/sql/create-table
+#[derive(Debug, Clone, PartialEq, PartialOrd, Eq, Ord, Hash)]
+#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
+#[cfg_attr(feature = "visitor", derive(Visit, VisitMut))]
+pub struct TagsColumnOption {
+    /// This flag indicates that the tags option is declared using the `WITH` prefix.
+    /// Example:
+    /// ```sql
+    /// WITH TAG (A = 'Tag A')
+    /// ```
+    /// [Snowflake]: https://docs.snowflake.com/en/sql-reference/sql/create-table
+    pub with: bool,
+    pub tags: Vec<Tag>,
+}
+
+impl fmt::Display for TagsColumnOption {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        if self.with {
+            write!(f, "WITH ")?;
+        }
+        write!(f, "TAG ({})", display_comma_separated(&self.tags))?;
+        Ok(())
     }
 }
 
@@ -1180,16 +1384,32 @@ pub enum ColumnOption {
     /// [1]: https://cloud.google.com/bigquery/docs/reference/standard-sql/data-definition-language#view_column_option_list
     /// [2]: https://cloud.google.com/bigquery/docs/reference/standard-sql/data-definition-language#column_option_list
     Options(Vec<SqlOption>),
-    /// MS SQL Server specific: Creates an identity column in a table.
+    /// Creates an identity or an autoincrement column in a table.
     /// Syntax
     /// ```sql
-    /// IDENTITY [ (seed , increment) ]
+    /// { IDENTITY | AUTOINCREMENT } [ (seed , increment) | START num INCREMENT num ] [ ORDER | NOORDER ]
     /// ```
     /// [MS SQL Server]: https://learn.microsoft.com/en-us/sql/t-sql/statements/create-table-transact-sql-identity-property
-    Identity(Option<IdentityProperty>),
+    /// [Snowflake]: https://docs.snowflake.com/en/sql-reference/sql/create-table
+    Identity(IdentityPropertyKind),
     /// SQLite specific: ON CONFLICT option on column definition
     /// <https://www.sqlite.org/lang_conflict.html>
     OnConflict(Keyword),
+    /// Snowflake specific: an option of specifying security masking or projection policy to set on a column.
+    /// Syntax:
+    /// ```sql
+    /// [ WITH ] MASKING POLICY <policy_name> [ USING ( <col_name> , <cond_col1> , ... ) ]
+    /// [ WITH ] PROJECTION POLICY <policy_name>
+    /// ```
+    /// [Snowflake]: https://docs.snowflake.com/en/sql-reference/sql/create-table
+    Policy(ColumnPolicy),
+    /// Snowflake specific: Specifies the tag name and the tag string value.
+    /// Syntax:
+    /// ```sql
+    /// [ WITH ] TAG ( <tag_name> = '<tag_value>' [ , <tag_name> = '<tag_value>' , ... ] )
+    /// ```
+    /// [Snowflake]: https://docs.snowflake.com/en/sql-reference/sql/create-table
+    Tags(TagsColumnOption),
 }
 
 impl fmt::Display for ColumnOption {
@@ -1292,16 +1512,18 @@ impl fmt::Display for ColumnOption {
                 write!(f, "OPTIONS({})", display_comma_separated(options))
             }
             Identity(parameters) => {
-                write!(f, "IDENTITY")?;
-                if let Some(parameters) = parameters {
-                    write!(f, "({parameters})")?;
-                }
-                Ok(())
+                write!(f, "{parameters}")
             }
             OnConflict(keyword) => {
                 write!(f, "ON CONFLICT {:?}", keyword)?;
                 Ok(())
             }
+            Policy(parameters) => {
+                write!(f, "{parameters}")
+            }
+            Tags(tags) => {
+                write!(f, "{tags}")
+            }
         }
     }
 }

src/ast/mod.rs

@@ -40,11 +40,12 @@ pub use self::data_type::{
 pub use self::dcl::{AlterRoleOperation, ResetConfig, RoleOption, SetConfigValue, Use};
 pub use self::ddl::{
     AlterColumnOperation, AlterIndexOperation, AlterPolicyOperation, AlterTableOperation,
-    ClusteredBy, ColumnDef, ColumnOption, ColumnOptionDef, ConstraintCharacteristics, Deduplicate,
-    DeferrableInitial, GeneratedAs, GeneratedExpressionMode, IdentityProperty, IndexOption,
-    IndexType, KeyOrIndexDisplay, Owner, Partition, ProcedureParam, ReferentialAction,
-    TableConstraint, UserDefinedTypeCompositeAttributeDef, UserDefinedTypeRepresentation,
-    ViewColumnDef,
+    ClusteredBy, ColumnDef, ColumnOption, ColumnOptionDef, ColumnPolicy, ColumnPolicyProperty,
+    ConstraintCharacteristics, Deduplicate, DeferrableInitial, GeneratedAs,
+    GeneratedExpressionMode, IdentityParameters, IdentityProperty, IdentityPropertyFormatKind,
+    IdentityPropertyKind, IdentityPropertyOrder, IndexOption, IndexType, KeyOrIndexDisplay, Owner,
+    Partition, ProcedureParam, ReferentialAction, TableConstraint, TagsColumnOption,
+    UserDefinedTypeCompositeAttributeDef, UserDefinedTypeRepresentation, ViewColumnDef,
 };
 pub use self::dml::{CreateIndex, CreateTable, Delete, Insert};
 pub use self::operator::{BinaryOperator, UnaryOperator};

src/dialect/mod.rs

@@ -49,7 +49,7 @@ pub use self::postgresql::PostgreSqlDialect;
 pub use self::redshift::RedshiftSqlDialect;
 pub use self::snowflake::SnowflakeDialect;
 pub use self::sqlite::SQLiteDialect;
-use crate::ast::{Expr, Statement};
+use crate::ast::{ColumnOption, Expr, Statement};
 pub use crate::keywords;
 use crate::keywords::Keyword;
 use crate::parser::{Parser, ParserError};
@@ -478,6 +478,19 @@ pub trait Dialect: Debug + Any {
         None
     }
 
+    /// Dialect-specific column option parser override
+    ///
+    /// This method is called to parse the next column option.
+    ///
+    /// If `None` is returned, falls back to the default behavior.
+    fn parse_column_option(
+        &self,
+        _parser: &mut Parser,
+    ) -> Option<Result<Option<ColumnOption>, ParserError>> {
+        // return None to fall back to the default behavior
+        None
+    }
+
     /// Decide the lexical Precedence of operators.
     ///
     /// Uses (APPROXIMATELY) <https://www.postgresql.org/docs/7.0/operators.htm#AEN2026> as a reference

src/dialect/snowflake.rs

@@ -22,7 +22,11 @@ use crate::ast::helpers::stmt_data_loading::{
     DataLoadingOption, DataLoadingOptionType, DataLoadingOptions, StageLoadSelectItem,
     StageParamsObject,
 };
-use crate::ast::{Ident, ObjectName, RowAccessPolicy, Statement, Tag, WrappedCollection};
+use crate::ast::{
+    ColumnOption, ColumnPolicy, ColumnPolicyProperty, Ident, IdentityParameters, IdentityProperty,
+    IdentityPropertyFormatKind, IdentityPropertyKind, IdentityPropertyOrder, ObjectName,
+    RowAccessPolicy, Statement, TagsColumnOption, WrappedCollection,
+};
 use crate::dialect::{Dialect, Precedence};
 use crate::keywords::Keyword;
 use crate::parser::{Parser, ParserError};
@@ -149,6 +153,36 @@ impl Dialect for SnowflakeDialect {
         None
     }
 
+    fn parse_column_option(
+        &self,
+        parser: &mut Parser,
+    ) -> Option<Result<Option<ColumnOption>, ParserError>> {
+        parser.maybe_parse(|parser| {
+            let with = parser.parse_keyword(Keyword::WITH);
+
+            if parser.parse_keyword(Keyword::IDENTITY) {
+                Ok(parse_identity_property(parser)
+                    .map(|p| Some(ColumnOption::Identity(IdentityPropertyKind::Identity(p)))))
+            } else if parser.parse_keyword(Keyword::AUTOINCREMENT) {
+                Ok(parse_identity_property(parser).map(|p| {
+                    Some(ColumnOption::Identity(IdentityPropertyKind::Autoincrement(
+                        p,
+                    )))
+                }))
+            } else if parser.parse_keywords(&[Keyword::MASKING, Keyword::POLICY]) {
+                Ok(parse_column_policy_property(parser, with)
+                    .map(|p| Some(ColumnOption::Policy(ColumnPolicy::MaskingPolicy(p)))))
+            } else if parser.parse_keywords(&[Keyword::PROJECTION, Keyword::POLICY]) {
+                Ok(parse_column_policy_property(parser, with)
+                    .map(|p| Some(ColumnOption::Policy(ColumnPolicy::ProjectionPolicy(p)))))
+            } else if parser.parse_keywords(&[Keyword::TAG]) {
+                Ok(parse_column_tags(parser, with).map(|p| Some(ColumnOption::Tags(p))))
+            } else {
+                Err(ParserError::ParserError("not found match".to_string()))
+            }
+        })
+    }
+
     fn get_next_precedence(&self, parser: &Parser) -> Option<Result<u8, ParserError>> {
         let token = parser.peek_token();
         // Snowflake supports the `:` cast operator unlike other dialects
@@ -307,16 +341,8 @@ pub fn parse_create_table(
                         builder.with_row_access_policy(Some(RowAccessPolicy::new(policy, columns)))
                 }
                 Keyword::TAG => {
-                    fn parse_tag(parser: &mut Parser) -> Result<Tag, ParserError> {
-                        let name = parser.parse_identifier(false)?;
-                        parser.expect_token(&Token::Eq)?;
-                        let value = parser.parse_literal_string()?;
-
-                        Ok(Tag::new(name, value))
-                    }
-
                     parser.expect_token(&Token::LParen)?;
-                    let tags = parser.parse_comma_separated(parse_tag)?;
+                    let tags = parser.parse_comma_separated(Parser::parse_tag)?;
                     parser.expect_token(&Token::RParen)?;
                     builder = builder.with_tags(Some(tags));
                 }
@@ -776,3 +802,79 @@ fn parse_parentheses_options(parser: &mut Parser) -> Result<Vec<DataLoadingOptio
     }
     Ok(options)
 }
+
+/// Parsing a property of identity or autoincrement column option
+/// Syntax:
+/// ```sql
+/// [ (seed , increment) | START num INCREMENT num ] [ ORDER | NOORDER ]
+/// ```
+/// [Snowflake]: https://docs.snowflake.com/en/sql-reference/sql/create-table
+fn parse_identity_property(parser: &mut Parser) -> Result<IdentityProperty, ParserError> {
+    let parameters = if parser.consume_token(&Token::LParen) {
+        let seed = parser.parse_number()?;
+        parser.expect_token(&Token::Comma)?;
+        let increment = parser.parse_number()?;
+        parser.expect_token(&Token::RParen)?;
+
+        Some(IdentityPropertyFormatKind::FunctionCall(
+            IdentityParameters { seed, increment },
+        ))
+    } else if parser.parse_keyword(Keyword::START) {
+        let seed = parser.parse_number()?;
+        parser.expect_keyword(Keyword::INCREMENT)?;
+        let increment = parser.parse_number()?;
+
+        Some(IdentityPropertyFormatKind::StartAndIncrement(
+            IdentityParameters { seed, increment },
+        ))
+    } else {
+        None
+    };
+    let order = match parser.parse_one_of_keywords(&[Keyword::ORDER, Keyword::NOORDER]) {
+        Some(Keyword::ORDER) => Some(IdentityPropertyOrder::Order),
+        Some(Keyword::NOORDER) => Some(IdentityPropertyOrder::NoOrder),
+        _ => None,
+    };
+    Ok(IdentityProperty { parameters, order })
+}
+
+/// Parsing a policy property of column option
+/// Syntax:
+/// ```sql
+/// <policy_name> [ USING ( <col_name> , <cond_col1> , ... )
+/// ```
+/// [Snowflake]: https://docs.snowflake.com/en/sql-reference/sql/create-table
+fn parse_column_policy_property(
+    parser: &mut Parser,
+    with: bool,
+) -> Result<ColumnPolicyProperty, ParserError> {
+    let policy_name = parser.parse_identifier(false)?;
+    let using_columns = if parser.parse_keyword(Keyword::USING) {
+        parser.expect_token(&Token::LParen)?;
+        let columns = parser.parse_comma_separated(|p| p.parse_identifier(false))?;
+        parser.expect_token(&Token::RParen)?;
+        Some(columns)
+    } else {
+        None
+    };
+
+    Ok(ColumnPolicyProperty {
+        with,
+        policy_name,
+        using_columns,
+    })
+}
+
+/// Parsing tags list of column
+/// Syntax:
+/// ```sql
+/// ( <tag_name> = '<tag_value>' [ , <tag_name> = '<tag_value>' , ... ] )
+/// ```
+/// [Snowflake]: https://docs.snowflake.com/en/sql-reference/sql/create-table
+fn parse_column_tags(parser: &mut Parser, with: bool) -> Result<TagsColumnOption, ParserError> {
+    parser.expect_token(&Token::LParen)?;
+    let tags = parser.parse_comma_separated(Parser::parse_tag)?;
+    parser.expect_token(&Token::RParen)?;
+
+    Ok(TagsColumnOption { with, tags })
+}

src/keywords.rs

@@ -453,6 +453,7 @@ define_keywords!(
     MACRO,
     MANAGEDLOCATION,
     MAP,
+    MASKING,
     MATCH,
     MATCHED,
     MATCHES,
@@ -504,6 +505,7 @@ define_keywords!(
     NOINHERIT,
     NOLOGIN,
     NONE,
+    NOORDER,
     NOREPLICATION,
     NORMALIZE,
     NOSCAN,

src/parser/mod.rs

@@ -6065,7 +6065,7 @@ impl<'a> Parser<'a> {
                 }
             } else if let Some(option) = self.parse_optional_column_option()? {
                 options.push(ColumnOptionDef { name: None, option });
-            } else if dialect_of!(self is MySqlDialect | GenericDialect)
+            } else if dialect_of!(self is MySqlDialect | SnowflakeDialect | GenericDialect)
                 && self.parse_keyword(Keyword::COLLATE)
             {
                 collation = Some(self.parse_object_name(false)?);
@@ -6105,6 +6105,10 @@ impl<'a> Parser<'a> {
     }
 
     pub fn parse_optional_column_option(&mut self) -> Result<Option<ColumnOption>, ParserError> {
+        if let Some(option) = self.dialect.parse_column_option(self) {
+            return option;
+        }
+
         if self.parse_keywords(&[Keyword::CHARACTER, Keyword::SET]) {
             Ok(Some(ColumnOption::CharacterSet(
                 self.parse_object_name(false)?,
@@ -6232,17 +6236,24 @@ impl<'a> Parser<'a> {
         } else if self.parse_keyword(Keyword::IDENTITY)
             && dialect_of!(self is MsSqlDialect | GenericDialect)
         {
-            let property = if self.consume_token(&Token::LParen) {
+            let parameters = if self.consume_token(&Token::LParen) {
                 let seed = self.parse_number()?;
                 self.expect_token(&Token::Comma)?;
                 let increment = self.parse_number()?;
                 self.expect_token(&Token::RParen)?;
 
-                Some(IdentityProperty { seed, increment })
+                Some(IdentityPropertyFormatKind::FunctionCall(
+                    IdentityParameters { seed, increment },
+                ))
             } else {
                 None
             };
-            Ok(Some(ColumnOption::Identity(property)))
+            Ok(Some(ColumnOption::Identity(
+                IdentityPropertyKind::Identity(IdentityProperty {
+                    parameters,
+                    order: None,
+                }),
+            )))
         } else if dialect_of!(self is SQLiteDialect | GenericDialect)
             && self.parse_keywords(&[Keyword::ON, Keyword::CONFLICT])
         {
@@ -6260,6 +6271,15 @@ impl<'a> Parser<'a> {
             Ok(None)
         }
     }
+
+    pub(crate) fn parse_tag(&mut self) -> Result<Tag, ParserError> {
+        let name = self.parse_identifier(false)?;
+        self.expect_token(&Token::Eq)?;
+        let value = self.parse_literal_string()?;
+
+        Ok(Tag::new(name, value))
+    }
+
     fn parse_optional_column_option_generated(
         &mut self,
     ) -> Result<Option<ColumnOption>, ParserError> {

tests/sqlparser_mssql.rs

@@ -921,7 +921,12 @@ fn parse_create_table_with_identity_column() {
             vec![
                 ColumnOptionDef {
                     name: None,
-                    option: ColumnOption::Identity(None),
+                    option: ColumnOption::Identity(IdentityPropertyKind::Identity(
+                        IdentityProperty {
+                            parameters: None,
+                            order: None,
+                        },
+                    )),
                 },
                 ColumnOptionDef {
                     name: None,
@@ -934,19 +939,17 @@ fn parse_create_table_with_identity_column() {
             vec![
                 ColumnOptionDef {
                     name: None,
-                    #[cfg(not(feature = "bigdecimal"))]
-                    option: ColumnOption::Identity(Some(IdentityProperty {
-                        seed: Expr::Value(Value::Number("1".to_string(), false)),
-                        increment: Expr::Value(Value::Number("1".to_string(), false)),
-                    })),
-                    #[cfg(feature = "bigdecimal")]
-                    option: ColumnOption::Identity(Some(IdentityProperty {
-                        seed: Expr::Value(Value::Number(bigdecimal::BigDecimal::from(1), false)),
-                        increment: Expr::Value(Value::Number(
-                            bigdecimal::BigDecimal::from(1),
-                            false,
-                        )),
-                    })),
+                    option: ColumnOption::Identity(IdentityPropertyKind::Identity(
+                        IdentityProperty {
+                            parameters: Some(IdentityPropertyFormatKind::FunctionCall(
+                                IdentityParameters {
+                                    seed: Expr::Value(number("1")),
+                                    increment: Expr::Value(number("1")),
+                                },
+                            )),
+                            order: None,
+                        },
+                    )),
                 },
                 ColumnOptionDef {
                     name: None,

tests/sqlparser_snowflake.rs

@@ -525,6 +525,321 @@ fn test_snowflake_single_line_tokenize() {
     assert_eq!(expected, tokens);
 }
 
+#[test]
+fn test_snowflake_create_table_with_autoincrement_columns() {
+    let sql = concat!(
+        "CREATE TABLE my_table (",
+        "a INT AUTOINCREMENT ORDER, ",
+        "b INT AUTOINCREMENT(100, 1) NOORDER, ",
+        "c INT IDENTITY, ",
+        "d INT IDENTITY START 100 INCREMENT 1 ORDER",
+        ")"
+    );
+    // it is a snowflake specific options (AUTOINCREMENT/IDENTITY)
+    match snowflake().verified_stmt(sql) {
+        Statement::CreateTable(CreateTable { columns, .. }) => {
+            assert_eq!(
+                columns,
+                vec![
+                    ColumnDef {
+                        name: "a".into(),
+                        data_type: DataType::Int(None),
+                        collation: None,
+                        options: vec![ColumnOptionDef {
+                            name: None,
+                            option: ColumnOption::Identity(IdentityPropertyKind::Autoincrement(
+                                IdentityProperty {
+                                    parameters: None,
+                                    order: Some(IdentityPropertyOrder::Order),
+                                }
+                            ))
+                        }]
+                    },
+                    ColumnDef {
+                        name: "b".into(),
+                        data_type: DataType::Int(None),
+                        collation: None,
+                        options: vec![ColumnOptionDef {
+                            name: None,
+                            option: ColumnOption::Identity(IdentityPropertyKind::Autoincrement(
+                                IdentityProperty {
+                                    parameters: Some(IdentityPropertyFormatKind::FunctionCall(
+                                        IdentityParameters {
+                                            seed: Expr::Value(number("100")),
+                                            increment: Expr::Value(number("1")),
+                                        }
+                                    )),
+                                    order: Some(IdentityPropertyOrder::NoOrder),
+                                }
+                            ))
+                        }]
+                    },
+                    ColumnDef {
+                        name: "c".into(),
+                        data_type: DataType::Int(None),
+                        collation: None,
+                        options: vec![ColumnOptionDef {
+                            name: None,
+                            option: ColumnOption::Identity(IdentityPropertyKind::Identity(
+                                IdentityProperty {
+                                    parameters: None,
+                                    order: None,
+                                }
+                            ))
+                        }]
+                    },
+                    ColumnDef {
+                        name: "d".into(),
+                        data_type: DataType::Int(None),
+                        collation: None,
+                        options: vec![ColumnOptionDef {
+                            name: None,
+                            option: ColumnOption::Identity(IdentityPropertyKind::Identity(
+                                IdentityProperty {
+                                    parameters: Some(
+                                        IdentityPropertyFormatKind::StartAndIncrement(
+                                            IdentityParameters {
+                                                seed: Expr::Value(number("100")),
+                                                increment: Expr::Value(number("1")),
+                                            }
+                                        )
+                                    ),
+                                    order: Some(IdentityPropertyOrder::Order),
+                                }
+                            ))
+                        }]
+                    },
+                ]
+            );
+        }
+        _ => unreachable!(),
+    }
+}
+
+#[test]
+fn test_snowflake_create_table_with_collated_column() {
+    match snowflake_and_generic().verified_stmt("CREATE TABLE my_table (a TEXT COLLATE 'de_DE')") {
+        Statement::CreateTable(CreateTable { columns, .. }) => {
+            assert_eq!(
+                columns,
+                vec![ColumnDef {
+                    name: "a".into(),
+                    data_type: DataType::Text,
+                    collation: Some(ObjectName(vec![Ident::with_quote('\'', "de_DE")])),
+                    options: vec![]
+                },]
+            );
+        }
+        _ => unreachable!(),
+    }
+}
+
+#[test]
+fn test_snowflake_create_table_with_columns_masking_policy() {
+    for (sql, with, using_columns) in [
+        (
+            "CREATE TABLE my_table (a INT WITH MASKING POLICY p)",
+            true,
+            None,
+        ),
+        (
+            "CREATE TABLE my_table (a INT MASKING POLICY p)",
+            false,
+            None,
+        ),
+        (
+            "CREATE TABLE my_table (a INT WITH MASKING POLICY p USING (a, b))",
+            true,
+            Some(vec!["a".into(), "b".into()]),
+        ),
+        (
+            "CREATE TABLE my_table (a INT MASKING POLICY p USING (a, b))",
+            false,
+            Some(vec!["a".into(), "b".into()]),
+        ),
+    ] {
+        match snowflake().verified_stmt(sql) {
+            Statement::CreateTable(CreateTable { columns, .. }) => {
+                assert_eq!(
+                    columns,
+                    vec![ColumnDef {
+                        name: "a".into(),
+                        data_type: DataType::Int(None),
+                        collation: None,
+                        options: vec![ColumnOptionDef {
+                            name: None,
+                            option: ColumnOption::Policy(ColumnPolicy::MaskingPolicy(
+                                ColumnPolicyProperty {
+                                    with,
+                                    policy_name: "p".into(),
+                                    using_columns,
+                                }
+                            ))
+                        }],
+                    },]
+                );
+            }
+            _ => unreachable!(),
+        }
+    }
+}
+
+#[test]
+fn test_snowflake_create_table_with_columns_projection_policy() {
+    for (sql, with) in [
+        (
+            "CREATE TABLE my_table (a INT WITH PROJECTION POLICY p)",
+            true,
+        ),
+        ("CREATE TABLE my_table (a INT PROJECTION POLICY p)", false),
+    ] {
+        match snowflake().verified_stmt(sql) {
+            Statement::CreateTable(CreateTable { columns, .. }) => {
+                assert_eq!(
+                    columns,
+                    vec![ColumnDef {
+                        name: "a".into(),
+                        data_type: DataType::Int(None),
+                        collation: None,
+                        options: vec![ColumnOptionDef {
+                            name: None,
+                            option: ColumnOption::Policy(ColumnPolicy::ProjectionPolicy(
+                                ColumnPolicyProperty {
+                                    with,
+                                    policy_name: "p".into(),
+                                    using_columns: None,
+                                }
+                            ))
+                        }],
+                    },]
+                );
+            }
+            _ => unreachable!(),
+        }
+    }
+}
+
+#[test]
+fn test_snowflake_create_table_with_columns_tags() {
+    for (sql, with) in [
+        (
+            "CREATE TABLE my_table (a INT WITH TAG (A='TAG A', B='TAG B'))",
+            true,
+        ),
+        (
+            "CREATE TABLE my_table (a INT TAG (A='TAG A', B='TAG B'))",
+            false,
+        ),
+    ] {
+        match snowflake().verified_stmt(sql) {
+            Statement::CreateTable(CreateTable { columns, .. }) => {
+                assert_eq!(
+                    columns,
+                    vec![ColumnDef {
+                        name: "a".into(),
+                        data_type: DataType::Int(None),
+                        collation: None,
+                        options: vec![ColumnOptionDef {
+                            name: None,
+                            option: ColumnOption::Tags(TagsColumnOption {
+                                with,
+                                tags: vec![
+                                    Tag::new("A".into(), "TAG A".into()),
+                                    Tag::new("B".into(), "TAG B".into()),
+                                ]
+                            }),
+                        }],
+                    },]
+                );
+            }
+            _ => unreachable!(),
+        }
+    }
+}
+
+#[test]
+fn test_snowflake_create_table_with_several_column_options() {
+    let sql = concat!(
+        "CREATE TABLE my_table (",
+        "a INT IDENTITY WITH MASKING POLICY p1 USING (a, b) WITH TAG (A='TAG A', B='TAG B'), ",
+        "b TEXT COLLATE 'de_DE' PROJECTION POLICY p2 TAG (C='TAG C', D='TAG D')",
+        ")"
+    );
+    match snowflake().verified_stmt(sql) {
+        Statement::CreateTable(CreateTable { columns, .. }) => {
+            assert_eq!(
+                columns,
+                vec![
+                    ColumnDef {
+                        name: "a".into(),
+                        data_type: DataType::Int(None),
+                        collation: None,
+                        options: vec![
+                            ColumnOptionDef {
+                                name: None,
+                                option: ColumnOption::Identity(IdentityPropertyKind::Identity(
+                                    IdentityProperty {
+                                        parameters: None,
+                                        order: None
+                                    }
+                                )),
+                            },
+                            ColumnOptionDef {
+                                name: None,
+                                option: ColumnOption::Policy(ColumnPolicy::MaskingPolicy(
+                                    ColumnPolicyProperty {
+                                        with: true,
+                                        policy_name: "p1".into(),
+                                        using_columns: Some(vec!["a".into(), "b".into()]),
+                                    }
+                                )),
+                            },
+                            ColumnOptionDef {
+                                name: None,
+                                option: ColumnOption::Tags(TagsColumnOption {
+                                    with: true,
+                                    tags: vec![
+                                        Tag::new("A".into(), "TAG A".into()),
+                                        Tag::new("B".into(), "TAG B".into()),
+                                    ]
+                                }),
+                            }
+                        ],
+                    },
+                    ColumnDef {
+                        name: "b".into(),
+                        data_type: DataType::Text,
+                        collation: Some(ObjectName(vec![Ident::with_quote('\'', "de_DE")])),
+                        options: vec![
+                            ColumnOptionDef {
+                                name: None,
+                                option: ColumnOption::Policy(ColumnPolicy::ProjectionPolicy(
+                                    ColumnPolicyProperty {
+                                        with: false,
+                                        policy_name: "p2".into(),
+                                        using_columns: None,
+                                    }
+                                )),
+                            },
+                            ColumnOptionDef {
+                                name: None,
+                                option: ColumnOption::Tags(TagsColumnOption {
+                                    with: false,
+                                    tags: vec![
+                                        Tag::new("C".into(), "TAG C".into()),
+                                        Tag::new("D".into(), "TAG D".into()),
+                                    ]
+                                }),
+                            }
+                        ],
+                    },
+                ]
+            );
+        }
+        _ => unreachable!(),
+    }
+}
+
 #[test]
 fn parse_sf_create_or_replace_view_with_comment_missing_equal() {
     assert!(snowflake_and_generic()