improve

Author: cloud-fan

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

@@ -302,7 +302,7 @@ object ScalaReflection extends ScalaReflection {
 
       case t if t <:< localTypeOf[Seq[_]] =>
         val TypeRef(_, _, Seq(elementType)) = t
-        val Schema(dataType, elementNullable) = schemaFor(elementType)
+        val Schema(_, elementNullable) = schemaFor(elementType)
         val className = getClassNameFromType(elementType)
         val newTypePath = s"""- array element class: "$className"""" +: walkedTypePath
 

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

@@ -452,7 +452,11 @@ case class UnresolvedMapObjects(
     function: Expression => Expression,
     child: Expression,
     customCollectionCls: Option[Class[_]] = None) extends UnaryExpression with Unevaluable {
-  override def dataType: DataType = throw new UnsupportedOperationException("not resolved")
+  override lazy val resolved = false
+
+  override def dataType: DataType = customCollectionCls.map(ObjectType.apply).getOrElse {
+    throw new UnsupportedOperationException("not resolved")
+  }
 }
 
 /**
@@ -588,17 +592,24 @@ case class MapObjects private(
           // collection
           val collObjectName = s"${cls.getName}$$.MODULE$$"
           val getBuilderVar = s"$collObjectName.newBuilder()"
-
-          (s"""${classOf[Builder[_, _]].getName} $builderValue = $getBuilderVar;
-        $builderValue.sizeHint($dataLength);""",
+          (
+            s"""
+               ${classOf[Builder[_, _]].getName} $builderValue = $getBuilderVar;
+               $builderValue.sizeHint($dataLength);
+             """,
             genValue => s"$builderValue.$$plus$$eq($genValue);",
-            s"(${cls.getName}) $builderValue.result();")
+            s"(${cls.getName}) $builderValue.result();"
+          )
         case None =>
           // array
-          (s"""$convertedType[] $convertedArray = null;
-        $convertedArray = $arrayConstructor;""",
+          (
+            s"""
+               $convertedType[] $convertedArray = null;
+               $convertedArray = $arrayConstructor;
+             """,
             genValue => s"$convertedArray[$loopIndex] = $genValue;",
-            s"new ${classOf[GenericArrayData].getName}($convertedArray);")
+            s"new ${classOf[GenericArrayData].getName}($convertedArray);"
+          )
       }
 
     val code = s"""

sql/core/src/main/scala/org/apache/spark/sql/expressions/ReduceAggregator.scala

@@ -27,17 +27,19 @@ import org.apache.spark.sql.catalyst.encoders.ExpressionEncoder
  *
  * This class currently assumes there is at least one input row.
  */
-private[sql] class ReduceAggregator[T](func: (T, T) => T)(@transient implicit val enc: Encoder[T])
+private[sql] class ReduceAggregator[T: Encoder](func: (T, T) => T)
   extends Aggregator[T, (Boolean, T), T] {
 
+  @transient private val encoder = implicitly[Encoder[T]]
+
   override def zero: (Boolean, T) = (false, null.asInstanceOf[T])
 
   override def bufferEncoder: Encoder[(Boolean, T)] =
     ExpressionEncoder.tuple(
       ExpressionEncoder[Boolean](),
-      enc.asInstanceOf[ExpressionEncoder[T]])
+      encoder.asInstanceOf[ExpressionEncoder[T]])
 
-  override def outputEncoder: Encoder[T] = enc
+  override def outputEncoder: Encoder[T] = encoder
 
   override def reduce(b: (Boolean, T), a: T): (Boolean, T) = {
     if (b._1) {