addressed review comments

Author: kiszk

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/ScalaReflection.scala

@@ -444,13 +444,7 @@ object ScalaReflection extends ScalaReflection {
          case dt @ (BooleanType | ByteType | ShortType | IntegerType | LongType |
                     FloatType | DoubleType) =>
           val cls = input.dataType.asInstanceOf[ObjectType].cls
-          if (cls.isAssignableFrom(classOf[Array[Boolean]]) ||
-              cls.isAssignableFrom(classOf[Array[Byte]]) ||
-              cls.isAssignableFrom(classOf[Array[Short]]) ||
-              cls.isAssignableFrom(classOf[Array[Int]]) ||
-              cls.isAssignableFrom(classOf[Array[Long]]) ||
-              cls.isAssignableFrom(classOf[Array[Float]]) ||
-              cls.isAssignableFrom(classOf[Array[Double]])) {
+          if (cls.isArray && cls.getComponentType.isPrimitive) {
             StaticInvoke(
               classOf[UnsafeArrayData],
               ArrayType(dt, false),

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/encoders/RowEncoder.scala

@@ -23,7 +23,7 @@ import scala.reflect.ClassTag
 import org.apache.spark.SparkException
 import org.apache.spark.sql.Row
 import org.apache.spark.sql.catalyst.expressions._
-import org.apache.spark.sql.catalyst.util.{ArrayBasedMapData, DateTimeUtils, GenericArrayData}
+import org.apache.spark.sql.catalyst.util.{ArrayBasedMapData, ArrayData, DateTimeUtils, GenericArrayData}
 import org.apache.spark.sql.catalyst.ScalaReflection
 import org.apache.spark.sql.catalyst.analysis.GetColumnByOrdinal
 import org.apache.spark.sql.catalyst.expressions.objects._
@@ -122,28 +122,12 @@ object RowEncoder {
     case t @ ArrayType(et, cn) =>
       val cls = inputObject.dataType.asInstanceOf[ObjectType].cls
       et match {
-/*
-        case BooleanType | ByteType | ShortType | IntegerType | LongType | FloatType | DoubleType
-          if !cn && (
-            cls.isAssignableFrom(classOf[Array[Boolean]]) ||
-            cls.isAssignableFrom(classOf[Array[Byte]]) ||
-            cls.isAssignableFrom(classOf[Array[Short]]) ||
-            cls.isAssignableFrom(classOf[Array[Int]]) ||
-            cls.isAssignableFrom(classOf[Array[Long]]) ||
-            cls.isAssignableFrom(classOf[Array[Float]]) ||
-            cls.isAssignableFrom(classOf[Array[Double]])) =>
-          print(s"1@ET: $et, $cn, $cls\n")
+        case BooleanType | ByteType | ShortType | IntegerType | LongType | FloatType | DoubleType =>
           StaticInvoke(
-            classOf[UnsafeArrayData],
-            ArrayType(et, false),
-            "fromPrimitiveArray",
+            classOf[ArrayData],
+            ObjectType(classOf[ArrayData]),
+            "toArrayData",
             inputObject :: Nil)
-*/
-        case BooleanType | ByteType | ShortType | IntegerType | LongType | FloatType | DoubleType =>
-          NewInstance(
-            classOf[GenericArrayData],
-            inputObject :: Nil,
-            dataType = t)
         case _ => MapObjects(
           element => serializerFor(ValidateExternalType(element, et), et),
           inputObject,
@@ -211,8 +195,7 @@ object RowEncoder {
     // as java.lang.Object.
     case _: DecimalType => ObjectType(classOf[java.lang.Object])
     // In order to support both Array and Seq in external row, we make this as java.lang.Object.
-    case a @ ArrayType(et, cn) =>
-      ObjectType(classOf[java.lang.Object])
+    case _: ArrayType => ObjectType(classOf[java.lang.Object])
     case _ => externalDataTypeFor(dt)
   }
 

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

@@ -993,7 +993,7 @@ case class ValidateExternalType(child: Expression, expected: DataType)
       case _: DecimalType =>
         Seq(classOf[java.math.BigDecimal], classOf[scala.math.BigDecimal], classOf[Decimal])
           .map(cls => s"$obj instanceof ${cls.getName}").mkString(" || ")
-      case a @ ArrayType(et, cn) =>
+      case _: ArrayType =>
         s"$obj instanceof ${classOf[Seq[_]].getName} || $obj.getClass().isArray()"
       case _ =>
         s"$obj instanceof ${ctx.boxedType(dataType)}"

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/ArrayData.scala

@@ -19,9 +19,22 @@ package org.apache.spark.sql.catalyst.util
 
 import scala.reflect.ClassTag
 
-import org.apache.spark.sql.catalyst.expressions.SpecializedGetters
+import org.apache.spark.sql.catalyst.expressions.{SpecializedGetters, UnsafeArrayData}
 import org.apache.spark.sql.types.DataType
 
+object ArrayData {
+  def toArrayData(input: Any): ArrayData = input match {
+    case a: Array[Boolean] => UnsafeArrayData.fromPrimitiveArray(a)
+    case a: Array[Byte] => UnsafeArrayData.fromPrimitiveArray(a)
+    case a: Array[Short] => UnsafeArrayData.fromPrimitiveArray(a)
+    case a: Array[Int] => UnsafeArrayData.fromPrimitiveArray(a)
+    case a: Array[Long] => UnsafeArrayData.fromPrimitiveArray(a)
+    case a: Array[Float] => UnsafeArrayData.fromPrimitiveArray(a)
+    case a: Array[Double] => UnsafeArrayData.fromPrimitiveArray(a)
+    case other => new GenericArrayData(other)
+  }
+}
+
 abstract class ArrayData extends SpecializedGetters with Serializable {
   def numElements(): Int