Set proper outputsUnsafeRows when projectList is None.

Author: viirya

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

@@ -204,7 +204,12 @@ case class TakeOrderedAndProject(
     projectOutput.getOrElse(child.output)
   }
 
-  override def outputsUnsafeRows: Boolean = true
+  override def outputsUnsafeRows: Boolean = if (projectList.isDefined) {
+    true
+  } else {
+    child.outputsUnsafeRows
+  }
+
   override def canProcessUnsafeRows: Boolean = true
   override def canProcessSafeRows: Boolean = true
 
@@ -216,19 +221,13 @@ 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 {
-      if (child.outputsUnsafeRows) {
-        data
-      } else {
-        data.map(unsafeProjection(_).copy().asInstanceOf[InternalRow])
-      }
+      data
     }
   }