branch-4.1: [fix](cast) Declare Variant casts to non-Variant targets nullable

be/src/exec/operator/scan_operator.cpp

@@ -943,6 +943,15 @@ Status ScanLocalStateBase::_normalize_is_null_predicate(
         return Status::OK();
     }
 
+    // A Variant cast can introduce NULLs for non-NULL stored values, so `CAST(v AS T) IS NULL`
+    // is not equivalent to `v IS NULL`. Keep the null test in the expression tree instead of
+    // pushing a NullPredicate down on the source Variant column.
+    if (root->children()[0]->node_type() == TExprNodeType::CAST_EXPR &&
+        VExpr::expr_without_cast(root->children()[0])->data_type()->get_primitive_type() ==
+                PrimitiveType::TYPE_VARIANT) {
+        return Status::OK();
+    }
+
     auto fn_call = assert_cast<VectorizedFnCall*>(root.get());
     if (fn_call->fn().name.function_name == "is_null_pred") {
         pred = NullPredicate::create_shared(

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

@@ -34,7 +34,6 @@
 import org.apache.doris.nereids.types.SmallIntType;
 import org.apache.doris.nereids.types.TimeStampTzType;
 import org.apache.doris.nereids.types.TinyIntType;
-import org.apache.doris.nereids.types.VariantType;
 import org.apache.doris.nereids.types.coercion.DateLikeType;
 import org.apache.doris.nereids.util.DateUtils;
 
@@ -207,12 +206,8 @@ public static boolean castNullable(boolean srcNullable, DataType srcType, DataTy
         } else if (childDataType.isJsonType() && !targetType.isJsonType()) {
             // Json to other type is always nullable
             return true;
-        } else if (childDataType instanceof VariantType
-                && ((VariantType) childDataType).isExecutionV2() && !targetType.isVariantType()) {
-            // Variant V2 values can have a shape that is incompatible with the target type.
-            return true;
-        } else if (childDataType.isVariantType() && targetType.isJsonType()) {
-            // Legacy Variant to Json is always nullable.
+        } else if (childDataType.isVariantType() && !targetType.isVariantType()) {
+            // Variant nulls and incompatible values can produce SQL NULL in both execution representations.
             return true;
         }
         return false;

fe/fe-core/src/test/java/org/apache/doris/nereids/rules/expression/rules/VariantCastNullableTest.java

@@ -0,0 +1,67 @@
+// 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.nereids.rules.expression.rules;
+
+import org.apache.doris.common.Config;
+import org.apache.doris.nereids.rules.expression.ExpressionRewriteTestHelper;
+import org.apache.doris.nereids.rules.expression.ExpressionRuleExecutor;
+import org.apache.doris.nereids.trees.expressions.Cast;
+import org.apache.doris.nereids.trees.expressions.IsNull;
+import org.apache.doris.nereids.trees.expressions.NullSafeEqual;
+import org.apache.doris.nereids.trees.expressions.SlotReference;
+import org.apache.doris.nereids.trees.expressions.functions.scalar.Coalesce;
+import org.apache.doris.nereids.trees.expressions.functions.scalar.Nvl;
+import org.apache.doris.nereids.trees.expressions.literal.NullLiteral;
+import org.apache.doris.nereids.trees.expressions.literal.StringLiteral;
+import org.apache.doris.nereids.types.StringType;
+import org.apache.doris.nereids.types.VariantType;
+
+import com.google.common.collect.ImmutableList;
+import org.junit.jupiter.api.Test;
+
+public class VariantCastNullableTest extends ExpressionRewriteTestHelper {
+    @Test
+    public void testNullPredicatesRetainVariantCast() {
+        boolean original = Config.enable_variant_v2;
+        try {
+            Config.enable_variant_v2 = false;
+            executor = new ExpressionRuleExecutor(ImmutableList.of(
+                    bottomUp(NullSafeEqualToEqual.INSTANCE)));
+            Cast cast = new Cast(new SlotReference("v", VariantType.INSTANCE, false), StringType.INSTANCE);
+            assertRewrite(new IsNull(cast), new IsNull(cast));
+            assertRewrite(new NullSafeEqual(cast, new NullLiteral(StringType.INSTANCE)), new IsNull(cast));
+        } finally {
+            Config.enable_variant_v2 = original;
+        }
+    }
+
+    @Test
+    public void testConditionalFunctionsRetainVariantCastFallback() {
+        boolean original = Config.enable_variant_v2;
+        try {
+            Config.enable_variant_v2 = false;
+            executor = new ExpressionRuleExecutor(ImmutableList.of(bottomUp(SimplifyConditionalFunction.INSTANCE)));
+            Cast cast = new Cast(new SlotReference("v", VariantType.INSTANCE, false), StringType.INSTANCE);
+            StringLiteral fallback = new StringLiteral("fallback");
+            assertRewrite(new Coalesce(cast, fallback), new Coalesce(cast, fallback));
+            assertRewrite(new Nvl(cast, fallback), new Nvl(cast, fallback));
+        } finally {
+            Config.enable_variant_v2 = original;
+        }
+    }
+}

fe/fe-core/src/test/java/org/apache/doris/nereids/trees/expressions/CastTest.java

@@ -641,11 +641,11 @@ public void testCastFromVariant() {
         boolean originalEnableVariantV2 = Config.enable_variant_v2;
         try {
             Config.enable_variant_v2 = false;
-            Assertions.assertFalse(castNullable(false, VariantType.INSTANCE, BooleanType.INSTANCE));
-            Assertions.assertFalse(castNullable(false, VariantType.INSTANCE, IntegerType.INSTANCE));
-            Assertions.assertFalse(castNullable(false, VariantType.INSTANCE, StringType.INSTANCE));
+            Assertions.assertTrue(castNullable(false, VariantType.INSTANCE, BooleanType.INSTANCE));
+            Assertions.assertTrue(castNullable(false, VariantType.INSTANCE, IntegerType.INSTANCE));
+            Assertions.assertTrue(castNullable(false, VariantType.INSTANCE, StringType.INSTANCE));
             Assertions.assertTrue(castNullable(false, VariantType.INSTANCE, JsonType.INSTANCE));
-            Assertions.assertFalse(castNullable(false, VariantType.INSTANCE, ArrayType.of(IntegerType.INSTANCE)));
+            Assertions.assertTrue(castNullable(false, VariantType.INSTANCE, ArrayType.of(IntegerType.INSTANCE)));
             Assertions.assertFalse(castNullable(false, VariantType.INSTANCE, VariantType.INSTANCE));
             Assertions.assertTrue(castNullable(true, VariantType.INSTANCE, VariantType.INSTANCE));
 

regression-test/data/variant_p0/test_variant_cast_isnull_pushdown.out

@@ -0,0 +1,17 @@
+-- This file is automatically generated. You should know what you did if you want to edit this
+-- !cast_int --
+1	1
+2	\N
+3	\N
+
+-- !is_null --
+2
+3
+
+-- !is_not_null --
+1
+
+-- !null_safe --
+2
+3
+

regression-test/data/variant_p0/test_variant_cast_nullable.out

@@ -0,0 +1,53 @@
+-- This file is automatically generated. You should know what you did if you want to edit this
+-- !casts --
+1	{}	\N	\N
+2	42	42	42
+3	{"a":1}	0	0
+4	"invalid"	\N	\N
+5	[1,2]	\N	\N
+
+-- !null_predicates --
+1	false	false	true	true	false
+2	false	false	false	false	false
+3	false	false	false	false	false
+4	false	false	true	true	false
+5	false	false	true	true	false
+
+-- !is_null_filter --
+
+-- !is_not_null_filter --
+2
+3
+
+-- !nullable_input_filter --
+1
+4
+5
+
+-- !null_safe_filter --
+1
+4
+5
+
+-- !fallbacks --
+1	{}	{}	99	99	99	99
+2	42	42	42	42	42	42
+3	{"a":1}	{"a":1}	0	0	0	0
+4	"invalid"	"invalid"	99	99	99	99
+5	[1,2]	[1,2]	99	99	99	99
+
+-- !alias_null_predicate --
+1	\N	99
+4	\N	99
+5	\N	99
+
+-- !all_null_filter --
+1
+2
+
+-- !all_not_null_filter --
+
+-- !all_null_roots --
+1	\N	true	true	fallback	fallback	99	99
+2	\N	true	true	fallback	fallback	99	99
+

regression-test/suites/variant_p0/test_variant_cast_isnull_pushdown.groovy

@@ -0,0 +1,42 @@
+// 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_variant_cast_isnull_pushdown", "p0") {
+    // A Variant cast can turn a non-NULL Variant value into NULL, so `CAST(v AS INT) IS NULL`
+    // is not equivalent to `v IS NULL`. The scan node must keep this null test as an expression
+    // instead of pushing a null predicate down on the source Variant column.
+    sql "DROP TABLE IF EXISTS variant_cast_isnull_pushdown"
+    sql """
+        CREATE TABLE variant_cast_isnull_pushdown (
+            id INT NOT NULL,
+            v VARIANT NOT NULL
+        ) DUPLICATE KEY(id)
+        DISTRIBUTED BY HASH(id) BUCKETS 1
+        PROPERTIES("replication_num" = "1")
+    """
+    sql """
+        INSERT INTO variant_cast_isnull_pushdown VALUES
+            (1, '1'),
+            (2, '"x"'),
+            (3, '[1,2]')
+    """
+
+    order_qt_cast_int "SELECT id, CAST(v AS INT) FROM variant_cast_isnull_pushdown ORDER BY id"
+    order_qt_is_null "SELECT id FROM variant_cast_isnull_pushdown WHERE CAST(v AS INT) IS NULL ORDER BY id"
+    order_qt_is_not_null "SELECT id FROM variant_cast_isnull_pushdown WHERE CAST(v AS INT) IS NOT NULL ORDER BY id"
+    order_qt_null_safe "SELECT id FROM variant_cast_isnull_pushdown WHERE CAST(v AS INT) <=> NULL ORDER BY id"
+}

regression-test/suites/variant_p0/test_variant_cast_nullable.groovy

@@ -0,0 +1,98 @@
+// 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_variant_cast_nullable", "p0") {
+    // This regression covers legacy Variant conversion results; V2 has different conversion semantics.
+    if (getFeConfig("enable_variant_v2").toBoolean()) {
+        return
+    }
+    sql "DROP TABLE IF EXISTS variant_cast_nullable"
+    sql """
+        CREATE TABLE variant_cast_nullable (
+            id INT NOT NULL,
+            v VARIANT NOT NULL,
+            nullable_v VARIANT NULL
+        ) DUPLICATE KEY(id)
+        DISTRIBUTED BY HASH(id) BUCKETS 1
+        PROPERTIES("replication_num" = "1")
+    """
+    sql """
+        INSERT INTO variant_cast_nullable VALUES
+        (1, 'null', NULL),
+        (2, '42', '42'),
+        (3, '{"a":1}', '{"a":1}'),
+        (4, '"invalid"', '"invalid"'),
+        (5, '[1,2]', '[1,2]')
+    """
+
+    qt_casts """
+        SELECT id, CAST(v AS STRING), CAST(v AS INT), CAST(nullable_v AS INT)
+        FROM variant_cast_nullable ORDER BY id
+    """
+    qt_null_predicates """
+        SELECT id, CAST(v AS STRING) IS NULL, CAST(v AS STRING) <=> NULL,
+               CAST(v AS INT) IS NULL, CAST(v AS INT) <=> NULL,
+               CAST(v AS VARIANT) IS NULL
+        FROM variant_cast_nullable ORDER BY id
+    """
+    qt_is_null_filter """
+        SELECT id FROM variant_cast_nullable WHERE CAST(v AS STRING) IS NULL ORDER BY id
+    """
+    qt_is_not_null_filter """
+        SELECT id FROM variant_cast_nullable WHERE CAST(v AS INT) IS NOT NULL ORDER BY id
+    """
+    qt_nullable_input_filter """
+        SELECT id FROM variant_cast_nullable WHERE CAST(nullable_v AS INT) IS NULL ORDER BY id
+    """
+    qt_null_safe_filter """
+        SELECT id FROM variant_cast_nullable WHERE CAST(v AS INT) <=> NULL ORDER BY id
+    """
+    qt_fallbacks """
+        SELECT id, COALESCE(CAST(v AS STRING), 'fallback'), NVL(CAST(v AS STRING), 'fallback'),
+               COALESCE(CAST(v AS INT), 99), NVL(CAST(v AS INT), 99),
+               COALESCE(CAST(nullable_v AS INT), 99), NVL(CAST(nullable_v AS INT), 99)
+        FROM variant_cast_nullable ORDER BY id
+    """
+    qt_alias_null_predicate """
+        WITH casts AS (SELECT id, CAST(v AS INT) AS value FROM variant_cast_nullable)
+        SELECT id, value, COALESCE(value, 99) FROM casts WHERE value IS NULL ORDER BY id
+    """
+
+    // Keep an all-Variant-null root to exercise the legacy all-default-values cast path.
+    sql "DROP TABLE IF EXISTS variant_cast_nullable_nulls"
+    sql """
+        CREATE TABLE variant_cast_nullable_nulls (
+            id INT NOT NULL,
+            v VARIANT NOT NULL
+        ) DUPLICATE KEY(id)
+        DISTRIBUTED BY HASH(id) BUCKETS 1
+        PROPERTIES("replication_num" = "1")
+    """
+    sql "INSERT INTO variant_cast_nullable_nulls VALUES (1, 'null'), (2, 'null')"
+    qt_all_null_filter """
+        SELECT id FROM variant_cast_nullable_nulls WHERE CAST(v AS STRING) IS NULL ORDER BY id
+    """
+    qt_all_not_null_filter """
+        SELECT id FROM variant_cast_nullable_nulls WHERE CAST(v AS STRING) IS NOT NULL ORDER BY id
+    """
+    qt_all_null_roots """
+        SELECT id, CAST(v AS STRING), CAST(v AS STRING) IS NULL, CAST(v AS STRING) <=> NULL,
+               COALESCE(CAST(v AS STRING), 'fallback'), NVL(CAST(v AS STRING), 'fallback'),
+               COALESCE(CAST(v AS INT), 99), NVL(CAST(v AS INT), 99)
+        FROM variant_cast_nullable_nulls ORDER BY id
+    """
+}