remove limit

Author: ulysses-you

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala

@@ -1458,15 +1458,7 @@ object PushPredicateThroughJoin extends Rule[LogicalPlan] with PredicateHelper {
  */
 object EliminateLimits extends Rule[LogicalPlan] {
   private def canEliminate(limitExpr: Expression, child: LogicalPlan): Boolean = {
-    // We skip such case that Sort is after Limit since
-    // SparkStrategies will convert them to TakeOrderedAndProjectExec
-    val skipEliminate = child match {
-      case Sort(_, true, _) => true
-      case Project(_, Sort(_, true, _)) => true
-      case _ => false
-    }
-    !skipEliminate && limitExpr.foldable &&
-      child.maxRows.exists { _ <= limitExpr.eval().toString.toLong }
+    limitExpr.foldable && child.maxRows.exists { _ <= limitExpr.eval().toString.toLong }
   }
 
   def apply(plan: LogicalPlan): LogicalPlan = plan transformDown {

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/CombiningLimitsSuite.scala

@@ -109,8 +109,11 @@ class CombiningLimitsSuite extends PlanTest {
     comparePlans(optimized3, query3)
 
     // test sort after limit
-    val query4 = testRelation.select(Symbol("a")).orderBy(Symbol("a").asc).limit(1).analyze
+    val query4 = testRelation.select().groupBy()(count(1))
+      .orderBy(Symbol("a").asc).limit(1).analyze
     val optimized4 = Optimize.execute(query4)
-    comparePlans(optimized4, optimized4)
+    val expected4 = testRelation.select().groupBy()(count(1))
+      .orderBy(Symbol("a").asc).analyze
+    comparePlans(optimized4, expected4)
   }
 }