Add extra unsafe projection for the case projectList is None.

Author: viirya

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

@@ -216,13 +216,19 @@ case class TakeOrderedAndProject(
 
   // TODO: remove @transient after figure out how to clean closure at InsertIntoHiveTable.
   @transient private val projection = projectList.map(UnsafeProjection.create(_, child.output))
+  @transient private lazy val unsafeProjection =
+    UnsafeProjection.create(child.output.map(_.dataType).toArray)
 
   private def collectData(): Array[InternalRow] = {
     val data = child.execute().map(_.copy()).takeOrdered(limit)(ord)
     if (projection.isDefined) {
       projection.map(p => data.map(p(_).copy().asInstanceOf[InternalRow])).get
     } else {
-      data
+      if (child.outputsUnsafeRows) {
+        data
+      } else {
+        data.map(unsafeProjection(_).copy().asInstanceOf[InternalRow])
+      }
     }
   }
 

sql/core/src/test/scala/org/apache/spark/sql/SQLQuerySuite.scala

@@ -571,6 +571,12 @@ class SQLQuerySuite extends QueryTest with SharedSQLContext {
       mapData.collect().take(1).map(Row.fromTuple).toSeq)
   }
 
+  test("sort and limit") {
+    checkAnswer(
+      sql("SELECT * FROM arrayData ORDER BY data[0] ASC LIMIT 1"),
+      arrayData.collect().sortBy(_.data(0)).map(Row.fromTuple).take(1).toSeq)
+  }
+
   test("CTE feature") {
     checkAnswer(
       sql("with q1 as (select * from testData limit 10) select * from q1"),