fix: allow passing extra table options (#3484)

* fix: do not check options in parser

* test: fix tests

* test: fix sqlness

* test: add sqlness test

* chore: log options

* chore: must specify compaction type

* feat: validate option key

* feat: add option key validation back

src/sql/src/error.rs

@@ -117,7 +117,7 @@ pub enum Error {
         source: datatypes::error::Error,
     },
 
-    #[snafu(display("Invalid table option key: {}", key))]
+    #[snafu(display("Unrecognized table option key: {}", key))]
     InvalidTableOption { key: String, location: Location },
 
     #[snafu(display("Failed to serialize column default constraint"))]

src/sql/src/parsers/create_parser.rs

@@ -23,7 +23,7 @@ use sqlparser::keywords::ALL_KEYWORDS;
 use sqlparser::parser::IsOptional::Mandatory;
 use sqlparser::parser::{Parser, ParserError};
 use sqlparser::tokenizer::{Token, TokenWithLocation, Word};
-use table::requests::valid_table_option;
+use table::requests::validate_table_option;
 
 use crate::ast::{ColumnDef, Ident, TableConstraint};
 use crate::error::{
@@ -84,7 +84,7 @@ impl<'a> ParserContext<'a> {
             .collect::<HashMap<String, String>>();
         for key in options.keys() {
             ensure!(
-                valid_table_option(key),
+                validate_table_option(key),
                 InvalidTableOptionSnafu {
                     key: key.to_string()
                 }
@@ -150,7 +150,7 @@ impl<'a> ParserContext<'a> {
             .context(error::SyntaxSnafu)?;
         for option in options.iter() {
             ensure!(
-                valid_table_option(&option.name.value),
+                validate_table_option(&option.name.value),
                 InvalidTableOptionSnafu {
                     key: option.name.value.to_string()
                 }
@@ -799,6 +799,17 @@ mod tests {
                 expected_engine: "foo",
                 expected_if_not_exist: true,
             },
+            Test {
+                sql: "CREATE EXTERNAL TABLE IF NOT EXISTS city ENGINE=foo with(location='/var/data/city.csv',format='csv','compaction.type'='bar');",
+                expected_table_name: "city",
+                expected_options: HashMap::from([
+                    ("location".to_string(), "/var/data/city.csv".to_string()),
+                    ("format".to_string(), "csv".to_string()),
+                    ("compaction.type".to_string(), "bar".to_string()),
+                ]),
+                expected_engine: "foo",
+                expected_if_not_exist: true,
+            },
         ];
 
         for test in tests {

src/sql/src/statements/create.rs

@@ -230,7 +230,7 @@ pub struct CreateTableLike {
 #[cfg(test)]
 mod tests {
     use crate::dialect::GreptimeDbDialect;
-    use crate::error::Error::InvalidTableOption;
+    use crate::error::Error;
     use crate::parser::{ParseOptions, ParserContext};
     use crate::statements::statement::Statement;
 
@@ -344,7 +344,29 @@ ENGINE=mito
     fn test_validate_table_options() {
         let sql = r"create table if not exists demo(
             host string,
-            ts bigint,
+            ts timestamp,
+            cpu double default 0,
+            memory double,
+            TIME INDEX (ts),
+            PRIMARY KEY(host)
+      )
+      PARTITION ON COLUMNS (host) ()
+      engine=mito
+      with(regions=1, ttl='7d', 'compaction.type'='world');
+";
+        let result =
+            ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default())
+                .unwrap();
+        match &result[0] {
+            Statement::CreateTable(c) => {
+                assert_eq!(3, c.options.len());
+            }
+            _ => unreachable!(),
+        }
+
+        let sql = r"create table if not exists demo(
+            host string,
+            ts timestamp,
             cpu double default 0,
             memory double,
             TIME INDEX (ts),
@@ -356,6 +378,6 @@ ENGINE=mito
 ";
         let result =
             ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default());
-        assert!(matches!(result, Err(InvalidTableOption { .. })))
+        assert!(matches!(result, Err(Error::InvalidTableOption { .. })));
     }
 }