feat: add CometMapSort serializer

Author: Shekharrajak

spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala

@@ -126,7 +126,8 @@ object QueryPlanSerde extends Logging with CometExprShim {
     classOf[MapKeys] -> CometMapKeys,
     classOf[MapEntries] -> CometMapEntries,
     classOf[MapValues] -> CometMapValues,
-    classOf[MapFromArrays] -> CometMapFromArrays)
+    classOf[MapFromArrays] -> CometMapFromArrays,
+    classOf[MapSort] -> CometMapSort)
 
   private val structExpressions: Map[Class[_ <: Expression], CometExpressionSerde[_]] = Map(
     classOf[CreateNamedStruct] -> CometCreateNamedStruct,

spark/src/main/scala/org/apache/comet/serde/maps.scala

@@ -20,8 +20,10 @@
 package org.apache.comet.serde
 
 import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.expressions.objects.RowOrdering
 import org.apache.spark.sql.types.{ArrayType, MapType}
 
+import org.apache.comet.CometSparkSessionExtensions.withInfo
 import org.apache.comet.serde.QueryPlanSerde.{exprToProtoInternal, optExprWithInfo, scalarFunctionExprToProto, scalarFunctionExprToProtoWithReturnType}
 
 object CometMapKeys extends CometExpressionSerde[MapKeys] {
@@ -89,3 +91,21 @@ object CometMapFromArrays extends CometExpressionSerde[MapFromArrays] {
     optExprWithInfo(mapFromArraysExpr, expr, expr.children: _*)
   }
 }
+
+object CometMapSort extends CometExpressionSerde[MapSort] {
+
+  override def convert(
+      expr: MapSort,
+      inputs: Seq[Attribute],
+      binding: Boolean): Option[ExprOuterClass.Expr] = {
+    val keyType = expr.base.dataType.asInstanceOf[MapType].keyType
+    if (!RowOrdering.isOrderable(keyType)) {
+      withInfo(expr, s"map_sort requires orderable key type, got: $keyType")
+      return None
+    }
+
+    val childExpr = exprToProtoInternal(expr.base, inputs, binding)
+    val mapSortScalarExpr = scalarFunctionExprToProto("map_sort", childExpr)
+    optExprWithInfo(mapSortScalarExpr, expr, expr.children: _*)
+  }
+}