fix

Author: wangyum

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

@@ -649,9 +649,9 @@ object LimitPushDown extends Rule[LogicalPlan] {
       LocalLimit(exp, project.copy(child = pushLocalLimitThroughJoin(exp, join)))
     // Push down limit 1 through Aggregate if it is group only.
     case Limit(le @ IntegerLiteral(1), a: Aggregate) if a.groupOnly =>
-      Limit(le, a.copy(child = LocalLimit(le, a.child)))
+      Limit(le, Project(a.output, LocalLimit(le, a.child)))
     case Limit(le @ IntegerLiteral(1), p @ Project(_, a: Aggregate)) if a.groupOnly =>
-      Limit(le, p.copy(child = a.copy(child = LocalLimit(le, a.child))))
+      Limit(le, p.copy(child = Project(a.output, LocalLimit(le, a.child))))
   }
 }
 

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

@@ -244,16 +244,15 @@ class LimitPushdownSuite extends PlanTest {
     // Push down when it is group only and limit 1.
     comparePlans(
       Optimize.execute(x.groupBy("x.a".attr)("x.a".attr).limit(1).analyze),
-      LocalLimit(1, x).groupBy("x.a".attr)("x.a".attr).limit(1).analyze)
+      LocalLimit(1, x).select("x.a".attr).limit(1).analyze)
 
     comparePlans(
       Optimize.execute(x.groupBy("x.a".attr)("x.a".attr).select("x.a".attr).limit(1).analyze),
-      LocalLimit(1, x).groupBy("x.a".attr)("x.a".attr).select("x.a".attr).limit(1).analyze)
+      LocalLimit(1, x).select("x.a".attr).select("x.a".attr).limit(1).analyze)
 
     comparePlans(
       Optimize.execute(x.union(y).groupBy("x.a".attr)("x.a".attr).limit(1).analyze),
-      LocalLimit(1, LocalLimit(1, x).union(LocalLimit(1, y)))
-        .groupBy("x.a".attr)("x.a".attr).limit(1).analyze)
+      LocalLimit(1, LocalLimit(1, x).union(LocalLimit(1, y))).select("x.a".attr).limit(1).analyze)
 
     // No push down
     comparePlans(