fix

Author: ulysses-you

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

@@ -1458,7 +1458,7 @@ object PushPredicateThroughJoin extends Rule[LogicalPlan] with PredicateHelper {
  */
 object EliminateLimits extends Rule[LogicalPlan] {
   private def canEliminate(limitExpr: Expression, child: LogicalPlan): Boolean = {
-    limitExpr.foldable && child.maxRows.exists { _ <= limitExpr.eval().toString.toLong }
+    limitExpr.foldable && child.maxRows.exists { _ <= limitExpr.eval().asInstanceOf[Int] }
   }
 
   def apply(plan: LogicalPlan): LogicalPlan = plan transformDown {

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

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