[fix](cluster key) some data type is not supported for cluster key

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/info/ColumnDefinition.java

@@ -199,10 +199,37 @@ private DataType updateCharacterTypeLength(DataType dataType) {
         }
     }
 
+    private void checkKeyColumnType(boolean isOlap) {
+        if (isOlap) {
+            if (type.isFloatLikeType()) {
+                throw new AnalysisException("Float or double can not used as a key, use decimal instead.");
+            } else if (type.isStringType()) {
+                throw new AnalysisException("String Type should not be used in key column[" + name + "]");
+            } else if (type.isArrayType()) {
+                throw new AnalysisException("Array can only be used in the non-key column of"
+                        + " the duplicate table at present.");
+            }
+        }
+        if (type.isBitmapType() || type.isHllType() || type.isQuantileStateType()) {
+            throw new AnalysisException("Key column can not set complex type:" + name);
+        } else if (type.isJsonType()) {
+            throw new AnalysisException("JsonType type should not be used in key column[" + getName() + "].");
+        } else if (type.isVariantType()) {
+            throw new AnalysisException("Variant type should not be used in key column[" + getName() + "].");
+        } else if (type.isMapType()) {
+            throw new AnalysisException("Map can only be used in the non-key column of"
+                    + " the duplicate table at present.");
+        } else if (type.isStructType()) {
+            throw new AnalysisException("Struct can only be used in the non-key column of"
+                    + " the duplicate table at present.");
+        }
+    }
+
     /**
      * validate column definition and analyze
      */
-    public void validate(boolean isOlap, Set<String> keysSet, boolean isEnableMergeOnWrite, KeysType keysType) {
+    public void validate(boolean isOlap, Set<String> keysSet, Set<String> clusterKeySet, boolean isEnableMergeOnWrite,
+            KeysType keysType) {
         try {
             FeNameFormat.checkColumnName(name);
         } catch (Exception e) {
@@ -234,33 +261,7 @@ public void validate(boolean isOlap, Set<String> keysSet, boolean isEnableMergeO
                 throw new AnalysisException(
                         String.format("Key column %s can not set aggregation type", name));
             }
-            if (isOlap) {
-                if (type.isFloatLikeType()) {
-                    throw new AnalysisException(
-                            "Float or double can not used as a key, use decimal instead.");
-                } else if (type.isStringType()) {
-                    throw new AnalysisException(
-                            "String Type should not be used in key column[" + name + "]");
-                } else if (type.isArrayType()) {
-                    throw new AnalysisException("Array can only be used in the non-key column of"
-                            + " the duplicate table at present.");
-                }
-            }
-            if (type.isBitmapType() || type.isHllType() || type.isQuantileStateType()) {
-                throw new AnalysisException("Key column can not set complex type:" + name);
-            } else if (type.isJsonType()) {
-                throw new AnalysisException(
-                        "JsonType type should not be used in key column[" + getName() + "].");
-            } else if (type.isVariantType()) {
-                throw new AnalysisException(
-                        "Variant type should not be used in key column[" + getName() + "].");
-            } else if (type.isMapType()) {
-                throw new AnalysisException("Map can only be used in the non-key column of"
-                        + " the duplicate table at present.");
-            } else if (type.isStructType()) {
-                throw new AnalysisException("Struct can only be used in the non-key column of"
-                        + " the duplicate table at present.");
-            }
+            checkKeyColumnType(isOlap);
         } else if (aggType == null && isOlap) {
             Preconditions.checkState(keysType != null, "keysType is null");
             if (keysType.equals(KeysType.DUP_KEYS)) {
@@ -274,6 +275,10 @@ public void validate(boolean isOlap, Set<String> keysSet, boolean isEnableMergeO
             }
         }
 
+        if (clusterKeySet.contains(name)) {
+            checkKeyColumnType(isOlap);
+        }
+
         if (aggType != null) {
             // check if aggregate type is valid
             if (aggType != AggregateType.GENERIC

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/info/CreateMTMVInfo.java

@@ -162,7 +162,7 @@ public void analyze(ConnectContext ctx) throws Exception {
         final boolean finalEnableMergeOnWrite = false;
         Set<String> keysSet = Sets.newTreeSet(String.CASE_INSENSITIVE_ORDER);
         keysSet.addAll(keys);
-        columns.forEach(c -> c.validate(true, keysSet, finalEnableMergeOnWrite, KeysType.DUP_KEYS));
+        columns.forEach(c -> c.validate(true, keysSet, Sets.newHashSet(), finalEnableMergeOnWrite, KeysType.DUP_KEYS));
 
         if (distribution == null) {
             throw new AnalysisException("Create async materialized view should contain distribution desc");

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/info/CreateTableInfo.java

@@ -558,7 +558,9 @@ public void validate(ConnectContext ctx) {
         final boolean finalEnableMergeOnWrite = isEnableMergeOnWrite;
         Set<String> keysSet = Sets.newTreeSet(String.CASE_INSENSITIVE_ORDER);
         keysSet.addAll(keys);
-        columns.forEach(c -> c.validate(engineName.equals(ENGINE_OLAP), keysSet, finalEnableMergeOnWrite,
+        Set<String> clusterKeySet = Sets.newTreeSet(String.CASE_INSENSITIVE_ORDER);
+        clusterKeySet.addAll(clusterKeysColumnNames);
+        columns.forEach(c -> c.validate(engineName.equals(ENGINE_OLAP), keysSet, clusterKeySet, finalEnableMergeOnWrite,
                 keysType));
 
         // validate index

regression-test/data/unique_with_mow_c_p0/test_delete_sign.out

@@ -0,0 +1,69 @@
+-- This file is automatically generated. You should know what you did if you want to edit this
+-- !select_0 --
+true	1	10	{"c":"c"}
+
+-- !select_1 --
+
+-- !select_2 --
+
+-- !select_3 --
+true	1	10	{"c":"c"}
+
+-- !select_4 --
+
+-- !select_5 --
+
+-- !select_6 --
+
+-- !select_7 --
+true	1	10	{"c":"c"}
+
+-- !select_8 --
+true	1	10	{"c":"c"}
+
+-- !select_9 --
+true	1	30	{"b":"b"}
+
+-- !select_10 --
+true	1	10	{"c":"c"}
+
+-- !select_11 --
+true	1	10	{"c":"c"}
+
+-- !select_12 --
+true	1	30	{"b":"b"}
+
+-- !select_0 --
+true	1	10	{"c":"c"}
+
+-- !select_1 --
+
+-- !select_2 --
+
+-- !select_3 --
+true	1	10	{"c":"c"}
+
+-- !select_4 --
+
+-- !select_5 --
+
+-- !select_6 --
+
+-- !select_7 --
+true	1	10	{"c":"c"}
+
+-- !select_8 --
+true	1	10	{"c":"c"}
+
+-- !select_9 --
+true	1	30	{"b":"b"}
+
+-- !select_10 --
+true	1	10	{"c":"c"}
+
+-- !select_11 --
+true	1	10	{"c":"c"}
+
+-- !select_12 --
+true	1	30	{"b":"b"}
+

regression-test/data/unique_with_mow_c_p0/test_mow_full_clone_exception.out

@@ -0,0 +1,37 @@
+-- This file is automatically generated. You should know what you did if you want to edit this
+-- !sql --
+1	10
+2	200
+3	30
+4	400
+5	500
+6	600
+7	7
+8	8
+9	9
+10	10
+
+-- !sql --
+1	10
+2	200
+3	30
+4	400
+5	500
+6	600
+7	7
+8	8
+9	9
+10	10
+
+-- !sql --
+1	10
+2	200
+3	30
+4	400
+5	500
+6	600
+7	7
+8	8
+9	9
+10	10
+

regression-test/suites/point_query_p0/test_point_query_cluster_key.groovy

@@ -239,7 +239,7 @@ suite("test_point_query_cluster_key") {
                 sql """CREATE TABLE ${tableName} (
                         `customer_key` bigint(20) NULL,
                         `customer_btm_value_0` text NULL,
-                        `customer_btm_value_1` text NULL,
+                        `customer_btm_value_1` VARCHAR(1000) NULL,
                         `customer_btm_value_2` text NULL
                     ) ENGINE=OLAP
                     UNIQUE KEY(`customer_key`)

regression-test/suites/unique_with_mow_c_p0/ssb_unique_sql_zstd/ddl/customer_create.sql

@@ -14,6 +14,5 @@ DISTRIBUTED BY HASH(`c_custkey`) BUCKETS 10
 PROPERTIES (
 "compression"="zstd",
 "replication_num" = "1",
-"disable_auto_compaction" = "true",
 "enable_unique_key_merge_on_write" = "true"
 );

regression-test/suites/unique_with_mow_c_p0/ssb_unique_sql_zstd/ddl/date_create.sql

@@ -23,6 +23,5 @@ DISTRIBUTED BY HASH(`d_datekey`) BUCKETS 1
 PROPERTIES (
 "compression"="zstd",
 "replication_num" = "1",
-"disable_auto_compaction" = "true",
 "enable_unique_key_merge_on_write" = "true"
 );

regression-test/suites/unique_with_mow_c_p0/ssb_unique_sql_zstd/ddl/lineorder_create.sql

@@ -31,6 +31,5 @@ DISTRIBUTED BY HASH(`lo_orderkey`) BUCKETS 48
 PROPERTIES (
 "compression"="zstd",
 "replication_num" = "1",
-"disable_auto_compaction" = "true",
 "enable_unique_key_merge_on_write" = "true"
 );

regression-test/suites/unique_with_mow_c_p0/ssb_unique_sql_zstd/ddl/part_create.sql

@@ -15,6 +15,5 @@ DISTRIBUTED BY HASH(`p_partkey`) BUCKETS 10
 PROPERTIES (
 "compression"="zstd",
 "replication_num" = "1",
-"disable_auto_compaction" = "true",
 "enable_unique_key_merge_on_write" = "true"
 );

regression-test/suites/unique_with_mow_c_p0/ssb_unique_sql_zstd/ddl/supplier_create.sql

@@ -13,6 +13,5 @@ DISTRIBUTED BY HASH(`s_suppkey`) BUCKETS 10
 PROPERTIES (
 "compression"="zstd",
 "replication_num" = "1",
-"disable_auto_compaction" = "true",
 "enable_unique_key_merge_on_write" = "true"
 );

regression-test/suites/unique_with_mow_c_p0/test_create_table.groovy

@@ -227,4 +227,24 @@ suite("test_create_table") {
         """
         exception "Cluster keys only support unique keys table"
     }
+
+    // cluster key contains complex type
+    test {
+        sql """
+            CREATE TABLE `$tableName` (
+                `c_custkey` int(11) NOT NULL COMMENT "",
+                `c_name` varchar(26) NOT NULL COMMENT "",
+                `c_address` varchar(41) NOT NULL COMMENT "",
+                `c_city` variant NOT NULL COMMENT ""
+            )
+            UNIQUE KEY (`c_custkey`)
+            CLUSTER BY (`c_name`, `c_city`, `c_address`)
+            DISTRIBUTED BY HASH(`c_custkey`) BUCKETS 1
+            PROPERTIES (
+                "replication_num" = "1",
+                "enable_unique_key_merge_on_write" = "true"
+            );
+        """
+        exception "Variant type should not be used in key column"
+    }
 }