Addresed review feedback

mllib/src/test/scala/org/apache/spark/ml/feature/TokenizerSuite.scala

@@ -25,10 +25,7 @@ import org.apache.spark.mllib.util.MLlibTestSparkContext
 import org.apache.spark.sql.{DataFrame, Row, SQLContext}
 
 @BeanInfo
-case class TokenizerTestData(rawText: String, wantedTokens: Seq[String]) {
-  /** Constructor used in [[org.apache.spark.ml.feature.JavaTokenizerSuite]] */
-  def this(rawText: String, wantedTokens: Array[String]) = this(rawText, wantedTokens.toSeq)
-}
+case class TokenizerTestData(rawText: String, wantedTokens: Array[String])
 
 class RegexTokenizerSuite extends FunSuite with MLlibTestSparkContext {
   import org.apache.spark.ml.feature.RegexTokenizerSuite._
@@ -46,14 +43,14 @@ class RegexTokenizerSuite extends FunSuite with MLlibTestSparkContext {
       .setOutputCol("tokens")
 
     val dataset0 = sqlContext.createDataFrame(Seq(
-      TokenizerTestData("Test for tokenization.", Seq("Test", "for", "tokenization", ".")),
-      TokenizerTestData("Te,st. punct", Seq("Te", ",", "st", ".", "punct"))
+      TokenizerTestData("Test for tokenization.", Array("Test", "for", "tokenization", ".")),
+      TokenizerTestData("Te,st. punct", Array("Te", ",", "st", ".", "punct"))
     ))
     testRegexTokenizer(tokenizer, dataset0)
 
     val dataset1 = sqlContext.createDataFrame(Seq(
-      TokenizerTestData("Test for tokenization.", Seq("Test", "for", "tokenization")),
-      TokenizerTestData("Te,st. punct", Seq("punct"))
+      TokenizerTestData("Test for tokenization.", Array("Test", "for", "tokenization")),
+      TokenizerTestData("Te,st. punct", Array("punct"))
     ))
 
     tokenizer.setMinTokenLength(3)
@@ -64,8 +61,8 @@ class RegexTokenizerSuite extends FunSuite with MLlibTestSparkContext {
       .setGaps(true)
       .setMinTokenLength(0)
     val dataset2 = sqlContext.createDataFrame(Seq(
-      TokenizerTestData("Test for tokenization.", Seq("Test", "for", "tokenization.")),
-      TokenizerTestData("Te,st.  punct", Seq("Te,st.", "", "punct"))
+      TokenizerTestData("Test for tokenization.", Array("Test", "for", "tokenization.")),
+      TokenizerTestData("Te,st.  punct", Array("Te,st.", "", "punct"))
     ))
     testRegexTokenizer(tokenizer, dataset2)
   }

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

@@ -33,7 +33,8 @@ object CatalystTypeConverters {
   import scala.collection.Map
 
   /**
-   * Converts Scala objects to catalyst rows / types.
+   * Converts Scala objects to catalyst rows / types. This method is slow, and for batch
+   * conversion you should be using converter produced by createToCatalystConverter.
    * Note: This is always called after schemaFor has been called.
    *       This ordering is important for UDT registration.
    */
@@ -97,6 +98,8 @@ object CatalystTypeConverters {
 
   /**
    * Creates a converter function that will convert Scala objects to the specified catalyst type.
+   * Typical use case would be converting a collection of rows that have the same schema. You will
+   * call this function once to get a converter, and apply it to every row.
    */
   private[sql] def createToCatalystConverter(dataType: DataType): Any => Any = {
     def extractOption(item: Any): Any = item match {
@@ -181,7 +184,10 @@ object CatalystTypeConverters {
     }
   }
 
-  /** Converts Catalyst types used internally in rows to standard Scala types */
+  /** Converts Catalyst types used internally in rows to standard Scala types
+    * This method is slow, and for batch conversion you should be using converter
+    * produced by createToScalaConverter.
+    */
   def convertToScala(a: Any, dataType: DataType): Any = (a, dataType) match {
     // Check UDT first since UDTs can override other types
     case (d, udt: UserDefinedType[_]) =>
@@ -210,6 +216,8 @@ object CatalystTypeConverters {
 
   /**
    * Creates a converter function that will convert Catalyst types to Scala type.
+   * Typical use case would be converting a collection of rows that have the same schema. You will
+   * call this function once to get a converter, and apply it to every row.
    */
   private[sql] def createToScalaConverter(dataType: DataType): Any => Any = dataType match {
     // Check UDT first since UDTs can override other types

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/LocalRelation.scala

@@ -18,7 +18,7 @@
 package org.apache.spark.sql.catalyst.plans.logical
 
 import org.apache.spark.sql.Row
-import org.apache.spark.sql.catalyst.analysis
+import org.apache.spark.sql.catalyst.{CatalystTypeConverters, analysis}
 import org.apache.spark.sql.catalyst.expressions.Attribute
 import org.apache.spark.sql.types.{DataTypeConversions, StructType, StructField}
 
@@ -31,7 +31,8 @@ object LocalRelation {
 
   def fromProduct(output: Seq[Attribute], data: Seq[Product]): LocalRelation = {
     val schema = StructType.fromAttributes(output)
-    LocalRelation(output, data.map(row => DataTypeConversions.productToRow(row, schema)))
+    val converter = CatalystTypeConverters.createToCatalystConverter(schema)
+    LocalRelation(output, data.map(converter(_).asInstanceOf[Row]))
   }
 }
 

sql/catalyst/src/main/scala/org/apache/spark/sql/types/DataTypeConversions.scala

@@ -25,18 +25,6 @@ import org.apache.spark.sql.catalyst.expressions.GenericRowWithSchema
 
 
 private[sql] object DataTypeConversions {
-
-  def productToRow(product: Product, schema: StructType): Row = {
-    val converted = new Array[Any](schema.length)
-    var i = 0
-    while (i < schema.length) {
-      converted(i) = CatalystTypeConverters.convertToCatalyst(product.productElement(i),
-        schema.fields(i).dataType)
-      i += 1
-    }
-    new GenericRowWithSchema(converted, schema)
-  }
-
   def stringToTime(s: String): java.util.Date = {
     if (!s.contains('T')) {
       // JDBC escape string

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

@@ -962,10 +962,8 @@ class DataFrame private[sql](
     // use a local variable to make sure the map closure doesn't capture the whole DataFrame
     val schema = this.schema
     queryExecution.executedPlan.execute().mapPartitions { rows =>
-      val converters = schema.fields.map {
-        f => CatalystTypeConverters.createToScalaConverter(f.dataType)
-      }
-      rows.map(CatalystTypeConverters.convertRowWithConverters(_, schema, converters))
+      val converter = CatalystTypeConverters.createToScalaConverter(schema)
+      rows.map(converter(_).asInstanceOf[Row])
     }
   }
 

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

@@ -404,10 +404,8 @@ class SQLContext(@transient val sparkContext: SparkContext)
     // TODO: use MutableProjection when rowRDD is another DataFrame and the applied
     // schema differs from the existing schema on any field data type.
     val catalystRows = if (needsConversion) {
-      val converters = schema.fields.map {
-        f => CatalystTypeConverters.createToCatalystConverter(f.dataType)
-      }
-      rowRDD.map(CatalystTypeConverters.convertRowWithConverters(_, schema, converters))
+      val converter = CatalystTypeConverters.createToCatalystConverter(schema)
+      rowRDD.map(converter(_).asInstanceOf[Row])
     } else {
       rowRDD
     }

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

@@ -34,18 +34,13 @@ case class LocalTableScan(output: Seq[Attribute], rows: Seq[Row]) extends LeafNo
 
 
   override def executeCollect(): Array[Row] = {
-    val converters = schema.fields.map {
-      f => CatalystTypeConverters.createToScalaConverter(f.dataType)
-    }
-    rows.map(CatalystTypeConverters.convertRowWithConverters(_, schema, converters)).toArray
+    val converter = CatalystTypeConverters.createToScalaConverter(schema)
+    rows.map(converter(_).asInstanceOf[Row]).toArray
   }
 
 
   override def executeTake(limit: Int): Array[Row] = {
-    val converters = schema.fields.map {
-      f => CatalystTypeConverters.createToScalaConverter(f.dataType)
-    }
-    rows.map(CatalystTypeConverters.convertRowWithConverters(_, schema, converters))
-      .take(limit).toArray
+    val converter = CatalystTypeConverters.createToScalaConverter(schema)
+    rows.map(converter(_).asInstanceOf[Row]).take(limit).toArray
   }
 }

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

@@ -83,10 +83,8 @@ abstract class SparkPlan extends QueryPlan[SparkPlan] with Logging with Serializ
 
   def executeCollect(): Array[Row] = {
     execute().mapPartitions { iter =>
-      val converters = schema.fields.map {
-        f => CatalystTypeConverters.createToScalaConverter(f.dataType)
-      }
-      iter.map(CatalystTypeConverters.convertRowWithConverters(_, schema, converters))
+      val converter = CatalystTypeConverters.createToScalaConverter(schema)
+      iter.map(converter(_).asInstanceOf[Row])
     }.collect()
   }
 
@@ -131,10 +129,8 @@ abstract class SparkPlan extends QueryPlan[SparkPlan] with Logging with Serializ
       partsScanned += numPartsToTry
     }
 
-    val converters = schema.fields.map {
-      f => CatalystTypeConverters.createToScalaConverter(f.dataType)
-    }
-    buf.toArray.map(CatalystTypeConverters.convertRowWithConverters(_, schema, converters))
+    val converter = CatalystTypeConverters.createToScalaConverter(schema)
+    buf.toArray.map(converter(_).asInstanceOf[Row])
   }
 
   protected def newProjection(

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

@@ -140,10 +140,8 @@ case class TakeOrdered(limit: Int, sortOrder: Seq[SortOrder], child: SparkPlan)
   private def collectData(): Array[Row] = child.execute().map(_.copy()).takeOrdered(limit)(ord)
 
   override def executeCollect(): Array[Row] = {
-    val converters = schema.fields.map {
-      f => CatalystTypeConverters.createToScalaConverter(f.dataType)
-    }
-    collectData().map(CatalystTypeConverters.convertRowWithConverters(_, schema, converters))
+    val converter = CatalystTypeConverters.createToScalaConverter(schema)
+    collectData().map(converter(_).asInstanceOf[Row])
   }
 
   // TODO: Terminal split should be implemented differently from non-terminal split.