[SPARK-60103][SQL] Keep XML MAP struct fields on convertField dispatch

Drop the direct convertObject MapType arm so empty, attribute-only, and
text-only MAP<STRING> elements stay null or malformed. Restrict convertMap
to StringType-family keys so format("xml") MAP<INT> does not ClassCast.

Author: srielau

docs/sql-migration-guide.md

@@ -29,7 +29,7 @@ license: |
 ## Upgrading from Spark SQL 4.3 to 4.4
 
 - Since Spark 4.4, when `spark.sql.preserveCharVarcharTypeInfo` is true and `spark.sql.charVarchar.standardSemantics.enabled` is false, ORC reads that apply a CHAR/VARCHAR schema over STRING storage return the stored values without ORC truncation, matching Parquet. Previously the ORC reader requested `char(n)`/`varchar(n)` and truncated STRING-stored values to `n`. Read-side length checks (`EXCEED_LIMIT_LENGTH`) apply only when `spark.sql.charVarchar.standardSemantics.enabled` is true.
-- Since Spark 4.4, when `spark.sql.charVarchar.standardSemantics.enabled` is true, XML element and attribute names used as `MAP<CHAR(n), _>` or `MAP<VARCHAR(n), _>` keys in `from_xml` and the XML datasource are length-checked without padding or trimming. A `CHAR(n)` key must already be exactly `n` characters, and a `VARCHAR(n)` key must already be at most `n` characters. Mismatched keys fail with `UNSUPPORTED_XML_CHAR_VARCHAR_MAP_KEY` (SQLSTATE `0A000`) and follow the XML parse mode (`PERMISSIVE` or `FAILFAST`). Exact repeated names last-win; `spark.sql.mapKeyDedupPolicy` is not applied. XML CHAR/VARCHAR values still use write-side pad and `EXCEED_LIMIT_LENGTH`. With the flag off, CHAR keys are still padded and VARCHAR overflow still uses `EXCEED_LIMIT_LENGTH`.
+- Since Spark 4.4, when `spark.sql.charVarchar.standardSemantics.enabled` is true, XML names used as `MAP<CHAR(n), _>` or `MAP<VARCHAR(n), _>` keys in `from_xml` and the XML datasource are length-checked without padding or trimming. The checked names are element names, `attributePrefix` plus attribute names (default prefix `_`), and the `valueTag` (default `_VALUE`) when mixed text is present. A `CHAR(n)` key must already be exactly `n` characters, and a `VARCHAR(n)` key must already be at most `n` characters. Mismatched keys fail with `UNSUPPORTED_XML_CHAR_VARCHAR_MAP_KEY` (SQLSTATE `0A000`) and follow the XML parse mode (`PERMISSIVE` or `FAILFAST`). Exact repeated names last-win; `spark.sql.mapKeyDedupPolicy` is not applied. XML CHAR/VARCHAR values use the same pad and `EXCEED_LIMIT_LENGTH` checks as other parsed text. Empty, attribute-only, and text-only map elements stay SQL NULL or a malformed record, matching 4.3; they do not become an empty map or a `valueTag` entry. In Spark 4.3, `MAP<CHAR(n), _>` and `MAP<VARCHAR(n), _>` XML maps were malformed records because only unbounded STRING keys were accepted. Padding of CHAR keys and VARCHAR `EXCEED_LIMIT_LENGTH` apply when the standard-semantics flag is false and `spark.sql.preserveCharVarcharTypeInfo` is true, so first-class CHAR/VARCHAR types still reach the parser.
 - Since Spark 4.4, the options maps passed to `from_csv`, `to_csv`, `schema_of_csv`, `from_json`, `to_json`, `schema_of_json`, `from_xml`, `to_xml`, and `schema_of_xml` must be foldable after replacing `RuntimeReplaceable` expressions. Previously, Spark evaluated non-foldable options during analysis, which allowed some constant expressions but could fail with an internal error or incorrectly evaluate row-dependent expressions. To allow deterministic and row-independent non-foldable options, set `spark.sql.legacy.allowNonFoldableOptions` to `true`. Row-dependent, unevaluable, and nondeterministic options are always rejected.
 - Since Spark 4.4, when an already-analyzed Data Source V2 query is refreshed after a compatible schema change, connectors can return more data columns from the current table schema in `Scan.readSchema()` than requested by `SupportsPushDownRequiredColumns.pruneColumns`. Previously, this partial pruning could fail planning because the scan reported columns absent from the analyzed relation output.
 - Since Spark 4.4, when an already-analyzed Data Source V2 query is refreshed, if rebinding an expanded struct used as a map key causes distinct current keys to collide under the analyzed schema, the query fails with `DUPLICATED_MAP_KEY` by default instead of returning duplicate keys. With `spark.sql.mapKeyDedupPolicy=LAST_WIN`, the last value is kept.

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/xml/StaxXmlParser.scala

@@ -385,7 +385,10 @@ class StaxXmlParser(
         startElementName: String,
         attributes: Array[Attribute]): Any = dt match {
       case st: StructType => convertObject(parser, st)
-      case MapType(kt, vt, _) => convertMap(parser, vt, attributes, kt)
+      // CHAR/VARCHAR extend StringType. Non-string keys (for example MAP<INT, _> through
+      // format("xml").load()) must not reach convertMap: applyTextParseSemantics would keep
+      // UTF8String keys and fail later with ClassCastException.
+      case MapType(kt: StringType, vt, _) => convertMap(parser, vt, attributes, kt)
       case ArrayType(st, _) => convertField(parser, st, startElementName)
       case VariantType =>
         StaxXmlParser.convertVariant(parser, attributes, options)
@@ -447,14 +450,18 @@ class StaxXmlParser(
   /**
    * Parse an object as a Map.
    *
-   * XML names used as CHAR/VARCHAR keys are length-checked without rewriting:
-   * CHAR keys must already be exactly n characters, and VARCHAR keys must already be
-   * at most n characters. Padding, trimming, and mapKeyDedupPolicy are not applied.
+   * When `spark.sql.charVarchar.standardSemantics.enabled` is true, XML names used as
+   * CHAR/VARCHAR keys are length-checked without rewriting: CHAR keys must already be
+   * exactly n characters, and VARCHAR keys must already be at most n characters.
+   * Padding, trimming, and mapKeyDedupPolicy are not applied. With the flag off,
+   * CHAR keys are padded and VARCHAR overflow uses EXCEED_LIMIT_LENGTH when
+   * first-class CHAR/VARCHAR types reach the parser.
    * Repeated names last-win on binary equality (collation is not consulted), matching
    * ordinary MAP<STRING, ...> XML maps.
    *
-   * Example: `from_xml('<ROW><m><ab>1</ab></m></ROW>', 'm MAP<CHAR(2), INT>')`
-   * keeps key `ab`; `MAP<CHAR(4), INT>` raises `UNSUPPORTED_XML_CHAR_VARCHAR_MAP_KEY`.
+   * Example (flag on): `from_xml('<ROW><m><ab>1</ab></m></ROW>',
+   * 'm MAP<CHAR(2), INT>')` keeps key `ab`; `MAP<CHAR(4), INT>` raises
+   * `UNSUPPORTED_XML_CHAR_VARCHAR_MAP_KEY`.
    *
    * This method owns element bounding so a key-check failure still consumes the current
    * map element (ARRAY<MAP<...>> and nested maps included).
@@ -647,9 +654,6 @@ class StaxXmlParser(
               case st: StructType =>
                 row(index) = convertNestedStruct(parser, st, field, attributes)
 
-              case mt: MapType =>
-                row(index) = convertMap(parser, mt.valueType, attributes, mt.keyType)
-
               case ArrayType(dt: DataType, _) =>
                 val values = Option(row(index))
                   .map(_.asInstanceOf[ArrayBuffer[Any]])

sql/core/src/test/scala/org/apache/spark/sql/CharVarcharTestSuite.scala

@@ -2978,13 +2978,14 @@ class BasicCharVarcharTestSuite extends SharedSparkSession {
         key = "a",
         dataTypeSql = "CHAR(2)")
 
-      // Default valueTag is "_VALUE" (6 characters), so CHAR(2) rejects it.
+      // Default valueTag is "_VALUE" (6 characters). Text-only elements are malformed
+      // records (convertField), so use mixed content to reach convertMap.
       checkAnswer(
-        sql("SELECT from_xml('<ROW><m>xy</m></ROW>', 'm MAP<CHAR(2), STRING>')"),
+        sql("SELECT from_xml('<ROW><m><ab>1</ab>xy</m></ROW>', 'm MAP<CHAR(2), STRING>')"),
         Row(Row(null)))
       assertUnsupportedXmlMapKey(
         """SELECT from_xml(
-          |  '<ROW><m>xy</m></ROW>',
+          |  '<ROW><m><ab>1</ab>xy</m></ROW>',
           |  'm MAP<CHAR(2), STRING>',
           |  map('mode', 'FAILFAST'))""".stripMargin,
         key = "_VALUE",
@@ -3049,20 +3050,38 @@ class BasicCharVarcharTestSuite extends SharedSparkSession {
         dataTypeSql = "VARCHAR(2)")
 
       // Default attributePrefix "_" is part of the key: local name ab is key _ab.
+      // Attribute-only elements are SQL NULL (convertField EndElement), so include a child.
       checkAnswer(
-        sql("SELECT from_xml('<ROW><m ab=\"1\"></m></ROW>', 'm MAP<CHAR(2), INT>')"),
+        sql("SELECT from_xml('<ROW><m ab=\"1\"><xy>2</xy></m></ROW>', 'm MAP<CHAR(2), INT>')"),
         Row(Row(null)))
       checkAnswer(
-        sql("SELECT from_xml('<ROW><m ab=\"1\"></m></ROW>', 'm MAP<CHAR(3), INT>')"),
-        Row(Row(Map("_ab" -> 1))))
+        sql("SELECT from_xml('<ROW><m ab=\"1\"><xyz>2</xyz></m></ROW>', 'm MAP<CHAR(3), INT>')"),
+        Row(Row(Map("_ab" -> 1, "xyz" -> 2))))
       assertUnsupportedXmlMapKey(
         """SELECT from_xml(
-          |  '<ROW><m ab="1"></m></ROW>',
+          |  '<ROW><m ab="1"><xy>2</xy></m></ROW>',
           |  'm MAP<CHAR(2), INT>',
           |  map('mode', 'FAILFAST'))""".stripMargin,
         key = "_ab",
         dataTypeSql = "CHAR(2)")
 
+      // MAP<STRING> empty, attribute-only, and text-only elements stay SQL NULL, matching
+      // convertField on master. ARRAY<MAP> empty elements are null entries, not {}.
+      checkAnswer(
+        sql("SELECT from_xml('<ROW><m></m></ROW>', 'm MAP<STRING, INT>')"),
+        Row(Row(null)))
+      checkAnswer(
+        sql("SELECT from_xml('<ROW><m a=\"1\"></m></ROW>', 'm MAP<STRING, INT>')"),
+        Row(Row(null)))
+      checkAnswer(
+        sql("SELECT from_xml('<ROW><m>xy</m></ROW>', 'm MAP<STRING, STRING>')"),
+        Row(Row(null)))
+      checkAnswer(
+        sql("""SELECT from_xml(
+          |  '<ROW><m></m><tail>9</tail></ROW>',
+          |  'm ARRAY<MAP<STRING, INT>>, tail INT')""".stripMargin),
+        Row(Row(Seq(null), 9)))
+
       withTempPath { path =>
         Seq(
           """<ROWS><ROW><m><ab>1</ab></m></ROW><ROW><m><a>1</a></m></ROW></ROWS>"""
@@ -3099,6 +3118,31 @@ class BasicCharVarcharTestSuite extends SharedSparkSession {
         checkAnswer(arrayPermissive, Seq(Row(null), Row(Seq(Map("ab" -> 2)))))
       }
 
+      // format("xml").load() skips INVALID_XML_MAP_KEY_TYPE. Non-string keys must stay a
+      // malformed record (MatchError), not a mistyped map that fails with ClassCastException.
+      withTempPath { path =>
+        Seq("""<ROWS><ROW><m><a>x</a></m></ROW></ROWS>""")
+          .toDS().write.text(path.getCanonicalPath)
+        val intKeySchema = "m MAP<INT, STRING>"
+        val permissiveInt = spark.read
+          .format("xml")
+          .option("rowTag", "ROW")
+          .schema(intKeySchema)
+          .load(path.getCanonicalPath)
+        checkAnswer(permissiveInt, Seq(Row(null)))
+        val failFastInt = spark.read
+          .format("xml")
+          .option("rowTag", "ROW")
+          .option("mode", "FAILFAST")
+          .schema(intKeySchema)
+          .load(path.getCanonicalPath)
+        val error = intercept[Exception] { failFastInt.collect() }
+        val causes = Iterator.iterate[Throwable](error)(_.getCause).takeWhile(_ != null)
+        assert(
+          !causes.exists(_.isInstanceOf[ClassCastException]),
+          s"MAP<INT, STRING> XML read must not ClassCastException: $error")
+      }
+
       withSQLConf(
           SQLConf.MAP_KEY_DEDUP_POLICY.key -> SQLConf.MapKeyDedupPolicy.EXCEPTION.toString) {
         checkAnswer(