[SPARK-9949] [SQL] Fix TakeOrderedAndProject's output.

https://issues.apache.org/jira/browse/SPARK-9949

Author: Yin Huai <[email protected]>

Closes #8179 from yhuai/SPARK-9949.

(cherry picked from commit 932b24f)
Signed-off-by: Reynold Xin <[email protected]>

Conflicts:
	sql/core/src/test/scala/org/apache/spark/sql/execution/PlannerSuite.scala

Author: yhuai

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

@@ -157,7 +157,10 @@ case class TakeOrderedAndProject(
     projectList: Option[Seq[NamedExpression]],
     child: SparkPlan) extends UnaryNode {
 
-  override def output: Seq[Attribute] = child.output
+  override def output: Seq[Attribute] = {
+    val projectOutput = projectList.map(_.map(_.toAttribute))
+    projectOutput.getOrElse(child.output)
+  }
 
   override def outputPartitioning: Partitioning = SinglePartition
 
@@ -181,6 +184,13 @@ case class TakeOrderedAndProject(
   protected override def doExecute(): RDD[Row] = sparkContext.makeRDD(collectData(), 1)
 
   override def outputOrdering: Seq[SortOrder] = sortOrder
+
+  override def simpleString: String = {
+    val orderByString = sortOrder.mkString("[", ",", "]")
+    val outputString = output.mkString("[", ",", "]")
+
+    s"TakeOrderedAndProject(limit=$limit, orderBy=$orderByString, output=$outputString)"
+  }
 }
 
 /**

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

@@ -144,8 +144,22 @@ class PlannerSuite extends SparkFunSuite {
   }
 
   test("efficient limit -> project -> sort") {
-    val query = testData.sort('key).select('value).limit(2).logicalPlan
-    val planned = planner.TakeOrderedAndProject(query)
-    assert(planned.head.isInstanceOf[execution.TakeOrderedAndProject])
+    {
+      val query =
+        testData.select('key, 'value).sort('key).limit(2).logicalPlan
+      val planned = planner.TakeOrderedAndProject(query)
+      assert(planned.head.isInstanceOf[execution.TakeOrderedAndProject])
+      assert(planned.head.output === testData.select('key, 'value).logicalPlan.output)
+    }
+
+    {
+      // We need to make sure TakeOrderedAndProject's output is correct when we push a project
+      // into it.
+      val query =
+        testData.select('key, 'value).sort('key).select('value, 'key).limit(2).logicalPlan
+      val planned = planner.TakeOrderedAndProject(query)
+      assert(planned.head.isInstanceOf[execution.TakeOrderedAndProject])
+      assert(planned.head.output === testData.select('value, 'key).logicalPlan.output)
+    }
   }
 }