feat: add MAP type support for first level

Author: comphead

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

@@ -70,6 +70,8 @@ object QueryPlanSerde extends Logging with ShimQueryPlanSerde with CometExprShim
       s.fields.map(_.dataType).forall(supportedDataType(_, allowComplex))
     case a: ArrayType if allowComplex =>
       supportedDataType(a.elementType, allowComplex)
+    case m: MapType if allowComplex =>
+      supportedDataType(m.keyType, allowComplex) && supportedDataType(m.valueType, allowComplex)
     case dt =>
       emitWarning(s"unsupported Spark data type: $dt")
       false

spark/src/main/scala/org/apache/spark/sql/comet/CometNativeScanExec.scala

@@ -230,10 +230,10 @@ object CometNativeScanExec extends DataTypeSupport {
   }
 
   override def isAdditionallySupported(dt: DataType): Boolean = {
-    // TODO add map
     dt match {
       case s: StructType => s.fields.map(_.dataType).forall(isTypeSupported)
       case a: ArrayType => isTypeSupported(a.elementType)
+      case m: MapType => isTypeSupported(m.keyType) && isTypeSupported(m.valueType)
       case _ => false
     }
   }

spark/src/main/scala/org/apache/spark/sql/comet/CometScanExec.scala

@@ -480,10 +480,10 @@ object CometScanExec extends DataTypeSupport {
 
   override def isAdditionallySupported(dt: DataType): Boolean = {
     if (CometConf.COMET_NATIVE_SCAN_IMPL.get() == CometConf.SCAN_NATIVE_ICEBERG_COMPAT) {
-      // TODO add map
       dt match {
         case s: StructType => s.fields.map(_.dataType).forall(isTypeSupported)
         case a: ArrayType => isTypeSupported(a.elementType)
+        case m: MapType => isTypeSupported(m.keyType) && isTypeSupported(m.valueType)
         case _ => false
       }
     } else {

spark/src/test/scala/org/apache/comet/exec/CometNativeReaderSuite.scala

@@ -134,4 +134,13 @@ class CometNativeReaderSuite extends CometTestBase with AdaptiveSparkPlanHelper
         |""".stripMargin,
       "select c0 from tbl")
   }
+
+  test("native reader - read simple MAP fields") {
+    testSingleLineQuery(
+      """
+        |select map('a', 1) as c0 union all
+        |select map('b', 2)
+        |""".stripMargin,
+      "select c0 from tbl")
+  }
 }