[SPARK-26138][SQL] Push limit to the right side

guoxiaolong · guoxiaolong · commit 588c1513906a · 2018-11-23T12:15:30.000+08:00
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala
@@ -459,7 +459,7 @@ object LimitPushDown extends Rule[LogicalPlan] {
       val newJoin = joinType match {
         case RightOuter => join.copy(right = maybePushLocalLimit(exp, right))
         case LeftOuter => join.copy(left = maybePushLocalLimit(exp, left))
-        case Cross => join.copy(left = maybePushLocalLimit(exp, left))
+        case Cross => join.copy(left = maybePushLocalLimit(exp, left), right = maybePushLocalLimit(exp, right))
         case _ => join
       }
       LocalLimit(exp, newJoin)
diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/LimitPushdownSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/LimitPushdownSuite.scala
@@ -175,21 +175,21 @@ class LimitPushdownSuite extends PlanTest {
   test("cross join") {
     val originalQuery = x.join(y, Cross).limit(1)
     val optimized = Optimize.execute(originalQuery.analyze)
-    val correctAnswer = Limit(1, LocalLimit(1, x).join(y, Cross)).analyze
+    val correctAnswer = Limit(1, LocalLimit(1, x).join(LocalLimit(1, y), Cross)).analyze
     comparePlans(optimized, correctAnswer)
   }
 
   test("cross join and left sides are limited") {
     val originalQuery = x.limit(2).join(y, Cross).limit(1)
     val optimized = Optimize.execute(originalQuery.analyze)
-    val correctAnswer = Limit(1, LocalLimit(1, x).join(y, Cross)).analyze
+    val correctAnswer = Limit(1, LocalLimit(1, x).join(LocalLimit(1, y), Cross)).analyze
     comparePlans(optimized, correctAnswer)
   }
 
   test("cross join and right sides are limited") {
     val originalQuery = x.join(y.limit(2), Cross).limit(1)
     val optimized = Optimize.execute(originalQuery.analyze)
-    val correctAnswer = Limit(1, LocalLimit(1, x).join(Limit(2, y), Cross)).analyze
+    val correctAnswer = Limit(1, LocalLimit(1, x).join(LocalLimit(1, y), Cross)).analyze
     comparePlans(optimized, correctAnswer)
   }
 }

Original file line number	Diff line number	Diff line change
`@@ -459,7 +459,7 @@ object LimitPushDown extends Rule[LogicalPlan] {`
`459`	`459`	`val newJoin = joinType match {`
`460`	`460`	`case RightOuter => join.copy(right = maybePushLocalLimit(exp, right))`
`461`	`461`	`case LeftOuter => join.copy(left = maybePushLocalLimit(exp, left))`
`462`		`- case Cross => join.copy(left = maybePushLocalLimit(exp, left))`
	`462`	`+ case Cross => join.copy(left = maybePushLocalLimit(exp, left), right = maybePushLocalLimit(exp, right))`
`463`	`463`	`case _ => join`
`464`	`464`	`}`
`465`	`465`	`LocalLimit(exp, newJoin)`
Original file line number	Diff line number	Diff line change
`@@ -175,21 +175,21 @@ class LimitPushdownSuite extends PlanTest {`
`175`	`175`	`test("cross join") {`
`176`	`176`	`val originalQuery = x.join(y, Cross).limit(1)`
`177`	`177`	`val optimized = Optimize.execute(originalQuery.analyze)`
`178`		`- val correctAnswer = Limit(1, LocalLimit(1, x).join(y, Cross)).analyze`
	`178`	`+ val correctAnswer = Limit(1, LocalLimit(1, x).join(LocalLimit(1, y), Cross)).analyze`
`179`	`179`	`comparePlans(optimized, correctAnswer)`
`180`	`180`	`}`
`181`	`181`
`182`	`182`	`test("cross join and left sides are limited") {`
`183`	`183`	`val originalQuery = x.limit(2).join(y, Cross).limit(1)`
`184`	`184`	`val optimized = Optimize.execute(originalQuery.analyze)`
`185`		`- val correctAnswer = Limit(1, LocalLimit(1, x).join(y, Cross)).analyze`
	`185`	`+ val correctAnswer = Limit(1, LocalLimit(1, x).join(LocalLimit(1, y), Cross)).analyze`
`186`	`186`	`comparePlans(optimized, correctAnswer)`
`187`	`187`	`}`
`188`	`188`
`189`	`189`	`test("cross join and right sides are limited") {`
`190`	`190`	`val originalQuery = x.join(y.limit(2), Cross).limit(1)`
`191`	`191`	`val optimized = Optimize.execute(originalQuery.analyze)`
`192`		`- val correctAnswer = Limit(1, LocalLimit(1, x).join(Limit(2, y), Cross)).analyze`
	`192`	`+ val correctAnswer = Limit(1, LocalLimit(1, x).join(LocalLimit(1, y), Cross)).analyze`
`193`	`193`	`comparePlans(optimized, correctAnswer)`
`194`	`194`	`}`
`195`	`195`	`}`