From 02b478ea07c8721a3186e03ead00b68e3452bddd Mon Sep 17 00:00:00 2001 From: srielau Date: Sat, 10 Oct 2026 04:57:43 +0000 Subject: [PATCH 1/4] [SPARK-60103][SQL] Length-check XML CHAR/VARCHAR map keys without pad or trim When standard CHAR/VARCHAR semantics are enabled, XML element and attribute names used as MAP keys are length-checked in place: CHAR keys must already be exactly n characters, and VARCHAR keys at most n. Failures use UNSUPPORTED_XML_CHAR_VARCHAR_MAP_KEY and follow PERMISSIVE/FAILFAST. convertMap drains the current element so ARRAY> key mismatches do not desync the parser. --- .../resources/error/error-conditions.json | 7 + docs/sql-migration-guide.md | 1 + .../catalyst/util/DuplicateMapKeyUtils.scala | 49 --- .../sql/catalyst/xml/StaxXmlParser.scala | 164 ++++---- .../sql/errors/QueryExecutionErrors.scala | 14 + .../apache/spark/sql/internal/SQLConf.scala | 8 +- .../spark/sql/CharVarcharTestSuite.scala | 379 +++++++++++------- 7 files changed, 330 insertions(+), 292 deletions(-) diff --git a/common/utils/src/main/resources/error/error-conditions.json b/common/utils/src/main/resources/error/error-conditions.json index 94d6b897b18c8..7d17404b29a43 100644 --- a/common/utils/src/main/resources/error/error-conditions.json +++ b/common/utils/src/main/resources/error/error-conditions.json @@ -9896,6 +9896,13 @@ ], "sqlState" : "0A000" }, + "UNSUPPORTED_XML_CHAR_VARCHAR_MAP_KEY" : { + "message" : [ + "XML map keys of CHAR or VARCHAR cannot be padded or trimmed.", + "The key is not valid for type ." + ], + "sqlState" : "0A000" + }, "UNTYPED_SCALA_UDF" : { "message" : [ "You're using untyped Scala UDF, which does not have the input type information. Spark may blindly pass null to the Scala closure with primitive-type argument, and the closure will see the default value of the Java type for the null argument, e.g. `udf((x: Int) => x, IntegerType)`, the result is 0 for null input. To get rid of this error, you could:", diff --git a/docs/sql-migration-guide.md b/docs/sql-migration-guide.md index f9c7bf4771227..be98ca88c26de 100644 --- a/docs/sql-migration-guide.md +++ b/docs/sql-migration-guide.md @@ -29,6 +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` or `MAP` 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, 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. diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/DuplicateMapKeyUtils.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/DuplicateMapKeyUtils.scala index fd179bc73d05c..5b9f3bf453511 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/DuplicateMapKeyUtils.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/DuplicateMapKeyUtils.scala @@ -17,13 +17,7 @@ package org.apache.spark.sql.catalyst.util -import scala.collection.mutable - import org.apache.spark.SparkRuntimeException -import org.apache.spark.sql.errors.QueryExecutionErrors -import org.apache.spark.sql.internal.SQLConf -import org.apache.spark.sql.types.{DataType, StringType} -import org.apache.spark.unsafe.types.UTF8String import org.apache.spark.util.SparkErrorUtils private[sql] object DuplicateMapKeyUtils { @@ -36,47 +30,4 @@ private[sql] object DuplicateMapKeyUtils { } def unapply(exception: Throwable): Option[SparkRuntimeException] = cause(exception) - - /** - * Builds an XML map with a constrained CHAR/VARCHAR key type from a raw-key last-wins - * accumulator. Failed values still occupy a slot so normalized collisions are visible. - * - * CHAR/VARCHAR keys: repeated raw XML keys keep the last value, then - * `spark.sql.mapKeyDedupPolicy` applies to normalized keys. - * - * Example: parsing `a` and `a ` as CHAR(2) keys raises - * DUPLICATED_MAP_KEY under EXCEPTION and keeps `a ` -> 2 under LAST_WIN. - */ - def buildConstrainedMap( - lastEntries: mutable.LinkedHashMap[UTF8String, (UTF8String, Option[Any])], - keyType: DataType, - valueType: DataType): MapData = { - if (SQLConf.get.getConf(SQLConf.MAP_KEY_DEDUP_POLICY) == - SQLConf.MapKeyDedupPolicy.EXCEPTION) { - val distinctKeys = keyType match { - case stringType: StringType if stringType.supportsBinaryEquality => - new java.util.HashSet[Any]() - case _ => - new java.util.TreeSet[Any](TypeUtils.getInterpretedOrdering(keyType)) - } - val keys = mutable.ArrayBuffer.empty[Any] - val values = mutable.ArrayBuffer.empty[Any] - lastEntries.valuesIterator.foreach { case (key, value) => - if (!distinctKeys.add(key)) { - throw QueryExecutionErrors.duplicateMapKeyFoundError(key) - } - value.foreach { v => - keys += key - values += v - } - } - ArrayBasedMapData(keys.toArray, values.toArray) - } else { - val builder = new ArrayBasedMapBuilder(keyType, valueType) - lastEntries.valuesIterator.foreach { case (normalizedKey, value) => - value.foreach(builder.put(normalizedKey, _)) - } - builder.build() - } - } } diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/xml/StaxXmlParser.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/xml/StaxXmlParser.scala index f0923d4054366..50f5404b56dc9 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/xml/StaxXmlParser.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/xml/StaxXmlParser.scala @@ -26,7 +26,6 @@ import javax.xml.stream.events._ import javax.xml.transform.stream.StreamSource import javax.xml.validation.Schema -import scala.collection.mutable import scala.collection.mutable.ArrayBuffer import scala.jdk.CollectionConverters._ import scala.util.Try @@ -86,6 +85,9 @@ class StaxXmlParser( private val caseSensitive = SQLConf.get.caseSensitiveAnalysis + // CHAR/VARCHAR XML map keys are length-checked without pad or trim under this flag. + private val charVarcharStandardSemantics = SQLConf.get.charVarcharStandardSemantics + /** * Limits a view of an event stream to one element whose start event has already been consumed. * Closing or draining this view consumes the matching end event without closing the underlying @@ -383,9 +385,7 @@ class StaxXmlParser( startElementName: String, attributes: Array[Attribute]): Any = dt match { case st: StructType => convertObject(parser, st) - case MapType(StringType, vt, _) => convertMap(parser, vt, attributes) - case MapType(kt @ (_: CharType | _: VarcharType), vt, _) => - convertConstrainedMap(parser, kt, vt, attributes) + case MapType(kt, vt, _) => convertMap(parser, vt, attributes, kt) case ArrayType(st, _) => convertField(parser, st, startElementName) case VariantType => StaxXmlParser.convertVariant(parser, attributes, options) @@ -445,109 +445,76 @@ class StaxXmlParser( } /** - * Parse an object as map. + * 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. + * Repeated names last-win on binary equality (collation is not consulted), matching + * ordinary MAP XML maps. + * + * Example: `from_xml('1', 'm MAP')` + * keeps key `ab`; `MAP` raises `UNSUPPORTED_XML_CHAR_VARCHAR_MAP_KEY`. + * + * This method owns element bounding so a key-check failure still consumes the current + * map element (ARRAY> and nested maps included). */ private def convertMap( parser: XMLEventReader, valueType: DataType, - attributes: Array[Attribute]): MapData = { - val kvPairs = ArrayBuffer.empty[(UTF8String, Any)] - attributes.foreach { attr => - kvPairs += (UTF8String.fromString(options.attributePrefix + attr.getName.getLocalPart) - -> convertTo(attr.getValue, valueType)) - } - var shouldStop = false - while (!shouldStop) { - parser.nextEvent match { - case e: StartElement => - val key = StaxXmlParserUtils.getName(e.asStartElement.getName, options) - kvPairs += - (UTF8String.fromString(key) -> convertField(parser, valueType, key)) - case c: Characters if !c.isWhiteSpace => - // Create a value tag field for it - kvPairs += - // TODO: We don't support array value tags in maps yet. - (UTF8String.fromString(options.valueTag) -> convertTo(c.getData, valueType)) - case _: EndElement | _: EndDocument => - shouldStop = true - case _ => // do nothing + attributes: Array[Attribute], + keyType: DataType): MapData = { + val bounded = new ElementBoundedEventReader(parser) + try { + val kvPairs = ArrayBuffer.empty[(UTF8String, Any)] + attributes.foreach { attr => + val key = convertXmlMapKey( + options.attributePrefix + attr.getName.getLocalPart, keyType) + kvPairs += (key -> convertTo(attr.getValue, valueType)) + } + var shouldStop = false + while (!shouldStop) { + bounded.nextEvent match { + case e: StartElement => + val rawName = StaxXmlParserUtils.getName(e.asStartElement.getName, options) + val key = convertXmlMapKey(rawName, keyType) + kvPairs += (key -> convertField(bounded, valueType, rawName)) + case c: Characters if !c.isWhiteSpace => + // Create a value tag field for it + kvPairs += + // TODO: We don't support array value tags in maps yet. + (convertXmlMapKey(options.valueTag, keyType) -> convertTo(c.getData, valueType)) + case _: EndElement | _: EndDocument => + shouldStop = true + case _ => // do nothing + } } + ArrayBasedMapData(kvPairs.toMap) + } finally { + bounded.drain() } - ArrayBasedMapData(kvPairs.toMap) } - private def convertConstrainedMap( - parser: XMLEventReader, - keyType: DataType, - valueType: DataType, - attributes: Array[Attribute]): MapData = { - val lastEntries = - mutable.LinkedHashMap.empty[UTF8String, (UTF8String, Option[Any])] - var badMapException: Option[Throwable] = None - def mapKey(raw: UTF8String): UTF8String = { - CharVarcharUtils.applyTextParseSemantics(raw, keyType) - } - def appendPair(rawKey: String, value: Option[Any]): Unit = { - try { - val rawKeyUtf8 = UTF8String.fromString(rawKey) - lastEntries.remove(rawKeyUtf8) - lastEntries.update(rawKeyUtf8, (mapKey(rawKeyUtf8), value)) - } catch { - case NonFatal(e) => badMapException = badMapException.orElse(Some(e)) - } - } - attributes.foreach { attr => - val value = try { - Some(convertTo(attr.getValue, valueType)) - } catch { - case e: SparkUpgradeException => throw e - case NonFatal(e) => - badMapException = badMapException.orElse(Some(e)) - None - } - appendPair(options.attributePrefix + attr.getName.getLocalPart, value) - } - var shouldStop = false - while (!shouldStop) { - parser.nextEvent match { - case e: StartElement => - val rawKey = StaxXmlParserUtils.getName(e.asStartElement.getName, options) - val entryParser = new ElementBoundedEventReader(parser) - val value = { - try { - Some(convertField(entryParser, valueType, rawKey)) - } catch { - case e: SparkUpgradeException => throw e - case DuplicateMapKeyUtils(e) => throw e - case NonFatal(e) => - badMapException = badMapException.orElse(Some(e)) - None - } finally { - entryParser.drain() - } - } - appendPair(rawKey, value) - case c: Characters if !c.isWhiteSpace => - // Create a value tag field for it - // TODO: We don't support array value tags in maps yet. - val value = try { - Some(convertTo(c.getData, valueType)) - } catch { - case e: SparkUpgradeException => throw e - case NonFatal(e) => - badMapException = badMapException.orElse(Some(e)) - None - } - appendPair(options.valueTag, value) - case _: EndElement | _: EndDocument => - shouldStop = true - case _ => // do nothing + /** + * Length-check a CHAR/VARCHAR XML name used as a map key. Unlike value assignment, + * this does not pad or trim. STRING keys and the flag-off path still use + * [[CharVarcharUtils.applyTextParseSemantics]]. + */ + private def convertXmlMapKey(rawName: String, keyType: DataType): UTF8String = { + val key = UTF8String.fromString(rawName) + if (charVarcharStandardSemantics) { + keyType match { + case c: CharType if key.numChars() != c.length => + throw QueryExecutionErrors.unsupportedXmlCharVarcharMapKey(key, c) + case _: CharType => key + case v: VarcharType if key.numChars() > v.length => + throw QueryExecutionErrors.unsupportedXmlCharVarcharMapKey(key, v) + case _: VarcharType => key + case _ => CharVarcharUtils.applyTextParseSemantics(key, keyType) } + } else { + CharVarcharUtils.applyTextParseSemantics(key, keyType) } - val mapData = DuplicateMapKeyUtils.buildConstrainedMap( - lastEntries, keyType, valueType) - badMapException.foreach(throw _) - mapData } /** @@ -680,6 +647,9 @@ 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]]) diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryExecutionErrors.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryExecutionErrors.scala index b8bfd00cb2c84..090e73c264cc6 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryExecutionErrors.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryExecutionErrors.scala @@ -2739,6 +2739,20 @@ private[sql] object QueryExecutionErrors extends QueryErrorsBase with ExecutionE ) } + /** + * Restricting CHAR/VARCHAR XML map keys is a parser limitation (SQLSTATE 0A000), + * but this is a SparkRuntimeException so XML parse modes can handle it: PERMISSIVE + * wraps a bad record and FAILFAST surfaces the failure. + */ + def unsupportedXmlCharVarcharMapKey( + key: UTF8String, dataType: DataType): SparkRuntimeException = { + new SparkRuntimeException( + errorClass = "UNSUPPORTED_XML_CHAR_VARCHAR_MAP_KEY", + messageParameters = Map( + "key" -> toSQLValue(key, StringType), + "dataType" -> toSQLType(dataType))) + } + def timestampAddOverflowError(micros: Long, amount: Long, unit: String): ArithmeticException = { new SparkArithmeticException( errorClass = "DATETIME_OVERFLOW", diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala index 3a72fd622a2af..a6ff1e548f7bb 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala @@ -6953,12 +6953,8 @@ object SQLConf { val MAP_KEY_DEDUP_POLICY = buildConf("spark.sql.mapKeyDedupPolicy") .doc("The policy to deduplicate map keys in built-in functions: CreateMap, MapFromArrays, " + - "MapFromEntries, StringToMap, MapConcat and TransformKeys. The policy also applies in " + - "schema-driven XML parsing, including from_xml and the XML data source, when CHAR/VARCHAR " + - "keys normalize to the same value. " + - "EXCEPTION fails the query when duplicate keys are detected. LAST_WIN makes the last " + - "inserted key take precedence. Repeated raw XML map keys retain their historical " + - "last-wins behavior. Ordinary STRING keys retain each parser's historical behavior.") + "MapFromEntries, StringToMap, MapConcat and TransformKeys. EXCEPTION fails the query " + + "when duplicate keys are detected. LAST_WIN makes the last inserted key take precedence.") .version("3.0.0") .enumConf(MapKeyDedupPolicy) .createWithDefault(MapKeyDedupPolicy.EXCEPTION) diff --git a/sql/core/src/test/scala/org/apache/spark/sql/CharVarcharTestSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/CharVarcharTestSuite.scala index c203fa1d8c344..8d23ce44fdf22 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/CharVarcharTestSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/CharVarcharTestSuite.scala @@ -1044,6 +1044,28 @@ class BasicCharVarcharTestSuite extends SharedSparkSession { parameters = Map("limit" -> expectedLimit)) } + private def assertUnsupportedXmlMapKey( + query: String, key: String, dataTypeSql: String): Unit = { + assertUnsupportedXmlMapKeyError(sql(query).collect(), key, dataTypeSql) + } + + private def assertUnsupportedXmlMapKeyError( + body: => Any, key: String, dataTypeSql: String): Unit = { + val error = intercept[Exception] { body } + val cause = Iterator.iterate[Throwable](error)(_.getCause) + .takeWhile(_ != null) + .collectFirst { + case e: SparkRuntimeException + if e.getCondition == "UNSUPPORTED_XML_CHAR_VARCHAR_MAP_KEY" => e + } + .getOrElse(fail("expected UNSUPPORTED_XML_CHAR_VARCHAR_MAP_KEY cause", error)) + checkError( + exception = cause, + condition = "UNSUPPORTED_XML_CHAR_VARCHAR_MAP_KEY", + sqlState = "0A000", + parameters = Map("key" -> s"'$key'", "dataType" -> s""""$dataTypeSql"""")) + } + private def assertDuplicateMapKey(query: String, expectedKey: String = "a "): Unit = { assertDuplicateMapKeyError(sql(query).collect(), expectedKey) } @@ -2795,9 +2817,12 @@ class BasicCharVarcharTestSuite extends SharedSparkSession { s"SELECT from_xml('abcdef', 'a $dataType', " + "map('mode', 'FAILFAST'))") } + checkAnswer( + sql("SELECT from_xml('1', 'm MAP')"), + Row(Row(Map("ab" -> 1)))) checkAnswer( sql("SELECT from_xml('1', 'm MAP')"), - Row(Row(Map("ab " -> 1)))) + Row(Row(null))) checkAnswer( sql( """SELECT from_xml( @@ -2837,7 +2862,7 @@ class BasicCharVarcharTestSuite extends SharedSparkSession { | 'xs_any ARRAY', | map('wildcardColName', 'xs_any', 'mode', 'FAILFAST'))""".stripMargin) withTempPath { path => - Seq("1").toDS().write.text(path.getCanonicalPath) + Seq("1").toDS().write.text(path.getCanonicalPath) val xmlDataFrame = spark.read .option("rowTag", "ROW") .schema("m MAP") @@ -2845,7 +2870,7 @@ class BasicCharVarcharTestSuite extends SharedSparkSession { assert( xmlDataFrame.schema("m").dataType === MapType(CharType(2), IntegerType, valueContainsNull = true)) - checkAnswer(xmlDataFrame, Row(Map("a " -> 1))) + checkAnswer(xmlDataFrame, Row(Map("ab" -> 1))) } withTempPath { path => Seq("""{"c":"ab"}""").toDS().write.text(path.getCanonicalPath) @@ -2896,155 +2921,241 @@ class BasicCharVarcharTestSuite extends SharedSparkSession { } } - test("SPARK-59274: normalized XML CHAR map keys honor the dedup policy") { + test("SPARK-60103: XML CHAR/VARCHAR map keys are not padded or trimmed") { withSQLConf(SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "true") { - val xmlQuery = - """SELECT from_xml( - | '19', - | 'm MAP', - | map('valueTag', 'a ')).m""".stripMargin - val varcharXmlQuery = - """SELECT from_xml( - | '12', - | 'm MAP', - | map('valueTag', 'ab ')).m""".stripMargin - val exactCharXmlQuery = + checkAnswer( + sql("SELECT from_xml('1', 'm MAP').m"), + Row(Map("ab" -> 1))) + checkAnswer( + sql("SELECT from_xml('1', 'm MAP').m"), + Row(Map("ab" -> 1))) + checkAnswer( + sql("""SELECT from_xml( + | '12', + | 'm MAP').m""".stripMargin), + Row(Map("ab" -> 1, "AB" -> 2))) + checkAnswer( + sql("""SELECT from_xml( + | '12', + | 'm MAP').m""".stripMargin), + Row(Map("ab" -> 2))) + checkAnswer( + sql("""SELECT from_xml( + | '1', + | 'm MAP>').m""".stripMargin), + Row(Map("outer" -> Map("xy" -> 1)))) + + // A key that fails length-check still lets later struct fields parse. + checkAnswer( + sql("""SELECT from_xml( + | '12', + | 'm MAP, tail INT')""".stripMargin), + Row(Row(null, 2))) + checkAnswer( + sql("""SELECT from_xml( + | '12', + | 'm MAP>, tail INT')""".stripMargin), + Row(Row(null, 2))) + assertUnsupportedXmlMapKey( """SELECT from_xml( - | '12', - | 'm MAP').m""".stripMargin - val exactVarcharXmlQuery = + | '1', + | 'm MAP>', + | map('mode', 'FAILFAST'))""".stripMargin, + key = "z", + dataTypeSql = "CHAR(2)") + + // ARRAY> uses convertField's MapType arm; a short key must still keep tail. + checkAnswer( + sql("""SELECT from_xml( + | '129', + | 'm ARRAY>, tail INT')""".stripMargin), + Row(Row(Seq(Map("ab" -> 2)), 9))) + assertUnsupportedXmlMapKey( """SELECT from_xml( - | '12', - | 'm MAP').m""".stripMargin - val interleavedCharXmlQuery = - """SELECT map_entries(from_xml( - | '123', - | 'm MAP', - | map('valueTag', 'a ')).m)""".stripMargin - val badXmlKeyThenSiblingQuery = + | '12', + | 'm ARRAY>', + | map('mode', 'FAILFAST'))""".stripMargin, + key = "a", + dataTypeSql = "CHAR(2)") + + // Default valueTag is "_VALUE" (6 characters), so CHAR(2) rejects it. + checkAnswer( + sql("SELECT from_xml('xy', 'm MAP')"), + Row(Row(null))) + assertUnsupportedXmlMapKey( """SELECT from_xml( - | '12', - | 'm MAP, tail INT').tail""".stripMargin - val badXmlKeyBeforeDuplicateQuery = + | 'xy', + | 'm MAP', + | map('mode', 'FAILFAST'))""".stripMargin, + key = "_VALUE", + dataTypeSql = "CHAR(2)") + + // Too-short CHAR, CHAR overflow, and VARCHAR overflow are not EXCEED_LIMIT_LENGTH. + checkAnswer( + sql("SELECT from_xml('1', 'm MAP')"), + Row(Row(null))) + checkAnswer( + sql("SELECT from_xml('1', 'm MAP')"), + Row(Row(null))) + checkAnswer( + sql("SELECT from_xml('1', 'm MAP')"), + Row(Row(null))) + assertUnsupportedXmlMapKey( """SELECT from_xml( - | '012', + | '1', | 'm MAP', - | map('valueTag', 'a ')).m""".stripMargin - val malformedXmlValueBeforeDuplicateQuery = + | map('mode', 'FAILFAST'))""".stripMargin, + key = "a", + dataTypeSql = "CHAR(2)") + assertUnsupportedXmlMapKey( """SELECT from_xml( - | 'bad2', + | '1', | 'm MAP', - | map('valueTag', 'a ')).m""".stripMargin - val ignoreCorruptXmlQuery = + | map('mode', 'FAILFAST'))""".stripMargin, + key = "abc", + dataTypeSql = "CHAR(2)") + assertUnsupportedXmlMapKey( """SELECT from_xml( - | '12', + | '1', + | 'm MAP', + | map('mode', 'FAILFAST'))""".stripMargin, + key = "abcd", + dataTypeSql = "VARCHAR(3)") + + checkAnswer( + sql("""SELECT from_xml( + | '2', | 'm MAP', - | map('valueTag', 'a ', 'ignoreCorruptFiles', 'true')).m""".stripMargin - val attributePaddingQuery = - """SELECT from_xml( - | '2', + | map('attributePrefix', ''))""".stripMargin), + Row(Row(null))) + checkAnswer( + sql("""SELECT from_xml( + | '2', | 'm MAP', - | map('attributePrefix', '')).m""".stripMargin - val attributeOverflowQuery = - """SELECT from_xml( - | '2', + | map('attributePrefix', ''))""".stripMargin), + Row(Row(Map("ab" -> 1, "cd" -> 2)))) + checkAnswer( + sql("""SELECT from_xml( + | '2', | 'm MAP', - | map('attributePrefix', '')).m""".stripMargin - val attributeOverflowFailfastQuery = + | map('attributePrefix', ''))""".stripMargin), + Row(Row(null))) + assertUnsupportedXmlMapKey( """SELECT from_xml( - | '2', + | '2', | 'm MAP', - | map('attributePrefix', '', 'mode', 'FAILFAST')).m""".stripMargin - val attributeCollisionQuery = + | map('attributePrefix', '', 'mode', 'FAILFAST'))""".stripMargin, + key = "abc", + dataTypeSql = "VARCHAR(2)") + + // Default attributePrefix "_" is part of the key: local name ab is key _ab. + checkAnswer( + sql("SELECT from_xml('', 'm MAP')"), + Row(Row(null))) + checkAnswer( + sql("SELECT from_xml('', 'm MAP')"), + Row(Row(Map("_ab" -> 1)))) + assertUnsupportedXmlMapKey( """SELECT from_xml( - | '02', + | '', | 'm MAP', - | map('attributePrefix', '', 'valueTag', 'a ')).m""".stripMargin - - assertDuplicateMapKey(xmlQuery) - assertDuplicateMapKey(varcharXmlQuery, expectedKey = "ab") - checkAnswer(sql(exactCharXmlQuery), Row(Map("a " -> 2))) - checkAnswer(sql(exactVarcharXmlQuery), Row(Map("ab" -> 2))) - assertDuplicateMapKey(interleavedCharXmlQuery) - assertDuplicateMapKey(badXmlKeyBeforeDuplicateQuery) - assertDuplicateMapKey(malformedXmlValueBeforeDuplicateQuery) - assertDuplicateMapKey(ignoreCorruptXmlQuery) - checkAnswer(sql(badXmlKeyThenSiblingQuery), Row(2)) - checkAnswer(sql(attributePaddingQuery), Row(Map("a " -> 1, "b " -> 2))) - checkAnswer(sql(attributeOverflowQuery), Row(null)) - assertParseExceedLimit(attributeOverflowFailfastQuery, expectedLimit = "2") - assertDuplicateMapKey(attributeCollisionQuery) + | map('mode', 'FAILFAST'))""".stripMargin, + key = "_ab", + dataTypeSql = "CHAR(2)") withTempPath { path => - Seq("12").toDS().write.text(path.getCanonicalPath) - def readXmlMap(): DataFrame = spark.read + Seq( + """11""" + ).toDS().write.text(path.getCanonicalPath) + val schema = "m MAP" + val permissive = spark.read .option("rowTag", "ROW") - .option("valueTag", "a ") - .schema("m MAP") + .schema(schema) .xml(path.getCanonicalPath) + checkAnswer(permissive, Seq(Row(Map("ab" -> 1)), Row(null))) + val failFast = spark.read + .option("rowTag", "ROW") + .option("mode", "FAILFAST") + .schema(schema) + .xml(path.getCanonicalPath) + assertUnsupportedXmlMapKeyError(failFast.collect(), key = "a", dataTypeSql = "CHAR(2)") + val ignoreCorrupt = spark.read + .option("rowTag", "ROW") + .option("ignoreCorruptFiles", "true") + .schema(schema) + .xml(path.getCanonicalPath) + checkAnswer(ignoreCorrupt, Seq(Row(Map("ab" -> 1)), Row(null))) + } - assertDuplicateMapKeyError(readXmlMap().collect()) - - withSQLConf( - SQLConf.MAP_KEY_DEDUP_POLICY.key -> SQLConf.MapKeyDedupPolicy.LAST_WIN.toString) { - checkAnswer(readXmlMap(), Row(Map("a " -> 2))) - } + withTempPath { path => + Seq( + """12""" + ).toDS().write.text(path.getCanonicalPath) + val arraySchema = "m ARRAY>" + val arrayPermissive = spark.read + .option("rowTag", "ROW") + .schema(arraySchema) + .xml(path.getCanonicalPath) + checkAnswer(arrayPermissive, Seq(Row(null), Row(Seq(Map("ab" -> 2))))) } withSQLConf( - SQLConf.MAP_KEY_DEDUP_POLICY.key -> SQLConf.MapKeyDedupPolicy.LAST_WIN.toString) { - checkAnswer(sql(xmlQuery), Row(Map("a " -> 9))) - checkAnswer(sql(varcharXmlQuery), Row(Map("ab" -> 2))) - checkAnswer(sql(exactCharXmlQuery), Row(Map("a " -> 2))) - checkAnswer(sql(exactVarcharXmlQuery), Row(Map("ab" -> 2))) - checkAnswer(sql(interleavedCharXmlQuery), Row(Seq(Row("a ", 3)))) - checkAnswer(sql(badXmlKeyBeforeDuplicateQuery), Row(null)) - checkAnswer(sql(malformedXmlValueBeforeDuplicateQuery), Row(null)) - checkAnswer(sql(ignoreCorruptXmlQuery), Row(Map("a " -> 2))) - checkAnswer(sql(attributeCollisionQuery), Row(Map("a " -> 2, "b " -> 0))) + SQLConf.MAP_KEY_DEDUP_POLICY.key -> SQLConf.MapKeyDedupPolicy.EXCEPTION.toString) { + checkAnswer( + sql("""SELECT from_xml( + | '12', + | 'm MAP').m""".stripMargin), + Row(Map("ab" -> 2))) } - } - } - test("SPARK-59274: pretty-printed XML map failures preserve parser position") { - withSQLConf(SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "true") { - val overflowQueries = Seq("CHAR(5)", "VARCHAR(5)").map { valueType => - s"""SELECT from_xml( - | ' - | - | abcdef - | xy - | - | 9 - | ', - | 'm MAP, tail INT', - | map('valueTag', 'a '))""".stripMargin - } - val nestedFailureQuery = + // Value overflow still uses EXCEED_LIMIT_LENGTH. A last-win duplicate whose winning + // value overflows nulls the map, matching the non-duplicate case. + checkAnswer( + sql("""SELECT from_xml( + | 'xabcdef', + | 'm MAP')""".stripMargin), + Row(Row(null))) + assertParseExceedLimit( """SELECT from_xml( - | ' - | - | bad - | 1ignored - | - | 9 - | ', - | 'm MAP>, tail INT', - | map('valueTag', 'a '))""".stripMargin - - (overflowQueries :+ nestedFailureQuery).foreach { query => - // A duplicate, rather than an AssertionError from XML recovery, wins under EXCEPTION. - withClue(s"$query: ") { - assertDuplicateMapKey(query) - } + | 'xabcdef', + | 'm MAP', + | map('mode', 'FAILFAST'))""".stripMargin, + expectedLimit = "2") + + withSQLConf(SQLConf.ALLOW_COLLATIONS_IN_MAP_KEYS.key -> "true") { + checkAnswer( + sql("""SELECT from_xml( + | '12', + | 'm MAP').m""".stripMargin), + Row(Map("ab" -> 1, "AB" -> 2))) } + } - withSQLConf( - SQLConf.MAP_KEY_DEDUP_POLICY.key -> SQLConf.MapKeyDedupPolicy.LAST_WIN.toString) { - // The failed map remains null, but its complete element is consumed and the `tail` field - // is parsed. - (overflowQueries :+ nestedFailureQuery).foreach { query => - checkAnswer(sql(query), Row(Row(null, 9))) + withSQLConf( + SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "false", + SQLConf.PRESERVE_CHAR_VARCHAR_TYPE_INFO.key -> "true") { + checkAnswer( + sql("SELECT from_xml('1', 'm MAP')"), + Row(Row(Map("a " -> 1)))) + assertParseExceedLimit( + """SELECT from_xml( + | '1', + | 'm MAP', + | map('mode', 'FAILFAST'))""".stripMargin, + expectedLimit = "3") + } + } + + test("SPARK-60103: XML map value overflow keeps siblings") { + withSQLConf(SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "true") { + Seq("CHAR(5)", "VARCHAR(5)").foreach { valueType => + val xml = "abcdef9" + val schema = s"m MAP, tail INT" + withClue(valueType) { + checkAnswer(sql(s"SELECT from_xml('$xml', '$schema')"), Row(Row(null, 9))) + assertParseExceedLimit( + s"SELECT from_xml('$xml', '$schema', map('mode', 'FAILFAST'))", + expectedLimit = "5") } } } @@ -3194,27 +3305,15 @@ class BasicCharVarcharTestSuite extends SharedSparkSession { } } - test("SPARK-59274: collated CHAR/VARCHAR map keys honor the dedup policy") { + test("SPARK-59274: collated XML CHAR/VARCHAR map keys last-win on exact names") { withSQLConf( SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "true", SQLConf.ALLOW_COLLATIONS_IN_MAP_KEYS.key -> "true") { - val xmlQuery = - """SELECT from_xml( - | '12', - | 'm MAP').m""".stripMargin - val varcharXmlQuery = - """SELECT from_xml( + checkAnswer( + sql("""SELECT from_xml( | '12', - | 'm MAP').m""".stripMargin - - assertDuplicateMapKey(xmlQuery, expectedKey = "A ") - assertDuplicateMapKey(varcharXmlQuery, expectedKey = "AB") - - withSQLConf( - SQLConf.MAP_KEY_DEDUP_POLICY.key -> SQLConf.MapKeyDedupPolicy.LAST_WIN.toString) { - checkAnswer(sql(xmlQuery), Row(Map("a " -> 2))) - checkAnswer(sql(varcharXmlQuery), Row(Map("ab" -> 2))) - } + | 'm MAP').m""".stripMargin), + Row(Map("ab" -> 1, "AB" -> 2))) } } } From 8e1c55b7b487aa7d8a3678089db4e1fd5b41f63c Mon Sep 17 00:00:00 2001 From: srielau Date: Sat, 10 Oct 2026 18:43:08 +0000 Subject: [PATCH 2/4] [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 elements stay null or malformed. Restrict convertMap to StringType-family keys so format("xml") MAP does not ClassCast. --- docs/sql-migration-guide.md | 2 +- .../sql/catalyst/xml/StaxXmlParser.scala | 22 ++++--- .../spark/sql/CharVarcharTestSuite.scala | 58 ++++++++++++++++--- 3 files changed, 65 insertions(+), 17 deletions(-) diff --git a/docs/sql-migration-guide.md b/docs/sql-migration-guide.md index be98ca88c26de..2525954897efe 100644 --- a/docs/sql-migration-guide.md +++ b/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` or `MAP` 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` or `MAP` 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` and `MAP` 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. diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/xml/StaxXmlParser.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/xml/StaxXmlParser.scala index 50f5404b56dc9..7766f2bc213b0 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/xml/StaxXmlParser.scala +++ b/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 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 XML maps. * - * Example: `from_xml('1', 'm MAP')` - * keeps key `ab`; `MAP` raises `UNSUPPORTED_XML_CHAR_VARCHAR_MAP_KEY`. + * Example (flag on): `from_xml('1', + * 'm MAP')` keeps key `ab`; `MAP` raises + * `UNSUPPORTED_XML_CHAR_VARCHAR_MAP_KEY`. * * This method owns element bounding so a key-check failure still consumes the current * map element (ARRAY> 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]]) diff --git a/sql/core/src/test/scala/org/apache/spark/sql/CharVarcharTestSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/CharVarcharTestSuite.scala index 8d23ce44fdf22..9278dff4fa19d 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/CharVarcharTestSuite.scala +++ b/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('xy', 'm MAP')"), + sql("SELECT from_xml('1xy', 'm MAP')"), Row(Row(null))) assertUnsupportedXmlMapKey( """SELECT from_xml( - | 'xy', + | '1xy', | 'm MAP', | 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('', 'm MAP')"), + sql("SELECT from_xml('2', 'm MAP')"), Row(Row(null))) checkAnswer( - sql("SELECT from_xml('', 'm MAP')"), - Row(Row(Map("_ab" -> 1)))) + sql("SELECT from_xml('2', 'm MAP')"), + Row(Row(Map("_ab" -> 1, "xyz" -> 2)))) assertUnsupportedXmlMapKey( """SELECT from_xml( - | '', + | '2', | 'm MAP', | map('mode', 'FAILFAST'))""".stripMargin, key = "_ab", dataTypeSql = "CHAR(2)") + // MAP empty, attribute-only, and text-only elements stay SQL NULL, matching + // convertField on master. ARRAY empty elements are null entries, not {}. + checkAnswer( + sql("SELECT from_xml('', 'm MAP')"), + Row(Row(null))) + checkAnswer( + sql("SELECT from_xml('', 'm MAP')"), + Row(Row(null))) + checkAnswer( + sql("SELECT from_xml('xy', 'm MAP')"), + Row(Row(null))) + checkAnswer( + sql("""SELECT from_xml( + | '9', + | 'm ARRAY>, tail INT')""".stripMargin), + Row(Row(Seq(null), 9))) + withTempPath { path => Seq( """11""" @@ -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("""x""") + .toDS().write.text(path.getCanonicalPath) + val intKeySchema = "m MAP" + 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 XML read must not ClassCastException: $error") + } + withSQLConf( SQLConf.MAP_KEY_DEDUP_POLICY.key -> SQLConf.MapKeyDedupPolicy.EXCEPTION.toString) { checkAnswer( From 83eb33194a873c9bf25ee1aed2daf62890263a97 Mon Sep 17 00:00:00 2001 From: srielau Date: Sun, 11 Oct 2026 02:27:51 +0000 Subject: [PATCH 3/4] [SPARK-60103][SQL] Drop dead XML DUPLICATED_MAP_KEY routing --- .../catalyst/util/DuplicateMapKeyUtils.scala | 33 ------------------- 1 file changed, 33 deletions(-) delete mode 100644 sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/DuplicateMapKeyUtils.scala diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/DuplicateMapKeyUtils.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/DuplicateMapKeyUtils.scala deleted file mode 100644 index 5b9f3bf453511..0000000000000 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/DuplicateMapKeyUtils.scala +++ /dev/null @@ -1,33 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.apache.spark.sql.catalyst.util - -import org.apache.spark.SparkRuntimeException -import org.apache.spark.util.SparkErrorUtils - -private[sql] object DuplicateMapKeyUtils { - def cause(exception: Throwable): Option[SparkRuntimeException] = { - SparkErrorUtils.getRootCause(exception) match { - case cause: SparkRuntimeException if cause.getCondition == "DUPLICATED_MAP_KEY" => - Some(cause) - case _ => None - } - } - - def unapply(exception: Throwable): Option[SparkRuntimeException] = cause(exception) -} From 38a81e32e268bd8795cd9cd99e7cce37c5befcd7 Mon Sep 17 00:00:00 2001 From: srielau Date: Sun, 11 Oct 2026 02:27:58 +0000 Subject: [PATCH 4/4] [SPARK-60103][SQL] Correct XML CHAR/VARCHAR map-key 4.3 and flag-off notes --- docs/sql-migration-guide.md | 2 +- .../sql/catalyst/xml/StaxXmlParser.scala | 16 +++--------- .../spark/sql/CharVarcharTestSuite.scala | 26 +++++-------------- 3 files changed, 11 insertions(+), 33 deletions(-) diff --git a/docs/sql-migration-guide.md b/docs/sql-migration-guide.md index 2525954897efe..c459a028da553 100644 --- a/docs/sql-migration-guide.md +++ b/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 names used as `MAP` or `MAP` 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` and `MAP` 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, when `spark.sql.charVarchar.standardSemantics.enabled` is true, XML names used as `MAP` or `MAP` 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; they do not become an empty map or a `valueTag` entry. In Spark 4.3, `from_xml` and XML reader schemas rejected CHAR/VARCHAR with `UNSUPPORTED_CHAR_OR_VARCHAR_AS_STRING`, or replaced them with STRING when `spark.sql.legacy.charVarcharAsString` was true, so CHAR/VARCHAR map keys did not reach the parser. When the standard-semantics flag is false and `spark.sql.preserveCharVarcharTypeInfo` is true, short CHAR keys are padded and over-length CHAR or VARCHAR keys raise `EXCEED_LIMIT_LENGTH`. - 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. diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/xml/StaxXmlParser.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/xml/StaxXmlParser.scala index 7766f2bc213b0..4f388fdf49335 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/xml/StaxXmlParser.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/xml/StaxXmlParser.scala @@ -41,7 +41,7 @@ import org.apache.spark.{SparkIllegalArgumentException, SparkUpgradeException} import org.apache.spark.internal.Logging import org.apache.spark.sql.catalyst.InternalRow import org.apache.spark.sql.catalyst.expressions.{ExprUtils, GenericInternalRow, ToStringBase} -import org.apache.spark.sql.catalyst.util.{ArrayBasedMapData, BadRecordException, CharVarcharUtils, DateFormatter, DropMalformedMode, DuplicateMapKeyUtils, FailureSafeParser, GenericArrayData, MapData, ParseMode, PartialResultArrayException, PartialResultException, PermissiveMode, TimeFormatter, TimestampFormatter} +import org.apache.spark.sql.catalyst.util.{ArrayBasedMapData, BadRecordException, CharVarcharUtils, DateFormatter, DropMalformedMode, FailureSafeParser, GenericArrayData, MapData, ParseMode, PartialResultArrayException, PartialResultException, PermissiveMode, TimeFormatter, TimestampFormatter} import org.apache.spark.sql.catalyst.util.LegacyDateFormats.FAST_DATE_FORMAT import org.apache.spark.sql.catalyst.xml.StaxXmlParser.convertStream import org.apache.spark.sql.errors.QueryExecutionErrors @@ -235,7 +235,6 @@ class StaxXmlParser( // ValidatorUtil.newValidator throws this when the JAXP implementation cannot // disable external access; that is an environment error, not a bad record. case e: UnsupportedOperationException => throw e - case DuplicateMapKeyUtils(e) => throw e case e@(_: RuntimeException | _: XMLStreamException | _: MalformedInputException | _: SAXException) => // XML parser currently doesn't support partial results for corrupted records. @@ -332,16 +331,11 @@ class StaxXmlParser( throw BadRecordException(xmlLiteral, () => Array.empty, wrappedCharException) case PartialResultException(row, cause) => - DuplicateMapKeyUtils.cause(cause) match { - case Some(e) => throw e - case None => - throw BadRecordException(record = xmlLiteral, partialResults = () => Array(row), cause) - } + throw BadRecordException(record = xmlLiteral, partialResults = () => Array(row), cause) case PartialResultArrayException(rows, cause) => throw BadRecordException(record = xmlLiteral, partialResults = () => rows, cause) case e: Throwable => SparkErrorUtils.getRootCause(e) match { - case DuplicateMapKeyUtils(duplicate) => throw duplicate case _: FileNotFoundException if options.ignoreMissingFiles => logWarning("Skipped missing file", e) parser.close() @@ -454,8 +448,8 @@ class StaxXmlParser( * 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. + * short CHAR keys are padded and over-length CHAR or VARCHAR keys raise + * 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 XML maps. * @@ -557,7 +551,6 @@ class StaxXmlParser( row(i) = convertTo(v, schema(i).dataType) } catch { case e: SparkUpgradeException => throw e - case DuplicateMapKeyUtils(e) => throw e case NonFatal(e) => firstError = firstError.orElse(Some(e)) } } @@ -696,7 +689,6 @@ class StaxXmlParser( } } catch { case e: SparkUpgradeException => throw e - case DuplicateMapKeyUtils(e) => throw e case NonFatal(e) => // TODO: we don't support partial results now badRecordException = badRecordException.orElse(Some(e)) diff --git a/sql/core/src/test/scala/org/apache/spark/sql/CharVarcharTestSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/CharVarcharTestSuite.scala index 9278dff4fa19d..f9f5062b1dfc9 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/CharVarcharTestSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/CharVarcharTestSuite.scala @@ -1066,26 +1066,6 @@ class BasicCharVarcharTestSuite extends SharedSparkSession { parameters = Map("key" -> s"'$key'", "dataType" -> s""""$dataTypeSql"""")) } - private def assertDuplicateMapKey(query: String, expectedKey: String = "a "): Unit = { - assertDuplicateMapKeyError(sql(query).collect(), expectedKey) - } - - private def assertDuplicateMapKeyError(body: => Any, expectedKey: String = "a "): Unit = { - val error = intercept[Exception] { body } - val duplicateError = Iterator.iterate[Throwable](error)(_.getCause) - .takeWhile(_ != null) - .collectFirst { - case e: SparkRuntimeException if e.getCondition == "DUPLICATED_MAP_KEY" => e - } - .getOrElse(fail("expected DUPLICATED_MAP_KEY cause", error)) - checkError( - exception = duplicateError, - condition = "DUPLICATED_MAP_KEY", - parameters = Map( - "key" -> expectedKey, - "mapKeyDedupPolicy" -> "\"spark.sql.mapKeyDedupPolicy\"")) - } - test("user-specified schema in cast") { def assertNoCharType(df: DataFrame): Unit = { checkAnswer(df, Row("0")) @@ -3181,6 +3161,12 @@ class BasicCharVarcharTestSuite extends SharedSparkSession { checkAnswer( sql("SELECT from_xml('1', 'm MAP')"), Row(Row(Map("a " -> 1)))) + assertParseExceedLimit( + """SELECT from_xml( + | '1', + | 'm MAP', + | map('mode', 'FAILFAST'))""".stripMargin, + expectedLimit = "3") assertParseExceedLimit( """SELECT from_xml( | '1',