Add all missing table options to be handled in any order (#1747)

Co-authored-by: Tomer Shani <tomer.shani@satoricyber.com>

src/ast/dml.rs

@@ -33,11 +33,11 @@ pub use super::ddl::{ColumnDef, TableConstraint};
 
 use super::{
     display_comma_separated, display_separated, query::InputFormatClause, Assignment, ClusteredBy,
-    CommentDef, Expr, FileFormat, FromTable, HiveDistributionStyle, HiveFormat, HiveIOFormat,
+    CommentDef, CreateTableOptions, Expr, FileFormat, FromTable, HiveDistributionStyle, HiveFormat,
-    HiveRowFormat, Ident, IndexType, InsertAliases, MysqlInsertPriority, ObjectName, OnCommit,
+    HiveIOFormat, HiveRowFormat, Ident, IndexType, InsertAliases, MysqlInsertPriority, ObjectName,
-    OnInsert, OneOrManyWithParens, OrderByExpr, Query, RowAccessPolicy, SelectItem, Setting,
+    OnCommit, OnInsert, OneOrManyWithParens, OrderByExpr, Query, RowAccessPolicy, SelectItem,
-    SqlOption, SqliteOnConflict, StorageSerializationPolicy, TableEngine, TableObject,
+    Setting, SqliteOnConflict, StorageSerializationPolicy, TableObject, TableWithJoins, Tag,
-    TableWithJoins, Tag, WrappedCollection,
+    WrappedCollection,
 };
 
 /// Index column type.
@@ -146,19 +146,17 @@ pub struct CreateTable {
     pub constraints: Vec<TableConstraint>,
     pub hive_distribution: HiveDistributionStyle,
     pub hive_formats: Option<HiveFormat>,
-    pub table_properties: Vec<SqlOption>,
+    pub table_options: CreateTableOptions,
-    pub with_options: Vec<SqlOption>,
     pub file_format: Option<FileFormat>,
     pub location: Option<String>,
     pub query: Option<Box<Query>>,
     pub without_rowid: bool,
     pub like: Option<ObjectName>,
     pub clone: Option<ObjectName>,
-    pub engine: Option<TableEngine>,
+    // For Hive dialect, the table comment is after the column definitions without `=`,
+    // so the `comment` field is optional and different than the comment field in the general options list.
+    // [Hive](https://cwiki.apache.org/confluence/display/Hive/LanguageManual+DDL#LanguageManualDDL-CreateTable)
     pub comment: Option<CommentDef>,
-    pub auto_increment_offset: Option<u32>,
-    pub default_charset: Option<String>,
-    pub collation: Option<String>,
     pub on_commit: Option<OnCommit>,
     /// ClickHouse "ON CLUSTER" clause:
     /// <https://clickhouse.com/docs/en/sql-reference/distributed-ddl/>
@@ -179,9 +177,6 @@ pub struct CreateTable {
     /// Hive: Table clustering column list.
     /// <https://cwiki.apache.org/confluence/display/Hive/LanguageManual+DDL#LanguageManualDDL-CreateTable>
     pub clustered_by: Option<ClusteredBy>,
-    /// BigQuery: Table options list.
-    /// <https://cloud.google.com/bigquery/docs/reference/standard-sql/data-definition-language#table_option_list>
-    pub options: Option<Vec<SqlOption>>,
     /// Postgres `INHERITs` clause, which contains the list of tables from which
     /// the new table inherits.
     /// <https://www.postgresql.org/docs/current/ddl-inherit.html>
@@ -282,7 +277,7 @@ impl Display for CreateTable {
 
         // Hive table comment should be after column definitions, please refer to:
         // [Hive](https://cwiki.apache.org/confluence/display/Hive/LanguageManual+DDL#LanguageManualDDL-CreateTable)
-        if let Some(CommentDef::AfterColumnDefsWithoutEq(comment)) = &self.comment {
+        if let Some(comment) = &self.comment {
             write!(f, " COMMENT '{comment}'")?;
         }
 
@@ -375,35 +370,14 @@ impl Display for CreateTable {
             }
             write!(f, " LOCATION '{}'", self.location.as_ref().unwrap())?;
         }
-        if !self.table_properties.is_empty() {
+
-            write!(
+        match &self.table_options {
-                f,
+            options @ CreateTableOptions::With(_)
-                " TBLPROPERTIES ({})",
+            | options @ CreateTableOptions::Plain(_)
-                display_comma_separated(&self.table_properties)
+            | options @ CreateTableOptions::TableProperties(_) => write!(f, " {}", options)?,
-            )?;
+            _ => (),
-        }
-        if !self.with_options.is_empty() {
-            write!(f, " WITH ({})", display_comma_separated(&self.with_options))?;
-        }
-        if let Some(engine) = &self.engine {
-            write!(f, " ENGINE={engine}")?;
-        }
-        if let Some(comment_def) = &self.comment {
-            match comment_def {
-                CommentDef::WithEq(comment) => {
-                    write!(f, " COMMENT = '{comment}'")?;
-                }
-                CommentDef::WithoutEq(comment) => {
-                    write!(f, " COMMENT '{comment}'")?;
-                }
-                // For CommentDef::AfterColumnDefsWithoutEq will be displayed after column definition
-                CommentDef::AfterColumnDefsWithoutEq(_) => (),
-            }
         }
 
-        if let Some(auto_increment_offset) = self.auto_increment_offset {
-            write!(f, " AUTO_INCREMENT {auto_increment_offset}")?;
-        }
         if let Some(primary_key) = &self.primary_key {
             write!(f, " PRIMARY KEY {}", primary_key)?;
         }
@@ -419,15 +393,9 @@ impl Display for CreateTable {
         if let Some(cluster_by) = self.cluster_by.as_ref() {
             write!(f, " CLUSTER BY {cluster_by}")?;
         }
-
+        if let options @ CreateTableOptions::Options(_) = &self.table_options {
-        if let Some(options) = self.options.as_ref() {
+            write!(f, " {}", options)?;
-            write!(
-                f,
-                " OPTIONS({})",
-                display_comma_separated(options.as_slice())
-            )?;
         }
-
         if let Some(external_volume) = self.external_volume.as_ref() {
             write!(f, " EXTERNAL_VOLUME = '{external_volume}'")?;
         }
@@ -503,13 +471,6 @@ impl Display for CreateTable {
             write!(f, " WITH TAG ({})", display_comma_separated(tag.as_slice()))?;
         }
 
-        if let Some(default_charset) = &self.default_charset {
-            write!(f, " DEFAULT CHARSET={default_charset}")?;
-        }
-        if let Some(collation) = &self.collation {
-            write!(f, " COLLATE={collation}")?;
-        }
-
         if self.on_commit.is_some() {
             let on_commit = match self.on_commit {
                 Some(OnCommit::DeleteRows) => "ON COMMIT DELETE ROWS",

src/ast/helpers/stmt_create_table.rs

@@ -26,10 +26,12 @@ use sqlparser_derive::{Visit, VisitMut};
 
 use super::super::dml::CreateTable;
 use crate::ast::{
-    ClusteredBy, ColumnDef, CommentDef, Expr, FileFormat, HiveDistributionStyle, HiveFormat, Ident,
+    ClusteredBy, ColumnDef, CommentDef, CreateTableOptions, Expr, FileFormat,
-    ObjectName, OnCommit, OneOrManyWithParens, Query, RowAccessPolicy, SqlOption, Statement,
+    HiveDistributionStyle, HiveFormat, Ident, ObjectName, OnCommit, OneOrManyWithParens, Query,
-    StorageSerializationPolicy, TableConstraint, TableEngine, Tag, WrappedCollection,
+    RowAccessPolicy, Statement, StorageSerializationPolicy, TableConstraint, Tag,
+    WrappedCollection,
 };
+
 use crate::parser::ParserError;
 
 /// Builder for create table statement variant ([1]).
@@ -76,19 +78,13 @@ pub struct CreateTableBuilder {
     pub constraints: Vec<TableConstraint>,
     pub hive_distribution: HiveDistributionStyle,
     pub hive_formats: Option<HiveFormat>,
-    pub table_properties: Vec<SqlOption>,
-    pub with_options: Vec<SqlOption>,
     pub file_format: Option<FileFormat>,
     pub location: Option<String>,
     pub query: Option<Box<Query>>,
     pub without_rowid: bool,
     pub like: Option<ObjectName>,
     pub clone: Option<ObjectName>,
-    pub engine: Option<TableEngine>,
     pub comment: Option<CommentDef>,
-    pub auto_increment_offset: Option<u32>,
-    pub default_charset: Option<String>,
-    pub collation: Option<String>,
     pub on_commit: Option<OnCommit>,
     pub on_cluster: Option<Ident>,
     pub primary_key: Option<Box<Expr>>,
@@ -96,7 +92,6 @@ pub struct CreateTableBuilder {
     pub partition_by: Option<Box<Expr>>,
     pub cluster_by: Option<WrappedCollection<Vec<Ident>>>,
     pub clustered_by: Option<ClusteredBy>,
-    pub options: Option<Vec<SqlOption>>,
     pub inherits: Option<Vec<ObjectName>>,
     pub strict: bool,
     pub copy_grants: bool,
@@ -113,6 +108,7 @@ pub struct CreateTableBuilder {
     pub catalog: Option<String>,
     pub catalog_sync: Option<String>,
     pub storage_serialization_policy: Option<StorageSerializationPolicy>,
+    pub table_options: CreateTableOptions,
 }
 
 impl CreateTableBuilder {
@@ -131,19 +127,13 @@ impl CreateTableBuilder {
             constraints: vec![],
             hive_distribution: HiveDistributionStyle::NONE,
             hive_formats: None,
-            table_properties: vec![],
-            with_options: vec![],
             file_format: None,
             location: None,
             query: None,
             without_rowid: false,
             like: None,
             clone: None,
-            engine: None,
             comment: None,
-            auto_increment_offset: None,
-            default_charset: None,
-            collation: None,
             on_commit: None,
             on_cluster: None,
             primary_key: None,
@@ -151,7 +141,6 @@ impl CreateTableBuilder {
             partition_by: None,
             cluster_by: None,
             clustered_by: None,
-            options: None,
             inherits: None,
             strict: false,
             copy_grants: false,
@@ -168,6 +157,7 @@ impl CreateTableBuilder {
             catalog: None,
             catalog_sync: None,
             storage_serialization_policy: None,
+            table_options: CreateTableOptions::None,
         }
     }
     pub fn or_replace(mut self, or_replace: bool) -> Self {
@@ -230,15 +220,6 @@ impl CreateTableBuilder {
         self
     }
 
-    pub fn table_properties(mut self, table_properties: Vec<SqlOption>) -> Self {
-        self.table_properties = table_properties;
-        self
-    }
-
-    pub fn with_options(mut self, with_options: Vec<SqlOption>) -> Self {
-        self.with_options = with_options;
-        self
-    }
     pub fn file_format(mut self, file_format: Option<FileFormat>) -> Self {
         self.file_format = file_format;
         self
@@ -268,31 +249,11 @@ impl CreateTableBuilder {
         self
     }
 
-    pub fn engine(mut self, engine: Option<TableEngine>) -> Self {
+    pub fn comment_after_column_def(mut self, comment: Option<CommentDef>) -> Self {
-        self.engine = engine;
-        self
-    }
-
-    pub fn comment(mut self, comment: Option<CommentDef>) -> Self {
         self.comment = comment;
         self
     }
 
-    pub fn auto_increment_offset(mut self, offset: Option<u32>) -> Self {
-        self.auto_increment_offset = offset;
-        self
-    }
-
-    pub fn default_charset(mut self, default_charset: Option<String>) -> Self {
-        self.default_charset = default_charset;
-        self
-    }
-
-    pub fn collation(mut self, collation: Option<String>) -> Self {
-        self.collation = collation;
-        self
-    }
-
     pub fn on_commit(mut self, on_commit: Option<OnCommit>) -> Self {
         self.on_commit = on_commit;
         self
@@ -328,11 +289,6 @@ impl CreateTableBuilder {
         self
     }
 
-    pub fn options(mut self, options: Option<Vec<SqlOption>>) -> Self {
-        self.options = options;
-        self
-    }
-
     pub fn inherits(mut self, inherits: Option<Vec<ObjectName>>) -> Self {
         self.inherits = inherits;
         self
@@ -422,6 +378,11 @@ impl CreateTableBuilder {
         self
     }
 
+    pub fn table_options(mut self, table_options: CreateTableOptions) -> Self {
+        self.table_options = table_options;
+        self
+    }
+
     pub fn build(self) -> Statement {
         Statement::CreateTable(CreateTable {
             or_replace: self.or_replace,
@@ -437,19 +398,13 @@ impl CreateTableBuilder {
             constraints: self.constraints,
             hive_distribution: self.hive_distribution,
             hive_formats: self.hive_formats,
-            table_properties: self.table_properties,
-            with_options: self.with_options,
             file_format: self.file_format,
             location: self.location,
             query: self.query,
             without_rowid: self.without_rowid,
             like: self.like,
             clone: self.clone,
-            engine: self.engine,
             comment: self.comment,
-            auto_increment_offset: self.auto_increment_offset,
-            default_charset: self.default_charset,
-            collation: self.collation,
             on_commit: self.on_commit,
             on_cluster: self.on_cluster,
             primary_key: self.primary_key,
@@ -457,7 +412,6 @@ impl CreateTableBuilder {
             partition_by: self.partition_by,
             cluster_by: self.cluster_by,
             clustered_by: self.clustered_by,
-            options: self.options,
             inherits: self.inherits,
             strict: self.strict,
             copy_grants: self.copy_grants,
@@ -474,6 +428,7 @@ impl CreateTableBuilder {
             catalog: self.catalog,
             catalog_sync: self.catalog_sync,
             storage_serialization_policy: self.storage_serialization_policy,
+            table_options: self.table_options,
         })
     }
 }
@@ -499,19 +454,13 @@ impl TryFrom<Statement> for CreateTableBuilder {
                 constraints,
                 hive_distribution,
                 hive_formats,
-                table_properties,
-                with_options,
                 file_format,
                 location,
                 query,
                 without_rowid,
                 like,
                 clone,
-                engine,
                 comment,
-                auto_increment_offset,
-                default_charset,
-                collation,
                 on_commit,
                 on_cluster,
                 primary_key,
@@ -519,7 +468,6 @@ impl TryFrom<Statement> for CreateTableBuilder {
                 partition_by,
                 cluster_by,
                 clustered_by,
-                options,
                 inherits,
                 strict,
                 copy_grants,
@@ -536,6 +484,7 @@ impl TryFrom<Statement> for CreateTableBuilder {
                 catalog,
                 catalog_sync,
                 storage_serialization_policy,
+                table_options,
             }) => Ok(Self {
                 or_replace,
                 temporary,
@@ -548,19 +497,13 @@ impl TryFrom<Statement> for CreateTableBuilder {
                 constraints,
                 hive_distribution,
                 hive_formats,
-                table_properties,
-                with_options,
                 file_format,
                 location,
                 query,
                 without_rowid,
                 like,
                 clone,
-                engine,
                 comment,
-                auto_increment_offset,
-                default_charset,
-                collation,
                 on_commit,
                 on_cluster,
                 primary_key,
@@ -568,7 +511,6 @@ impl TryFrom<Statement> for CreateTableBuilder {
                 partition_by,
                 cluster_by,
                 clustered_by,
-                options,
                 inherits,
                 strict,
                 iceberg,
@@ -587,6 +529,7 @@ impl TryFrom<Statement> for CreateTableBuilder {
                 catalog,
                 catalog_sync,
                 storage_serialization_policy,
+                table_options,
             }),
             _ => Err(ParserError::ParserError(format!(
                 "Expected create table statement, but received: {stmt}"
@@ -600,8 +543,8 @@ impl TryFrom<Statement> for CreateTableBuilder {
 pub(crate) struct CreateTableConfiguration {
     pub partition_by: Option<Box<Expr>>,
     pub cluster_by: Option<WrappedCollection<Vec<Ident>>>,
-    pub options: Option<Vec<SqlOption>>,
     pub inherits: Option<Vec<ObjectName>>,
+    pub table_options: CreateTableOptions,
 }
 
 #[cfg(test)]

src/ast/helpers/stmt_data_loading.rs

@@ -21,8 +21,6 @@
 
 #[cfg(not(feature = "std"))]
 use alloc::string::String;
-#[cfg(not(feature = "std"))]
-use alloc::vec::Vec;
 use core::fmt;
 
 #[cfg(feature = "serde")]

src/ast/mod.rs

@@ -2681,6 +2681,18 @@ pub enum CreateTableOptions {
     ///
     /// <https://cloud.google.com/bigquery/docs/reference/standard-sql/data-definition-language#table_option_list>
     Options(Vec<SqlOption>),
+
+    /// Plain options, options which are not part on any declerative statement e.g. WITH/OPTIONS/...
+    /// <https://dev.mysql.com/doc/refman/8.4/en/create-table.html>
+    Plain(Vec<SqlOption>),
+
+    TableProperties(Vec<SqlOption>),
+}
+
+impl Default for CreateTableOptions {
+    fn default() -> Self {
+        Self::None
+    }
 }
 
 impl fmt::Display for CreateTableOptions {
@@ -2692,6 +2704,12 @@ impl fmt::Display for CreateTableOptions {
             CreateTableOptions::Options(options) => {
                 write!(f, "OPTIONS({})", display_comma_separated(options))
             }
+            CreateTableOptions::TableProperties(options) => {
+                write!(f, "TBLPROPERTIES ({})", display_comma_separated(options))
+            }
+            CreateTableOptions::Plain(options) => {
+                write!(f, "{}", display_separated(options, " "))
+            }
             CreateTableOptions::None => Ok(()),
         }
     }
@@ -7560,6 +7578,18 @@ pub enum SqlOption {
         range_direction: Option<PartitionRangeDirection>,
         for_values: Vec<Expr>,
     },
+    /// Comment parameter (supports `=` and no `=` syntax)
+    Comment(CommentDef),
+    /// MySQL TableSpace option
+    /// <https://dev.mysql.com/doc/refman/8.4/en/create-table.html>
+    TableSpace(TablespaceOption),
+    /// An option representing a key value pair, where the value is a parenthesized list and with an optional name
+    /// e.g.
+    ///
+    ///   UNION  = (tbl_name\[,tbl_name\]...) <https://dev.mysql.com/doc/refman/8.4/en/create-table.html>
+    ///   ENGINE = ReplicatedMergeTree('/table_name','{replica}', ver) <https://clickhouse.com/docs/engines/table-engines/mergetree-family/replication>
+    ///   ENGINE = SummingMergeTree(\[columns\]) <https://clickhouse.com/docs/engines/table-engines/mergetree-family/summingmergetree>
+    NamedParenthesizedList(NamedParenthesizedList),
 }
 
 impl fmt::Display for SqlOption {
@@ -7591,10 +7621,54 @@ impl fmt::Display for SqlOption {
                     display_comma_separated(for_values)
                 )
             }
+            SqlOption::TableSpace(tablespace_option) => {
+                write!(f, "TABLESPACE {}", tablespace_option.name)?;
+                match tablespace_option.storage {
+                    Some(StorageType::Disk) => write!(f, " STORAGE DISK"),
+                    Some(StorageType::Memory) => write!(f, " STORAGE MEMORY"),
+                    None => Ok(()),
+                }
+            }
+            SqlOption::Comment(comment) => match comment {
+                CommentDef::WithEq(comment) => {
+                    write!(f, "COMMENT = '{comment}'")
+                }
+                CommentDef::WithoutEq(comment) => {
+                    write!(f, "COMMENT '{comment}'")
+                }
+            },
+            SqlOption::NamedParenthesizedList(value) => {
+                write!(f, "{} = ", value.key)?;
+                if let Some(key) = &value.name {
+                    write!(f, "{}", key)?;
+                }
+                if !value.values.is_empty() {
+                    write!(f, "({})", display_comma_separated(&value.values))?
+                }
+                Ok(())
+            }
         }
     }
 }
 
+#[derive(Debug, Clone, PartialEq, Eq, Hash, PartialOrd, Ord)]
+#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
+#[cfg_attr(feature = "visitor", derive(Visit, VisitMut))]
+pub enum StorageType {
+    Disk,
+    Memory,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Hash, PartialOrd, Ord)]
+#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
+#[cfg_attr(feature = "visitor", derive(Visit, VisitMut))]
+/// MySql TableSpace option
+/// <https://dev.mysql.com/doc/refman/8.4/en/create-table.html>
+pub struct TablespaceOption {
+    pub name: String,
+    pub storage: Option<StorageType>,
+}
+
 #[derive(Debug, Clone, PartialEq, PartialOrd, Eq, Ord, Hash)]
 #[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
 #[cfg_attr(feature = "visitor", derive(Visit, VisitMut))]
@@ -8860,27 +8934,20 @@ impl Display for CreateViewParams {
     }
 }
 
-/// Engine of DB. Some warehouse has parameters of engine, e.g. [ClickHouse]
-///
-/// [ClickHouse]: https://clickhouse.com/docs/en/engines/table-engines
 #[derive(Debug, Clone, PartialEq, PartialOrd, Eq, Ord, Hash)]
 #[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
 #[cfg_attr(feature = "visitor", derive(Visit, VisitMut))]
-pub struct TableEngine {
+/// Key/Value, where the value is a (optionally named) list of identifiers
-    pub name: String,
+///
-    pub parameters: Option<Vec<Ident>>,
+/// ```sql
-}
+/// UNION = (tbl_name[,tbl_name]...)
-
+/// ENGINE = ReplicatedMergeTree('/table_name','{replica}', ver)
-impl Display for TableEngine {
+/// ENGINE = SummingMergeTree([columns])
-    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+/// ```
-        write!(f, "{}", self.name)?;
+pub struct NamedParenthesizedList {
-
+    pub key: Ident,
-        if let Some(parameters) = self.parameters.as_ref() {
+    pub name: Option<Ident>,
-            write!(f, "({})", display_comma_separated(parameters))?;
+    pub values: Vec<Ident>,
-        }
-
-        Ok(())
-    }
 }
 
 /// Snowflake `WITH ROW ACCESS POLICY policy_name ON (identifier, ...)`
@@ -8944,18 +9011,12 @@ pub enum CommentDef {
     /// Does not include `=` when printing the comment, as `COMMENT 'comment'`
     WithEq(String),
     WithoutEq(String),
-    // For Hive dialect, the table comment is after the column definitions without `=`,
-    // so we need to add an extra variant to allow to identify this case when displaying.
-    // [Hive](https://cwiki.apache.org/confluence/display/Hive/LanguageManual+DDL#LanguageManualDDL-CreateTable)
-    AfterColumnDefsWithoutEq(String),
 }
 
 impl Display for CommentDef {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         match self {
-            CommentDef::WithEq(comment)
+            CommentDef::WithEq(comment) | CommentDef::WithoutEq(comment) => write!(f, "{comment}"),
-            | CommentDef::WithoutEq(comment)
-            | CommentDef::AfterColumnDefsWithoutEq(comment) => write!(f, "{comment}"),
         }
     }
 }

src/ast/spans.rs

@@ -30,10 +30,10 @@ use super::{
     Function, FunctionArg, FunctionArgExpr, FunctionArgumentClause, FunctionArgumentList,
     FunctionArguments, GroupByExpr, HavingBound, IfStatement, IlikeSelectItem, Insert, Interpolate,
     InterpolateExpr, Join, JoinConstraint, JoinOperator, JsonPath, JsonPathElem, LateralView,
-    LimitClause, MatchRecognizePattern, Measure, NamedWindowDefinition, ObjectName, ObjectNamePart,
+    LimitClause, MatchRecognizePattern, Measure, NamedParenthesizedList, NamedWindowDefinition,
-    Offset, OnConflict, OnConflictAction, OnInsert, OpenStatement, OrderBy, OrderByExpr,
+    ObjectName, ObjectNamePart, Offset, OnConflict, OnConflictAction, OnInsert, OpenStatement,
-    OrderByKind, Partition, PivotValueSource, ProjectionSelect, Query, RaiseStatement,
+    OrderBy, OrderByExpr, OrderByKind, Partition, PivotValueSource, ProjectionSelect, Query,
-    RaiseStatementValue, ReferentialAction, RenameSelectItem, ReplaceSelectElement,
+    RaiseStatement, RaiseStatementValue, ReferentialAction, RenameSelectItem, ReplaceSelectElement,
     ReplaceSelectItem, Select, SelectInto, SelectItem, SetExpr, SqlOption, Statement, Subscript,
     SymbolDefinition, TableAlias, TableAliasColumnDef, TableConstraint, TableFactor, TableObject,
     TableOptionsClustered, TableWithJoins, UpdateTableFromKind, Use, Value, Values, ViewColumnDef,
@@ -567,27 +567,20 @@ impl Spanned for CreateTable {
             constraints,
             hive_distribution: _, // hive specific
             hive_formats: _,      // hive specific
-            table_properties,
+            file_format: _,       // enum
-            with_options,
+            location: _,          // string, no span
-            file_format: _, // enum
-            location: _,    // string, no span
             query,
             without_rowid: _, // bool
             like,
             clone,
-            engine: _,                          // todo
+            comment: _, // todo, no span
-            comment: _,                         // todo, no span
+            on_commit: _,
-            auto_increment_offset: _,           // u32, no span
-            default_charset: _,                 // string, no span
-            collation: _,                       // string, no span
-            on_commit: _,                       // enum
             on_cluster: _,                      // todo, clickhouse specific
             primary_key: _,                     // todo, clickhouse specific
             order_by: _,                        // todo, clickhouse specific
             partition_by: _,                    // todo, BigQuery specific
             cluster_by: _,                      // todo, BigQuery specific
             clustered_by: _,                    // todo, Hive specific
-            options: _,                         // todo, BigQuery specific
             inherits: _,                        // todo, PostgreSQL specific
             strict: _,                          // bool
             copy_grants: _,                     // bool
@@ -603,15 +596,15 @@ impl Spanned for CreateTable {
             base_location: _,                   // todo, Snowflake specific
             catalog: _,                         // todo, Snowflake specific
             catalog_sync: _,                    // todo, Snowflake specific
-            storage_serialization_policy: _,    // todo, Snowflake specific
+            storage_serialization_policy: _,
+            table_options,
         } = self;
 
         union_spans(
             core::iter::once(name.span())
+                .chain(core::iter::once(table_options.span()))
                 .chain(columns.iter().map(|i| i.span()))
                 .chain(constraints.iter().map(|i| i.span()))
-                .chain(table_properties.iter().map(|i| i.span()))
-                .chain(with_options.iter().map(|i| i.span()))
                 .chain(query.iter().map(|i| i.span()))
                 .chain(like.iter().map(|i| i.span()))
                 .chain(clone.iter().map(|i| i.span())),
@@ -1004,6 +997,14 @@ impl Spanned for SqlOption {
             } => union_spans(
                 core::iter::once(column_name.span).chain(for_values.iter().map(|i| i.span())),
             ),
+            SqlOption::TableSpace(_) => Span::empty(),
+            SqlOption::Comment(_) => Span::empty(),
+            SqlOption::NamedParenthesizedList(NamedParenthesizedList {
+                key: name,
+                name: value,
+                values,
+            }) => union_spans(core::iter::once(name.span).chain(values.iter().map(|i| i.span)))
+                .union_opt(&value.as_ref().map(|i| i.span)),
         }
     }
 }
@@ -1041,6 +1042,8 @@ impl Spanned for CreateTableOptions {
             CreateTableOptions::None => Span::empty(),
             CreateTableOptions::With(vec) => union_spans(vec.iter().map(|i| i.span())),
             CreateTableOptions::Options(vec) => union_spans(vec.iter().map(|i| i.span())),
+            CreateTableOptions::Plain(vec) => union_spans(vec.iter().map(|i| i.span())),
+            CreateTableOptions::TableProperties(vec) => union_spans(vec.iter().map(|i| i.span())),
         }
     }
 }

src/dialect/snowflake.rs

@@ -25,8 +25,8 @@ use crate::ast::helpers::stmt_data_loading::{
 use crate::ast::{
     ColumnOption, ColumnPolicy, ColumnPolicyProperty, CopyIntoSnowflakeKind, Ident,
     IdentityParameters, IdentityProperty, IdentityPropertyFormatKind, IdentityPropertyKind,
-    IdentityPropertyOrder, ObjectName, RowAccessPolicy, ShowObjects, Statement, TagsColumnOption,
+    IdentityPropertyOrder, ObjectName, RowAccessPolicy, ShowObjects, SqlOption, Statement,
-    WrappedCollection,
+    TagsColumnOption, WrappedCollection,
 };
 use crate::dialect::{Dialect, Precedence};
 use crate::keywords::Keyword;
@@ -417,6 +417,8 @@ pub fn parse_create_table(
     // "CREATE TABLE x COPY GRANTS (c INT)" and "CREATE TABLE x (c INT) COPY GRANTS" are both
     // accepted by Snowflake
 
+    let mut plain_options = vec![];
+
     loop {
         let next_token = parser.next_token();
         match &next_token.token {
@@ -428,7 +430,9 @@ pub fn parse_create_table(
                 Keyword::COMMENT => {
                     // Rewind the COMMENT keyword
                     parser.prev_token();
-                    builder = builder.comment(parser.parse_optional_inline_comment()?);
+                    if let Some(comment_def) = parser.parse_optional_inline_comment()? {
+                        plain_options.push(SqlOption::Comment(comment_def))
+                    }
                 }
                 Keyword::AS => {
                     let query = parser.parse_query()?;
@@ -589,6 +593,13 @@ pub fn parse_create_table(
             }
         }
     }
+    let table_options = if !plain_options.is_empty() {
+        crate::ast::CreateTableOptions::Plain(plain_options)
+    } else {
+        crate::ast::CreateTableOptions::None
+    };
+
+    builder = builder.table_options(table_options);
 
     if iceberg && builder.base_location.is_none() {
         return Err(ParserError::ParserError(

src/keywords.rs

@@ -116,9 +116,11 @@ define_keywords!(
     AUTHENTICATION,
     AUTHORIZATION,
     AUTO,
+    AUTOEXTEND_SIZE,
     AUTOINCREMENT,
     AUTO_INCREMENT,
     AVG,
+    AVG_ROW_LENGTH,
     AVRO,
     BACKWARD,
     BASE64,
@@ -180,6 +182,7 @@ define_keywords!(
     CHARSET,
     CHAR_LENGTH,
     CHECK,
+    CHECKSUM,
     CIRCLE,
     CLEAR,
     CLOB,
@@ -269,6 +272,7 @@ define_keywords!(
     DEFINED,
     DEFINER,
     DELAYED,
+    DELAY_KEY_WRITE,
     DELETE,
     DELIMITED,
     DELIMITER,
@@ -313,6 +317,7 @@ define_keywords!(
     END_PARTITION,
     ENFORCED,
     ENGINE,
+    ENGINE_ATTRIBUTE,
     ENUM,
     ENUM16,
     ENUM8,
@@ -444,6 +449,7 @@ define_keywords!(
     INPUTFORMAT,
     INSENSITIVE,
     INSERT,
+    INSERT_METHOD,
     INSTALL,
     INSTANT,
     INSTEAD,
@@ -480,6 +486,7 @@ define_keywords!(
     JULIAN,
     KEY,
     KEYS,
+    KEY_BLOCK_SIZE,
     KILL,
     LAG,
     LANGUAGE,
@@ -533,6 +540,7 @@ define_keywords!(
     MAX,
     MAXVALUE,
     MAX_DATA_EXTENSION_TIME_IN_DAYS,
+    MAX_ROWS,
     MEASURES,
     MEDIUMBLOB,
     MEDIUMINT,
@@ -554,6 +562,7 @@ define_keywords!(
     MINUTE,
     MINUTES,
     MINVALUE,
+    MIN_ROWS,
     MOD,
     MODE,
     MODIFIES,
@@ -651,6 +660,7 @@ define_keywords!(
     OWNERSHIP,
     PACKAGE,
     PACKAGES,
+    PACK_KEYS,
     PARALLEL,
     PARAMETER,
     PARQUET,
@@ -773,6 +783,7 @@ define_keywords!(
     ROW,
     ROWID,
     ROWS,
+    ROW_FORMAT,
     ROW_NUMBER,
     RULE,
     RUN,
@@ -787,6 +798,7 @@ define_keywords!(
     SEARCH,
     SECOND,
     SECONDARY,
+    SECONDARY_ENGINE_ATTRIBUTE,
     SECONDS,
     SECRET,
     SECURITY,
@@ -838,12 +850,16 @@ define_keywords!(
     STATEMENT,
     STATIC,
     STATISTICS,
+    STATS_AUTO_RECALC,
+    STATS_PERSISTENT,
+    STATS_SAMPLE_PAGES,
     STATUS,
     STDDEV_POP,
     STDDEV_SAMP,
     STDIN,
     STDOUT,
     STEP,
+    STORAGE,
     STORAGE_INTEGRATION,
     STORAGE_SERIALIZATION_POLICY,
     STORED,
@@ -870,6 +886,7 @@ define_keywords!(
     TABLE,
     TABLES,
     TABLESAMPLE,
+    TABLESPACE,
     TAG,
     TARGET,
     TASK,

src/parser/mod.rs

@@ -5524,12 +5524,17 @@ impl<'a> Parser<'a> {
         };
         let location = hive_formats.location.clone();
         let table_properties = self.parse_options(Keyword::TBLPROPERTIES)?;
+        let table_options = if !table_properties.is_empty() {
+            CreateTableOptions::TableProperties(table_properties)
+        } else {
+            CreateTableOptions::None
+        };
         Ok(CreateTableBuilder::new(table_name)
             .columns(columns)
             .constraints(constraints)
             .hive_distribution(hive_distribution)
             .hive_formats(Some(hive_formats))
-            .table_properties(table_properties)
+            .table_options(table_options)
             .or_replace(or_replace)
             .if_not_exists(if_not_exists)
             .external(true)
@@ -7041,17 +7046,16 @@ impl<'a> Parser<'a> {
 
         // parse optional column list (schema)
         let (columns, constraints) = self.parse_columns()?;
-        let mut comment = if dialect_of!(self is HiveDialect)
+        let comment_after_column_def =
-            && self.parse_keyword(Keyword::COMMENT)
+            if dialect_of!(self is HiveDialect) && self.parse_keyword(Keyword::COMMENT) {
-        {
+                let next_token = self.next_token();
-            let next_token = self.next_token();
+                match next_token.token {
-            match next_token.token {
+                    Token::SingleQuotedString(str) => Some(CommentDef::WithoutEq(str)),
-                Token::SingleQuotedString(str) => Some(CommentDef::AfterColumnDefsWithoutEq(str)),
+                    _ => self.expected("comment", next_token)?,
-                _ => self.expected("comment", next_token)?,
+                }
-            }
+            } else {
-        } else {
+                None
-            None
+            };
-        };
 
         // SQLite supports `WITHOUT ROWID` at the end of `CREATE TABLE`
         let without_rowid = self.parse_keywords(&[Keyword::WITHOUT, Keyword::ROWID]);
@@ -7059,39 +7063,8 @@ impl<'a> Parser<'a> {
         let hive_distribution = self.parse_hive_distribution()?;
         let clustered_by = self.parse_optional_clustered_by()?;
         let hive_formats = self.parse_hive_formats()?;
-        // PostgreSQL supports `WITH ( options )`, before `AS`
-        let with_options = self.parse_options(Keyword::WITH)?;
-        let table_properties = self.parse_options(Keyword::TBLPROPERTIES)?;
 
-        let engine = if self.parse_keyword(Keyword::ENGINE) {
+        let create_table_config = self.parse_optional_create_table_config()?;
-            self.expect_token(&Token::Eq)?;
-            let next_token = self.next_token();
-            match next_token.token {
-                Token::Word(w) => {
-                    let name = w.value;
-                    let parameters = if self.peek_token() == Token::LParen {
-                        Some(self.parse_parenthesized_identifiers()?)
-                    } else {
-                        None
-                    };
-                    Some(TableEngine { name, parameters })
-                }
-                _ => self.expected("identifier", next_token)?,
-            }
-        } else {
-            None
-        };
-
-        let auto_increment_offset = if self.parse_keyword(Keyword::AUTO_INCREMENT) {
-            let _ = self.consume_token(&Token::Eq);
-            let next_token = self.next_token();
-            match next_token.token {
-                Token::Number(s, _) => Some(Self::parse::<u32>(s, next_token.span.start)?),
-                _ => self.expected("literal int", next_token)?,
-            }
-        } else {
-            None
-        };
 
         // ClickHouse supports `PRIMARY KEY`, before `ORDER BY`
         // https://clickhouse.com/docs/en/sql-reference/statements/create/table#primary-key
@@ -7119,30 +7092,6 @@ impl<'a> Parser<'a> {
             None
         };
 
-        let create_table_config = self.parse_optional_create_table_config()?;
-
-        let default_charset = if self.parse_keywords(&[Keyword::DEFAULT, Keyword::CHARSET]) {
-            self.expect_token(&Token::Eq)?;
-            let next_token = self.next_token();
-            match next_token.token {
-                Token::Word(w) => Some(w.value),
-                _ => self.expected("identifier", next_token)?,
-            }
-        } else {
-            None
-        };
-
-        let collation = if self.parse_keywords(&[Keyword::COLLATE]) {
-            self.expect_token(&Token::Eq)?;
-            let next_token = self.next_token();
-            match next_token.token {
-                Token::Word(w) => Some(w.value),
-                _ => self.expected("identifier", next_token)?,
-            }
-        } else {
-            None
-        };
-
         let on_commit = if self.parse_keywords(&[Keyword::ON, Keyword::COMMIT]) {
             Some(self.parse_create_table_on_commit()?)
         } else {
@@ -7151,13 +7100,6 @@ impl<'a> Parser<'a> {
 
         let strict = self.parse_keyword(Keyword::STRICT);
 
-        // Excludes Hive dialect here since it has been handled after table column definitions.
-        if !dialect_of!(self is HiveDialect) && self.parse_keyword(Keyword::COMMENT) {
-            // rewind the COMMENT keyword
-            self.prev_token();
-            comment = self.parse_optional_inline_comment()?
-        };
-
         // Parse optional `AS ( query )`
         let query = if self.parse_keyword(Keyword::AS) {
             Some(self.parse_query()?)
@@ -7174,8 +7116,6 @@ impl<'a> Parser<'a> {
             .temporary(temporary)
             .columns(columns)
             .constraints(constraints)
-            .with_options(with_options)
-            .table_properties(table_properties)
             .or_replace(or_replace)
             .if_not_exists(if_not_exists)
             .transient(transient)
@@ -7186,19 +7126,15 @@ impl<'a> Parser<'a> {
             .without_rowid(without_rowid)
             .like(like)
             .clone_clause(clone)
-            .engine(engine)
+            .comment_after_column_def(comment_after_column_def)
-            .comment(comment)
-            .auto_increment_offset(auto_increment_offset)
             .order_by(order_by)
-            .default_charset(default_charset)
-            .collation(collation)
             .on_commit(on_commit)
             .on_cluster(on_cluster)
             .clustered_by(clustered_by)
             .partition_by(create_table_config.partition_by)
             .cluster_by(create_table_config.cluster_by)
-            .options(create_table_config.options)
             .inherits(create_table_config.inherits)
+            .table_options(create_table_config.table_options)
             .primary_key(primary_key)
             .strict(strict)
             .build())
@@ -7222,17 +7158,29 @@ impl<'a> Parser<'a> {
     /// Parse configuration like inheritance, partitioning, clustering information during the table creation.
     ///
     /// [BigQuery](https://cloud.google.com/bigquery/docs/reference/standard-sql/data-definition-language#syntax_2)
-    /// [PostgreSQL Partitioning](https://www.postgresql.org/docs/current/ddl-partitioning.html)
+    /// [PostgreSQL](https://www.postgresql.org/docs/current/ddl-partitioning.html)
-    /// [PostgreSQL Inheritance](https://www.postgresql.org/docs/current/ddl-inherit.html)
+    /// [MySql](https://dev.mysql.com/doc/refman/8.4/en/create-table.html)
     fn parse_optional_create_table_config(
         &mut self,
     ) -> Result<CreateTableConfiguration, ParserError> {
+        let mut table_options = CreateTableOptions::None;
+
         let inherits = if self.parse_keyword(Keyword::INHERITS) {
             Some(self.parse_parenthesized_qualified_column_list(IsOptional::Mandatory, false)?)
         } else {
             None
         };
 
+        // PostgreSQL supports `WITH ( options )`, before `AS`
+        let with_options = self.parse_options(Keyword::WITH)?;
+        if !with_options.is_empty() {
+            table_options = CreateTableOptions::With(with_options)
+        }
+
+        let table_properties = self.parse_options(Keyword::TBLPROPERTIES)?;
+        if !table_properties.is_empty() {
+            table_options = CreateTableOptions::TableProperties(table_properties);
+        }
         let partition_by = if dialect_of!(self is BigQueryDialect | PostgreSqlDialect | GenericDialect)
             && self.parse_keywords(&[Keyword::PARTITION, Keyword::BY])
         {
@@ -7242,7 +7190,6 @@ impl<'a> Parser<'a> {
         };
 
         let mut cluster_by = None;
-        let mut options = None;
         if dialect_of!(self is BigQueryDialect | GenericDialect) {
             if self.parse_keywords(&[Keyword::CLUSTER, Keyword::BY]) {
                 cluster_by = Some(WrappedCollection::NoWrapping(
@@ -7252,19 +7199,230 @@ impl<'a> Parser<'a> {
 
             if let Token::Word(word) = self.peek_token().token {
                 if word.keyword == Keyword::OPTIONS {
-                    options = Some(self.parse_options(Keyword::OPTIONS)?);
+                    table_options =
+                        CreateTableOptions::Options(self.parse_options(Keyword::OPTIONS)?)
                 }
             };
         }
 
+        if !dialect_of!(self is HiveDialect) && table_options == CreateTableOptions::None {
+            let plain_options = self.parse_plain_options()?;
+            if !plain_options.is_empty() {
+                table_options = CreateTableOptions::Plain(plain_options)
+            }
+        };
+
         Ok(CreateTableConfiguration {
             partition_by,
             cluster_by,
-            options,
             inherits,
+            table_options,
         })
     }
 
+    fn parse_plain_option(&mut self) -> Result<Option<SqlOption>, ParserError> {
+        // Single parameter option
+        // <https://dev.mysql.com/doc/refman/8.4/en/create-table.html>
+        if self.parse_keywords(&[Keyword::START, Keyword::TRANSACTION]) {
+            return Ok(Some(SqlOption::Ident(Ident::new("START TRANSACTION"))));
+        }
+
+        // Custom option
+        // <https://dev.mysql.com/doc/refman/8.4/en/create-table.html>
+        if self.parse_keywords(&[Keyword::COMMENT]) {
+            let has_eq = self.consume_token(&Token::Eq);
+            let value = self.next_token();
+
+            let comment = match (has_eq, value.token) {
+                (true, Token::SingleQuotedString(s)) => {
+                    Ok(Some(SqlOption::Comment(CommentDef::WithEq(s))))
+                }
+                (false, Token::SingleQuotedString(s)) => {
+                    Ok(Some(SqlOption::Comment(CommentDef::WithoutEq(s))))
+                }
+                (_, token) => {
+                    self.expected("Token::SingleQuotedString", TokenWithSpan::wrap(token))
+                }
+            };
+            return comment;
+        }
+
+        // <https://dev.mysql.com/doc/refman/8.4/en/create-table.html>
+        // <https://clickhouse.com/docs/sql-reference/statements/create/table>
+        if self.parse_keywords(&[Keyword::ENGINE]) {
+            let _ = self.consume_token(&Token::Eq);
+            let value = self.next_token();
+
+            let engine = match value.token {
+                Token::Word(w) => {
+                    let parameters = if self.peek_token() == Token::LParen {
+                        self.parse_parenthesized_identifiers()?
+                    } else {
+                        vec![]
+                    };
+
+                    Ok(Some(SqlOption::NamedParenthesizedList(
+                        NamedParenthesizedList {
+                            key: Ident::new("ENGINE"),
+                            name: Some(Ident::new(w.value)),
+                            values: parameters,
+                        },
+                    )))
+                }
+                _ => {
+                    return self.expected("Token::Word", value)?;
+                }
+            };
+
+            return engine;
+        }
+
+        // <https://dev.mysql.com/doc/refman/8.4/en/create-table.html>
+        if self.parse_keywords(&[Keyword::TABLESPACE]) {
+            let _ = self.consume_token(&Token::Eq);
+            let value = self.next_token();
+
+            let tablespace = match value.token {
+                Token::Word(Word { value: name, .. }) | Token::SingleQuotedString(name) => {
+                    let storage = match self.parse_keyword(Keyword::STORAGE) {
+                        true => {
+                            let _ = self.consume_token(&Token::Eq);
+                            let storage_token = self.next_token();
+                            match &storage_token.token {
+                                Token::Word(w) => match w.value.to_uppercase().as_str() {
+                                    "DISK" => Some(StorageType::Disk),
+                                    "MEMORY" => Some(StorageType::Memory),
+                                    _ => self
+                                        .expected("Storage type (DISK or MEMORY)", storage_token)?,
+                                },
+                                _ => self.expected("Token::Word", storage_token)?,
+                            }
+                        }
+                        false => None,
+                    };
+
+                    Ok(Some(SqlOption::TableSpace(TablespaceOption {
+                        name,
+                        storage,
+                    })))
+                }
+                _ => {
+                    return self.expected("Token::Word", value)?;
+                }
+            };
+
+            return tablespace;
+        }
+
+        // <https://dev.mysql.com/doc/refman/8.4/en/create-table.html>
+        if self.parse_keyword(Keyword::UNION) {
+            let _ = self.consume_token(&Token::Eq);
+            let value = self.next_token();
+
+            match value.token {
+                Token::LParen => {
+                    let tables: Vec<Ident> =
+                        self.parse_comma_separated0(Parser::parse_identifier, Token::RParen)?;
+                    self.expect_token(&Token::RParen)?;
+
+                    return Ok(Some(SqlOption::NamedParenthesizedList(
+                        NamedParenthesizedList {
+                            key: Ident::new("UNION"),
+                            name: None,
+                            values: tables,
+                        },
+                    )));
+                }
+                _ => {
+                    return self.expected("Token::LParen", value)?;
+                }
+            }
+        }
+
+        // Key/Value parameter option
+        let key = if self.parse_keywords(&[Keyword::DEFAULT, Keyword::CHARSET]) {
+            Ident::new("DEFAULT CHARSET")
+        } else if self.parse_keyword(Keyword::CHARSET) {
+            Ident::new("CHARSET")
+        } else if self.parse_keywords(&[Keyword::DEFAULT, Keyword::CHARACTER, Keyword::SET]) {
+            Ident::new("DEFAULT CHARACTER SET")
+        } else if self.parse_keywords(&[Keyword::CHARACTER, Keyword::SET]) {
+            Ident::new("CHARACTER SET")
+        } else if self.parse_keywords(&[Keyword::DEFAULT, Keyword::COLLATE]) {
+            Ident::new("DEFAULT COLLATE")
+        } else if self.parse_keyword(Keyword::COLLATE) {
+            Ident::new("COLLATE")
+        } else if self.parse_keywords(&[Keyword::DATA, Keyword::DIRECTORY]) {
+            Ident::new("DATA DIRECTORY")
+        } else if self.parse_keywords(&[Keyword::INDEX, Keyword::DIRECTORY]) {
+            Ident::new("INDEX DIRECTORY")
+        } else if self.parse_keyword(Keyword::KEY_BLOCK_SIZE) {
+            Ident::new("KEY_BLOCK_SIZE")
+        } else if self.parse_keyword(Keyword::ROW_FORMAT) {
+            Ident::new("ROW_FORMAT")
+        } else if self.parse_keyword(Keyword::PACK_KEYS) {
+            Ident::new("PACK_KEYS")
+        } else if self.parse_keyword(Keyword::STATS_AUTO_RECALC) {
+            Ident::new("STATS_AUTO_RECALC")
+        } else if self.parse_keyword(Keyword::STATS_PERSISTENT) {
+            Ident::new("STATS_PERSISTENT")
+        } else if self.parse_keyword(Keyword::STATS_SAMPLE_PAGES) {
+            Ident::new("STATS_SAMPLE_PAGES")
+        } else if self.parse_keyword(Keyword::DELAY_KEY_WRITE) {
+            Ident::new("DELAY_KEY_WRITE")
+        } else if self.parse_keyword(Keyword::COMPRESSION) {
+            Ident::new("COMPRESSION")
+        } else if self.parse_keyword(Keyword::ENCRYPTION) {
+            Ident::new("ENCRYPTION")
+        } else if self.parse_keyword(Keyword::MAX_ROWS) {
+            Ident::new("MAX_ROWS")
+        } else if self.parse_keyword(Keyword::MIN_ROWS) {
+            Ident::new("MIN_ROWS")
+        } else if self.parse_keyword(Keyword::AUTOEXTEND_SIZE) {
+            Ident::new("AUTOEXTEND_SIZE")
+        } else if self.parse_keyword(Keyword::AVG_ROW_LENGTH) {
+            Ident::new("AVG_ROW_LENGTH")
+        } else if self.parse_keyword(Keyword::CHECKSUM) {
+            Ident::new("CHECKSUM")
+        } else if self.parse_keyword(Keyword::CONNECTION) {
+            Ident::new("CONNECTION")
+        } else if self.parse_keyword(Keyword::ENGINE_ATTRIBUTE) {
+            Ident::new("ENGINE_ATTRIBUTE")
+        } else if self.parse_keyword(Keyword::PASSWORD) {
+            Ident::new("PASSWORD")
+        } else if self.parse_keyword(Keyword::SECONDARY_ENGINE_ATTRIBUTE) {
+            Ident::new("SECONDARY_ENGINE_ATTRIBUTE")
+        } else if self.parse_keyword(Keyword::INSERT_METHOD) {
+            Ident::new("INSERT_METHOD")
+        } else if self.parse_keyword(Keyword::AUTO_INCREMENT) {
+            Ident::new("AUTO_INCREMENT")
+        } else {
+            return Ok(None);
+        };
+
+        let _ = self.consume_token(&Token::Eq);
+
+        let value = match self
+            .maybe_parse(|parser| parser.parse_value())?
+            .map(Expr::Value)
+        {
+            Some(expr) => expr,
+            None => Expr::Identifier(self.parse_identifier()?),
+        };
+
+        Ok(Some(SqlOption::KeyValue { key, value }))
+    }
+
+    pub fn parse_plain_options(&mut self) -> Result<Vec<SqlOption>, ParserError> {
+        let mut options = Vec::new();
+
+        while let Some(option) = self.parse_plain_option()? {
+            options.push(option);
+        }
+
+        Ok(options)
+    }
+
     pub fn parse_optional_inline_comment(&mut self) -> Result<Option<CommentDef>, ParserError> {
         let comment = if self.parse_keyword(Keyword::COMMENT) {
             let has_eq = self.consume_token(&Token::Eq);

tests/sqlparser_bigquery.rs

@@ -484,7 +484,7 @@ fn parse_create_table_with_options() {
             columns,
             partition_by,
             cluster_by,
-            options,
+            table_options,
             ..
         }) => {
             assert_eq!(
@@ -539,7 +539,7 @@ fn parse_create_table_with_options() {
                         Ident::new("userid"),
                         Ident::new("age"),
                     ])),
-                    Some(vec![
+                    CreateTableOptions::Options(vec![
                         SqlOption::KeyValue {
                             key: Ident::new("partition_expiration_days"),
                             value: Expr::Value(
@@ -561,7 +561,7 @@ fn parse_create_table_with_options() {
                         },
                     ])
                 ),
-                (partition_by, cluster_by, options)
+                (partition_by, cluster_by, table_options)
             )
         }
         _ => unreachable!(),

tests/sqlparser_clickhouse.rs

@@ -219,10 +219,10 @@ fn parse_delimited_identifiers() {
 
 #[test]
 fn parse_create_table() {
-    clickhouse().verified_stmt(r#"CREATE TABLE "x" ("a" "int") ENGINE=MergeTree ORDER BY ("x")"#);
+    clickhouse().verified_stmt(r#"CREATE TABLE "x" ("a" "int") ENGINE = MergeTree ORDER BY ("x")"#);
-    clickhouse().verified_stmt(r#"CREATE TABLE "x" ("a" "int") ENGINE=MergeTree ORDER BY "x""#);
+    clickhouse().verified_stmt(r#"CREATE TABLE "x" ("a" "int") ENGINE = MergeTree ORDER BY "x""#);
     clickhouse().verified_stmt(
-        r#"CREATE TABLE "x" ("a" "int") ENGINE=MergeTree ORDER BY "x" AS SELECT * FROM "t" WHERE true"#,
+        r#"CREATE TABLE "x" ("a" "int") ENGINE = MergeTree ORDER BY "x" AS SELECT * FROM "t" WHERE true"#,
     );
 }
 
@@ -589,7 +589,7 @@ fn parse_clickhouse_data_types() {
 
 #[test]
 fn parse_create_table_with_nullable() {
-    let sql = r#"CREATE TABLE table (k UInt8, `a` Nullable(String), `b` Nullable(DateTime64(9, 'UTC')), c Nullable(DateTime64(9)), d Date32 NULL) ENGINE=MergeTree ORDER BY (`k`)"#;
+    let sql = r#"CREATE TABLE table (k UInt8, `a` Nullable(String), `b` Nullable(DateTime64(9, 'UTC')), c Nullable(DateTime64(9)), d Date32 NULL) ENGINE = MergeTree ORDER BY (`k`)"#;
     // ClickHouse has a case-sensitive definition of data type, but canonical representation is not
     let canonical_sql = sql.replace("String", "STRING");
 
@@ -714,14 +714,14 @@ fn parse_create_table_with_nested_data_types() {
 fn parse_create_table_with_primary_key() {
     match clickhouse_and_generic().verified_stmt(concat!(
         r#"CREATE TABLE db.table (`i` INT, `k` INT)"#,
-        " ENGINE=SharedMergeTree('/clickhouse/tables/{uuid}/{shard}', '{replica}')",
+        " ENGINE = SharedMergeTree('/clickhouse/tables/{uuid}/{shard}', '{replica}')",
         " PRIMARY KEY tuple(i)",
         " ORDER BY tuple(i)",
     )) {
         Statement::CreateTable(CreateTable {
             name,
             columns,
-            engine,
+            table_options,
             primary_key,
             order_by,
             ..
@@ -742,16 +742,23 @@ fn parse_create_table_with_primary_key() {
                 ],
                 columns
             );
-            assert_eq!(
+
-                engine,
+            let plain_options = match table_options {
-                Some(TableEngine {
+                CreateTableOptions::Plain(options) => options,
-                    name: "SharedMergeTree".to_string(),
+                _ => unreachable!(),
-                    parameters: Some(vec![
+            };
+
+            assert!(plain_options.contains(&SqlOption::NamedParenthesizedList(
+                NamedParenthesizedList {
+                    key: Ident::new("ENGINE"),
+                    name: Some(Ident::new("SharedMergeTree")),
+                    values: vec![
                         Ident::with_quote('\'', "/clickhouse/tables/{uuid}/{shard}"),
                         Ident::with_quote('\'', "{replica}"),
-                    ]),
+                    ]
-                })
+                }
-            );
+            )));
+
             fn assert_function(actual: &Function, name: &str, arg: &str) -> bool {
                 assert_eq!(actual.name, ObjectName::from(vec![Ident::new(name)]));
                 assert_eq!(
@@ -798,7 +805,7 @@ fn parse_create_table_with_variant_default_expressions() {
         " b DATETIME EPHEMERAL now(),",
         " c DATETIME EPHEMERAL,",
         " d STRING ALIAS toString(c)",
-        ") ENGINE=MergeTree"
+        ") ENGINE = MergeTree"
     );
     match clickhouse_and_generic().verified_stmt(sql) {
         Statement::CreateTable(CreateTable { columns, .. }) => {

tests/sqlparser_common.rs

@@ -3657,7 +3657,7 @@ fn parse_create_table() {
             name,
             columns,
             constraints,
-            with_options,
+            table_options,
             if_not_exists: false,
             external: false,
             file_format: None,
@@ -3795,7 +3795,7 @@ fn parse_create_table() {
                     },
                 ]
             );
-            assert_eq!(with_options, vec![]);
+            assert_eq!(table_options, CreateTableOptions::None);
         }
         _ => unreachable!(),
     }
@@ -3840,7 +3840,7 @@ fn parse_create_table_with_constraint_characteristics() {
             name,
             columns,
             constraints,
-            with_options,
+            table_options,
             if_not_exists: false,
             external: false,
             file_format: None,
@@ -3934,7 +3934,7 @@ fn parse_create_table_with_constraint_characteristics() {
                     },
                 ]
             );
-            assert_eq!(with_options, vec![]);
+            assert_eq!(table_options, CreateTableOptions::None);
         }
         _ => unreachable!(),
     }
@@ -4421,7 +4421,11 @@ fn parse_create_table_with_options() {
 
     let sql = "CREATE TABLE t (c INT) WITH (foo = 'bar', a = 123)";
     match generic.verified_stmt(sql) {
-        Statement::CreateTable(CreateTable { with_options, .. }) => {
+        Statement::CreateTable(CreateTable { table_options, .. }) => {
+            let with_options = match table_options {
+                CreateTableOptions::With(options) => options,
+                _ => unreachable!(),
+            };
             assert_eq!(
                 vec![
                     SqlOption::KeyValue {
@@ -4482,7 +4486,7 @@ fn parse_create_external_table() {
             name,
             columns,
             constraints,
-            with_options,
+            table_options,
             if_not_exists,
             external,
             file_format,
@@ -4525,7 +4529,7 @@ fn parse_create_external_table() {
             assert_eq!(FileFormat::TEXTFILE, file_format.unwrap());
             assert_eq!("/tmp/example.csv", location.unwrap());
 
-            assert_eq!(with_options, vec![]);
+            assert_eq!(table_options, CreateTableOptions::None);
             assert!(!if_not_exists);
         }
         _ => unreachable!(),
@@ -4550,7 +4554,7 @@ fn parse_create_or_replace_external_table() {
             name,
             columns,
             constraints,
-            with_options,
+            table_options,
             if_not_exists,
             external,
             file_format,
@@ -4579,7 +4583,7 @@ fn parse_create_or_replace_external_table() {
             assert_eq!(FileFormat::TEXTFILE, file_format.unwrap());
             assert_eq!("/tmp/example.csv", location.unwrap());
 
-            assert_eq!(with_options, vec![]);
+            assert_eq!(table_options, CreateTableOptions::None);
             assert!(!if_not_exists);
             assert!(or_replace);
         }
@@ -11420,7 +11424,9 @@ fn test_parse_inline_comment() {
     // [Hive](https://cwiki.apache.org/confluence/display/Hive/LanguageManual+DDL#LanguageManualDDL-CreateTable)
     match all_dialects_except(|d| d.is::<HiveDialect>()).verified_stmt(sql) {
         Statement::CreateTable(CreateTable {
-            columns, comment, ..
+            columns,
+            table_options,
+            ..
         }) => {
             assert_eq!(
                 columns,
@@ -11434,8 +11440,10 @@ fn test_parse_inline_comment() {
                 }]
             );
             assert_eq!(
-                comment.unwrap(),
+                table_options,
-                CommentDef::WithEq("comment with equal".to_string())
+                CreateTableOptions::Plain(vec![SqlOption::Comment(CommentDef::WithEq(
+                    "comment with equal".to_string()
+                ))])
             );
         }
         _ => unreachable!(),
@@ -12460,21 +12468,6 @@ fn parse_select_wildcard_with_except() {
     );
 }
 
-#[test]
-fn parse_auto_increment_too_large() {
-    let dialect = GenericDialect {};
-    let u64_max = u64::MAX;
-    let sql =
-        format!("CREATE TABLE foo (bar INT NOT NULL AUTO_INCREMENT) AUTO_INCREMENT=1{u64_max}");
-
-    let res = Parser::new(&dialect)
-        .try_with_sql(&sql)
-        .expect("tokenize to work")
-        .parse_statements();
-
-    assert!(res.is_err(), "{res:?}");
-}
-
 #[test]
 fn test_group_by_nothing() {
     let Select { group_by, .. } = all_dialects_where(|d| d.supports_group_by_expr())

tests/sqlparser_duckdb.rs

@@ -735,19 +735,13 @@ fn test_duckdb_union_datatype() {
                 storage: Default::default(),
                 location: Default::default()
             }),
-            table_properties: Default::default(),
-            with_options: Default::default(),
             file_format: Default::default(),
             location: Default::default(),
             query: Default::default(),
             without_rowid: Default::default(),
             like: Default::default(),
             clone: Default::default(),
-            engine: Default::default(),
             comment: Default::default(),
-            auto_increment_offset: Default::default(),
-            default_charset: Default::default(),
-            collation: Default::default(),
             on_commit: Default::default(),
             on_cluster: Default::default(),
             primary_key: Default::default(),
@@ -755,7 +749,6 @@ fn test_duckdb_union_datatype() {
             partition_by: Default::default(),
             cluster_by: Default::default(),
             clustered_by: Default::default(),
-            options: Default::default(),
             inherits: Default::default(),
             strict: Default::default(),
             copy_grants: Default::default(),
@@ -772,6 +765,7 @@ fn test_duckdb_union_datatype() {
             catalog: Default::default(),
             catalog_sync: Default::default(),
             storage_serialization_policy: Default::default(),
+            table_options: CreateTableOptions::None
         }),
         stmt
     );

tests/sqlparser_hive.rs

@@ -133,9 +133,7 @@ fn create_table_with_comment() {
         Statement::CreateTable(CreateTable { comment, .. }) => {
             assert_eq!(
                 comment,
-                Some(CommentDef::AfterColumnDefsWithoutEq(
+                Some(CommentDef::WithoutEq("table comment".to_string()))
-                    "table comment".to_string()
-                ))
             )
         }
         _ => unreachable!(),

tests/sqlparser_mssql.rs

@@ -1725,7 +1725,6 @@ fn parse_create_table_with_valid_options() {
                             span: Span::empty(),
                         },
                         data_type: Int(None,),
-
                         options: vec![],
                     },
                     ColumnDef {
@@ -1735,7 +1734,6 @@ fn parse_create_table_with_valid_options() {
                             span: Span::empty(),
                         },
                         data_type: Int(None,),
-
                         options: vec![],
                     },
                 ],
@@ -1747,19 +1745,13 @@ fn parse_create_table_with_valid_options() {
                     storage: None,
                     location: None,
                 },),
-                table_properties: vec![],
-                with_options,
                 file_format: None,
                 location: None,
                 query: None,
                 without_rowid: false,
                 like: None,
                 clone: None,
-                engine: None,
                 comment: None,
-                auto_increment_offset: None,
-                default_charset: None,
-                collation: None,
                 on_commit: None,
                 on_cluster: None,
                 primary_key: None,
@@ -1767,7 +1759,6 @@ fn parse_create_table_with_valid_options() {
                 partition_by: None,
                 cluster_by: None,
                 clustered_by: None,
-                options: None,
                 inherits: None,
                 strict: false,
                 iceberg: false,
@@ -1785,6 +1776,7 @@ fn parse_create_table_with_valid_options() {
                 catalog: None,
                 catalog_sync: None,
                 storage_serialization_policy: None,
+                table_options: CreateTableOptions::With(with_options)
             })
         );
     }
@@ -1918,19 +1910,13 @@ fn parse_create_table_with_identity_column() {
                     storage: None,
                     location: None,
                 },),
-                table_properties: vec![],
-                with_options: vec![],
                 file_format: None,
                 location: None,
                 query: None,
                 without_rowid: false,
                 like: None,
                 clone: None,
-                engine: None,
                 comment: None,
-                auto_increment_offset: None,
-                default_charset: None,
-                collation: None,
                 on_commit: None,
                 on_cluster: None,
                 primary_key: None,
@@ -1938,7 +1924,6 @@ fn parse_create_table_with_identity_column() {
                 partition_by: None,
                 cluster_by: None,
                 clustered_by: None,
-                options: None,
                 inherits: None,
                 strict: false,
                 copy_grants: false,
@@ -1955,6 +1940,7 @@ fn parse_create_table_with_identity_column() {
                 catalog: None,
                 catalog_sync: None,
                 storage_serialization_policy: None,
+                table_options: CreateTableOptions::None
             }),
         );
     }

tests/sqlparser_mysql.rs

@@ -848,9 +848,23 @@ fn parse_create_table_comment() {
 
     for sql in [without_equal, with_equal] {
         match mysql().verified_stmt(sql) {
-            Statement::CreateTable(CreateTable { name, comment, .. }) => {
+            Statement::CreateTable(CreateTable {
+                name,
+                table_options,
+                ..
+            }) => {
                 assert_eq!(name.to_string(), "foo");
-                assert_eq!(comment.expect("Should exist").to_string(), "baz");
+
+                let plain_options = match table_options {
+                    CreateTableOptions::Plain(options) => options,
+                    _ => unreachable!(),
+                };
+                let comment = match plain_options.first().unwrap() {
+                    SqlOption::Comment(CommentDef::WithEq(c))
+                    | SqlOption::Comment(CommentDef::WithoutEq(c)) => c,
+                    _ => unreachable!(),
+                };
+                assert_eq!(comment, "baz");
             }
             _ => unreachable!(),
         }
@@ -859,29 +873,226 @@ fn parse_create_table_comment() {
 
 #[test]
 fn parse_create_table_auto_increment_offset() {
-    let canonical =
+    let sql =
-        "CREATE TABLE foo (bar INT NOT NULL AUTO_INCREMENT) ENGINE=InnoDB AUTO_INCREMENT 123";
+        "CREATE TABLE foo (bar INT NOT NULL AUTO_INCREMENT) ENGINE = InnoDB AUTO_INCREMENT = 123";
-    let with_equal =
-        "CREATE TABLE foo (bar INT NOT NULL AUTO_INCREMENT) ENGINE=InnoDB AUTO_INCREMENT=123";
 
-    for sql in [canonical, with_equal] {
+    match mysql().verified_stmt(sql) {
-        match mysql().one_statement_parses_to(sql, canonical) {
+        Statement::CreateTable(CreateTable {
+            name,
+            table_options,
+            ..
+        }) => {
+            assert_eq!(name.to_string(), "foo");
+
+            let plain_options = match table_options {
+                CreateTableOptions::Plain(options) => options,
+                _ => unreachable!(),
+            };
+
+            assert!(plain_options.contains(&SqlOption::KeyValue {
+                key: Ident::new("AUTO_INCREMENT"),
+                value: Expr::Value(test_utils::number("123").with_empty_span())
+            }));
+        }
+        _ => unreachable!(),
+    }
+}
+
+#[test]
+fn parse_create_table_multiple_options_order_independent() {
+    let sql1 = "CREATE TABLE mytable (id INT) ENGINE=InnoDB ROW_FORMAT=DYNAMIC KEY_BLOCK_SIZE=8 COMMENT='abc'";
+    let sql2 = "CREATE TABLE mytable (id INT) KEY_BLOCK_SIZE=8 COMMENT='abc' ENGINE=InnoDB ROW_FORMAT=DYNAMIC";
+    let sql3 = "CREATE TABLE mytable (id INT) ROW_FORMAT=DYNAMIC KEY_BLOCK_SIZE=8 COMMENT='abc' ENGINE=InnoDB";
+
+    for sql in [sql1, sql2, sql3] {
+        match mysql().parse_sql_statements(sql).unwrap().pop().unwrap() {
             Statement::CreateTable(CreateTable {
                 name,
-                auto_increment_offset,
+                table_options,
                 ..
             }) => {
-                assert_eq!(name.to_string(), "foo");
+                assert_eq!(name.to_string(), "mytable");
-                assert_eq!(
+
-                    auto_increment_offset.expect("Should exist").to_string(),
+                let plain_options = match table_options {
-                    "123"
+                    CreateTableOptions::Plain(options) => options,
-                );
+                    _ => unreachable!(),
+                };
+
+                assert!(plain_options.contains(&SqlOption::NamedParenthesizedList(
+                    NamedParenthesizedList {
+                        key: Ident::new("ENGINE"),
+                        name: Some(Ident::new("InnoDB")),
+                        values: vec![]
+                    }
+                )));
+
+                assert!(plain_options.contains(&SqlOption::KeyValue {
+                    key: Ident::new("KEY_BLOCK_SIZE"),
+                    value: Expr::Value(test_utils::number("8").with_empty_span())
+                }));
+
+                assert!(plain_options
+                    .contains(&SqlOption::Comment(CommentDef::WithEq("abc".to_owned()))));
+
+                assert!(plain_options.contains(&SqlOption::KeyValue {
+                    key: Ident::new("ROW_FORMAT"),
+                    value: Expr::Identifier(Ident::new("DYNAMIC".to_owned()))
+                }));
             }
             _ => unreachable!(),
         }
     }
 }
 
+#[test]
+fn parse_create_table_with_all_table_options() {
+    let sql =
+        "CREATE TABLE foo (bar INT NOT NULL AUTO_INCREMENT) ENGINE = InnoDB AUTO_INCREMENT = 123 DEFAULT CHARSET = utf8mb4 COLLATE = utf8mb4_0900_ai_ci INSERT_METHOD = FIRST KEY_BLOCK_SIZE = 8 ROW_FORMAT = DYNAMIC DATA DIRECTORY = '/var/lib/mysql/data' INDEX DIRECTORY = '/var/lib/mysql/index' PACK_KEYS = 1 STATS_AUTO_RECALC = 1 STATS_PERSISTENT = 0 STATS_SAMPLE_PAGES = 128 DELAY_KEY_WRITE = 1 COMPRESSION = 'ZLIB' ENCRYPTION = 'Y' MAX_ROWS = 10000 MIN_ROWS = 10 AUTOEXTEND_SIZE = 64 AVG_ROW_LENGTH = 128 CHECKSUM = 1 CONNECTION = 'mysql://localhost' ENGINE_ATTRIBUTE = 'primary' PASSWORD = 'secure_password' SECONDARY_ENGINE_ATTRIBUTE = 'secondary_attr' START TRANSACTION TABLESPACE my_tablespace STORAGE DISK UNION = (table1, table2, table3)";
+
+    match mysql().verified_stmt(sql) {
+        Statement::CreateTable(CreateTable {
+            name,
+            table_options,
+            ..
+        }) => {
+            assert_eq!(name, vec![Ident::new("foo".to_owned())].into());
+
+            let plain_options = match table_options {
+                CreateTableOptions::Plain(options) => options,
+                _ => unreachable!(),
+            };
+
+            assert!(plain_options.contains(&SqlOption::NamedParenthesizedList(
+                NamedParenthesizedList {
+                    key: Ident::new("ENGINE"),
+                    name: Some(Ident::new("InnoDB")),
+                    values: vec![]
+                }
+            )));
+
+            assert!(plain_options.contains(&SqlOption::KeyValue {
+                key: Ident::new("COLLATE"),
+                value: Expr::Identifier(Ident::new("utf8mb4_0900_ai_ci".to_owned()))
+            }));
+            assert!(plain_options.contains(&SqlOption::KeyValue {
+                key: Ident::new("DEFAULT CHARSET"),
+                value: Expr::Identifier(Ident::new("utf8mb4".to_owned()))
+            }));
+            assert!(plain_options.contains(&SqlOption::KeyValue {
+                key: Ident::new("AUTO_INCREMENT"),
+                value: Expr::value(test_utils::number("123"))
+            }));
+            assert!(plain_options.contains(&SqlOption::KeyValue {
+                key: Ident::new("KEY_BLOCK_SIZE"),
+                value: Expr::value(test_utils::number("8"))
+            }));
+            assert!(plain_options.contains(&SqlOption::KeyValue {
+                key: Ident::new("ROW_FORMAT"),
+                value: Expr::Identifier(Ident::new("DYNAMIC".to_owned()))
+            }));
+            assert!(plain_options.contains(&SqlOption::KeyValue {
+                key: Ident::new("PACK_KEYS"),
+                value: Expr::value(test_utils::number("1"))
+            }));
+            assert!(plain_options.contains(&SqlOption::KeyValue {
+                key: Ident::new("STATS_AUTO_RECALC"),
+                value: Expr::value(test_utils::number("1"))
+            }));
+            assert!(plain_options.contains(&SqlOption::KeyValue {
+                key: Ident::new("STATS_PERSISTENT"),
+                value: Expr::value(test_utils::number("0"))
+            }));
+            assert!(plain_options.contains(&SqlOption::KeyValue {
+                key: Ident::new("STATS_SAMPLE_PAGES"),
+                value: Expr::value(test_utils::number("128"))
+            }));
+            assert!(plain_options.contains(&SqlOption::KeyValue {
+                key: Ident::new("STATS_SAMPLE_PAGES"),
+                value: Expr::value(test_utils::number("128"))
+            }));
+            assert!(plain_options.contains(&SqlOption::KeyValue {
+                key: Ident::new("INSERT_METHOD"),
+                value: Expr::Identifier(Ident::new("FIRST".to_owned()))
+            }));
+            assert!(plain_options.contains(&SqlOption::KeyValue {
+                key: Ident::new("COMPRESSION"),
+                value: Expr::value(Value::SingleQuotedString("ZLIB".to_owned()))
+            }));
+            assert!(plain_options.contains(&SqlOption::KeyValue {
+                key: Ident::new("ENCRYPTION"),
+                value: Expr::value(Value::SingleQuotedString("Y".to_owned()))
+            }));
+            assert!(plain_options.contains(&SqlOption::KeyValue {
+                key: Ident::new("MAX_ROWS"),
+                value: Expr::value(test_utils::number("10000"))
+            }));
+            assert!(plain_options.contains(&SqlOption::KeyValue {
+                key: Ident::new("MIN_ROWS"),
+                value: Expr::value(test_utils::number("10"))
+            }));
+            assert!(plain_options.contains(&SqlOption::KeyValue {
+                key: Ident::new("AUTOEXTEND_SIZE"),
+                value: Expr::value(test_utils::number("64"))
+            }));
+            assert!(plain_options.contains(&SqlOption::KeyValue {
+                key: Ident::new("AVG_ROW_LENGTH"),
+                value: Expr::value(test_utils::number("128"))
+            }));
+            assert!(plain_options.contains(&SqlOption::KeyValue {
+                key: Ident::new("CHECKSUM"),
+                value: Expr::value(test_utils::number("1"))
+            }));
+            assert!(plain_options.contains(&SqlOption::KeyValue {
+                key: Ident::new("CONNECTION"),
+                value: Expr::value(Value::SingleQuotedString("mysql://localhost".to_owned()))
+            }));
+            assert!(plain_options.contains(&SqlOption::KeyValue {
+                key: Ident::new("ENGINE_ATTRIBUTE"),
+                value: Expr::value(Value::SingleQuotedString("primary".to_owned()))
+            }));
+            assert!(plain_options.contains(&SqlOption::KeyValue {
+                key: Ident::new("PASSWORD"),
+                value: Expr::value(Value::SingleQuotedString("secure_password".to_owned()))
+            }));
+            assert!(plain_options.contains(&SqlOption::KeyValue {
+                key: Ident::new("SECONDARY_ENGINE_ATTRIBUTE"),
+                value: Expr::value(Value::SingleQuotedString("secondary_attr".to_owned()))
+            }));
+            assert!(plain_options.contains(&SqlOption::Ident(Ident::new(
+                "START TRANSACTION".to_owned()
+            ))));
+            assert!(
+                plain_options.contains(&SqlOption::TableSpace(TablespaceOption {
+                    name: "my_tablespace".to_string(),
+                    storage: Some(StorageType::Disk),
+                }))
+            );
+
+            assert!(plain_options.contains(&SqlOption::NamedParenthesizedList(
+                NamedParenthesizedList {
+                    key: Ident::new("UNION"),
+                    name: None,
+                    values: vec![
+                        Ident::new("table1".to_string()),
+                        Ident::new("table2".to_string()),
+                        Ident::new("table3".to_string())
+                    ]
+                }
+            )));
+
+            assert!(plain_options.contains(&SqlOption::KeyValue {
+                key: Ident::new("DATA DIRECTORY"),
+                value: Expr::value(Value::SingleQuotedString("/var/lib/mysql/data".to_owned()))
+            }));
+            assert!(plain_options.contains(&SqlOption::KeyValue {
+                key: Ident::new("INDEX DIRECTORY"),
+                value: Expr::value(Value::SingleQuotedString("/var/lib/mysql/index".to_owned()))
+            }));
+        }
+        _ => unreachable!(),
+    }
+}
+
 #[test]
 fn parse_create_table_set_enum() {
     let sql = "CREATE TABLE foo (bar SET('a', 'b'), baz ENUM('a', 'b'))";
@@ -916,13 +1127,12 @@ fn parse_create_table_set_enum() {
 
 #[test]
 fn parse_create_table_engine_default_charset() {
-    let sql = "CREATE TABLE foo (id INT(11)) ENGINE=InnoDB DEFAULT CHARSET=utf8mb3";
+    let sql = "CREATE TABLE foo (id INT(11)) ENGINE = InnoDB DEFAULT CHARSET = utf8mb3";
     match mysql().verified_stmt(sql) {
         Statement::CreateTable(CreateTable {
             name,
             columns,
-            engine,
+            table_options,
-            default_charset,
             ..
         }) => {
             assert_eq!(name.to_string(), "foo");
@@ -934,14 +1144,24 @@ fn parse_create_table_engine_default_charset() {
                 },],
                 columns
             );
-            assert_eq!(
+
-                engine,
+            let plain_options = match table_options {
-                Some(TableEngine {
+                CreateTableOptions::Plain(options) => options,
-                    name: "InnoDB".to_string(),
+                _ => unreachable!(),
-                    parameters: None
+            };
-                })
+
-            );
+            assert!(plain_options.contains(&SqlOption::KeyValue {
-            assert_eq!(default_charset, Some("utf8mb3".to_string()));
+                key: Ident::new("DEFAULT CHARSET"),
+                value: Expr::Identifier(Ident::new("utf8mb3".to_owned()))
+            }));
+
+            assert!(plain_options.contains(&SqlOption::NamedParenthesizedList(
+                NamedParenthesizedList {
+                    key: Ident::new("ENGINE"),
+                    name: Some(Ident::new("InnoDB")),
+                    values: vec![]
+                }
+            )));
         }
         _ => unreachable!(),
     }
@@ -949,12 +1169,12 @@ fn parse_create_table_engine_default_charset() {
 
 #[test]
 fn parse_create_table_collate() {
-    let sql = "CREATE TABLE foo (id INT(11)) COLLATE=utf8mb4_0900_ai_ci";
+    let sql = "CREATE TABLE foo (id INT(11)) COLLATE = utf8mb4_0900_ai_ci";
     match mysql().verified_stmt(sql) {
         Statement::CreateTable(CreateTable {
             name,
             columns,
-            collation,
+            table_options,
             ..
         }) => {
             assert_eq!(name.to_string(), "foo");
@@ -966,7 +1186,16 @@ fn parse_create_table_collate() {
                 },],
                 columns
             );
-            assert_eq!(collation, Some("utf8mb4_0900_ai_ci".to_string()));
+
+            let plain_options = match table_options {
+                CreateTableOptions::Plain(options) => options,
+                _ => unreachable!(),
+            };
+
+            assert!(plain_options.contains(&SqlOption::KeyValue {
+                key: Ident::new("COLLATE"),
+                value: Expr::Identifier(Ident::new("utf8mb4_0900_ai_ci".to_owned()))
+            }));
         }
         _ => unreachable!(),
     }
@@ -974,16 +1203,26 @@ fn parse_create_table_collate() {
 
 #[test]
 fn parse_create_table_both_options_and_as_query() {
-    let sql = "CREATE TABLE foo (id INT(11)) ENGINE=InnoDB DEFAULT CHARSET=utf8mb3 COLLATE=utf8mb4_0900_ai_ci AS SELECT 1";
+    let sql = "CREATE TABLE foo (id INT(11)) ENGINE = InnoDB DEFAULT CHARSET = utf8mb3 COLLATE = utf8mb4_0900_ai_ci AS SELECT 1";
     match mysql_and_generic().verified_stmt(sql) {
         Statement::CreateTable(CreateTable {
             name,
-            collation,
             query,
+            table_options,
             ..
         }) => {
             assert_eq!(name.to_string(), "foo");
-            assert_eq!(collation, Some("utf8mb4_0900_ai_ci".to_string()));
+
+            let plain_options = match table_options {
+                CreateTableOptions::Plain(options) => options,
+                _ => unreachable!(),
+            };
+
+            assert!(plain_options.contains(&SqlOption::KeyValue {
+                key: Ident::new("COLLATE"),
+                value: Expr::Identifier(Ident::new("utf8mb4_0900_ai_ci".to_owned()))
+            }));
+
             assert_eq!(
                 query.unwrap().body.as_select().unwrap().projection,
                 vec![SelectItem::UnnamedExpr(Expr::Value(
@@ -994,7 +1233,8 @@ fn parse_create_table_both_options_and_as_query() {
         _ => unreachable!(),
     }
 
-    let sql = r"CREATE TABLE foo (id INT(11)) ENGINE=InnoDB AS SELECT 1 DEFAULT CHARSET=utf8mb3";
+    let sql =
+        r"CREATE TABLE foo (id INT(11)) ENGINE = InnoDB AS SELECT 1 DEFAULT CHARSET = utf8mb3";
     assert!(matches!(
         mysql_and_generic().parse_sql_statements(sql),
         Err(ParserError::ParserError(_))

tests/sqlparser_postgres.rs

@@ -348,7 +348,7 @@ fn parse_create_table_with_defaults() {
             name,
             columns,
             constraints,
-            with_options,
+            table_options,
             if_not_exists: false,
             external: false,
             file_format: None,
@@ -485,6 +485,11 @@ fn parse_create_table_with_defaults() {
                 ]
             );
             assert!(constraints.is_empty());
+
+            let with_options = match table_options {
+                CreateTableOptions::With(options) => options,
+                _ => unreachable!(),
+            };
             assert_eq!(
                 with_options,
                 vec![
@@ -4668,7 +4673,6 @@ fn parse_create_table_with_alias() {
             name,
             columns,
             constraints,
-            with_options: _with_options,
             if_not_exists: false,
             external: false,
             file_format: None,
@@ -5078,7 +5082,11 @@ fn parse_at_time_zone() {
 fn parse_create_table_with_options() {
     let sql = "CREATE TABLE t (c INT) WITH (foo = 'bar', a = 123)";
     match pg().verified_stmt(sql) {
-        Statement::CreateTable(CreateTable { with_options, .. }) => {
+        Statement::CreateTable(CreateTable { table_options, .. }) => {
+            let with_options = match table_options {
+                CreateTableOptions::With(options) => options,
+                _ => unreachable!(),
+            };
             assert_eq!(
                 vec![
                     SqlOption::KeyValue {
@@ -5506,19 +5514,13 @@ fn parse_trigger_related_functions() {
                 storage: None,
                 location: None
             }),
-            table_properties: vec![],
-            with_options: vec![],
             file_format: None,
             location: None,
             query: None,
             without_rowid: false,
             like: None,
             clone: None,
-            engine: None,
             comment: None,
-            auto_increment_offset: None,
-            default_charset: None,
-            collation: None,
             on_commit: None,
             on_cluster: None,
             primary_key: None,
@@ -5526,7 +5528,6 @@ fn parse_trigger_related_functions() {
             partition_by: None,
             cluster_by: None,
             clustered_by: None,
-            options: None,
             inherits: None,
             strict: false,
             copy_grants: false,
@@ -5543,6 +5544,7 @@ fn parse_trigger_related_functions() {
             catalog: None,
             catalog_sync: None,
             storage_serialization_policy: None,
+            table_options: CreateTableOptions::None
         }
     );
 

tests/sqlparser_snowflake.rs

@@ -470,9 +470,22 @@ fn test_snowflake_create_table_cluster_by() {
 #[test]
 fn test_snowflake_create_table_comment() {
     match snowflake().verified_stmt("CREATE TABLE my_table (a INT) COMMENT = 'some comment'") {
-        Statement::CreateTable(CreateTable { name, comment, .. }) => {
+        Statement::CreateTable(CreateTable {
+            name,
+            table_options,
+            ..
+        }) => {
             assert_eq!("my_table", name.to_string());
-            assert_eq!("some comment", comment.unwrap().to_string());
+            let plain_options = match table_options {
+                CreateTableOptions::Plain(options) => options,
+                _ => unreachable!(),
+            };
+            let comment = match plain_options.first().unwrap() {
+                SqlOption::Comment(CommentDef::WithEq(c))
+                | SqlOption::Comment(CommentDef::WithoutEq(c)) => c,
+                _ => unreachable!(),
+            };
+            assert_eq!("some comment", comment);
         }
         _ => unreachable!(),
     }