[fix](agg_state) adjust nullable should apply on agg_state inner type too (#37489)

fe/fe-core/src/main/java/org/apache/doris/nereids/analyzer/MappingSlot.java

@@ -102,7 +102,12 @@ public Slot withName(String name) {
     }
 
     @Override
-    public Slot withNullable(boolean newNullable) {
+    public Slot withNullable(boolean nullable) {
+        return this;
+    }
+
+    @Override
+    public Slot withNullableAndDataType(boolean nullable, DataType dataType) {
         return this;
     }
 

fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/AdjustNullable.java

@@ -293,20 +293,20 @@ private <T extends Expression> Set<T> updateExpressions(Set<T> inputs, Map<ExprI
         return result.build();
     }
 
-    private Map<ExprId, Slot> collectChildrenOutputMap(LogicalPlan plan) {
-        return plan.children().stream()
-                .map(Plan::getOutputSet)
-                .flatMap(Set::stream)
-                .collect(Collectors.toMap(NamedExpression::getExprId, s -> s));
-    }
-
     private static class SlotReferenceReplacer extends DefaultExpressionRewriter<Map<ExprId, Slot>> {
         public static SlotReferenceReplacer INSTANCE = new SlotReferenceReplacer();
 
         @Override
         public Expression visitSlotReference(SlotReference slotReference, Map<ExprId, Slot> context) {
             if (context.containsKey(slotReference.getExprId())) {
-                return slotReference.withNullable(context.get(slotReference.getExprId()).nullable());
+                Slot slot = context.get(slotReference.getExprId());
+                if (slot.getDataType().isAggStateType()) {
+                    // we must replace data type, because nested type and agg state contains nullable of their children.
+                    // TODO: remove if statement after we ensure be constant folding do not change expr type at all.
+                    return slotReference.withNullableAndDataType(slot.nullable(), slot.getDataType());
+                } else {
+                    return slotReference.withNullable(slot.nullable());
+                }
             } else {
                 return slotReference;
             }

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/ArrayItemReference.java

@@ -158,7 +158,12 @@ public ArrayItemSlot withName(String name) {
         }
 
         @Override
-        public SlotReference withNullable(boolean newNullable) {
+        public SlotReference withNullable(boolean nullable) {
+            return new ArrayItemSlot(exprId, name.get(), dataType, this.nullable);
+        }
+
+        @Override
+        public Slot withNullableAndDataType(boolean nullable, DataType dataType) {
             return new ArrayItemSlot(exprId, name.get(), dataType, nullable);
         }
 

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/Slot.java

@@ -19,6 +19,7 @@
 
 import org.apache.doris.common.Pair;
 import org.apache.doris.nereids.trees.expressions.shape.LeafExpression;
+import org.apache.doris.nereids.types.DataType;
 
 import com.google.common.collect.ImmutableList;
 
@@ -43,7 +44,11 @@ public Slot toSlot() {
         return this;
     }
 
-    public Slot withNullable(boolean newNullable) {
+    public Slot withNullable(boolean nullable) {
+        throw new RuntimeException("Do not implement");
+    }
+
+    public Slot withNullableAndDataType(boolean nullable, DataType dataType) {
         throw new RuntimeException("Do not implement");
     }
 

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/SlotReference.java

@@ -249,12 +249,20 @@ public SlotReference withChildren(List<Expression> children) {
     }
 
     @Override
-    public SlotReference withNullable(boolean newNullable) {
-        if (this.nullable == newNullable) {
+    public SlotReference withNullable(boolean nullable) {
+        if (this.nullable == nullable) {
             return this;
         }
+        return new SlotReference(exprId, name, dataType, nullable,
+                qualifier, table, column, internalName, subPath, indexInSqlString);
+    }
 
-        return new SlotReference(exprId, name, dataType, newNullable,
+    @Override
+    public Slot withNullableAndDataType(boolean nullable, DataType dataType) {
+        if (this.nullable == nullable && this.dataType.equals(dataType)) {
+            return this;
+        }
+        return new SlotReference(exprId, name, dataType, nullable,
                 qualifier, table, column, internalName, subPath, indexInSqlString);
     }
 

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/VirtualSlotReference.java

@@ -121,11 +121,20 @@ public boolean nullable() {
         return false;
     }
 
-    public VirtualSlotReference withNullable(boolean newNullable) {
-        if (this.nullable == newNullable) {
+    public VirtualSlotReference withNullable(boolean nullable) {
+        if (this.nullable == nullable) {
             return this;
         }
-        return new VirtualSlotReference(exprId, name.get(), dataType, newNullable, qualifier,
+        return new VirtualSlotReference(exprId, name.get(), dataType, nullable, qualifier,
+                originExpression, computeLongValueMethod);
+    }
+
+    @Override
+    public Slot withNullableAndDataType(boolean nullable, DataType dataType) {
+        if (this.nullable == nullable && this.dataType.equals(dataType)) {
+            return this;
+        }
+        return new VirtualSlotReference(exprId, name.get(), dataType, nullable, qualifier,
                 originExpression, computeLongValueMethod);
     }
 

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

@@ -107,6 +107,8 @@ public void run(ConnectContext ctx, StmtExecutor executor) throws Exception {
         LogicalPlan query = ctasQuery.get();
         List<String> ctasCols = createTableInfo.getCtasColumns();
         NereidsPlanner planner = new NereidsPlanner(ctx.getStatementContext());
+        // must disable constant folding by be, because be constant folding may return wrong type
+        ctx.getSessionVariable().disableConstantFoldingByBEOnce();
         Plan plan = planner.plan(new UnboundResultSink<>(query), PhysicalProperties.ANY, ExplainLevel.NONE);
         if (ctasCols == null) {
             // we should analyze the plan firstly to get the columns' name.

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

@@ -141,7 +141,7 @@ public CreateMTMVInfo(boolean ifNotExists, TableNameInfo mvName,
     /**
      * analyze create table info
      */
-    public void analyze(ConnectContext ctx) {
+    public void analyze(ConnectContext ctx) throws Exception {
         // analyze table name
         mvName.analyze(ctx);
         try {
@@ -201,12 +201,14 @@ private void analyzeProperties() {
     /**
      * analyzeQuery
      */
-    public void analyzeQuery(ConnectContext ctx, Map<String, String> mvProperties) {
+    public void analyzeQuery(ConnectContext ctx, Map<String, String> mvProperties) throws Exception {
         // create table as select
         StatementContext statementContext = ctx.getStatementContext();
         NereidsPlanner planner = new NereidsPlanner(statementContext);
         // this is for expression column name infer when not use alias
         LogicalSink<Plan> logicalSink = new UnboundResultSink<>(logicalQuery);
+        // must disable constant folding by be, because be constant folding may return wrong type
+        ctx.getSessionVariable().disableConstantFoldingByBEOnce();
         Plan plan = planner.plan(logicalSink, PhysicalProperties.ANY, ExplainLevel.ALL_PLAN);
         if (plan.anyMatch(node -> node instanceof OneRowRelation)) {
             throw new AnalysisException("at least contain one table");

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

@@ -3702,6 +3702,15 @@ public void enableFallbackToOriginalPlannerOnce() throws DdlException {
                 new SetVar(SessionVariable.ENABLE_FALLBACK_TO_ORIGINAL_PLANNER, new StringLiteral("true")));
     }
 
+    public void disableConstantFoldingByBEOnce() throws DdlException {
+        if (!enableFoldConstantByBe) {
+            return;
+        }
+        setIsSingleSetVar(true);
+        VariableMgr.setVar(this,
+                new SetVar(SessionVariable.ENABLE_FOLD_CONSTANT_BY_BE, new StringLiteral("false")));
+    }
+
     public void disableNereidsPlannerOnce() throws DdlException {
         if (!enableNereidsPlanner) {
             return;

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

@@ -24,3 +24,6 @@ r2	{"title":"Amount","value":2.1}
 -- !desc --
 __substring_0	VARCHAR(30)	Yes	true	\N	
 
+-- !desc --
+__substring_0	VARCHAR(30)	Yes	true	\N	
+

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

@@ -270,6 +270,11 @@ suite("nereids_test_ctas") {
 
     sql """DROP TABLE IF EXISTS test_varchar_length"""
     sql """set use_max_length_of_varchar_in_ctas = false"""
+    sql """set enable_fold_constant_by_be = 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 enable_fold_constant_by_be = true"""
     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"""

regression-test/suites/nereids_rules_p0/adjust_nullable/agg_state_type.groovy

@@ -0,0 +1,39 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+suite("test_agg_state_type_adjust_nullable") {
+
+    sql """
+        DROP TABLE IF EXISTS test_agg_state_type
+    """
+
+    sql """
+        CREATE TABLE test_agg_state_type(`id` INT NOT NULL, `c1` INT NOT NULL) DISTRIBUTED BY hash(id) PROPERTIES ("replication_num" = "1")
+    """
+
+    sql """
+        insert into test_agg_state_type values (1, 1)
+    """
+
+    sql """
+        select * from (select sum_state(c1) as c2, 1 as c1 from (select avg(id) as c1 from test_agg_state_type) v) v join test_agg_state_type on v.c1 = test_agg_state_type.id
+    """
+
+    sql """
+        DROP TABLE IF EXISTS test_agg_state_type
+    """
+}