fix test failures for handling Map with primitive types

Author: kiszk

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

@@ -603,7 +603,14 @@ case class ExternalMapToCatalyst private(
 
   override def foldable: Boolean = false
 
-  override def dataType: MapType = MapType(keyConverter.dataType, valueConverter.dataType)
+  override def dataType: MapType = {
+    val isPrimitiveType = valueType match {
+      case BooleanType | ByteType | ShortType | IntegerType | LongType |
+            FloatType | DoubleType => true
+      case _ => false
+    }
+    MapType(keyConverter.dataType, valueConverter.dataType, !isPrimitiveType)
+  }
 
   override def eval(input: InternalRow): Any =
     throw new UnsupportedOperationException("Only code-generated evaluation is supported")

sql/catalyst/src/main/scala/org/apache/spark/sql/types/MapType.scala

@@ -94,10 +94,5 @@ object MapType extends AbstractDataType {
    * The `valueContainsNull` is true.
    */
   def apply(keyType: DataType, valueType: DataType): MapType =
-    MapType(keyType: DataType, valueType: DataType, valueContainsNull =
-      valueType match {
-        case BooleanType | ByteType | ShortType | IntegerType | LongType |
-              FloatType | DoubleType => false
-        case _ => true
-      })
+    MapType(keyType: DataType, valueType: DataType, valueContainsNull = true)
 }

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

@@ -961,14 +961,22 @@ class DatasetSuite extends QueryTest with SharedSQLContext {
     assert(df7.schema(0).nullable == true)
     assert(df7.schema(0).dataType.asInstanceOf[ArrayType].containsNull == false)
 
-    val df8 = (Tuple1(Map(2 -> 3)) :: Nil).toDF("m")
+    val df8 = (Tuple1(Array((null: Integer), (null: Integer))) :: Nil).toDF("a")
     assert(df8.schema(0).nullable == true)
-    assert(df8.schema(0).dataType.asInstanceOf[MapType].valueContainsNull == false)
+    assert(df8.schema(0).dataType.asInstanceOf[ArrayType].containsNull == true)
 
-    val df9 = Seq(TestDataPoint(1, 2.2, "a"), TestDataPoint(3, 4.4, "null")).toDF
-    assert(df9.schema(0).nullable == false)
-    assert(df9.schema(1).nullable == false)
-    assert(df9.schema(2).nullable == true)
+    val df9 = (Tuple1(Map(2 -> 3)) :: Nil).toDF("m")
+    assert(df9.schema(0).nullable == true)
+    assert(df9.schema(0).dataType.asInstanceOf[MapType].valueContainsNull == false)
+
+    val df10 = (Tuple1(Map(1 -> (null: Integer))) :: Nil).toDF("m")
+    assert(df10.schema(0).nullable == true)
+    assert(df10.schema(0).dataType.asInstanceOf[MapType].valueContainsNull == true)
+
+    val df11 = Seq(TestDataPoint(1, 2.2, "a"), TestDataPoint(3, 4.4, "null")).toDF
+    assert(df11.schema(0).nullable == false)
+    assert(df11.schema(1).nullable == false)
+    assert(df11.schema(2).nullable == true)
   }
 
   Seq(true, false).foreach { eager =>