[fix](tvf) Partition columns in CTAS need to be compatible with the STRING type of external tables/TVF

fe/fe-core/src/main/java/org/apache/doris/analysis/DistributionDesc.java

@@ -65,4 +65,12 @@ public String toSql() {
     public DistributionInfo toDistributionInfo(List<Column> columns) throws DdlException {
         throw new NotImplementedException("toDistributionInfo not implemented");
     }
+
+    public List<String> getDistributionColumnNames() {
+        throw new NotImplementedException("getDistributionColumnNames not implemented");
+    }
+
+    public boolean inDistributionColumns(String columnName) {
+        return getDistributionColumnNames() != null && getDistributionColumnNames().contains(columnName);
+    }
 }

fe/fe-core/src/main/java/org/apache/doris/analysis/HashDistributionDesc.java

@@ -47,6 +47,7 @@ public HashDistributionDesc(int numBucket, boolean autoBucket, List<String> dist
         this.distributionColumnNames = distributionColumnNames;
     }
 
+    @Override
     public List<String> getDistributionColumnNames() {
         return distributionColumnNames;
     }

fe/fe-core/src/main/java/org/apache/doris/analysis/PartitionDesc.java

@@ -285,4 +285,8 @@ public PartitionInfo toPartitionInfo(List<Column> schema, Map<String, Long> part
             throws DdlException, AnalysisException {
         throw new NotImplementedException("toPartitionInfo not implemented");
     }
+
+    public boolean inIdentifierPartitions(String colName) {
+        return partitionColNames != null && partitionColNames.contains(colName);
+    }
 }

fe/fe-core/src/main/java/org/apache/doris/analysis/RandomDistributionDesc.java

@@ -62,4 +62,9 @@ public DistributionInfo toDistributionInfo(List<Column> columns) {
         RandomDistributionInfo randomDistributionInfo = new RandomDistributionInfo(numBucket, autoBucket);
         return randomDistributionInfo;
     }
+
+    @Override
+    public List<String> getDistributionColumnNames() {
+        return null;
+    }
 }

fe/fe-core/src/main/java/org/apache/doris/datasource/InternalCatalog.java

@@ -1325,7 +1325,19 @@ public void createTableAsSelect(CreateTableAsSelectStmt stmt) throws DdlExceptio
                     if (resultExpr.getSrcSlotRef() != null
                             && resultExpr.getSrcSlotRef().getTable() != null
                             && !resultExpr.getSrcSlotRef().getTable().isManagedTable()) {
-                        typeDef = new TypeDef(ScalarType.createStringType());
+                        if (createTableStmt.getPartitionDesc().inIdentifierPartitions(
+                                resultExpr.getSrcSlotRef().getColumnName())
+                                || (createTableStmt.getDistributionDesc() != null
+                                && createTableStmt.getDistributionDesc().inDistributionColumns(
+                                        resultExpr.getSrcSlotRef().getColumnName()))) {
+                            // String type can not be used in partition/distributed column
+                            // so we replace it to varchar
+                            if (resultType.getPrimitiveType() == PrimitiveType.STRING) {
+                                typeDef = new TypeDef(ScalarType.createVarchar(ScalarType.MAX_VARCHAR_LENGTH));
+                            }
+                        } else {
+                            typeDef = new TypeDef(ScalarType.createStringType());
+                        }
                     }
                 } else if (resultType.isDecimalV2() && resultType.equals(ScalarType.DECIMALV2)) {
                     typeDef = new TypeDef(ScalarType.createDecimalType(27, 9));

fe/fe-core/src/main/java/org/apache/doris/datasource/hive/ThriftHMSCachedClient.java

@@ -206,7 +206,7 @@ public void createTable(TableMetadata tbl, boolean ignoreIfExists) {
                 throw e;
             }
         } catch (Exception e) {
-            throw new HMSClientException("failed to create database from hms client", e);
+            throw new HMSClientException("failed to create table from hms client", e);
         }
     }
 

fe/fe-core/src/main/java/org/apache/doris/nereids/parser/PartitionTableInfo.java

@@ -63,10 +63,14 @@ public class PartitionTableInfo {
                 null);
 
     private boolean isAutoPartition;
+    // for PartitionType
     private String partitionType;
-    private List<String> partitionColumns;
     private List<PartitionDefinition> partitionDefs;
+    // save all list partition expressions, including identifier and function
     private List<Expression> partitionList;
+    // save identifier expressions in partitionList,
+    // facilitates subsequent verification process
+    private List<String> identifierPartitionColumns;
 
     /**
      * struct for partition definition
@@ -85,7 +89,7 @@ public PartitionTableInfo(
         this.partitionDefs = partitionDefs;
         this.partitionList = partitionFields;
         if (this.partitionList != null) {
-            this.partitionColumns = this.partitionList.stream()
+            this.identifierPartitionColumns = this.partitionList.stream()
                 .filter(UnboundSlot.class::isInstance)
                 .map(partition -> ((UnboundSlot) partition).getName())
                 .collect(Collectors.toList());
@@ -100,10 +104,6 @@ public String getPartitionType() {
         return partitionType;
     }
 
-    public List<String> getPartitionColumns() {
-        return partitionColumns;
-    }
-
     /**
      * check partitions types.
      */
@@ -169,16 +169,16 @@ public void validatePartitionInfo(
             boolean isEnableMergeOnWrite,
             boolean isExternal) {
 
-        if (partitionColumns != null) {
+        if (identifierPartitionColumns != null) {
 
-            if (partitionColumns.size() != partitionList.size()) {
+            if (identifierPartitionColumns.size() != partitionList.size()) {
                 if (!isExternal && partitionType.equalsIgnoreCase(PartitionType.LIST.name())) {
                     throw new AnalysisException("internal catalog does not support functions in 'LIST' partition");
                 }
                 isAutoPartition = true;
             }
 
-            partitionColumns.forEach(p -> {
+            identifierPartitionColumns.forEach(p -> {
                 if (!columnMap.containsKey(p)) {
                     throw new AnalysisException(
                             String.format("partition key %s is not exists", p));
@@ -187,7 +187,7 @@ public void validatePartitionInfo(
             });
 
             Set<String> partitionColumnSets = Sets.newHashSet();
-            List<String> duplicatesKeys = partitionColumns.stream()
+            List<String> duplicatesKeys = identifierPartitionColumns.stream()
                     .filter(c -> !partitionColumnSets.add(c)).collect(Collectors.toList());
             if (!duplicatesKeys.isEmpty()) {
                 throw new AnalysisException(
@@ -199,16 +199,16 @@ public void validatePartitionInfo(
                 // 2. The partition field must be at the end of the schema
                 // 3. The order of partition fields in the schema
                 //    must be consistent with the order defined in `PARTITIONED BY LIST()`
-                if (partitionColumns.size() == columns.size()) {
+                if (identifierPartitionColumns.size() == columns.size()) {
                     throw new AnalysisException("Cannot set all columns as partitioning columns.");
                 }
                 List<ColumnDefinition> partitionInSchema = columns.subList(
-                        columns.size() - partitionColumns.size(), columns.size());
-                if (partitionInSchema.stream().anyMatch(p -> !partitionColumns.contains(p.getName()))) {
+                        columns.size() - identifierPartitionColumns.size(), columns.size());
+                if (partitionInSchema.stream().anyMatch(p -> !identifierPartitionColumns.contains(p.getName()))) {
                     throw new AnalysisException("The partition field must be at the end of the schema.");
                 }
                 for (int i = 0; i < partitionInSchema.size(); i++) {
-                    if (!partitionInSchema.get(i).getName().equals(partitionColumns.get(i))) {
+                    if (!partitionInSchema.get(i).getName().equals(identifierPartitionColumns.get(i))) {
                         throw new AnalysisException("The order of partition fields in the schema "
                             + "must be consistent with the order defined in `PARTITIONED BY LIST()`");
                     }
@@ -234,7 +234,7 @@ public void validatePartitionInfo(
                     partitionNames.add(partitionName);
                 }
                 partitionDefs.forEach(p -> {
-                    p.setPartitionTypes(partitionColumns.stream()
+                    p.setPartitionTypes(identifierPartitionColumns.stream()
                             .map(s -> columnMap.get(s).getType()).collect(Collectors.toList()));
                     p.validate(Maps.newHashMap(properties));
                 });
@@ -269,18 +269,18 @@ public PartitionDesc convertToPartitionDesc(boolean isExternal) {
 
             try {
                 ArrayList<Expr> exprs = convertToLegacyAutoPartitionExprs(partitionList);
-                // here we have already extracted partitionColumns
+                // here we have already extracted identifierPartitionColumns
                 if (partitionType.equals(PartitionType.RANGE.name())) {
                     if (isAutoPartition) {
-                        partitionDesc = new RangePartitionDesc(exprs, partitionColumns, partitionDescs);
+                        partitionDesc = new RangePartitionDesc(exprs, identifierPartitionColumns, partitionDescs);
                     } else {
-                        partitionDesc = new RangePartitionDesc(partitionColumns, partitionDescs);
+                        partitionDesc = new RangePartitionDesc(identifierPartitionColumns, partitionDescs);
                     }
                 } else {
                     if (isAutoPartition) {
-                        partitionDesc = new ListPartitionDesc(exprs, partitionColumns, partitionDescs);
+                        partitionDesc = new ListPartitionDesc(exprs, identifierPartitionColumns, partitionDescs);
                     } else {
-                        partitionDesc = new ListPartitionDesc(partitionColumns, partitionDescs);
+                        partitionDesc = new ListPartitionDesc(identifierPartitionColumns, partitionDescs);
                     }
                 }
             } catch (Exception e) {
@@ -319,18 +319,22 @@ private static List<Expr> convertToLegacyArguments(List<Expression> children) {
     }
 
     /**
-     *  Get column names and put in partitionColumns
+     *  Get column names and put in identifierPartitionColumns
      */
     public void extractPartitionColumns() throws AnalysisException {
         if (partitionList == null) {
             return;
         }
         ArrayList<Expr> exprs = convertToLegacyAutoPartitionExprs(partitionList);
         try {
-            partitionColumns = PartitionDesc.getColNamesFromExpr(exprs,
+            identifierPartitionColumns = PartitionDesc.getColNamesFromExpr(exprs,
                     partitionType.equalsIgnoreCase(PartitionType.LIST.name()), isAutoPartition);
         } catch (Exception e) {
             throw new AnalysisException(e.getMessage(), e.getCause());
         }
     }
+
+    public boolean inIdentifierPartitions(String columnName) {
+        return identifierPartitionColumns != null && identifierPartitionColumns.contains(columnName);
+    }
 }

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

@@ -128,10 +128,19 @@ public void run(ConnectContext ctx, StmtExecutor executor) throws Exception {
                 dataType = TypeCoercionUtils.replaceSpecifiedType(dataType,
                         DecimalV2Type.class, DecimalV2Type.SYSTEM_DEFAULT);
                 if (s.isColumnFromTable()) {
-                    if (!((SlotReference) s).getTable().isPresent()
-                            || !((SlotReference) s).getTable().get().isManagedTable()) {
-                        dataType = TypeCoercionUtils.replaceSpecifiedType(dataType,
-                                CharacterType.class, StringType.INSTANCE);
+                    if ((!((SlotReference) s).getTable().isPresent()
+                            || !((SlotReference) s).getTable().get().isManagedTable())) {
+                        if (createTableInfo.getPartitionTableInfo().inIdentifierPartitions(s.getName())
+                                || (createTableInfo.getDistribution() != null
+                                && createTableInfo.getDistribution().inDistributionColumns(s.getName()))) {
+                            // 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);
+                        } else {
+                            dataType = TypeCoercionUtils.replaceSpecifiedType(dataType,
+                                    CharacterType.class, StringType.INSTANCE);
+                        }
                     }
                 } else {
                     dataType = TypeCoercionUtils.replaceSpecifiedType(dataType,

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

@@ -1048,5 +1048,13 @@ public boolean isGeneratedColumn() {
             return isGeneratedColumn;
         }
     }
+
+    public PartitionTableInfo getPartitionTableInfo() {
+        return partitionTableInfo;
+    }
+
+    public DistributionDescriptor getDistribution() {
+        return distribution;
+    }
 }
 

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

@@ -107,4 +107,8 @@ public DistributionDesc translateToCatalogStyle() {
         }
         return new RandomDistributionDesc(bucketNum, isAutoBucket);
     }
+
+    public boolean inDistributionColumns(String columnName) {
+        return cols != null && cols.contains(columnName);
+    }
 }

fe/fe-core/src/main/java/org/apache/doris/tablefunction/ExternalFileTableValuedFunction.java

@@ -465,7 +465,7 @@ private void fillColumns(InternalService.PFetchTableSchemaResult result) {
         // HACK(tsy): path columns are all treated as STRING type now, after BE supports reading all columns
         //  types by all format readers from file meta, maybe reading path columns types from BE then.
         for (String colName : pathPartitionKeys) {
-            columns.add(new Column(colName, Type.STRING, false));
+            columns.add(new Column(colName, ScalarType.createVarcharType(ScalarType.MAX_VARCHAR_LENGTH), false));
         }
     }
 

regression-test/data/external_table_p0/hive/write/test_hive_ctas_to_doris.out

@@ -0,0 +1,37 @@
+-- This file is automatically generated. You should know what you did if you want to edit this
+-- !q01 --
+65540	65533	4
+
+-- !q02 --
+id	INT	Yes	true	\N	
+str1	TEXT	Yes	true	\N	
+str2	TEXT	Yes	true	\N	
+str3	TEXT	Yes	true	\N	
+
+-- !q03 --
+65540	65533	4
+
+-- !q04 --
+id	INT	Yes	true	\N	
+str1	TEXT	Yes	false	\N	NONE
+str2	TEXT	Yes	false	\N	NONE
+str3	VARCHAR(65533)	Yes	false	\N	NONE
+
+-- !q05 --
+65540	65533	4
+
+-- !q06 --
+id	INT	Yes	true	\N	
+str1	TEXT	Yes	false	\N	NONE
+str2	VARCHAR(65533)	Yes	false	\N	NONE
+str3	TEXT	Yes	false	\N	NONE
+
+-- !q07 --
+65540	65533	4
+
+-- !q08 --
+id	INT	Yes	true	\N	
+str1	TEXT	Yes	false	\N	NONE
+str2	TEXT	Yes	false	\N	NONE
+str3	VARCHAR(65533)	Yes	false	\N	NONE
+