[fix](mtmv)transfer col in mysql varchar to text when create MTMV

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

@@ -136,7 +136,7 @@ public void run(ConnectContext ctx, StmtExecutor executor) throws Exception {
                             // String type can not be used in partition/distributed column
                             // so we replace it to varchar
                             dataType = TypeCoercionUtils.replaceSpecifiedType(dataType,
-                                    StringType.class, VarcharType.MAX_VARCHAR_TYPE);
+                                    CharacterType.class, VarcharType.MAX_VARCHAR_TYPE);
                         } else {
                             dataType = TypeCoercionUtils.replaceSpecifiedType(dataType,
                                     CharacterType.class, StringType.INSTANCE);

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

@@ -29,6 +29,7 @@
 import org.apache.doris.catalog.Env;
 import org.apache.doris.catalog.KeysType;
 import org.apache.doris.catalog.PartitionType;
+import org.apache.doris.catalog.ScalarType;
 import org.apache.doris.catalog.TableIf;
 import org.apache.doris.catalog.Type;
 import org.apache.doris.catalog.View;
@@ -57,6 +58,7 @@
 import org.apache.doris.nereids.rules.exploration.mv.MaterializedViewUtils;
 import org.apache.doris.nereids.trees.expressions.Expression;
 import org.apache.doris.nereids.trees.expressions.Slot;
+import org.apache.doris.nereids.trees.expressions.SlotReference;
 import org.apache.doris.nereids.trees.plans.Plan;
 import org.apache.doris.nereids.trees.plans.algebra.OneRowRelation;
 import org.apache.doris.nereids.trees.plans.commands.ExplainCommand.ExplainLevel;
@@ -66,9 +68,14 @@
 import org.apache.doris.nereids.trees.plans.logical.LogicalSink;
 import org.apache.doris.nereids.trees.plans.logical.LogicalSubQueryAlias;
 import org.apache.doris.nereids.types.AggStateType;
+import org.apache.doris.nereids.types.CharType;
 import org.apache.doris.nereids.types.DataType;
+import org.apache.doris.nereids.types.DecimalV2Type;
 import org.apache.doris.nereids.types.NullType;
+import org.apache.doris.nereids.types.StringType;
 import org.apache.doris.nereids.types.TinyIntType;
+import org.apache.doris.nereids.types.VarcharType;
+import org.apache.doris.nereids.types.coercion.CharacterType;
 import org.apache.doris.nereids.util.TypeCoercionUtils;
 import org.apache.doris.nereids.util.Utils;
 import org.apache.doris.qe.ConnectContext;
@@ -245,11 +252,12 @@ public void analyzeQuery(ConnectContext ctx, Map<String, String> mvProperties) t
             throw new AnalysisException("can not contain invalid expression");
         }
         getRelation(planner);
-        getColumns(plan);
-        analyzeKeys();
         this.mvPartitionInfo = mvPartitionDefinition
                 .analyzeAndTransferToMTMVPartitionInfo(planner, ctx, logicalQuery);
         this.partitionDesc = generatePartitionDesc(ctx);
+        getColumns(plan, ctx, mvPartitionInfo.getPartitionCol(), distribution);
+        analyzeKeys();
+
     }
 
     private void analyzeKeys() {
@@ -378,7 +386,7 @@ private void analyzeExpressions(Plan plan, Map<String, String> mvProperties) {
         }
     }
 
-    private void getColumns(Plan plan) {
+    private void getColumns(Plan plan, ConnectContext ctx, String partitionCol, DistributionDescriptor distribution) {
         List<Slot> slots = plan.getOutput();
         if (slots.isEmpty()) {
             throw new AnalysisException("table should contain at least one column");
@@ -400,11 +408,11 @@ private void getColumns(Plan plan) {
             } else {
                 colNames.add(colName);
             }
+            DataType dataType = getDataType(slots.get(i), i, ctx, partitionCol, distribution);
             // If datatype is AggStateType, AggregateType should be generic, or column definition check will fail
             columns.add(new ColumnDefinition(
                     colName,
-                    TypeCoercionUtils.replaceSpecifiedType(slots.get(i).getDataType(),
-                            NullType.class, TinyIntType.INSTANCE),
+                    dataType,
                     false,
                     slots.get(i).getDataType() instanceof AggStateType ? AggregateType.GENERIC : null,
                     slots.get(i).nullable(),
@@ -426,6 +434,42 @@ private void getColumns(Plan plan) {
         }
     }
 
+    private DataType getDataType(Slot s, int i, ConnectContext ctx, String partitionCol,
+            DistributionDescriptor distribution) {
+        DataType dataType = s.getDataType().conversion();
+        if (i == 0 && dataType.isStringType()) {
+            dataType = VarcharType.createVarcharType(ScalarType.MAX_VARCHAR_LENGTH);
+        } else {
+            dataType = TypeCoercionUtils.replaceSpecifiedType(dataType,
+                    NullType.class, TinyIntType.INSTANCE);
+            dataType = TypeCoercionUtils.replaceSpecifiedType(dataType,
+                    DecimalV2Type.class, DecimalV2Type.SYSTEM_DEFAULT);
+            if (s.isColumnFromTable()) {
+                if ((!((SlotReference) s).getTable().isPresent()
+                        || !((SlotReference) s).getTable().get().isManagedTable())) {
+                    if (s.getName().equals(partitionCol) || (distribution != null && distribution.inDistributionColumns(
+                            s.getName()))) {
+                        // String type can not be used in partition/distributed column
+                        // so we replace it to varchar
+                        dataType = TypeCoercionUtils.replaceSpecifiedType(dataType,
+                                CharacterType.class, VarcharType.MAX_VARCHAR_TYPE);
+                    } else {
+                        dataType = TypeCoercionUtils.replaceSpecifiedType(dataType,
+                                CharacterType.class, StringType.INSTANCE);
+                    }
+                }
+            } else {
+                if (ctx.getSessionVariable().useMaxLengthOfVarcharInCtas) {
+                    dataType = TypeCoercionUtils.replaceSpecifiedType(dataType,
+                            VarcharType.class, VarcharType.MAX_VARCHAR_TYPE);
+                    dataType = TypeCoercionUtils.replaceSpecifiedType(dataType,
+                            CharType.class, VarcharType.MAX_VARCHAR_TYPE);
+                }
+            }
+        }
+        return dataType;
+    }
+
     /**
      * translate to catalog CreateMultiTableMaterializedViewStmt
      */

regression-test/data/external_table_p0/tvf/test_ctas_with_hdfs.out

@@ -94,7 +94,7 @@ varchar_col	text	Yes	false	\N	NONE
 bigint_col	bigint	Yes	false	\N	NONE
 binary_col	text	Yes	false	\N	NONE
 boolean_col	boolean	Yes	false	\N	NONE
-char_col	char(50)	Yes	false	\N	NONE
+char_col	varchar(65533)	Yes	false	\N	NONE
 date_col	date	Yes	false	\N	NONE
 decimal_col	decimal(12,4)	Yes	false	\N	NONE
 double_col	double	Yes	false	\N	NONE

regression-test/data/mtmv_p0/test_build_mtmv.out

@@ -61,7 +61,7 @@ zhangsang	200
 11	111
 
 -- !desc_mv --
-field_1	varchar(16)	No	true	\N	
+field_1	varchar(65533)	No	true	\N	
 
 -- !query_mv_with_cte --
 2	3

regression-test/data/mtmv_p0/test_mysql_mtmv.out

@@ -4,8 +4,16 @@
 123	15
 123	20
 
+-- !desc_random --
+count_value	text	Yes	false	\N	NONE
+id	int	Yes	true	\N	
+
 -- !mtmv --
 123	10
 123	15
 123	20
 
+-- !desc_hash --
+count_value	varchar(65533)	Yes	true	\N	
+id	int	Yes	true	\N	
+

regression-test/suites/mtmv_p0/test_mysql_mtmv.groovy

@@ -40,7 +40,7 @@ suite("test_mysql_mtmv", "p0,external,mysql,external_docker,external_docker_hive
             "type"="jdbc",
             "user"="root",
             "password"="123456",
-            "jdbc_url" = "jdbc:mysql://${externalEnvIp}:${mysql_port}/${mysqlDb}?useSSL=false&zeroDateTimeBehavior=convertToNull",
+            "jdbc_url" = "jdbc:mysql://${externalEnvIp}:${mysql_port}/${mysqlDb}?useSSL=false&zeroDateTimeBehavior=convertToNull&allowPublicKeyRetrieval=true",
             "driver_url" = "${driver_url}",
             "driver_class" = "com.mysql.cj.jdbc.Driver"
         );"""
@@ -56,14 +56,24 @@ suite("test_mysql_mtmv", "p0,external,mysql,external_docker,external_docker_hive
                 AS
                 SELECT * FROM ${catalog_name}.${mysqlDb}.${mysqlTable};
             """
-
+        order_qt_desc_random "desc ${mvName}"
         sql """
                 REFRESH MATERIALIZED VIEW ${mvName} AUTO
             """
         def jobName = getJobName(dbName, mvName);
         waitingMTMVTaskFinished(jobName)
         order_qt_mtmv "SELECT * FROM ${mvName} order by id"
 
+        sql """drop materialized view if exists ${mvName};"""
+        sql """
+            CREATE MATERIALIZED VIEW ${mvName}
+                BUILD DEFERRED REFRESH COMPLETE ON MANUAL
+                DISTRIBUTED BY hash(count_value) BUCKETS 2
+                PROPERTIES ('replication_num' = '1')
+                AS
+                SELECT * FROM ${catalog_name}.${mysqlDb}.${mysqlTable};
+            """
+        order_qt_desc_hash "desc ${mvName}"
         sql """drop materialized view if exists ${mvName};"""
         sql """ drop catalog if exists ${catalog_name} """
     }