Non-nullable null type should not coerce to nullable type

Author: HyukjinKwon

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/TypeCoercion.scala

@@ -160,7 +160,7 @@ object TypeCoercion {
       }
     case (MapType(kt1, vt1, valueContainsNull1), MapType(kt2, vt2, valueContainsNull2)) =>
       findTypeFunc(kt1, kt2)
-        .filter { kt => Cast.canCastMapKeyNullSafe(kt1, kt) && Cast.canCastMapKeyNullSafe(kt2, kt) }
+        .filter { kt => !Cast.forceNullable(kt1, kt) && !Cast.forceNullable(kt2, kt) }
         .flatMap { kt =>
           findTypeFunc(vt1, vt2).map { vt =>
             MapType(kt, vt, valueContainsNull1 || valueContainsNull2 ||

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/Cast.scala

@@ -77,7 +77,8 @@ object Cast {
         resolvableNullability(fn || forceNullable(fromType, toType), tn)
 
     case (MapType(fromKey, fromValue, fn), MapType(toKey, toValue, tn)) =>
-      canCast(fromKey, toKey) && canCastMapKeyNullSafe(fromKey, toKey) &&
+      canCast(fromKey, toKey) &&
+        (!forceNullable(fromKey, toKey)) &&
         canCast(fromValue, toValue) &&
         resolvableNullability(fn || forceNullable(fromValue, toValue), tn)
 
@@ -97,11 +98,6 @@ object Cast {
     case _ => false
   }
 
-  def canCastMapKeyNullSafe(fromType: DataType, toType: DataType): Boolean = {
-    // If the original map key type is NullType, it's OK as the map must be empty.
-    fromType == NullType || !forceNullable(fromType, toType)
-  }
-
   /**
    * Return true if we need to use the `timeZone` information casting `from` type to `to` type.
    * The patterns matched reflect the current implementation in the Cast node.
@@ -210,8 +206,15 @@ object Cast {
     case _ => false  // overflow
   }
 
+  /**
+   * Returns `true` iff it should change the nullability of this type in map type,
+   * array type, expressions such as cast, etc.
+   *
+   * Note that you should take the nullability context into account.
+   * For example, it should not force to nullable type when null type in array
+   * type is non-nullable which means an empty array of null type.
+   */
   def forceNullable(from: DataType, to: DataType): Boolean = (from, to) match {
-    case (NullType, _) => true
     case (_, _) if from == to => false
 
     case (StringType, BinaryType) => false

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/TypeCoercionSuite.scala

@@ -497,6 +497,23 @@ class TypeCoercionSuite extends AnalysisTest {
         .add("null", IntegerType, nullable = false),
       Some(new StructType()
         .add("null", IntegerType, nullable = true)))
+
+    widenTest(
+      ArrayType(NullType, containsNull = false),
+      ArrayType(IntegerType, containsNull = false),
+      Some(ArrayType(IntegerType, containsNull = false)))
+
+    widenTest(MapType(NullType, NullType, false),
+      MapType(IntegerType, StringType, false),
+      Some(MapType(IntegerType, StringType, false)))
+
+    widenTest(
+      new StructType()
+        .add("null", NullType, nullable = false),
+      new StructType()
+        .add("null", IntegerType, nullable = false),
+      Some(new StructType()
+        .add("null", IntegerType, nullable = false)))
   }
 
   test("wider common type for decimal and array") {

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/CastSuite.scala

@@ -412,6 +412,14 @@ abstract class CastSuiteBase extends SparkFunSuite with ExpressionEvalHelper {
       assert(ret.resolved)
       checkEvaluation(ret, Seq(null, true, false, null))
     }
+
+    {
+      val array = Literal.create(Seq.empty, ArrayType(NullType, containsNull = false))
+      val ret = cast(array, ArrayType(IntegerType, containsNull = false))
+      assert(ret.resolved)
+      checkEvaluation(ret, Seq.empty)
+    }
+
     {
       val ret = cast(array, ArrayType(BooleanType, containsNull = false))
       assert(ret.resolved === false)
@@ -1157,6 +1165,17 @@ class CastSuite extends CastSuiteBase {
       StructType(StructField("a", IntegerType, true) :: Nil)))
   }
 
+  test("SPARK-31227: Non-nullable null type should not coerce to nullable type") {
+    assert(Cast.canCast(ArrayType(NullType, false), ArrayType(IntegerType, false)))
+
+    assert(Cast.canCast(
+      MapType(NullType, NullType, false), MapType(IntegerType, IntegerType, false)))
+
+    assert(Cast.canCast(
+      StructType(StructField("a", NullType, false) :: Nil),
+      StructType(StructField("a", IntegerType, false) :: Nil)))
+  }
+
   test("Cast should output null for invalid strings when ANSI is not enabled.") {
     withSQLConf(SQLConf.ANSI_ENABLED.key -> "false") {
       checkEvaluation(cast("abdef", DecimalType.USER_DEFAULT), null)

sql/core/src/test/scala/org/apache/spark/sql/DataFrameFunctionsSuite.scala

@@ -1532,6 +1532,13 @@ class DataFrameFunctionsSuite extends QueryTest with SharedSparkSession {
     assert(e.getMessage.contains("string, binary or array"))
   }
 
+  test("SPARK-31227: Non-nullable null type should not coerce to nullable type in concat") {
+    val actual = spark.range(1).selectExpr("concat(array(), array(1)) as arr")
+    val expected = spark.range(1).selectExpr("array(1) as arr")
+    checkAnswer(actual, expected)
+    assert(actual.schema === expected.schema)
+  }
+
   test("flatten function") {
     // Test cases with a primitive type
     val intDF = Seq(