[SPARK-12293][SQL] Support UnsafeRow in LocalTableScan

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala

@@ -937,8 +937,9 @@ object DecimalAggregates extends Rule[LogicalPlan] {
 object ConvertToLocalRelation extends Rule[LogicalPlan] {
   def apply(plan: LogicalPlan): LogicalPlan = plan transform {
     case Project(projectList, LocalRelation(output, data)) =>
-      val projection = new InterpretedProjection(projectList, output)
-      LocalRelation(projectList.map(_.toAttribute), data.map(projection))
+      val projection = UnsafeProjection.create(projectList, output)
+      LocalRelation(projectList.map(_.toAttribute),
+        data.map(projection(_).copy().asInstanceOf[UnsafeRow]))
   }
 }
 

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

@@ -18,7 +18,8 @@
 package org.apache.spark.sql.catalyst.plans.logical
 
 import org.apache.spark.sql.Row
-import org.apache.spark.sql.catalyst.expressions.Attribute
+import org.apache.spark.sql.catalyst.encoders.{ExpressionEncoder, RowEncoder}
+import org.apache.spark.sql.catalyst.expressions.{Attribute, BindReferences, UnsafeProjection, UnsafeRow}
 import org.apache.spark.sql.catalyst.{CatalystTypeConverters, InternalRow, analysis}
 import org.apache.spark.sql.types.{StructField, StructType}
 
@@ -29,20 +30,27 @@ object LocalRelation {
     new LocalRelation(StructType(output1 +: output).toAttributes)
   }
 
+  def fromInternalRows(output: Seq[Attribute], data: Seq[InternalRow]): LocalRelation = {
+    val projection = UnsafeProjection.create(output.map(_.dataType).toArray)
+    new LocalRelation(output, data.map(projection(_).copy()))
+  }
+
   def fromExternalRows(output: Seq[Attribute], data: Seq[Row]): LocalRelation = {
     val schema = StructType.fromAttributes(output)
-    val converter = CatalystTypeConverters.createToCatalystConverter(schema)
-    LocalRelation(output, data.map(converter(_).asInstanceOf[InternalRow]))
+    val encoder = RowEncoder(schema)
+    LocalRelation(output, data.map(encoder.toRow(_).copy().asInstanceOf[UnsafeRow]))
   }
 
-  def fromProduct(output: Seq[Attribute], data: Seq[Product]): LocalRelation = {
+  def fromProduct[T <: Product : ExpressionEncoder](
+      output: Seq[Attribute],
+      data: Seq[T]): LocalRelation = {
+    val encoder = implicitly[ExpressionEncoder[T]]
     val schema = StructType.fromAttributes(output)
-    val converter = CatalystTypeConverters.createToCatalystConverter(schema)
-    LocalRelation(output, data.map(converter(_).asInstanceOf[InternalRow]))
+    new LocalRelation(output, data.map(encoder.toRow(_).copy().asInstanceOf[UnsafeRow]))
   }
 }
 
-case class LocalRelation(output: Seq[Attribute], data: Seq[InternalRow] = Nil)
+case class LocalRelation(output: Seq[Attribute], data: Seq[UnsafeRow] = Nil)
   extends LeafNode with analysis.MultiInstanceRelation {
 
   /**

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/ConvertToLocalRelationSuite.scala

@@ -35,11 +35,11 @@ class ConvertToLocalRelationSuite extends PlanTest {
   }
 
   test("Project on LocalRelation should be turned into a single LocalRelation") {
-    val testRelation = LocalRelation(
+    val testRelation = LocalRelation.fromInternalRows(
       LocalRelation('a.int, 'b.int).output,
       InternalRow(1, 2) :: InternalRow(4, 5) :: Nil)
 
-    val correctAnswer = LocalRelation(
+    val correctAnswer = LocalRelation.fromInternalRows(
       LocalRelation('a1.int, 'b1.int).output,
       InternalRow(1, 3) :: InternalRow(4, 6) :: Nil)
 

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

@@ -33,6 +33,7 @@ import org.apache.spark.scheduler.{SparkListener, SparkListenerApplicationEnd}
 import org.apache.spark.sql.SQLConf.SQLConfEntry
 import org.apache.spark.sql.catalyst.analysis._
 import org.apache.spark.sql.catalyst.encoders.encoderFor
+import org.apache.spark.sql.catalyst.encoders.ExpressionEncoder
 import org.apache.spark.sql.catalyst.errors.DialectException
 import org.apache.spark.sql.catalyst.expressions._
 import org.apache.spark.sql.catalyst.optimizer.{DefaultOptimizer, Optimizer}
@@ -426,6 +427,7 @@ class SQLContext private[sql](
    */
   @Experimental
   def createDataFrame[A <: Product : TypeTag](data: Seq[A]): DataFrame = {
+    implicit def encoder[T : TypeTag]: ExpressionEncoder[T] = ExpressionEncoder()
     SQLContext.setActive(self)
     val schema = ScalaReflection.schemaFor[A].dataType.asInstanceOf[StructType]
     val attributeSeq = schema.toAttributes
@@ -501,7 +503,7 @@ class SQLContext private[sql](
   def createDataset[T : Encoder](data: Seq[T]): Dataset[T] = {
     val enc = encoderFor[T]
     val attributes = enc.schema.toAttributes
-    val encoded = data.map(d => enc.toRow(d).copy())
+    val encoded = data.map(d => enc.toRow(d).copy().asInstanceOf[UnsafeRow])
     val plan = new LocalRelation(attributes, encoded)
 
     new Dataset[T](this, plan)
@@ -604,7 +606,9 @@ class SQLContext private[sql](
     val className = beanClass.getName
     val beanInfo = Introspector.getBeanInfo(beanClass)
     val rows = SQLContext.beansToRows(data.asScala.iterator, beanInfo, attrSeq)
-    DataFrame(self, LocalRelation(attrSeq, rows.toSeq))
+    val projection = UnsafeProjection.create(attrSeq)
+    DataFrame(self,
+      LocalRelation(attrSeq, rows.toSeq.map(projection(_).copy().asInstanceOf[UnsafeRow])))
   }
 
 

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

@@ -19,17 +19,26 @@ package org.apache.spark.sql.execution
 
 import org.apache.spark.rdd.RDD
 import org.apache.spark.sql.catalyst.InternalRow
-import org.apache.spark.sql.catalyst.expressions.Attribute
+import org.apache.spark.sql.catalyst.expressions.{Attribute, BindReferences, UnsafeProjection, UnsafeRow}
 
+private[sql] object LocalTableScan {
+  def fromInternalRows(output: Seq[Attribute], data: Seq[InternalRow]): LocalTableScan = {
+    val projection = UnsafeProjection.create(output.map(_.dataType).toArray)
+    new LocalTableScan(output, data.map(projection(_).copy()))
+  }
+}
 
 /**
  * Physical plan node for scanning data from a local collection.
  */
 private[sql] case class LocalTableScan(
     output: Seq[Attribute],
-    rows: Seq[InternalRow]) extends LeafNode {
+    rows: Seq[UnsafeRow]) extends LeafNode {
+
+  override def outputsUnsafeRows: Boolean = true
+  override def canProcessUnsafeRows: Boolean = true
 
-  private lazy val rdd = sqlContext.sparkContext.parallelize(rows)
+  private lazy val rdd = sqlContext.sparkContext.parallelize(rows).asInstanceOf[RDD[InternalRow]]
 
   protected override def doExecute(): RDD[InternalRow] = rdd
 

sql/core/src/main/scala/org/apache/spark/sql/execution/stat/StatFunctions.scala

@@ -144,6 +144,7 @@ private[sql] object StatFunctions extends Logging {
     }
     val schema = StructType(StructField(tableName, StringType) +: headerNames)
 
-    new DataFrame(df.sqlContext, LocalRelation(schema.toAttributes, table)).na.fill(0.0)
+    new DataFrame(df.sqlContext,
+      LocalRelation.fromInternalRows(schema.toAttributes, table)).na.fill(0.0)
   }
 }

sql/core/src/test/scala/org/apache/spark/sql/execution/RowFormatConvertersSuite.scala

@@ -99,12 +99,11 @@ class RowFormatConvertersSuite extends SparkPlanTest with SharedSQLContext {
     val rows = (1 to 100).map { i =>
       InternalRow(new GenericArrayData(Array[Any](UTF8String.fromString(i.toString))))
     }
-    val relation = LocalTableScan(Seq(AttributeReference("t", schema)()), rows)
+    val relation = LocalTableScan.fromInternalRows(Seq(AttributeReference("t", schema)()), rows)
 
     val plan =
       DummyPlan(
-        ConvertToSafe(
-          ConvertToUnsafe(relation)))
+        ConvertToSafe(relation))
     assert(plan.execute().collect().map(_.getUTF8String(0).toString) === (1 to 100).map(_.toString))
   }
 }

sql/core/src/test/scala/org/apache/spark/sql/execution/local/DummyNode.scala

@@ -17,11 +17,26 @@
 
 package org.apache.spark.sql.execution.local
 
+import scala.reflect.runtime.universe.TypeTag
+
 import org.apache.spark.sql.SQLConf
 import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.encoders.ExpressionEncoder
 import org.apache.spark.sql.catalyst.expressions.Attribute
 import org.apache.spark.sql.catalyst.plans.logical.LocalRelation
 
+private[local] object DummyNode {
+  val CLOSED: Int = Int.MinValue
+
+  def apply[A <: Product : TypeTag](
+      output: Seq[Attribute],
+      data: Seq[A],
+      conf: SQLConf = new SQLConf): DummyNode = {
+    implicit def encoder[T : TypeTag]: ExpressionEncoder[T] = ExpressionEncoder()
+    new DummyNode(output, LocalRelation.fromProduct(output, data), conf)
+  }
+}
+
 /**
  * A dummy [[LocalNode]] that just returns rows from a [[LocalRelation]].
  */
@@ -36,10 +51,6 @@ private[local] case class DummyNode(
   private var index: Int = CLOSED
   private val input: Seq[InternalRow] = relation.data
 
-  def this(output: Seq[Attribute], data: Seq[Product], conf: SQLConf = new SQLConf) {
-    this(output, LocalRelation.fromProduct(output, data), conf)
-  }
-
   def isOpen: Boolean = index != CLOSED
 
   override def children: Seq[LocalNode] = Seq.empty
@@ -62,7 +73,3 @@ private[local] case class DummyNode(
     index = CLOSED
   }
 }
-
-private object DummyNode {
-  val CLOSED: Int = Int.MinValue
-}

sql/core/src/test/scala/org/apache/spark/sql/execution/local/ExpandNodeSuite.scala

@@ -23,7 +23,7 @@ import org.apache.spark.sql.catalyst.dsl.expressions._
 class ExpandNodeSuite extends LocalNodeTest {
 
   private def testExpand(inputData: Array[(Int, Int)] = Array.empty): Unit = {
-    val inputNode = new DummyNode(kvIntAttributes, inputData)
+    val inputNode = DummyNode(kvIntAttributes, inputData)
     val projections = Seq(Seq('k + 'v, 'k - 'v), Seq('k * 'v, 'k / 'v))
     val expandNode = new ExpandNode(conf, projections, inputNode.output, inputNode)
     val resolvedNode = resolveExpressions(expandNode)

sql/core/src/test/scala/org/apache/spark/sql/execution/local/FilterNodeSuite.scala

@@ -24,7 +24,7 @@ class FilterNodeSuite extends LocalNodeTest {
 
   private def testFilter(inputData: Array[(Int, Int)] = Array.empty): Unit = {
     val cond = 'k % 2 === 0
-    val inputNode = new DummyNode(kvIntAttributes, inputData)
+    val inputNode = DummyNode(kvIntAttributes, inputData)
     val filterNode = new FilterNode(conf, cond, inputNode)
     val resolvedNode = resolveExpressions(filterNode)
     val expectedOutput = inputData.filter { case (k, _) => k % 2 == 0 }

sql/core/src/test/scala/org/apache/spark/sql/execution/local/HashJoinNodeSuite.scala

@@ -62,8 +62,8 @@ class HashJoinNodeSuite extends LocalNodeTest {
     // Actual test body
     def runTest(leftInput: Array[(Int, String)], rightInput: Array[(Int, String)]): Unit = {
       val rightInputMap = rightInput.toMap
-      val leftNode = new DummyNode(joinNameAttributes, leftInput)
-      val rightNode = new DummyNode(joinNicknameAttributes, rightInput)
+      val leftNode = DummyNode(joinNameAttributes, leftInput)
+      val rightNode = DummyNode(joinNicknameAttributes, rightInput)
       val makeBinaryHashJoinNode = (node1: LocalNode, node2: LocalNode) => {
         val binaryHashJoinNode =
           BinaryHashJoinNode(conf, Seq('id1), Seq('id2), buildSide, node1, node2)

sql/core/src/test/scala/org/apache/spark/sql/execution/local/IntersectNodeSuite.scala

@@ -24,8 +24,8 @@ class IntersectNodeSuite extends LocalNodeTest {
     val n = 100
     val leftData = (1 to n).filter { i => i % 2 == 0 }.map { i => (i, i) }.toArray
     val rightData = (1 to n).filter { i => i % 3 == 0 }.map { i => (i, i) }.toArray
-    val leftNode = new DummyNode(kvIntAttributes, leftData)
-    val rightNode = new DummyNode(kvIntAttributes, rightData)
+    val leftNode = DummyNode(kvIntAttributes, leftData)
+    val rightNode = DummyNode(kvIntAttributes, rightData)
     val intersectNode = new IntersectNode(conf, leftNode, rightNode)
     val expectedOutput = leftData.intersect(rightData)
     val actualOutput = intersectNode.collect().map { case row =>

sql/core/src/test/scala/org/apache/spark/sql/execution/local/LimitNodeSuite.scala

@@ -21,7 +21,7 @@ package org.apache.spark.sql.execution.local
 class LimitNodeSuite extends LocalNodeTest {
 
   private def testLimit(inputData: Array[(Int, Int)] = Array.empty, limit: Int = 10): Unit = {
-    val inputNode = new DummyNode(kvIntAttributes, inputData)
+    val inputNode = DummyNode(kvIntAttributes, inputData)
     val limitNode = new LimitNode(conf, limit, inputNode)
     val expectedOutput = inputData.take(limit)
     val actualOutput = limitNode.collect().map { case row =>

sql/core/src/test/scala/org/apache/spark/sql/execution/local/LocalNodeSuite.scala

@@ -22,7 +22,7 @@ class LocalNodeSuite extends LocalNodeTest {
   private val data = (1 to 100).map { i => (i, i) }.toArray
 
   test("basic open, next, fetch, close") {
-    val node = new DummyNode(kvIntAttributes, data)
+    val node = DummyNode(kvIntAttributes, data)
     assert(!node.isOpen)
     node.open()
     assert(node.isOpen)
@@ -42,7 +42,7 @@ class LocalNodeSuite extends LocalNodeTest {
   }
 
   test("asIterator") {
-    val node = new DummyNode(kvIntAttributes, data)
+    val node = DummyNode(kvIntAttributes, data)
     val iter = node.asIterator
     node.open()
     data.foreach { case (k, v) =>
@@ -61,7 +61,7 @@ class LocalNodeSuite extends LocalNodeTest {
   }
 
   test("collect") {
-    val node = new DummyNode(kvIntAttributes, data)
+    val node = DummyNode(kvIntAttributes, data)
     node.open()
     val collected = node.collect()
     assert(collected.size === data.size)

sql/core/src/test/scala/org/apache/spark/sql/execution/local/NestedLoopJoinNodeSuite.scala

@@ -47,8 +47,8 @@ class NestedLoopJoinNodeSuite extends LocalNodeTest {
         joinType: JoinType,
         leftInput: Array[(Int, String)],
         rightInput: Array[(Int, String)]): Unit = {
-      val leftNode = new DummyNode(joinNameAttributes, leftInput)
-      val rightNode = new DummyNode(joinNicknameAttributes, rightInput)
+      val leftNode = DummyNode(joinNameAttributes, leftInput)
+      val rightNode = DummyNode(joinNicknameAttributes, rightInput)
       val cond = 'id1 === 'id2
       val makeNode = (node1: LocalNode, node2: LocalNode) => {
         resolveExpressions(

sql/core/src/test/scala/org/apache/spark/sql/execution/local/ProjectNodeSuite.scala

@@ -28,7 +28,7 @@ class ProjectNodeSuite extends LocalNodeTest {
     AttributeReference("name", StringType)())
 
   private def testProject(inputData: Array[(Int, Int, String)] = Array.empty): Unit = {
-    val inputNode = new DummyNode(pieAttributes, inputData)
+    val inputNode = DummyNode(pieAttributes, inputData)
     val columns = Seq[NamedExpression](inputNode.output(0), inputNode.output(2))
     val projectNode = new ProjectNode(conf, columns, inputNode)
     val expectedOutput = inputData.map { case (id, age, name) => (id, name) }

sql/core/src/test/scala/org/apache/spark/sql/execution/local/SampleNodeSuite.scala

@@ -29,7 +29,7 @@ class SampleNodeSuite extends LocalNodeTest {
     val maybeOut = if (withReplacement) "" else "out"
     test(s"with$maybeOut replacement") {
       val inputData = (1 to 1000).map { i => (i, i) }.toArray
-      val inputNode = new DummyNode(kvIntAttributes, inputData)
+      val inputNode = DummyNode(kvIntAttributes, inputData)
       val sampleNode = new SampleNode(conf, lowerb, upperb, withReplacement, seed, inputNode)
       val sampler =
         if (withReplacement) {

sql/core/src/test/scala/org/apache/spark/sql/execution/local/TakeOrderedAndProjectNodeSuite.scala

@@ -30,7 +30,7 @@ class TakeOrderedAndProjectNodeSuite extends LocalNodeTest {
     val ascOrDesc = if (desc) "desc" else "asc"
     test(ascOrDesc) {
       val inputData = Random.shuffle((1 to 100).toList).map { i => (i, i) }.toArray
-      val inputNode = new DummyNode(kvIntAttributes, inputData)
+      val inputNode = DummyNode(kvIntAttributes, inputData)
       val firstColumn = inputNode.output(0)
       val sortDirection = if (desc) Descending else Ascending
       val sortOrder = SortOrder(firstColumn, sortDirection)

sql/core/src/test/scala/org/apache/spark/sql/execution/local/UnionNodeSuite.scala

@@ -22,7 +22,7 @@ class UnionNodeSuite extends LocalNodeTest {
 
   private def testUnion(inputData: Seq[Array[(Int, Int)]]): Unit = {
     val inputNodes = inputData.map { data =>
-      new DummyNode(kvIntAttributes, data)
+      DummyNode(kvIntAttributes, data)
     }
     val unionNode = new UnionNode(conf, inputNodes)
     val expectedOutput = inputData.flatten