feat: support map_entries (#2059)

Author: comphead

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

@@ -123,6 +123,7 @@ object QueryPlanSerde extends Logging with CometExprShim {
     classOf[Murmur3Hash] -> CometMurmur3Hash,
     classOf[XxHash64] -> CometXxHash64,
     classOf[MapKeys] -> CometMapKeys,
+    classOf[MapEntries] -> CometMapEntries,
     classOf[MapValues] -> CometMapValues,
     classOf[MapFromArrays] -> CometMapFromArrays,
     classOf[GetMapValue] -> CometMapExtract,

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

@@ -37,6 +37,19 @@ object CometMapKeys extends CometExpressionSerde {
   }
 }
 
+object CometMapEntries extends CometExpressionSerde {
+
+  override def convert(
+      expr: Expression,
+      inputs: Seq[Attribute],
+      binding: Boolean): Option[ExprOuterClass.Expr] = {
+    val mk = expr.asInstanceOf[MapEntries]
+    val childExpr = exprToProtoInternal(mk.child, inputs, binding)
+    val mapEntriesScalarExpr = scalarFunctionExprToProto("map_entries", childExpr)
+    optExprWithInfo(mapEntriesScalarExpr, expr, expr.children: _*)
+  }
+}
+
 object CometMapValues extends CometExpressionSerde {
 
   override def convert(

spark/src/test/scala/org/apache/comet/parquet/ParquetReadSuite.scala

@@ -1970,6 +1970,7 @@ class ParquetReadV1Suite extends ParquetReadSuite with AdaptiveSparkPlanHelper {
                       "optional_map.key, " +
                       "optional_map.value, " +
                       "map_keys(complex_map), " +
+                      "map_entries(complex_map), " +
                       "map_values(complex_map) " +
                       "from complex_types"))
                 // leaf fields
@@ -1982,6 +1983,8 @@ class ParquetReadV1Suite extends ParquetReadSuite with AdaptiveSparkPlanHelper {
                       "optional_map.value, " +
                       "map_keys(complex_map)[0].key_field1, " +
                       "map_keys(complex_map)[0].key_field2, " +
+                      "map_entries(complex_map)[0].key, " +
+                      "map_entries(complex_map)[0].value, " +
                       "map_values(complex_map)[0].value_field1, " +
                       "map_values(complex_map)[0].value_field2 " +
                       "from complex_types"))