[fix](fe) Infer CAST nullability when building schema change tasks

fe/fe-core/src/main/java/org/apache/doris/alter/SchemaChangeJobV2.java

@@ -52,6 +52,7 @@
 import org.apache.doris.common.util.DebugPointUtil;
 import org.apache.doris.common.util.TimeUtils;
 import org.apache.doris.nereids.rules.expression.check.CheckCast;
+import org.apache.doris.nereids.trees.expressions.Cast;
 import org.apache.doris.nereids.types.DataType;
 import org.apache.doris.persist.gson.GsonPostProcessable;
 import org.apache.doris.persist.gson.GsonUtils;
@@ -544,6 +545,8 @@ protected void runWaitingTxnJob() throws AlterCancelException {
                                 if (!(srcType.isDecimalV2Type() && destType.isDecimalV2Type()
                                         || srcType.isStringLikeType() && destType.isStringLikeType())) {
                                     defineExpr = new CastExpr(newColumn.getType(), defineExpr, null);
+                                    defineExpr.setNullableFromNereids(
+                                            Cast.castNullable(slot.isNullable(), srcType, destType));
                                 }
                                 defineExprs.put(column.getName(), defineExpr);
                             }

fe/fe-core/src/test/java/org/apache/doris/alter/SchemaChangeJobV2Test.java

@@ -23,6 +23,7 @@
 import org.apache.doris.analysis.ColumnDef;
 import org.apache.doris.analysis.ColumnDef.DefaultValue;
 import org.apache.doris.analysis.ColumnPosition;
+import org.apache.doris.analysis.ModifyColumnClause;
 import org.apache.doris.analysis.ModifyTablePropertiesClause;
 import org.apache.doris.analysis.TypeDef;
 import org.apache.doris.backup.CatalogMocker;
@@ -58,6 +59,10 @@
 import org.apache.doris.meta.MetaContext;
 import org.apache.doris.task.AgentTask;
 import org.apache.doris.task.AgentTaskQueue;
+import org.apache.doris.task.AlterReplicaTask;
+import org.apache.doris.thrift.TAlterTabletReqV2;
+import org.apache.doris.thrift.TExprNode;
+import org.apache.doris.thrift.TExprNodeType;
 import org.apache.doris.thrift.TStorageFormat;
 import org.apache.doris.thrift.TTaskType;
 import org.apache.doris.transaction.FakeTransactionIDGenerator;
@@ -124,6 +129,49 @@ public void setUp()
         AgentTaskQueue.clearAllTasks();
     }
 
+    @Test
+    public void testDatetimeScaleIncreaseCastNullable() throws Exception {
+        assertDatetimeScaleCastNullable(3, 6, false);
+    }
+
+    @Test
+    public void testDatetimeScaleDecreaseCastNullable() throws Exception {
+        assertDatetimeScaleCastNullable(6, 3, false);
+    }
+
+    @Test
+    public void testDatetimeScaleNullableInput() throws Exception {
+        assertDatetimeScaleCastNullable(3, 6, true);
+    }
+
+    private void assertDatetimeScaleCastNullable(int sourceScale, int targetScale, boolean sourceNullable)
+            throws Exception {
+        FakeEnv.setEnv(masterEnv);
+        Database db = masterEnv.getInternalCatalog().getDbOrDdlException(CatalogTestUtil.testDbId1);
+        OlapTable table = (OlapTable) db.getTableOrDdlException(CatalogTestUtil.testTableId2);
+        table.renameIndexForSchemaChange(CatalogTestUtil.testIndex2, table.getName());
+        Column source = table.getColumn("v1");
+        source.setType(ScalarType.createDatetimeV2Type(sourceScale));
+        source.setIsAllowNull(sourceNullable);
+        Column target = new Column(source);
+        target.setType(ScalarType.createDatetimeV2Type(targetScale));
+        ModifyColumnClause clause = new ModifyColumnClause("", target, null, null, null);
+        SchemaChangeHandler handler = Env.getCurrentEnv().getSchemaChangeHandler();
+        handler.process(Lists.newArrayList(clause), db, table);
+        SchemaChangeJobV2 job = (SchemaChangeJobV2) handler.getAlterJobsV2().values().iterator().next();
+        handler.runAfterCatalogReady();
+        Assert.assertEquals(JobState.WAITING_TXN, job.getJobState());
+        handler.runAfterCatalogReady();
+        Assert.assertEquals(JobState.RUNNING, job.getJobState());
+        List<AgentTask> tasks = AgentTaskQueue.getTask(TTaskType.ALTER);
+        Assert.assertEquals(1, tasks.size());
+        TAlterTabletReqV2 request = ((AlterReplicaTask) tasks.get(0)).toThrift();
+        Assert.assertEquals(1, request.getMaterializedViewParamsSize());
+        TExprNode cast = request.getMaterializedViewParams().get(0).getMvExpr().getNodes().get(0);
+        Assert.assertEquals(TExprNodeType.CAST_EXPR, cast.getNodeType());
+        Assert.assertTrue(cast.isIsNullable());
+    }
+
     @Test
     public void testAddSchemaChange() throws UserException {
         fakeEnv = new FakeEnv();

fe/fe-core/src/test/java/org/apache/doris/analysis/CastExprTest.java

@@ -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.
+
+package org.apache.doris.analysis;
+
+import org.apache.doris.catalog.Type;
+import org.apache.doris.common.Config;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class CastExprTest {
+    @Test
+    public void testLegacyVariantDateFallbackRemainsNullable() {
+        boolean original = Config.enable_variant_v2;
+        try {
+            Config.enable_variant_v2 = false;
+            CastExpr cast = new CastExpr(Type.DATEV2, new SlotRef(Type.VARIANT, false), null);
+            Assertions.assertTrue(cast.isNullable());
+            Assertions.assertTrue(cast.treeToThrift().getNodes().get(0).isIsNullable());
+        } finally {
+            Config.enable_variant_v2 = original;
+        }
+    }
+}

regression-test/data/schema_change_p0/test_schema_change_cast_nullable.out

@@ -0,0 +1,29 @@
+-- This file is automatically generated. You should know what you did if you want to edit this
+-- !before --
+1	2024-02-29T12:34:56.123	2024-02-29T12:34:56.123
+2	9999-12-31T23:59:59.999	\N
+3	2000-01-01T00:00	2000-01-01T00:00:00.001
+
+-- !after --
+1	2024-02-29T12:34:56.123	2024-02-29T12:34:56.123
+2	9999-12-31T23:59:59.999	\N
+3	2000-01-01T00:00	2000-01-01T00:00:00.001
+
+-- !insert_after --
+1	2024-02-29T12:34:56.123	2024-02-29T12:34:56.123
+2	9999-12-31T23:59:59.999	\N
+3	2000-01-01T00:00	2000-01-01T00:00:00.001
+4	2024-02-29T12:34:56.123456	2024-02-29T12:34:56.654321
+
+-- !expected_downscale --
+1	2024-02-29T12:34:56.123	2024-02-29T12:34:56.123
+2	9999-12-31T23:59:59.999	\N
+3	2000-01-01T00:00	2000-01-01T00:00:00.001
+4	2024-02-29T12:34:56.123	2024-02-29T12:34:56.654
+
+-- !after_downscale --
+1	2024-02-29T12:34:56.123	2024-02-29T12:34:56.123
+2	9999-12-31T23:59:59.999	\N
+3	2000-01-01T00:00	2000-01-01T00:00:00.001
+4	2024-02-29T12:34:56.123	2024-02-29T12:34:56.654
+

regression-test/suites/schema_change_p0/test_schema_change_cast_nullable.groovy

@@ -0,0 +1,68 @@
+// 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_schema_change_cast_nullable", "p0") {
+    sql "DROP TABLE IF EXISTS schema_change_cast_nullable"
+    sql """
+        CREATE TABLE schema_change_cast_nullable (
+            id INT NOT NULL,
+            required_dt DATETIMEV2(3) NOT NULL,
+            nullable_dt DATETIMEV2(3) NULL
+        ) DUPLICATE KEY(id)
+        DISTRIBUTED BY HASH(id) BUCKETS 1
+        PROPERTIES("replication_num" = "1", "light_schema_change" = "false")
+    """
+    sql """
+        INSERT INTO schema_change_cast_nullable VALUES
+        (1, '2024-02-29 12:34:56.123', '2024-02-29 12:34:56.123'),
+        (2, '9999-12-31 23:59:59.999', NULL),
+        (3, '2000-01-01 00:00:00.000', '2000-01-01 00:00:00.001')
+    """
+    qt_before "SELECT * FROM schema_change_cast_nullable ORDER BY id"
+    sql """
+        ALTER TABLE schema_change_cast_nullable
+        MODIFY COLUMN required_dt DATETIMEV2(6) NOT NULL,
+        MODIFY COLUMN nullable_dt DATETIMEV2(6) NULL
+    """
+    waitForSchemaChangeDone {
+        sql """SHOW ALTER TABLE COLUMN WHERE IndexName='schema_change_cast_nullable'
+               ORDER BY JobId DESC LIMIT 1"""
+        time 120
+    }
+    qt_after "SELECT * FROM schema_change_cast_nullable ORDER BY id"
+    // Exercise writes using the new schema as well as the rewritten historical rowsets.
+    sql """
+        INSERT INTO schema_change_cast_nullable VALUES
+        (4, '2024-02-29 12:34:56.123456', '2024-02-29 12:34:56.654321')
+    """
+    qt_insert_after "SELECT * FROM schema_change_cast_nullable ORDER BY id"
+    qt_expected_downscale """
+        SELECT id, CAST(required_dt AS DATETIMEV2(3)), CAST(nullable_dt AS DATETIMEV2(3))
+        FROM schema_change_cast_nullable ORDER BY id
+    """
+    sql """
+        ALTER TABLE schema_change_cast_nullable
+        MODIFY COLUMN required_dt DATETIMEV2(3) NOT NULL,
+        MODIFY COLUMN nullable_dt DATETIMEV2(3) NULL
+    """
+    waitForSchemaChangeDone {
+        sql """SHOW ALTER TABLE COLUMN WHERE IndexName='schema_change_cast_nullable'
+               ORDER BY JobId DESC LIMIT 1"""
+        time 120
+    }
+    qt_after_downscale "SELECT * FROM schema_change_cast_nullable ORDER BY id"
+}