[opt](ctas) add a variable to control varchar length in ctas (#37069) (#37284)

pick from master #37069

add a new session variable: use_max_length_of_varchar_in_ctas

In CTAS (Create Table As Select), if CHAR/VARCHAR columns do not
originate from the source table, whether to set the length of such a
column to MAX, which is 65533. The default is true.

Author: morrySnow

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

@@ -143,10 +143,12 @@ public void run(ConnectContext ctx, StmtExecutor executor) throws Exception {
                         }
                     }
                 } else {
-                    dataType = TypeCoercionUtils.replaceSpecifiedType(dataType,
-                            VarcharType.class, VarcharType.MAX_VARCHAR_TYPE);
-                    dataType = TypeCoercionUtils.replaceSpecifiedType(dataType,
-                            CharType.class, VarcharType.MAX_VARCHAR_TYPE);
+                    if (ctx.getSessionVariable().useMaxLengthOfVarcharInCtas) {
+                        dataType = TypeCoercionUtils.replaceSpecifiedType(dataType,
+                                VarcharType.class, VarcharType.MAX_VARCHAR_TYPE);
+                        dataType = TypeCoercionUtils.replaceSpecifiedType(dataType,
+                                CharType.class, VarcharType.MAX_VARCHAR_TYPE);
+                    }
                 }
             }
             // if the column is an expression, we set it to nullable, otherwise according to the nullable of the slot.

fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java

@@ -572,6 +572,8 @@ public class SessionVariable implements Serializable, Writable {
 
     public static final String MAX_COLUMN_READER_NUM = "max_column_reader_num";
 
+    public static final String USE_MAX_LENGTH_OF_VARCHAR_IN_CTAS = "use_max_length_of_varchar_in_ctas";
+
     public static final List<String> DEBUG_VARIABLES = ImmutableList.of(
             SKIP_DELETE_PREDICATE,
             SKIP_DELETE_BITMAP,
@@ -1902,6 +1904,13 @@ public void setIgnoreShapePlanNodes(String ignoreShapePlanNodes) {
             checker = "checkExternalAggPartitionBits", fuzzy = true)
     public int externalAggPartitionBits = 5; // means that the hash table will be partitioned into 32 blocks.
 
+    @VariableMgr.VarAttr(name = USE_MAX_LENGTH_OF_VARCHAR_IN_CTAS, description = {
+            "在CTAS中，如果 CHAR / VARCHAR 列不来自于源表，是否是将这一列的长度设置为 MAX，即65533。默认为 true。",
+            "In CTAS (Create Table As Select), if CHAR/VARCHAR columns do not originate from the source table,"
+                    + " whether to set the length of such a column to MAX, which is 65533. The default is true."
+    })
+    public boolean useMaxLengthOfVarcharInCtas = true;
+
     public boolean isEnableJoinSpill() {
         return enableJoinSpill;
     }

regression-test/data/nereids_p0/create_table/test_ctas.out

@@ -21,3 +21,6 @@ r2	{"title":"Amount","value":2.1}
 2.1	2.20000	2.3	2.400000	2.500000	2.600000
 2.1	2.20000	2.3	2.400000	2.500000	2.600000
 
+-- !desc --
+__substring_0	VARCHAR(30)	Yes	true	\N	
+

regression-test/suites/nereids_p0/create_table/test_ctas.groovy

@@ -16,10 +16,6 @@
 // under the License.
 
 suite("nereids_test_ctas") {
-    sql 'set enable_nereids_planner=true'
-    sql 'set enable_fallback_to_original_planner=false'
-    sql 'set enable_nereids_dml=true'
-
     sql """ DROP TABLE IF EXISTS test_ctas """
     sql """ DROP TABLE IF EXISTS test_ctas1 """
     sql """ DROP TABLE IF EXISTS test_ctas2 """
@@ -271,5 +267,13 @@ suite("nereids_test_ctas") {
         sql 'drop table c'
         sql 'drop table test_date_v2'
     }
+
+    sql """DROP TABLE IF EXISTS test_varchar_length"""
+    sql """set use_max_length_of_varchar_in_ctas = false"""
+    sql """CREATE TABLE test_varchar_length properties ("replication_num"="1") AS SELECT CAST("1" AS VARCHAR(30))"""
+    qt_desc """desc test_varchar_length"""
+    sql """DROP TABLE IF EXISTS test_varchar_length"""
+    sql """set use_max_length_of_varchar_in_ctas = true"""
+
 }