address review comments

python/pyspark/sql/functions.py

@@ -1821,14 +1821,14 @@ def create_map(*cols):
 
 @ignore_unicode_prefix
 @since(2.4)
-def create_map_from_arrays(col1, col2):
+def map_from_arrays(col1, col2):
     """Creates a new map from two arrays.
 
     :param col1: name of column containing a set of keys. All elements should not be null
     :param col2: name of column containing a set of values
 
     >>> df = spark.createDataFrame([([2, 5], ["Alice", "Bob"])], ['k', 'v'])
-    >>> df.select(create_map_from_arrays(df.k, df.v).alias("map")).collect()
+    >>> df.select(map_from_arrays(df.k, df.v).alias("map")).collect()
     [Row(map={2: u'Alice', 5: u'Bob'})]
     """
     sc = SparkContext._active_spark_context

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

@@ -415,9 +415,9 @@ object FunctionRegistry {
     expression[ArrayPosition]("array_position"),
     expression[ArraySort]("array_sort"),
     expression[CreateMap]("map"),
-    expression[CreateMapFromArrays]("map_from_arrays"),
     expression[CreateNamedStruct]("named_struct"),
     expression[ElementAt]("element_at"),
+    expression[MapFromArrays]("map_from_arrays"),
     expression[MapKeys]("map_keys"),
     expression[MapValues]("map_values"),
     expression[MapEntries]("map_entries"),

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

@@ -248,29 +248,18 @@ case class CreateMap(children: Seq[Expression]) extends Expression {
       > SELECT _FUNC_([1.0, 3.0], ['2', '4']);
        {1.0:"2",3.0:"4"}
   """, since = "2.4.0")
-case class CreateMapFromArrays(left: Expression, right: Expression)
+case class MapFromArrays(left: Expression, right: Expression)
     extends BinaryExpression with ExpectsInputTypes {
 
   override def inputTypes: Seq[AbstractDataType] = Seq(ArrayType, ArrayType)
 
-  override def checkInputDataTypes(): TypeCheckResult = {
-    (left.dataType, right.dataType) match {
-      case (ArrayType(_, _), ArrayType(_, _)) =>
-        TypeCheckResult.TypeCheckSuccess
-      case _ =>
-        TypeCheckResult.TypeCheckFailure("The given two arguments should be an array")
-    }
-  }
-
   override def dataType: DataType = {
     MapType(
       keyType = left.dataType.asInstanceOf[ArrayType].elementType,
       valueType = right.dataType.asInstanceOf[ArrayType].elementType,
       valueContainsNull = right.dataType.asInstanceOf[ArrayType].containsNull)
   }
 
-  override def nullable: Boolean = left.nullable || right.nullable
-
   override def nullSafeEval(keyArray: Any, valueArray: Any): Any = {
     val keyArrayData = keyArray.asInstanceOf[ArrayData]
     val valueArrayData = valueArray.asInstanceOf[ArrayData]
@@ -279,8 +268,12 @@ case class CreateMapFromArrays(left: Expression, right: Expression)
     }
     val leftArrayType = left.dataType.asInstanceOf[ArrayType]
     if (leftArrayType.containsNull) {
-      if (keyArrayData.toArray(leftArrayType.elementType).contains(null)) {
-        throw new RuntimeException("Cannot use null as map key!")
+      var i = 0
+      while (i < keyArrayData.numElements) {
+        if (keyArrayData.isNullAt(i)) {
+          throw new RuntimeException("Cannot use null as map key!")
+        }
+        i += 1
       }
     }
     new ArrayBasedMapData(keyArrayData.copy(), valueArrayData.copy())
@@ -291,13 +284,10 @@ case class CreateMapFromArrays(left: Expression, right: Expression)
       val arrayBasedMapData = classOf[ArrayBasedMapData].getName
       val leftArrayType = left.dataType.asInstanceOf[ArrayType]
       val keyArrayElemNullCheck = if (!leftArrayType.containsNull) "" else {
-        val leftArrayTypeTerm = ctx.addReferenceObj("leftArrayType", leftArrayType.elementType)
-        val array = ctx.freshName("array")
         val i = ctx.freshName("i")
         s"""
-           |Object[] $array = $keyArrayData.toObjectArray($leftArrayTypeTerm);
-           |for (int $i = 0; $i < $array.length; $i++) {
-           |  if ($array[$i] == null) {
+           |for (int $i = 0; $i < $keyArrayData.numElements(); $i++) {
+           |  if ($keyArrayData.isNullAt($i)) {
            |    throw new RuntimeException("Cannot use null as map key!");
            |  }
            |}

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

@@ -186,7 +186,7 @@ class ComplexTypeSuite extends SparkFunSuite with ExpressionEvalHelper {
     }
   }
 
-  test("CreateMapFromArrays") {
+  test("MapFromArrays") {
     def createMap(keys: Seq[Any], values: Seq[Any]): Map[Any, Any] = {
       // catalyst map is order-sensitive, so we create ListMap here to preserve the elements order.
       scala.collection.immutable.ListMap(keys.zip(values): _*)
@@ -209,24 +209,24 @@ class ComplexTypeSuite extends SparkFunSuite with ExpressionEvalHelper {
 
     val nullArray = Literal.create(null, ArrayType(StringType, false))
 
-    checkEvaluation(CreateMapFromArrays(intArray, longArray), createMap(intSeq, longSeq))
-    checkEvaluation(CreateMapFromArrays(intArray, strArray), createMap(intSeq, strSeq))
-    checkEvaluation(CreateMapFromArrays(integerArray, strArray), createMap(integerSeq, strSeq))
+    checkEvaluation(MapFromArrays(intArray, longArray), createMap(intSeq, longSeq))
+    checkEvaluation(MapFromArrays(intArray, strArray), createMap(intSeq, strSeq))
+    checkEvaluation(MapFromArrays(integerArray, strArray), createMap(integerSeq, strSeq))
 
     checkEvaluation(
-      CreateMapFromArrays(strArray, intwithNullArray), createMap(strSeq, intWithNullSeq))
+      MapFromArrays(strArray, intwithNullArray), createMap(strSeq, intWithNullSeq))
     checkEvaluation(
-      CreateMapFromArrays(strArray, longwithNullArray), createMap(strSeq, longWithNullSeq))
+      MapFromArrays(strArray, longwithNullArray), createMap(strSeq, longWithNullSeq))
     checkEvaluation(
-      CreateMapFromArrays(strArray, longwithNullArray), createMap(strSeq, longWithNullSeq))
-    checkEvaluation(CreateMapFromArrays(nullArray, nullArray), null)
+      MapFromArrays(strArray, longwithNullArray), createMap(strSeq, longWithNullSeq))
+    checkEvaluation(MapFromArrays(nullArray, nullArray), null)
 
     intercept[RuntimeException] {
-      checkEvaluation(CreateMapFromArrays(intwithNullArray, strArray), null)
+      checkEvaluation(MapFromArrays(intwithNullArray, strArray), null)
     }
     intercept[RuntimeException] {
       checkEvaluation(
-        CreateMapFromArrays(intArray, Literal.create(Seq(1), ArrayType(IntegerType))), null)
+        MapFromArrays(intArray, Literal.create(Seq(1), ArrayType(IntegerType))), null)
     }
   }
 

sql/core/src/main/scala/org/apache/spark/sql/functions.scala

@@ -1078,7 +1078,7 @@ object functions {
    * @since 2.4
    */
   def map_from_arrays(keys: Column, values: Column): Column = withExpr {
-    CreateMapFromArrays(keys.expr, values.expr)
+    MapFromArrays(keys.expr, values.expr)
   }
 
   /**