fix: check compression options when binding CREATE TABLE statement (#18261)

* check create table compression options when create `Plan`

Author: chagelo

src/query/sql/src/planner/binder/ddl/table.rs

@@ -88,6 +88,7 @@ use databend_common_storage::init_operator;
 use databend_common_storages_view::view_table::QUERY;
 use databend_common_storages_view::view_table::VIEW_ENGINE;
 use databend_storages_common_table_meta::table::is_reserved_opt_key;
+use databend_storages_common_table_meta::table::TableCompression;
 use databend_storages_common_table_meta::table::OPT_KEY_CLUSTER_TYPE;
 use databend_storages_common_table_meta::table::OPT_KEY_DATABASE_ID;
 use databend_storages_common_table_meta::table::OPT_KEY_ENGINE_META;
@@ -742,6 +743,15 @@ impl Binder {
                     OPT_KEY_TABLE_COMPRESSION.to_owned(),
                     default_compression.to_owned(),
                 );
+            } else {
+                // validate the compression type
+                let _: TableCompression = options
+                    .get(OPT_KEY_TABLE_COMPRESSION)
+                    .ok_or_else(|| {
+                        ErrorCode::BadArguments("Table compression type is not specified")
+                    })?
+                    .as_str()
+                    .try_into()?;
             }
         } else if table_indexes.is_some() {
             return Err(ErrorCode::UnsupportedIndex(format!(

src/query/storages/common/table_meta/src/table/table_compression.rs

@@ -40,7 +40,7 @@ impl TryFrom<&str> for TableCompression {
             "zstd" => Ok(TableCompression::Zstd),
             "lz4" => Ok(TableCompression::LZ4),
             "snappy" => Ok(TableCompression::Snappy),
-            other => Err(ErrorCode::UnknownFormat(format!(
+            other => Err(ErrorCode::UnknownCompressionType(format!(
                 "unsupported table compression: {}",
                 other
             ))),

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

@@ -681,3 +681,6 @@ query
 select '\'0\'' = a from t
 ----
 1
+
+statement error 1075
+create or replace table t(id int) compression='xxxx';