[SPARK-51820][SQL][FOLLOWUP][CONNECT] Don't add UnresolvedOrdinal when appending grouping to aggregate expressions

### What changes were proposed in this pull request?
Don't add `UnresolvedOrdinal` when appending grouping to aggregate expressions in Spark Connect.

### Why are the changes needed?
Change is needed to fix a regression caused by #50606 where `UnresolvedOrdinal` would end up in aggregate expression and propagate all the way to `CheckAnalysis` where it would throw an error

### Does this PR introduce _any_ user-facing change?
No

### How was this patch tested?
Added a test case for the correct behavior.

### Was this patch authored or co-authored using generative AI tooling?
No.

Closes #50958 from mihailotim-db/mihailotim-db/fix_unresolved_ordinal.

Authored-by: Mihailo Timotic <mihailo.timotic@databricks.com>
Signed-off-by: Wenchen Fan <wenchen@databricks.com>

Author: mihailotim-db

sql/connect/server/src/main/scala/org/apache/spark/sql/connect/planner/SparkConnectPlanner.scala

@@ -2457,11 +2457,13 @@ class SparkConnectPlanner(
         input
       }
 
-    val groupingExpressionsWithOrdinals = rel.getGroupingExpressionsList.asScala.toSeq
-      .map(transformGroupingExpressionAndReplaceOrdinals)
+    val groupingExpressions =
+      rel.getGroupingExpressionsList.asScala.toSeq.map(transformExpression)
+    val groupingExpressionsWithOrdinals =
+      groupingExpressions.map(replaceOrdinalsInGroupingExpressions)
     val aggExprs = rel.getAggregateExpressionsList.asScala.toSeq
       .map(expr => transformExpressionWithTypedReduceExpression(expr, logicalPlan))
-    val aliasedAgg = (groupingExpressionsWithOrdinals ++ aggExprs).map(toNamedExpression)
+    val aliasedAgg = (groupingExpressions ++ aggExprs).map(toNamedExpression)
 
     rel.getGroupType match {
       case proto.Aggregate.GroupType.GROUP_TYPE_GROUPBY =>
@@ -2506,7 +2508,10 @@ class SparkConnectPlanner(
         val groupingSetsExpressionsWithOrdinals =
           rel.getGroupingSetsList.asScala.toSeq.map { getGroupingSets =>
             getGroupingSets.getGroupingSetList.asScala.toSeq
-              .map(transformGroupingExpressionAndReplaceOrdinals)
+              .map(groupingExpressions => {
+                val transformedGroupingExpression = transformExpression(groupingExpressions)
+                replaceOrdinalsInGroupingExpressions(transformedGroupingExpression)
+              })
           }
         logical.Aggregate(
           groupingExpressions = Seq(
@@ -2521,18 +2526,15 @@ class SparkConnectPlanner(
   }
 
   /**
-   * Transforms an input protobuf grouping expression into the Catalyst expression and converts
-   * top-level integer [[Literal]]s to [[UnresolvedOrdinal]]s, if `groupByOrdinal` is enabled.
+   * Replaces top-level integer [[Literal]]s to [[UnresolvedOrdinal]]s, if `groupByOrdinal` is
+   * enabled.
    */
-  private def transformGroupingExpressionAndReplaceOrdinals(
-      groupingExpression: proto.Expression) = {
-    val transformedGroupingExpression = transformExpression(groupingExpression)
+  private def replaceOrdinalsInGroupingExpressions(groupingExpression: Expression) =
     if (session.sessionState.conf.groupByOrdinal) {
-      replaceIntegerLiteralWithOrdinal(transformedGroupingExpression)
+      replaceIntegerLiteralWithOrdinal(groupingExpression)
     } else {
-      transformedGroupingExpression
+      groupingExpression
     }
-  }
 
   @deprecated("TypedReduce is now implemented using a normal UDAF aggregator.", "4.0.0")
   private def transformTypedReduceExpression(

sql/connect/server/src/test/scala/org/apache/spark/sql/connect/planner/SparkConnectPlannerSuite.scala

@@ -29,7 +29,8 @@ import org.apache.spark.sql.catalyst.InternalRow
 import org.apache.spark.sql.catalyst.analysis.{UnresolvedAlias, UnresolvedFunction, UnresolvedRelation}
 import org.apache.spark.sql.catalyst.expressions.{AttributeReference, UnsafeProjection}
 import org.apache.spark.sql.catalyst.plans.logical
-import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.catalyst.plans.logical.{Aggregate, LogicalPlan}
+import org.apache.spark.sql.catalyst.trees.TreePattern
 import org.apache.spark.sql.catalyst.types.DataTypeUtils
 import org.apache.spark.sql.classic.Dataset
 import org.apache.spark.sql.connect.SparkConnectTestUtils
@@ -922,4 +923,34 @@ class SparkConnectPlannerSuite extends SparkFunSuite with SparkConnectPlanTest {
     assert(fn3.nameParts.head == "abcde")
     assert(fn3.isInternal)
   }
+
+  test("SPARK-51820 aggregate list should not contain UnresolvedOrdinal") {
+    val ordinal = proto.Expression
+      .newBuilder()
+      .setLiteral(proto.Expression.Literal.newBuilder().setInteger(1).build())
+      .build()
+
+    val sum =
+      proto.Expression
+        .newBuilder()
+        .setUnresolvedFunction(
+          proto.Expression.UnresolvedFunction
+            .newBuilder()
+            .setFunctionName("sum")
+            .addArguments(ordinal))
+        .build()
+
+    val aggregate = proto.Aggregate.newBuilder
+      .setInput(readRel)
+      .addAggregateExpressions(sum)
+      .addGroupingExpressions(ordinal)
+      .setGroupType(proto.Aggregate.GroupType.GROUP_TYPE_ROLLUP)
+      .build()
+
+    val plan =
+      transform(proto.Relation.newBuilder.setAggregate(aggregate).build()).asInstanceOf[Aggregate]
+
+    assert(plan.aggregateExpressions.forall(aggregateExpression =>
+      !aggregateExpression.containsPattern(TreePattern.UNRESOLVED_ORDINAL)))
+  }
 }