fix(create): fix table options validation (#11290)

* table options validation

* fix in build_request

* delete useless codes

* delete useless codes

* delete useless codes

* add transient

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

@@ -12,6 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use std::collections::HashSet;
 use std::sync::Arc;
 
 use common_config::GlobalConfig;
@@ -27,11 +28,25 @@ use common_sql::binder::INTERNAL_COLUMN_FACTORY;
 use common_sql::field_default_value;
 use common_sql::plans::CreateTablePlan;
 use common_storages_fuse::io::MetaReaders;
+use common_storages_fuse::FUSE_OPT_KEY_BLOCK_IN_MEM_SIZE_THRESHOLD;
+use common_storages_fuse::FUSE_OPT_KEY_BLOCK_PER_SEGMENT;
+use common_storages_fuse::FUSE_OPT_KEY_ROW_AVG_DEPTH_THRESHOLD;
+use common_storages_fuse::FUSE_OPT_KEY_ROW_PER_BLOCK;
+use common_storages_fuse::FUSE_OPT_KEY_ROW_PER_PAGE;
 use common_users::UserApiProvider;
+use once_cell::sync::Lazy;
 use storages_common_cache::LoadParams;
 use storages_common_table_meta::meta::TableSnapshot;
 use storages_common_table_meta::meta::Versioned;
+use storages_common_table_meta::table::OPT_KEY_COMMENT;
+use storages_common_table_meta::table::OPT_KEY_DATABASE_ID;
+use storages_common_table_meta::table::OPT_KEY_ENGINE;
+use storages_common_table_meta::table::OPT_KEY_EXTERNAL_LOCATION;
+use storages_common_table_meta::table::OPT_KEY_LEGACY_SNAPSHOT_LOC;
 use storages_common_table_meta::table::OPT_KEY_SNAPSHOT_LOCATION;
+use storages_common_table_meta::table::OPT_KEY_STORAGE_FORMAT;
+use storages_common_table_meta::table::OPT_KEY_TABLE_COMPRESSION;
+use tracing::error;
 
 use crate::interpreters::InsertInterpreter;
 use crate::interpreters::Interpreter;
@@ -224,6 +239,17 @@ impl CreateTableInterpreter {
             },
             ..Default::default()
         };
+
+        for table_option in table_meta.options.iter() {
+            let key = table_option.0.to_lowercase();
+            if !is_valid_create_opt(&key) {
+                error!("invalid opt for fuse table in create table statement");
+                return Err(ErrorCode::TableOptionInvalid(format!(
+                    "table option {key} is invalid for create table statement",
+                )));
+            }
+        }
+
         if let Some(cluster_key) = &self.plan.cluster_key {
             table_meta = table_meta.push_cluster_key(cluster_key.clone());
         }
@@ -241,3 +267,30 @@ impl CreateTableInterpreter {
         Ok(req)
     }
 }
+
+/// Table option keys that can occur in 'create table statement'.
+pub static CREATE_TABLE_OPTIONS: Lazy<HashSet<&'static str>> = Lazy::new(|| {
+    let mut r = HashSet::new();
+    r.insert(FUSE_OPT_KEY_ROW_PER_PAGE);
+    r.insert(FUSE_OPT_KEY_BLOCK_PER_SEGMENT);
+    r.insert(FUSE_OPT_KEY_ROW_PER_BLOCK);
+    r.insert(FUSE_OPT_KEY_BLOCK_IN_MEM_SIZE_THRESHOLD);
+    r.insert(FUSE_OPT_KEY_ROW_AVG_DEPTH_THRESHOLD);
+
+    r.insert(OPT_KEY_SNAPSHOT_LOCATION);
+    r.insert(OPT_KEY_LEGACY_SNAPSHOT_LOC);
+    r.insert(OPT_KEY_TABLE_COMPRESSION);
+    r.insert(OPT_KEY_STORAGE_FORMAT);
+    r.insert(OPT_KEY_DATABASE_ID);
+
+    r.insert(OPT_KEY_COMMENT);
+    r.insert(OPT_KEY_EXTERNAL_LOCATION);
+    r.insert(OPT_KEY_ENGINE);
+
+    r.insert("transient");
+    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())
+}

src/query/storages/common/table-meta/src/table/table_keys.rs

@@ -15,11 +15,13 @@
 use std::collections::HashSet;
 
 use once_cell::sync::Lazy;
-
 pub const OPT_KEY_DATABASE_ID: &str = "database_id";
 pub const OPT_KEY_SNAPSHOT_LOCATION: &str = "snapshot_location";
 pub const OPT_KEY_STORAGE_FORMAT: &str = "storage_format";
 pub const OPT_KEY_TABLE_COMPRESSION: &str = "compression";
+pub const OPT_KEY_COMMENT: &str = "comment";
+pub const OPT_KEY_EXTERNAL_LOCATION: &str = "external_location";
+pub const OPT_KEY_ENGINE: &str = "engine";
 
 /// Legacy table snapshot location key
 ///

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

@@ -253,3 +253,6 @@ drop table if exists tt_v2
 
 statement error 3001
 create table tt_v2 (id int) engine=fuse SNAPSHOT_LOCATION='xx'
+
+statement error 1301
+create table t(a int) x=x