feat: add CometMapSort serializer

Author: Shekharrajak

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

@@ -131,7 +131,8 @@ object QueryPlanSerde extends Logging with CometExprShim {
     classOf[MapValues] -> CometMapValues,
     classOf[MapFromArrays] -> CometMapFromArrays,
     classOf[MapContainsKey] -> CometMapContainsKey,
-    classOf[MapFromEntries] -> CometMapFromEntries)
+    classOf[MapFromEntries] -> CometMapFromEntries,
+    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,9 +20,11 @@
 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._
 
-import org.apache.comet.serde.QueryPlanSerde.{createBinaryExpr, exprToProtoInternal, optExprWithInfo, scalarFunctionExprToProto}
+import org.apache.comet.CometSparkSessionExtensions.withInfo
+import org.apache.comet.serde.QueryPlanSerde.{createBinaryExpr, exprToProtoInternal, optExprWithInfo, scalarFunctionExprToProto, scalarFunctionExprToProtoWithReturnType}
 
 object CometMapKeys extends CometExpressionSerde[MapKeys] {
 
@@ -156,3 +158,21 @@ object CometMapFromEntries extends CometScalarFunction[MapFromEntries]("map_from
     Compatible(None)
   }
 }
+
+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: _*)
+  }
+}