fix: check if the table option is valid according to the engine

src/query/ast/src/ast/statements/table.rs

@@ -745,6 +745,21 @@ impl Display for Engine {
     }
 }
 
+impl From<&str> for Engine {
+    fn from(s: &str) -> Self {
+        match s.to_lowercase().as_str() {
+            "null" => Engine::Null,
+            "memory" => Engine::Memory,
+            "fuse" => Engine::Fuse,
+            "view" => Engine::View,
+            "random" => Engine::Random,
+            "iceberg" => Engine::Iceberg,
+            "delta" => Engine::Delta,
+            _ => unreachable!("invalid engine: {}", s),
+        }
+    }
+}
+
 #[derive(Debug, Clone, PartialEq, Eq, Drive, DriveMut)]
 pub enum CompactTarget {
     Block,

src/query/service/src/interpreters/common/table_option_validation.rs

@@ -17,6 +17,7 @@ use std::collections::HashSet;
 use std::sync::LazyLock;
 
 use chrono::Duration;
+use databend_common_ast::ast::Engine;
 use databend_common_exception::ErrorCode;
 use databend_common_expression::TableSchemaRef;
 use databend_common_io::constants::DEFAULT_BLOCK_MAX_ROWS;
@@ -45,7 +46,7 @@ use databend_storages_common_table_meta::table::OPT_KEY_TEMP_PREFIX;
 use log::error;
 
 /// Table option keys that can occur in 'create table statement'.
-pub static CREATE_TABLE_OPTIONS: LazyLock<HashSet<&'static str>> = LazyLock::new(|| {
+pub static CREATE_FUSE_OPTIONS: LazyLock<HashSet<&'static str>> = LazyLock::new(|| {
     let mut r = HashSet::new();
     r.insert(FUSE_OPT_KEY_ROW_PER_PAGE);
     r.insert(FUSE_OPT_KEY_BLOCK_PER_SEGMENT);
@@ -64,12 +65,32 @@ pub static CREATE_TABLE_OPTIONS: LazyLock<HashSet<&'static str>> = LazyLock::new
 
     r.insert(OPT_KEY_ENGINE);
 
+    r.insert(OPT_KEY_CONNECTION_NAME);
+
+    r.insert("transient");
+    r.insert(OPT_KEY_TEMP_PREFIX);
+    r
+});
+
+pub static CREATE_LAKE_OPTIONS: LazyLock<HashSet<&'static str>> = LazyLock::new(|| {
+    let mut r = HashSet::new();
+    r.insert(OPT_KEY_ENGINE);
     r.insert(OPT_KEY_LOCATION);
     r.insert(OPT_KEY_CONNECTION_NAME);
+    r
+});
 
+pub static CREATE_RANDOM_OPTIONS: LazyLock<HashSet<&'static str>> = LazyLock::new(|| {
+    let mut r = HashSet::new();
+    r.insert(OPT_KEY_ENGINE);
     r.insert(OPT_KEY_RANDOM_SEED);
+    r
+});
 
-    r.insert("transient");
+pub static CREATE_MEMORY_OPTIONS: LazyLock<HashSet<&'static str>> = LazyLock::new(|| {
+    let mut r = HashSet::new();
+    r.insert(OPT_KEY_ENGINE);
+    r.insert(OPT_KEY_DATABASE_ID);
     r.insert(OPT_KEY_TEMP_PREFIX);
     r
 });
@@ -85,8 +106,16 @@ pub static UNSET_TABLE_OPTIONS_WHITE_LIST: LazyLock<HashSet<&'static str>> = Laz
     r
 });
 
-pub fn is_valid_create_opt<S: AsRef<str>>(opt_key: S) -> bool {
-    CREATE_TABLE_OPTIONS.contains(opt_key.as_ref().to_lowercase().as_str())
+pub fn is_valid_create_opt<S: AsRef<str>>(opt_key: S, engine: &Engine) -> bool {
+    let opt_key = opt_key.as_ref().to_lowercase();
+    let opt_key = opt_key.as_str();
+    match engine {
+        Engine::Fuse => CREATE_FUSE_OPTIONS.contains(opt_key),
+        Engine::Iceberg | Engine::Delta => CREATE_LAKE_OPTIONS.contains(&opt_key),
+        Engine::Random => CREATE_RANDOM_OPTIONS.contains(&opt_key),
+        Engine::Memory => CREATE_MEMORY_OPTIONS.contains(&opt_key),
+        Engine::Null | Engine::View => opt_key == OPT_KEY_ENGINE,
+    }
 }
 
 pub fn is_valid_block_per_segment(

src/query/service/src/interpreters/interpreter_table_create.rs

@@ -402,7 +402,7 @@ impl CreateTableInterpreter {
 
         for table_option in table_meta.options.iter() {
             let key = table_option.0.to_lowercase();
-            if !is_valid_create_opt(&key) {
+            if !is_valid_create_opt(&key, &self.plan.engine) {
                 error!("invalid opt for fuse table in create table statement");
                 return Err(ErrorCode::TableOptionInvalid(format!(
                     "table option {key} is invalid for create table statement",

src/query/service/src/interpreters/interpreter_table_set_options.rs

@@ -15,6 +15,7 @@
 use std::collections::HashMap;
 use std::sync::Arc;
 
+use databend_common_ast::ast::Engine;
 use databend_common_catalog::table::TableExt;
 use databend_common_exception::ErrorCode;
 use databend_common_exception::Result;
@@ -101,22 +102,24 @@ impl Interpreter for SetOptionsInterpreter {
                 OPT_KEY_CLUSTER_TYPE
             )));
         }
+        let catalog = self.ctx.get_catalog(self.plan.catalog.as_str()).await?;
+        let database = self.plan.database.as_str();
+        let table_name = self.plan.table.as_str();
+        let table = catalog
+            .get_table(&self.ctx.get_tenant(), database, table_name)
+            .await?;
+
         for table_option in self.plan.set_options.iter() {
             let key = table_option.0.to_lowercase();
-            if !is_valid_create_opt(&key) {
+            let engine = Engine::from(table.engine());
+            if !is_valid_create_opt(&key, &engine) {
                 error!("{}", &error_str);
                 return Err(ErrorCode::TableOptionInvalid(format!(
                     "table option {key} is invalid for alter table statement",
                 )));
             }
             options_map.insert(key, Some(table_option.1.clone()));
         }
-        let catalog = self.ctx.get_catalog(self.plan.catalog.as_str()).await?;
-        let database = self.plan.database.as_str();
-        let table_name = self.plan.table.as_str();
-        let table = catalog
-            .get_table(&self.ctx.get_tenant(), database, table_name)
-            .await?;
 
         let table_version = table.get_table_info().ident.seq;
         if let Some(value) = self.plan.set_options.get(OPT_KEY_CHANGE_TRACKING) {

src/query/service/src/interpreters/interpreter_table_show_create.rs

@@ -281,7 +281,7 @@ impl ShowCreateTableInterpreter {
 
         if engine != "ICEBERG" && engine != "DELTA" {
             if let Some(sp) = &table_info.meta.storage_params {
-                table_create_sql.push_str(format!(" LOCATION = '{}'", sp).as_str());
+                table_create_sql.push_str(format!(" '{}' ", sp).as_str());
             }
         }
 

tests/sqllogictests/suites/base/05_ddl/05_0000_ddl_create_tables.test

@@ -303,6 +303,15 @@ create table t(a int) x=x
 statement error 1301
 create table t(a int) external_location='xxx'
 
+statement error 1301
+create table t(a int) location='xxx'
+
+statement error 1301
+create table t(a int) seed='123'
+
+statement ok
+create table t_random(a int) engine=RANDOM seed='123'
+
 statement error 1301
 create table t(a int) snapshot_loc='xxx'
 

tests/sqllogictests/suites/base/06_show/06_0001_show_create_table.test

@@ -53,7 +53,7 @@ CREATE TABLE test.d (a int not null) ENGINE=FUSE 'fs:///tmp/load/files/' CONNECT
 query TT
 SHOW CREATE TABLE `test`.`d`
 ----
-d CREATE TABLE d ( a INT NOT NULL ) ENGINE=FUSE CLUSTER BY linear(a, a % 3) CLUSTER_TYPE='linear' COMPRESSION='lz4' STORAGE_FORMAT='parquet' LOCATION = 'fs | root=/tmp/load/files/'
+d CREATE TABLE d ( a INT NOT NULL ) ENGINE=FUSE CLUSTER BY linear(a, a % 3) CLUSTER_TYPE='linear' COMPRESSION='lz4' STORAGE_FORMAT='parquet' 'fs | root=/tmp/load/files/'
 
 query TT
 SHOW CREATE TABLE `test`.`c`