Optimize code

beliefer · beliefer · commit 44b1861cac33 · 2020-02-25T15:22:49.000+08:00
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/limit.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/limit.scala
@@ -181,8 +181,7 @@ case class GlobalLimitAndOffsetExec(
 
   override def doExecute(): RDD[InternalRow] = {
     val rdd = child.execute().mapPartitions { iter => iter.take(limit + offset)}
-    val skips = rdd.take(offset)
-    rdd.filter(!skips.contains(_))
+    rdd.zipWithIndex().filter(_._2 >= offset).map(_._1)
   }
 
   private lazy val skipTerm = BaseLimitExec.newLimitCountTerm()

Original file line number	Diff line number	Diff line change
`@@ -181,8 +181,7 @@ case class GlobalLimitAndOffsetExec(`
`181`	`181`
`182`	`182`	`override def doExecute(): RDD[InternalRow] = {`
`183`	`183`	`val rdd = child.execute().mapPartitions { iter => iter.take(limit + offset)}`
`184`		`- val skips = rdd.take(offset)`
`185`		`- rdd.filter(!skips.contains(_))`
	`184`	`+ rdd.zipWithIndex().filter(_._2 >= offset).map(_._1)`
`186`	`185`	`}`
`187`	`186`
`188`	`187`	`private lazy val skipTerm = BaseLimitExec.newLimitCountTerm()`