Use existing function

Author: marmbrus

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

@@ -19,8 +19,7 @@ package org.apache.spark.sql.catalyst
 
 import java.sql.Timestamp
 
-import org.apache.spark.sql.catalyst.expressions.Attribute
-import org.apache.spark.sql.catalyst.expressions.AttributeReference
+import org.apache.spark.sql.catalyst.expressions.{GenericRow, Attribute, AttributeReference}
 import org.apache.spark.sql.catalyst.plans.logical.LocalRelation
 import org.apache.spark.sql.catalyst.types._
 
@@ -32,6 +31,14 @@ object ScalaReflection {
 
   case class Schema(dataType: DataType, nullable: Boolean)
 
+  /** Converts Scala objects to catalyst rows / types */
+  def convertToCatalyst(a: Any): Any = a match {
+    case o: Option[_] => o.orNull
+    case s: Seq[Any] => s.map(convertToCatalyst)
+    case p: Product => new GenericRow(p.productIterator.map(convertToCatalyst).toArray)
+    case other => other
+  }
+
   /** Returns a Sequence of attributes for the given case class type. */
   def attributesFor[T: TypeTag]: Seq[Attribute] = schemaFor[T] match {
     case Schema(s: StructType, _) =>

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

@@ -17,6 +17,7 @@
 
 package org.apache.spark.sql.catalyst.expressions
 
+import org.apache.spark.sql.catalyst.ScalaReflection
 import org.apache.spark.sql.catalyst.types.DataType
 
 case class ScalaUdf(function: AnyRef, dataType: DataType, children: Seq[Expression])
@@ -44,11 +45,6 @@ case class ScalaUdf(function: AnyRef, dataType: DataType, children: Seq[Expressi
 
   */
 
-  def convert(a: Any): Any = a match {
-    case p: Product => Row.fromSeq(p.productIterator.map(convert).toSeq)
-    case other => other
-  }
-
   // scalastyle:off
   override def eval(input: Row): Any = {
     val result = children.size match {
@@ -351,6 +347,6 @@ case class ScalaUdf(function: AnyRef, dataType: DataType, children: Seq[Expressi
     }
     // scalastyle:on
 
-    convert(result)
+    ScalaReflection.convertToCatalyst(result)
   }
 }

sql/core/src/main/scala/org/apache/spark/sql/execution/basicOperators.scala

@@ -204,13 +204,6 @@ case class Sort(
  */
 @DeveloperApi
 object ExistingRdd {
-  def convertToCatalyst(a: Any): Any = a match {
-    case o: Option[_] => o.orNull
-    case s: Seq[Any] => s.map(convertToCatalyst)
-    case p: Product => new GenericRow(p.productIterator.map(convertToCatalyst).toArray)
-    case other => other
-  }
-
   def productToRowRdd[A <: Product](data: RDD[A]): RDD[Row] = {
     data.mapPartitions { iterator =>
       if (iterator.isEmpty) {
@@ -222,7 +215,7 @@ object ExistingRdd {
         bufferedIterator.map { r =>
           var i = 0
           while (i < mutableRow.length) {
-            mutableRow(i) = convertToCatalyst(r.productElement(i))
+            mutableRow(i) = ScalaReflection.convertToCatalyst(r.productElement(i))
             i += 1
           }
 
@@ -244,6 +237,7 @@ object ExistingRdd {
 case class ExistingRdd(output: Seq[Attribute], rdd: RDD[Row]) extends LeafNode {
   override def execute() = rdd
 }
+
 /**
  * :: DeveloperApi ::
  * Computes the set of distinct input rows using a HashSet.