This is an automated email from the ASF dual-hosted git repository.
HyukjinKwon pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new ccc8f69ce6be [SPARK-60103][SQL] Length-check XML CHAR/VARCHAR map keys
without pad or trim
ccc8f69ce6be is described below
commit ccc8f69ce6be5a879c193788c1d12d7ea9073e22
Author: Serge Rielau <[email protected]>
AuthorDate: Sun Oct 11 13:45:01 2026 +0900
[SPARK-60103][SQL] Length-check XML CHAR/VARCHAR map keys without pad or
trim
### What changes were proposed in this pull request?
When `spark.sql.charVarchar.standardSemantics.enabled` is true, `from_xml`
and the XML datasource length-check XML names used as `MAP<CHAR(n), _>` /
`MAP<VARCHAR(n), _>` keys without rewriting them. This matches SPARK-60108 for
JSON.
The checked names are element names, `attributePrefix` plus attribute names
(default `_`), and the `valueTag` (default `_VALUE`) when mixed text is present.
- `CHAR(n)` keys must already be exactly `n` characters (no pad).
- `VARCHAR(n)` keys must already be at most `n` characters (no trim).
- Exact repeated names last-win. `spark.sql.mapKeyDedupPolicy` is not
applied.
- Length failures follow the XML parse mode (`PERMISSIVE` / `FAILFAST`).
Mismatched keys raise a new `UNSUPPORTED_XML_CHAR_VARCHAR_MAP_KEY`
condition (SQLSTATE `0A000`) instead of `EXCEED_LIMIT_LENGTH`, because pad/trim
of map keys is not supported. XML CHAR/VARCHAR values still use the same pad
and `EXCEED_LIMIT_LENGTH` checks as other parsed text. JSON, CSV, and TRANSFORM
are unchanged.
Struct-field maps still go through `convertField`. Empty, attribute-only,
and text-only `MAP<STRING, _>` elements stay SQL NULL or a malformed record, as
on master. `convertMap` is used for non-empty string-family keys (including
CHAR/VARCHAR) and owns element bounding so a key-check failure still drains
`ARRAY<MAP<...>>` siblings.
This PR drops `convertConstrainedMap` and `DuplicateMapKeyUtils`.
SPARK-59722-style assign-after-parse is not used.
JIRA: https://issues.apache.org/jira/browse/SPARK-60103
### Why are the changes needed?
SPARK-59274 padded and trimmed CHAR/VARCHAR XML map keys during the walk,
then applied `mapKeyDedupPolicy` to invented collisions. SPARK-59722 would have
done the same via write-side assignment after a STRING parse; that approach was
abandoned for JSON in favor of SPARK-60108. XML names are the key identity, so
they should be length-checked in place rather than rewritten.
### Does this PR introduce _any_ user-facing change?
Yes, under `spark.sql.charVarchar.standardSemantics.enabled=true`:
```sql
SELECT from_xml('<ROW><m><a>1</a></m></ROW>', 'm MAP<CHAR(2), INT>',
map('mode', 'FAILFAST'));
-- [UNSUPPORTED_XML_CHAR_VARCHAR_MAP_KEY] XML map keys of CHAR or VARCHAR
cannot be padded or trimmed.
-- The key 'a' is not valid for type "CHAR(2)". SQLSTATE: 0A000
```
Exact-width CHAR keys and in-limit VARCHAR keys are kept as-is. Repeated
names last-win. 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`.
Documented in `docs/sql-migration-guide.md` (Spark SQL 4.3 to 4.4). 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.
### How was this patch tested?
Added SPARK-60103 coverage in `BasicCharVarcharTestSuite` (too-short CHAR,
exact-width CHAR, CHAR/VARCHAR overflow, last-wins duplicates, prefixed
attributes and `valueTag` in mixed content, PERMISSIVE vs FAILFAST, nested
maps, `ARRAY<MAP>`, collated keys, `from_xml` and the XML datasource,
empty/attribute-only/text-only `MAP<STRING>`, `format("xml")` `MAP<INT>`,
flag-off pad/overflow including over-length CHAR). Also re-ran SPARK-59274 XML
value, sibling, and `rowTag` tests.
Ran:
```
JAVA_HOME=/usr/lib/jvm/java-17-openjdk-amd64 build/sbt
-Dsbt.override.build.repos=true \
'sql/testOnly org.apache.spark.sql.BasicCharVarcharTestSuite -- -z
SPARK-60103 -z "SPARK-59274: from_json/csv/xml" -z "SPARK-59274: ordinary
STRING" -z "SPARK-59274: collated" -z "SPARK-59274: XML rowTag" -z
"SPARK-59274: mixed XML" -z "SPARK-59274: nested XML"'
```
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Cursor Grok 4.6
Closes #59316 from srielau/SPARK-60103.
Authored-by: Serge Rielau <[email protected]>
Signed-off-by: Hyukjin Kwon <[email protected]>
---
.../src/main/resources/error/error-conditions.json | 7 +
docs/sql-migration-guide.md | 1 +
.../sql/catalyst/util/DuplicateMapKeyUtils.scala | 82 ----
.../spark/sql/catalyst/xml/StaxXmlParser.scala | 180 ++++-----
.../spark/sql/errors/QueryExecutionErrors.scala | 14 +
.../org/apache/spark/sql/internal/SQLConf.scala | 8 +-
.../apache/spark/sql/CharVarcharTestSuite.scala | 429 ++++++++++++++-------
7 files changed, 376 insertions(+), 345 deletions(-)
diff --git a/common/utils/src/main/resources/error/error-conditions.json
b/common/utils/src/main/resources/error/error-conditions.json
index 94d6b897b18c..7d17404b29a4 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 <key> is not valid for type <dataType>."
+ ],
+ "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 f9c7bf477122..c459a028da55 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 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 [...]
- 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 optio [...]
- 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
deleted file mode 100644
index fd179bc73d05..000000000000
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/DuplicateMapKeyUtils.scala
+++ /dev/null
@@ -1,82 +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 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 {
- 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)
-
- /**
- * 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 f0923d405436..4f388fdf4933 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
@@ -42,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
@@ -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
@@ -233,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.
@@ -330,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()
@@ -383,9 +379,10 @@ 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)
+ // 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)
@@ -445,109 +442,80 @@ class StaxXmlParser(
}
/**
- * Parse an object as map.
+ * Parse an object as a Map.
+ *
+ * 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,
+ * 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<STRING, ...> XML maps.
+ *
+ * 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).
*/
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
}
/**
@@ -583,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))
}
}
@@ -722,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/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 b8bfd00cb2c8..090e73c264cc 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 3a72fd622a2a..a6ff1e548f7b 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 c203fa1d8c34..f9f5062b1dfc 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,24 +1044,26 @@ class BasicCharVarcharTestSuite extends
SharedSparkSession {
parameters = Map("limit" -> expectedLimit))
}
- private def assertDuplicateMapKey(query: String, expectedKey: String = "a
"): Unit = {
- assertDuplicateMapKeyError(sql(query).collect(), expectedKey)
+ private def assertUnsupportedXmlMapKey(
+ query: String, key: String, dataTypeSql: String): Unit = {
+ assertUnsupportedXmlMapKeyError(sql(query).collect(), key, dataTypeSql)
}
- private def assertDuplicateMapKeyError(body: => Any, expectedKey: String =
"a "): Unit = {
+ private def assertUnsupportedXmlMapKeyError(
+ body: => Any, key: String, dataTypeSql: String): Unit = {
val error = intercept[Exception] { body }
- val duplicateError = Iterator.iterate[Throwable](error)(_.getCause)
+ val cause = Iterator.iterate[Throwable](error)(_.getCause)
.takeWhile(_ != null)
.collectFirst {
- case e: SparkRuntimeException if e.getCondition ==
"DUPLICATED_MAP_KEY" => e
+ case e: SparkRuntimeException
+ if e.getCondition == "UNSUPPORTED_XML_CHAR_VARCHAR_MAP_KEY" => e
}
- .getOrElse(fail("expected DUPLICATED_MAP_KEY cause", error))
+ .getOrElse(fail("expected UNSUPPORTED_XML_CHAR_VARCHAR_MAP_KEY cause",
error))
checkError(
- exception = duplicateError,
- condition = "DUPLICATED_MAP_KEY",
- parameters = Map(
- "key" -> expectedKey,
- "mapKeyDedupPolicy" -> "\"spark.sql.mapKeyDedupPolicy\""))
+ exception = cause,
+ condition = "UNSUPPORTED_XML_CHAR_VARCHAR_MAP_KEY",
+ sqlState = "0A000",
+ parameters = Map("key" -> s"'$key'", "dataType" ->
s""""$dataTypeSql""""))
}
test("user-specified schema in cast") {
@@ -2795,9 +2797,12 @@ class BasicCharVarcharTestSuite extends
SharedSparkSession {
s"SELECT from_xml('<ROW><a>abcdef</a></ROW>', 'a $dataType', " +
"map('mode', 'FAILFAST'))")
}
+ checkAnswer(
+ sql("SELECT from_xml('<ROW><m><ab>1</ab></m></ROW>', 'm MAP<CHAR(2),
INT>')"),
+ Row(Row(Map("ab" -> 1))))
checkAnswer(
sql("SELECT from_xml('<ROW><m><ab>1</ab></m></ROW>', 'm MAP<CHAR(4),
INT>')"),
- Row(Row(Map("ab " -> 1))))
+ Row(Row(null)))
checkAnswer(
sql(
"""SELECT from_xml(
@@ -2837,7 +2842,7 @@ class BasicCharVarcharTestSuite extends
SharedSparkSession {
| 'xs_any ARRAY<CHAR(5)>',
| map('wildcardColName', 'xs_any', 'mode',
'FAILFAST'))""".stripMargin)
withTempPath { path =>
-
Seq("<ROW><m><a>1</a></m></ROW>").toDS().write.text(path.getCanonicalPath)
+
Seq("<ROW><m><ab>1</ab></m></ROW>").toDS().write.text(path.getCanonicalPath)
val xmlDataFrame = spark.read
.option("rowTag", "ROW")
.schema("m MAP<CHAR(2), INT>")
@@ -2845,7 +2850,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 +2901,291 @@ 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(
- | '<ROW><m><a>1</a>9</m></ROW>',
- | 'm MAP<CHAR(2), INT>',
- | map('valueTag', 'a ')).m""".stripMargin
- val varcharXmlQuery =
- """SELECT from_xml(
- | '<ROW><m><ab>1</ab>2</m></ROW>',
- | 'm MAP<VARCHAR(2), INT>',
- | map('valueTag', 'ab ')).m""".stripMargin
- val exactCharXmlQuery =
+ checkAnswer(
+ sql("SELECT from_xml('<ROW><m><ab>1</ab></m></ROW>', 'm MAP<CHAR(2),
INT>').m"),
+ Row(Map("ab" -> 1)))
+ checkAnswer(
+ sql("SELECT from_xml('<ROW><m><ab>1</ab></m></ROW>', 'm
MAP<VARCHAR(3), INT>').m"),
+ Row(Map("ab" -> 1)))
+ checkAnswer(
+ sql("""SELECT from_xml(
+ | '<ROW><m><ab>1</ab><AB>2</AB></m></ROW>',
+ | 'm MAP<VARCHAR(2), INT>').m""".stripMargin),
+ Row(Map("ab" -> 1, "AB" -> 2)))
+ checkAnswer(
+ sql("""SELECT from_xml(
+ | '<ROW><m><ab>1</ab><ab>2</ab></m></ROW>',
+ | 'm MAP<CHAR(2), INT>').m""".stripMargin),
+ Row(Map("ab" -> 2)))
+ checkAnswer(
+ sql("""SELECT from_xml(
+ | '<ROW><m><outer><xy>1</xy></outer></m></ROW>',
+ | 'm MAP<STRING, MAP<CHAR(2), INT>>').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(
+ | '<ROW><m><a>1</a></m><tail>2</tail></ROW>',
+ | 'm MAP<CHAR(2), INT>, tail INT')""".stripMargin),
+ Row(Row(null, 2)))
+ checkAnswer(
+ sql("""SELECT from_xml(
+ | '<ROW><m><outer><z>1</z></outer></m><tail>2</tail></ROW>',
+ | 'm MAP<STRING, MAP<CHAR(2), INT>>, tail INT')""".stripMargin),
+ Row(Row(null, 2)))
+ assertUnsupportedXmlMapKey(
"""SELECT from_xml(
- | '<ROW><m><a>1</a><a>2</a></m></ROW>',
- | 'm MAP<CHAR(2), INT>').m""".stripMargin
- val exactVarcharXmlQuery =
+ | '<ROW><m><outer><z>1</z></outer></m></ROW>',
+ | 'm MAP<STRING, MAP<CHAR(2), INT>>',
+ | map('mode', 'FAILFAST'))""".stripMargin,
+ key = "z",
+ dataTypeSql = "CHAR(2)")
+
+ // ARRAY<MAP<...>> uses convertField's MapType arm; a short key must
still keep tail.
+ checkAnswer(
+ sql("""SELECT from_xml(
+ | '<ROW><m><a>1</a></m><m><ab>2</ab></m><tail>9</tail></ROW>',
+ | 'm ARRAY<MAP<CHAR(2), INT>>, tail INT')""".stripMargin),
+ Row(Row(Seq(Map("ab" -> 2)), 9)))
+ assertUnsupportedXmlMapKey(
"""SELECT from_xml(
- | '<ROW><m><ab>1</ab><ab>2</ab></m></ROW>',
- | 'm MAP<VARCHAR(2), INT>').m""".stripMargin
- val interleavedCharXmlQuery =
- """SELECT map_entries(from_xml(
- | '<ROW><m><a>1</a>2<a>3</a></m></ROW>',
- | 'm MAP<CHAR(2), INT>',
- | map('valueTag', 'a ')).m)""".stripMargin
- val badXmlKeyThenSiblingQuery =
+ | '<ROW><m><a>1</a></m><m><ab>2</ab></m></ROW>',
+ | 'm ARRAY<MAP<CHAR(2), INT>>',
+ | map('mode', 'FAILFAST'))""".stripMargin,
+ key = "a",
+ dataTypeSql = "CHAR(2)")
+
+ // 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><ab>1</ab>xy</m></ROW>', 'm MAP<CHAR(2),
STRING>')"),
+ Row(Row(null)))
+ assertUnsupportedXmlMapKey(
"""SELECT from_xml(
- | '<ROW><m><abc>1</abc></m><tail>2</tail></ROW>',
- | 'm MAP<CHAR(2), INT>, tail INT').tail""".stripMargin
- val badXmlKeyBeforeDuplicateQuery =
+ | '<ROW><m><ab>1</ab>xy</m></ROW>',
+ | 'm MAP<CHAR(2), STRING>',
+ | 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('<ROW><m><a>1</a></m></ROW>', 'm MAP<CHAR(2),
INT>')"),
+ Row(Row(null)))
+ checkAnswer(
+ sql("SELECT from_xml('<ROW><m><abc>1</abc></m></ROW>', 'm MAP<CHAR(2),
INT>')"),
+ Row(Row(null)))
+ checkAnswer(
+ sql("SELECT from_xml('<ROW><m><abcd>1</abcd></m></ROW>', 'm
MAP<VARCHAR(3), INT>')"),
+ Row(Row(null)))
+ assertUnsupportedXmlMapKey(
"""SELECT from_xml(
- | '<ROW><m><abc>0</abc><a>1</a>2</m></ROW>',
+ | '<ROW><m><a>1</a></m></ROW>',
| 'm MAP<CHAR(2), INT>',
- | map('valueTag', 'a ')).m""".stripMargin
- val malformedXmlValueBeforeDuplicateQuery =
+ | map('mode', 'FAILFAST'))""".stripMargin,
+ key = "a",
+ dataTypeSql = "CHAR(2)")
+ assertUnsupportedXmlMapKey(
"""SELECT from_xml(
- | '<ROW><m><a>bad</a>2</m></ROW>',
+ | '<ROW><m><abc>1</abc></m></ROW>',
| 'm MAP<CHAR(2), INT>',
- | map('valueTag', 'a ')).m""".stripMargin
- val ignoreCorruptXmlQuery =
+ | map('mode', 'FAILFAST'))""".stripMargin,
+ key = "abc",
+ dataTypeSql = "CHAR(2)")
+ assertUnsupportedXmlMapKey(
"""SELECT from_xml(
- | '<ROW><m><a>1</a>2</m></ROW>',
+ | '<ROW><m><abcd>1</abcd></m></ROW>',
+ | 'm MAP<VARCHAR(3), INT>',
+ | map('mode', 'FAILFAST'))""".stripMargin,
+ key = "abcd",
+ dataTypeSql = "VARCHAR(3)")
+
+ checkAnswer(
+ sql("""SELECT from_xml(
+ | '<ROW><m a="1"><bc>2</bc></m></ROW>',
| 'm MAP<CHAR(2), INT>',
- | map('valueTag', 'a ', 'ignoreCorruptFiles',
'true')).m""".stripMargin
- val attributePaddingQuery =
- """SELECT from_xml(
- | '<ROW><m a="1"><b>2</b></m></ROW>',
+ | map('attributePrefix', ''))""".stripMargin),
+ Row(Row(null)))
+ checkAnswer(
+ sql("""SELECT from_xml(
+ | '<ROW><m ab="1"><cd>2</cd></m></ROW>',
| 'm MAP<CHAR(2), INT>',
- | map('attributePrefix', '')).m""".stripMargin
- val attributeOverflowQuery =
- """SELECT from_xml(
- | '<ROW><m abc="1"><b>2</b></m></ROW>',
+ | map('attributePrefix', ''))""".stripMargin),
+ Row(Row(Map("ab" -> 1, "cd" -> 2))))
+ checkAnswer(
+ sql("""SELECT from_xml(
+ | '<ROW><m abc="1"><cd>2</cd></m></ROW>',
| 'm MAP<VARCHAR(2), INT>',
- | map('attributePrefix', '')).m""".stripMargin
- val attributeOverflowFailfastQuery =
+ | map('attributePrefix', ''))""".stripMargin),
+ Row(Row(null)))
+ assertUnsupportedXmlMapKey(
"""SELECT from_xml(
- | '<ROW><m abc="1"><b>2</b></m></ROW>',
+ | '<ROW><m abc="1"><cd>2</cd></m></ROW>',
| 'm MAP<VARCHAR(2), INT>',
- | 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.
+ // Attribute-only elements are SQL NULL (convertField EndElement), so
include a child.
+ checkAnswer(
+ 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\"><xyz>2</xyz></m></ROW>', 'm
MAP<CHAR(3), INT>')"),
+ Row(Row(Map("_ab" -> 1, "xyz" -> 2))))
+ assertUnsupportedXmlMapKey(
"""SELECT from_xml(
- | '<ROW><m a="1"><b>0</b>2</m></ROW>',
+ | '<ROW><m ab="1"><xy>2</xy></m></ROW>',
| 'm MAP<CHAR(2), INT>',
- | 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)")
+
+ // 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("<ROW><m><a>1</a>2</m></ROW>").toDS().write.text(path.getCanonicalPath)
- def readXmlMap(): DataFrame = spark.read
+ Seq(
+
"""<ROWS><ROW><m><ab>1</ab></m></ROW><ROW><m><a>1</a></m></ROW></ROWS>"""
+ ).toDS().write.text(path.getCanonicalPath)
+ val schema = "m MAP<CHAR(2), INT>"
+ val permissive = spark.read
.option("rowTag", "ROW")
- .option("valueTag", "a ")
- .schema("m MAP<CHAR(2), INT>")
+ .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())
+ withTempPath { path =>
+ Seq(
+
"""<ROWS><ROW><m><a>1</a></m></ROW><ROW><m><ab>2</ab></m></ROW></ROWS>"""
+ ).toDS().write.text(path.getCanonicalPath)
+ val arraySchema = "m ARRAY<MAP<CHAR(2), INT>>"
+ 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(readXmlMap(), Row(Map("a " -> 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.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(
+ | '<ROW><m><ab>1</ab><ab>2</ab></m></ROW>',
+ | 'm MAP<VARCHAR(2), INT>').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(
- | '<ROW>
- | <m>
- | <b>abcdef</b>
- | <a>x</a>y
- | </m>
- | <tail>9</tail>
- | </ROW>',
- | 'm MAP<CHAR(2), $valueType>, 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(
+ | '<ROW><m><ab>x</ab><ab>abcdef</ab></m></ROW>',
+ | 'm MAP<CHAR(2), CHAR(2)>')""".stripMargin),
+ Row(Row(null)))
+ assertParseExceedLimit(
"""SELECT from_xml(
- | '<ROW>
- | <m>
- | <b><x>bad</x></b>
- | <a><x>1</x></a>ignored
- | </m>
- | <tail>9</tail>
- | </ROW>',
- | 'm MAP<CHAR(2), MAP<CHAR(2), INT>>, 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)
- }
+ | '<ROW><m><ab>x</ab><ab>abcdef</ab></m></ROW>',
+ | 'm MAP<CHAR(2), CHAR(2)>',
+ | map('mode', 'FAILFAST'))""".stripMargin,
+ expectedLimit = "2")
+
+ withSQLConf(SQLConf.ALLOW_COLLATIONS_IN_MAP_KEYS.key -> "true") {
+ checkAnswer(
+ sql("""SELECT from_xml(
+ | '<ROW><m><ab>1</ab><AB>2</AB></m></ROW>',
+ | 'm MAP<VARCHAR(2) COLLATE UTF8_LCASE, INT>').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('<ROW><m><a>1</a></m></ROW>', 'm MAP<CHAR(3),
INT>')"),
+ Row(Row(Map("a " -> 1))))
+ assertParseExceedLimit(
+ """SELECT from_xml(
+ | '<ROW><m><abcd>1</abcd></m></ROW>',
+ | 'm MAP<CHAR(3), INT>',
+ | map('mode', 'FAILFAST'))""".stripMargin,
+ expectedLimit = "3")
+ assertParseExceedLimit(
+ """SELECT from_xml(
+ | '<ROW><m><abcd>1</abcd></m></ROW>',
+ | 'm MAP<VARCHAR(3), INT>',
+ | 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 = "<ROW><m><xy>abcdef</xy></m><tail>9</tail></ROW>"
+ val schema = s"m MAP<CHAR(2), $valueType>, 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 +3335,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(
- | '<ROW><m><a>1</a><A>2</A></m></ROW>',
- | 'm MAP<CHAR(2) COLLATE UTF8_LCASE, INT>').m""".stripMargin
- val varcharXmlQuery =
- """SELECT from_xml(
+ checkAnswer(
+ sql("""SELECT from_xml(
| '<ROW><m><ab>1</ab><AB>2</AB></m></ROW>',
- | 'm MAP<VARCHAR(2) COLLATE UTF8_LCASE, INT>').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<CHAR(2) COLLATE UTF8_LCASE, INT>').m""".stripMargin),
+ Row(Map("ab" -> 1, "AB" -> 2)))
}
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
