Address comments.

Author: viirya

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

@@ -24,7 +24,7 @@ import scala.util.Properties
 import org.apache.commons.lang3.reflect.ConstructorUtils
 
 import org.apache.spark.internal.Logging
-import org.apache.spark.sql.catalyst.analysis.{GetColumnByOrdinal, UnresolvedAttribute, UnresolvedExtractValue}
+import org.apache.spark.sql.catalyst.analysis.{GetColumnByOrdinal, UnresolvedExtractValue}
 import org.apache.spark.sql.catalyst.expressions._
 import org.apache.spark.sql.catalyst.expressions.objects._
 import org.apache.spark.sql.catalyst.util.{ArrayData, DateTimeUtils, GenericArrayData, MapData}
@@ -426,33 +426,20 @@ object ScalaReflection extends ScalaReflection {
    *  * the element type of [[Array]] or [[Seq]]: `array element class: "abc.xyz.MyClass"`
    *  * the field of [[Product]]: `field (class: "abc.xyz.MyClass", name: "myField")`
    */
-  def serializerForType(tpe: `Type`,
-      cls: RuntimeClass): Expression = ScalaReflection.cleanUpReflectionObjects {
+  def serializerForType(tpe: `Type`): Expression = ScalaReflection.cleanUpReflectionObjects {
     val clsName = getClassNameFromType(tpe)
     val walkedTypePath = s"""- root class: "$clsName"""" :: Nil
 
     // The input object to `ExpressionEncoder` is located at first column of an row.
     val inputObject = BoundReference(0, dataTypeFor(tpe),
-      nullable = !cls.isPrimitive)
+      nullable = !tpe.typeSymbol.asClass.isPrimitive)
 
     serializerFor(inputObject, tpe, walkedTypePath)
   }
 
   /**
    * Returns an expression for serializing the value of an input expression into Spark SQL
    * internal representation.
-   *
-   * The expression generated by this method will be used by `ExpressionEncoder` as serializer
-   * to convert a JVM object to Spark SQL representation.
-   *
-   * The returned serializer generally converts a JVM object to corresponding Spark SQL
-   * representation. For example, `Seq[_]` is converted to a Spark SQL array, `Product` is
-   * converted to a Spark SQL struct.
-   *
-   * If input object is not of ObjectType, it means that the input object is already in a form
-   * of Spark's internal representation. We simply return the input object.
-   *
-   * For unsupported types, an `UnsupportedOperationException` will be thrown.
    */
   private def serializerFor(
       inputObject: Expression,

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

@@ -18,7 +18,7 @@
 package org.apache.spark.sql.catalyst.encoders
 
 import scala.reflect.ClassTag
-import scala.reflect.runtime.universe.{`Type`, typeTag, RuntimeClass, TypeTag}
+import scala.reflect.runtime.universe.{typeTag, TypeTag}
 
 import org.apache.spark.sql.Encoder
 import org.apache.spark.sql.catalyst.{InternalRow, JavaTypeInference, ScalaReflection}
@@ -48,7 +48,6 @@ object ExpressionEncoder {
   def apply[T : TypeTag](): ExpressionEncoder[T] = {
     val mirror = ScalaReflection.mirror
     val tpe = typeTag[T].in(mirror).tpe
-    val cls = mirror.runtimeClass(tpe)
 
     if (ScalaReflection.optionOfProductType(tpe)) {
       throw new UnsupportedOperationException(
@@ -59,7 +58,8 @@ object ExpressionEncoder {
           "`val ds: Dataset[Tuple1[MyClass]] = Seq(Tuple1(MyClass(...)), Tuple1(null)).toDS`")
     }
 
-    val serializer = ScalaReflection.serializerForType(tpe, cls)
+    val cls = mirror.runtimeClass(tpe)
+    val serializer = ScalaReflection.serializerForType(tpe)
     val deserializer = ScalaReflection.deserializerForType(tpe)
 
     new ExpressionEncoder[T](
@@ -204,10 +204,9 @@ case class ExpressionEncoder[T](
    * 2. For other cases, we create a struct to wrap the `serializer`.
    */
   val serializer: Seq[NamedExpression] = {
-    val serializedAsStruct = objSerializer.dataType.isInstanceOf[StructType]
     val clsName = Utils.getSimpleName(clsTag.runtimeClass)
 
-    if (serializedAsStruct) {
+    if (isSerializedAsStruct) {
       val nullSafeSerializer = objSerializer.transformUp {
         case r: BoundReference =>
           // For input object of Product type, we can't encode it to row if it's null, as Spark SQL
@@ -236,9 +235,7 @@ case class ExpressionEncoder[T](
    * `GetColumnByOrdinal` with corresponding ordinal.
    */
   val deserializer: Expression = {
-    val serializedAsStruct = objSerializer.dataType.isInstanceOf[StructType]
-
-    if (serializedAsStruct) {
+    if (isSerializedAsStruct) {
       // We serialized this kind of objects to root-level row. The input of general deserializer
       // is a `GetColumnByOrdinal(0)` expression to extract first column of a row. We need to
       // transform attributes accessors.
@@ -264,6 +261,11 @@ case class ExpressionEncoder[T](
     StructField(s.name, s.dataType, s.nullable)
   })
 
+  /**
+   * Returns true if the type `T` is serialized as a struct.
+   */
+  def isSerializedAsStruct: Boolean = objSerializer.dataType.isInstanceOf[StructType]
+
   // serializer expressions are used to encode an object to a row, while the object is usually an
   // intermediate value produced inside an operator, not from the output of the child operator. This
   // is quite different from normal expressions, and `AttributeReference` doesn't work here

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/ScalaReflectionSuite.scala

@@ -262,8 +262,7 @@ class ScalaReflectionSuite extends SparkFunSuite {
 
   test("SPARK-15062: Get correct serializer for List[_]") {
     val list = List(1, 2, 3)
-    val serializer = serializerForType(ScalaReflection.localTypeOf[List[Int]],
-        classOf[List[Int]])
+    val serializer = serializerForType(ScalaReflection.localTypeOf[List[Int]])
     assert(serializer.isInstanceOf[NewInstance])
     assert(serializer.asInstanceOf[NewInstance]
       .cls.isAssignableFrom(classOf[org.apache.spark.sql.catalyst.util.GenericArrayData]))
@@ -276,42 +275,37 @@ class ScalaReflectionSuite extends SparkFunSuite {
 
   test("serialize and deserialize arbitrary sequence types") {
     import scala.collection.immutable.Queue
-    val queueSerializer = serializerForType(ScalaReflection.localTypeOf[Queue[Int]],
-        classOf[Queue[Int]])
+    val queueSerializer = serializerForType(ScalaReflection.localTypeOf[Queue[Int]])
     assert(queueSerializer.dataType ==
       ArrayType(IntegerType, containsNull = false))
     val queueDeserializer = deserializerForType(ScalaReflection.localTypeOf[Queue[Int]])
     assert(queueDeserializer.dataType == ObjectType(classOf[Queue[_]]))
 
     import scala.collection.mutable.ArrayBuffer
-    val arrayBufferSerializer = serializerForType(ScalaReflection.localTypeOf[ArrayBuffer[Int]],
-        classOf[ArrayBuffer[Int]])
+    val arrayBufferSerializer = serializerForType(ScalaReflection.localTypeOf[ArrayBuffer[Int]])
     assert(arrayBufferSerializer.dataType ==
       ArrayType(IntegerType, containsNull = false))
     val arrayBufferDeserializer = deserializerForType(ScalaReflection.localTypeOf[ArrayBuffer[Int]])
     assert(arrayBufferDeserializer.dataType == ObjectType(classOf[ArrayBuffer[_]]))
   }
 
   test("serialize and deserialize arbitrary map types") {
-    val mapSerializer = serializerForType(ScalaReflection.localTypeOf[Map[Int, Int]],
-        classOf[Map[Int, Int]])
+    val mapSerializer = serializerForType(ScalaReflection.localTypeOf[Map[Int, Int]])
     assert(mapSerializer.dataType ==
       MapType(IntegerType, IntegerType, valueContainsNull = false))
     val mapDeserializer = deserializerForType(ScalaReflection.localTypeOf[Map[Int, Int]])
     assert(mapDeserializer.dataType == ObjectType(classOf[Map[_, _]]))
 
     import scala.collection.immutable.HashMap
-    val hashMapSerializer = serializerForType(ScalaReflection.localTypeOf[HashMap[Int, Int]],
-        classOf[HashMap[Int, Int]])
+    val hashMapSerializer = serializerForType(ScalaReflection.localTypeOf[HashMap[Int, Int]])
     assert(hashMapSerializer.dataType ==
       MapType(IntegerType, IntegerType, valueContainsNull = false))
     val hashMapDeserializer = deserializerForType(ScalaReflection.localTypeOf[HashMap[Int, Int]])
     assert(hashMapDeserializer.dataType == ObjectType(classOf[HashMap[_, _]]))
 
     import scala.collection.mutable.{LinkedHashMap => LHMap}
     val linkedHashMapSerializer = serializerForType(
-        ScalaReflection.localTypeOf[LHMap[Long, String]],
-        classOf[LHMap[Long, String]])
+        ScalaReflection.localTypeOf[LHMap[Long, String]])
     assert(linkedHashMapSerializer.dataType ==
       MapType(LongType, StringType, valueContainsNull = true))
     val linkedHashMapDeserializer = deserializerForType(
@@ -320,10 +314,10 @@ class ScalaReflectionSuite extends SparkFunSuite {
   }
 
   test("SPARK-22442: Generate correct field names for special characters") {
-    val serializer = serializerForType(ScalaReflection.localTypeOf[SpecialCharAsFieldData],
-        classOf[SpecialCharAsFieldData]).collect {
-      case If(_, _, s: CreateNamedStruct) => s
-    }.head
+    val serializer = serializerForType(ScalaReflection.localTypeOf[SpecialCharAsFieldData])
+      .collect {
+        case If(_, _, s: CreateNamedStruct) => s
+      }.head
     val deserializer = deserializerForType(ScalaReflection.localTypeOf[SpecialCharAsFieldData])
     assert(serializer.dataType(0).name == "field.1")
     assert(serializer.dataType(1).name == "field 2")

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

@@ -1087,7 +1087,7 @@ class Dataset[T] private[sql](
     // Note that we do this before joining them, to enable the join operator to return null for one
     // side, in cases like outer-join.
     val left = {
-      val combined = if (!this.exprEnc.objSerializer.dataType.isInstanceOf[StructType]) {
+      val combined = if (!this.exprEnc.isSerializedAsStruct) {
         assert(joined.left.output.length == 1)
         Alias(joined.left.output.head, "_1")()
       } else {
@@ -1097,7 +1097,7 @@ class Dataset[T] private[sql](
     }
 
     val right = {
-      val combined = if (!other.exprEnc.objSerializer.dataType.isInstanceOf[StructType]) {
+      val combined = if (!other.exprEnc.isSerializedAsStruct) {
         assert(joined.right.output.length == 1)
         Alias(joined.right.output.head, "_2")()
       } else {
@@ -1110,14 +1110,14 @@ class Dataset[T] private[sql](
     // combine the outputs of each join side.
     val conditionExpr = joined.condition.get transformUp {
       case a: Attribute if joined.left.outputSet.contains(a) =>
-        if (!this.exprEnc.objSerializer.dataType.isInstanceOf[StructType]) {
+        if (!this.exprEnc.isSerializedAsStruct) {
           left.output.head
         } else {
           val index = joined.left.output.indexWhere(_.exprId == a.exprId)
           GetStructField(left.output.head, index)
         }
       case a: Attribute if joined.right.outputSet.contains(a) =>
-        if (!other.exprEnc.objSerializer.dataType.isInstanceOf[StructType]) {
+        if (!other.exprEnc.isSerializedAsStruct) {
           right.output.head
         } else {
           val index = joined.right.output.indexWhere(_.exprId == a.exprId)
@@ -1390,7 +1390,7 @@ class Dataset[T] private[sql](
     implicit val encoder = c1.encoder
     val project = Project(c1.withInputType(exprEnc, logicalPlan.output).named :: Nil, logicalPlan)
 
-    if (!encoder.objSerializer.dataType.isInstanceOf[StructType]) {
+    if (!encoder.isSerializedAsStruct) {
       new Dataset[U1](sparkSession, project, encoder)
     } else {
       // Flattens inner fields of U1

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

@@ -27,7 +27,6 @@ import org.apache.spark.sql.catalyst.plans.logical._
 import org.apache.spark.sql.execution.QueryExecution
 import org.apache.spark.sql.expressions.ReduceAggregator
 import org.apache.spark.sql.streaming.{GroupState, GroupStateTimeout, OutputMode}
-import org.apache.spark.sql.types.StructType
 
 /**
  * :: Experimental ::
@@ -458,8 +457,7 @@ class KeyValueGroupedDataset[K, V] private[sql](
     val encoders = columns.map(_.encoder)
     val namedColumns =
       columns.map(_.withInputType(vExprEnc, dataAttributes).named)
-
-    val keyColumn = if (!kExprEnc.objSerializer.dataType.isInstanceOf[StructType]) {
+    val keyColumn = if (!kExprEnc.isSerializedAsStruct) {
       assert(groupingAttributes.length == 1)
       groupingAttributes.head
     } else {

sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/TypedAggregateExpression.scala

@@ -32,7 +32,6 @@ import org.apache.spark.sql.types._
 import org.apache.spark.util.Utils
 
 object TypedAggregateExpression {
-
   def apply[BUF : Encoder, OUT : Encoder](
       aggregator: Aggregator[_, BUF, OUT]): TypedAggregateExpression = {
     val bufferEncoder = encoderFor[BUF]
@@ -46,8 +45,7 @@ object TypedAggregateExpression {
     // serialization.
     val isSimpleBuffer = {
       bufferSerializer.head match {
-        case Alias(_: BoundReference, _)
-          if !bufferEncoder.objSerializer.dataType.isInstanceOf[StructType] => true
+        case Alias(_: BoundReference, _) if !bufferEncoder.isSerializedAsStruct => true
         case _ => false
       }
     }