Copy tuples in Limit before shuffle.

rxin · rxin · commit 47d33278af0a · 2014-03-25T22:16:59.000-07:00
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/basicOperators.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/basicOperators.scala
@@ -76,7 +76,7 @@ case class Limit(limit: Int, child: SparkPlan)(@transient sc: SparkContext) exte
 
   override def execute() = {
     child.execute()
-      .mapPartitions(_.take(limit))
+      .mapPartitions(_.take(limit).map(_.copy()))
       .coalesce(1, shuffle = true)
       .mapPartitions(_.take(limit))
   }

Original file line number	Diff line number	Diff line change
`@@ -76,7 +76,7 @@ case class Limit(limit: Int, child: SparkPlan)(@transient sc: SparkContext) exte`
`76`	`76`
`77`	`77`	`override def execute() = {`
`78`	`78`	`child.execute()`
`79`		`- .mapPartitions(_.take(limit))`
	`79`	`+ .mapPartitions(_.take(limit).map(_.copy()))`
`80`	`80`	`.coalesce(1, shuffle = true)`
`81`	`81`	`.mapPartitions(_.take(limit))`
`82`	`82`	`}`