[fix](Nereids) null type in result set will be cast to tinyint (#37019)

fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/BindExpression.java

@@ -42,6 +42,7 @@
 import org.apache.doris.nereids.rules.expression.ExpressionRewriteContext;
 import org.apache.doris.nereids.trees.expressions.Alias;
 import org.apache.doris.nereids.trees.expressions.BoundStar;
+import org.apache.doris.nereids.trees.expressions.Cast;
 import org.apache.doris.nereids.trees.expressions.DefaultValueSlot;
 import org.apache.doris.nereids.trees.expressions.EqualTo;
 import org.apache.doris.nereids.trees.expressions.ExprId;
@@ -89,8 +90,11 @@
 import org.apache.doris.nereids.trees.plans.logical.UsingJoin;
 import org.apache.doris.nereids.trees.plans.visitor.InferPlanOutputAlias;
 import org.apache.doris.nereids.types.BooleanType;
+import org.apache.doris.nereids.types.DataType;
+import org.apache.doris.nereids.types.NullType;
 import org.apache.doris.nereids.types.StructField;
 import org.apache.doris.nereids.types.StructType;
+import org.apache.doris.nereids.types.TinyIntType;
 import org.apache.doris.nereids.util.ExpressionUtils;
 import org.apache.doris.nereids.util.PlanUtils;
 import org.apache.doris.nereids.util.TypeCoercionUtils;
@@ -208,22 +212,38 @@ protected boolean condition(Rule rule, Plan plan) {
 
     private LogicalResultSink<Plan> bindResultSink(MatchingContext<UnboundResultSink<Plan>> ctx) {
         LogicalSink<Plan> sink = ctx.root;
+        Plan child = sink.child();
+        List<Slot> output = child.getOutput();
+        List<NamedExpression> castNullToTinyInt = Lists.newArrayListWithCapacity(output.size());
+        boolean needProject = false;
+        for (Slot slot : output) {
+            DataType newType = TypeCoercionUtils.replaceSpecifiedType(
+                    slot.getDataType(), NullType.class, TinyIntType.INSTANCE);
+            if (!newType.equals(slot.getDataType())) {
+                needProject = true;
+                castNullToTinyInt.add(new Alias(new Cast(slot, newType)));
+            } else {
+                castNullToTinyInt.add(slot);
+            }
+        }
+        if (needProject) {
+            child = new LogicalProject<>(castNullToTinyInt, child);
+        }
         if (ctx.connectContext.getState().isQuery()) {
-            List<NamedExpression> outputExprs = sink.child().getOutput().stream()
+            List<NamedExpression> outputExprs = child.getOutput().stream()
                     .map(NamedExpression.class::cast)
                     .collect(ImmutableList.toImmutableList());
-            return new LogicalResultSink<>(outputExprs, sink.child());
+            return new LogicalResultSink<>(outputExprs, child);
         }
         // Should infer column name for expression when query command
-        final ImmutableListMultimap.Builder<ExprId, Integer> exprIdToIndexMapBuilder =
-                ImmutableListMultimap.builder();
-        List<Slot> childOutput = sink.child().getOutput();
+        final ImmutableListMultimap.Builder<ExprId, Integer> exprIdToIndexMapBuilder = ImmutableListMultimap.builder();
+        List<Slot> childOutput = child.getOutput();
         for (int index = 0; index < childOutput.size(); index++) {
             exprIdToIndexMapBuilder.put(childOutput.get(index).getExprId(), index);
         }
         InferPlanOutputAlias aliasInfer = new InferPlanOutputAlias(childOutput);
-        List<NamedExpression> output = aliasInfer.infer(sink.child(), exprIdToIndexMapBuilder.build());
-        return new LogicalResultSink<>(output, sink.child());
+        List<NamedExpression> sinkExpr = aliasInfer.infer(child, exprIdToIndexMapBuilder.build());
+        return new LogicalResultSink<>(sinkExpr, child);
     }
 
     private LogicalSubQueryAlias<Plan> bindSubqueryAlias(MatchingContext<LogicalSubQueryAlias<Plan>> ctx) {

fe/fe-core/src/main/java/org/apache/doris/nereids/util/TypeCoercionUtils.java

@@ -341,7 +341,8 @@ public static DataType replaceDateTimeV2WithMax(DataType dataType) {
     public static DataType replaceSpecifiedType(DataType dataType,
             Class<? extends DataType> specifiedType, DataType newType) {
         if (dataType instanceof ArrayType) {
-            return ArrayType.of(replaceSpecifiedType(((ArrayType) dataType).getItemType(), specifiedType, newType));
+            return ArrayType.of(replaceSpecifiedType(((ArrayType) dataType).getItemType(), specifiedType, newType),
+                    ((ArrayType) dataType).containsNull());
         } else if (dataType instanceof MapType) {
             return MapType.of(replaceSpecifiedType(((MapType) dataType).getKeyType(), specifiedType, newType),
                     replaceSpecifiedType(((MapType) dataType).getValueType(), specifiedType, newType));

regression-test/data/mtmv_p0/test_create_with_null_type.out

@@ -0,0 +1,7 @@
+-- This file is automatically generated. You should know what you did if you want to edit this
+-- !select --
+\N
+
+-- !desc --
+test_create_with_null_type	DUP_KEYS	__cast_0	TINYINT	TINYINT	Yes	true	\N		true		
+

regression-test/suites/mtmv_p0/test_create_with_null_type.groovy

@@ -0,0 +1,64 @@
+// 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.
+
+import org.junit.Assert;
+
+suite("test_create_with_null_type") {
+    def tableName = "t_test_create_with_null_type"
+    def mvName = "test_create_with_null_type"
+    def dbName = "regression_test_mtmv_p0"
+    sql """drop table if exists `${tableName}`"""
+    sql """drop materialized view if exists ${mvName};"""
+
+    sql """
+        CREATE TABLE `${tableName}` (
+          `user_id` LARGEINT NOT NULL COMMENT '\"用户id\"',
+          `num` SMALLINT SUM NOT NULL COMMENT '\"数量\"'
+        ) ENGINE=OLAP
+        AGGREGATE KEY(`user_id`)
+        COMMENT 'OLAP'
+        DISTRIBUTED BY HASH(`user_id`) BUCKETS 2
+        PROPERTIES ('replication_num' = '1') ;
+        """
+    sql """
+        insert into ${tableName} values (1,1),(1,2);
+        """
+
+    sql """
+        CREATE MATERIALIZED VIEW ${mvName}
+            BUILD DEFERRED REFRESH AUTO ON MANUAL
+            DISTRIBUTED BY RANDOM BUCKETS 2
+            PROPERTIES ('replication_num' = '1')
+            AS
+            SELECT null FROM ${tableName};
+    """
+
+    sql """
+        REFRESH MATERIALIZED VIEW ${mvName} AUTO;
+    """
+
+    def jobName = getJobName(dbName, mvName);
+    log.info(jobName)
+    waitingMTMVTaskFinished(jobName)
+
+    order_qt_select "SELECT * FROM ${mvName}"
+
+    qt_desc "desc ${mvName} all"
+
+    sql """drop table if exists `${tableName}`"""
+    sql """drop materialized view if exists ${mvName};"""
+}