fix: pass return type for MapSort serde

The MapSort serde for Spark 4.0 called scalarFunctionExprToProto without a
return type. The Rust planner then looked up "map_sort" in the session
UDF registry to infer the type, but map_sort is only handled via the
create_comet_physical_fun match dispatch, not registered as a UDF, causing
"There is no UDF named 'map_sort' in the registry" at execution time
(e.g., group-by on a map column in CollationSuite).

Pass ms.dataType explicitly via scalarFunctionExprToProtoWithReturnType,
matching the pattern used by ceil, floor, and other scalar functions.

Author: andygrove

spark/src/main/spark-4.0/org/apache/comet/shims/CometExprShim.scala

@@ -30,7 +30,7 @@ import org.apache.comet.CometSparkSessionExtensions.withInfo
 import org.apache.comet.expressions.{CometCast, CometEvalMode}
 import org.apache.comet.serde.{CommonStringExprs, Compatible, ExprOuterClass, Incompatible}
 import org.apache.comet.serde.ExprOuterClass.{BinaryOutputStyle, Expr}
-import org.apache.comet.serde.QueryPlanSerde.{exprToProtoInternal, optExprWithInfo, scalarFunctionExprToProto}
+import org.apache.comet.serde.QueryPlanSerde.{exprToProtoInternal, optExprWithInfo, scalarFunctionExprToProto, scalarFunctionExprToProtoWithReturnType}
 
 /**
  * `CometExprShim` acts as a shim for parsing expressions from different Spark versions.
@@ -130,7 +130,11 @@ trait CometExprShim extends CommonStringExprs {
 
       case ms: MapSort =>
         val childExpr = exprToProtoInternal(ms.child, inputs, binding)
-        val mapSortExpr = scalarFunctionExprToProto("map_sort", childExpr)
+        val mapSortExpr = scalarFunctionExprToProtoWithReturnType(
+          "map_sort",
+          ms.dataType,
+          failOnError = false,
+          childExpr)
         optExprWithInfo(mapSortExpr, ms, ms.child)
 
       case _ => None

spark/src/test/scala/org/apache/comet/CometMapExpressionSuite.scala

@@ -27,6 +27,7 @@ import org.apache.spark.sql.functions._
 import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.types.BinaryType
 
+import org.apache.comet.CometSparkSessionExtensions.isSpark40Plus
 import org.apache.comet.serde.CometMapFromEntries
 import org.apache.comet.testing.{DataGenOptions, ParquetGenerator, SchemaGenOptions}
 
@@ -221,6 +222,18 @@ class CometMapExpressionSuite extends CometTestBase {
     }
   }
 
+  test("group by map column with string values") {
+    assume(isSpark40Plus, "Spark 4.0 inserts MapSort for group-by on map keys")
+    withTable("t_map_group") {
+      sql("""
+        |CREATE TABLE t_map_group USING parquet AS
+        |SELECT map(cast(id as string), cast(id + 100 as string)) as m
+        |FROM range(5)
+      """.stripMargin)
+      checkSparkAnswer(sql("SELECT m, count(*) FROM t_map_group GROUP BY m"))
+    }
+  }
+
   test("map_from_entries - fallback for binary type") {
     def fallbackReason(reason: String) = reason
     val table = "t2"