[SUPPOCaused by: java.lang.ClassCastException: class org.apache.spark.sql.types.StructType cannot be cast to class org.apache.spark.sql.types.MapType RT]

**_Tips before filing an issue_**

- Have you gone through our [FAQs](https://hudi.apache.org/learn/faq/)? yes

- Join the mailing list to engage in conversations and get faster support at dev-subscribe@hudi.apache.org.

- If you have triaged this as a bug, then file an [issue](https://issues.apache.org/jira/projects/HUDI/issues) directly.

**Describe the problem you faced**

Caused by: java.lang.ClassCastException: class org.apache.spark.sql.types.StructType cannot be cast to class org.apache.spark.sql.types.MapType

A clear and concise description of the problem.

**To Reproduce**

Steps to reproduce the behavior:

gcloud dataproc clusters create m0s0d56-test-2 \
--region us-central1 \
--project gcp-project-name \
--master-machine-type e2-standard-8 \
--num-workers 0 \
--worker-machine-type e2-standard-8 \
--no-address \
--max-idle 1800 \
--scopes cloud-platform \
--image 2.1.18-debian11 \
--properties spark:spark.jars.packages=org.apache.hudi:hudi-spark3.3-bundle_2.12:0.14.0 \

spark-shell --conf 'spark.serializer=org.apache.spark.serializer.KryoSerializer' --conf 'spark.sql.extensions=org.apache.spark.sql.hudi.HoodieSparkSessionExtension'

import org.apache.spark.sql.types._
import org.apache.spark.sql._

val schema = StructType(Seq(
| | StructField("payload", MapType(StringType,
| | StructType(Seq(
| | StructField("member0", StringType, true),
| | StructField("member1", MapType(StringType,
| | StructType(Seq(
| | StructField("member0", StringType, true),
| | StructField("member1", MapType(StringType, StringType, true), true)
| | )), true), true),
| | StructField("member2", ArrayType(MapType(StringType, StringType, true), true), true)
| | )), true), true)
| | ))

val data = Seq(Row(Map("key1" -> Row("value0", Map("subkey1" -> Row("subvalue0", Map("subsubkey1" -> "subsubvalue1"))), Array(Map("arraykey1" -> "arrayvalue1"))))))

val testDF = spark.createDataFrame(spark.sparkContext.parallelize(data), schema)

testDF.write.format("hudi").mode("overwrite").option("hoodie.table.name","hudi_test_poc").save("gs://test-bucket/notification_1")

**Expected behavior**

A clear and concise description of what you expected to happen.

**Environment Description**

* Hudi version : hudi-spark3.3-bundle_2.12:0.14.0

* Spark version : 3.3.2/3.1.3 same issue on both spark version

* Hive version :

* Hadoop version :

* Storage (HDFS/S3/GCS..) :

* Running on Docker? (yes/no) :

**Additional context**

Add any other context about the problem here.

**Stacktrace**

```Add the stacktrace of the error.```

Start by reproducing the nested MapType and StructType write with Hudi 0.14.0 on Spark 3.3.2 or 3.1.3 using the supplied Dataproc and spark-shell commands. No source file or test is named; investigate the resulting ClassCastException and consider the issue resolved when the shown DataFrame writes successfully without that exception.

Written by the indexing model from the issue text.