srielau commented on code in PR #59167:
URL: https://github.com/apache/spark/pull/59167#discussion_r4190362425
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala:
##########
@@ -659,6 +721,63 @@ class JacksonParser(
}
}
+ /**
+ * CHAR/VARCHAR JSON maps apply assignment to object names, then
+ * `spark.sql.mapKeyDedupPolicy` to collisions created by that normalization.
+ * Exact repeated names stay historical last-wins.
+ */
+ private def convertConstrainedMap(
+ parser: JsonParser,
+ fieldConverter: ValueConverter,
+ keyType: DataType,
+ valueType: DataType,
+ mapDepth: Int): MapData = {
+ val lastEntries =
+ mutable.LinkedHashMap.empty[UTF8String, (UTF8String, Option[Any])]
+ var partialResultException: Option[Throwable] = None
+ var badMapException: Option[Throwable] = None
+
+ while (nextUntil(parser, JsonToken.END_OBJECT)) {
+ val rawKey = UTF8String.fromString(parser.currentName)
+ val value = try {
+ Some(fieldConverter.apply(parser))
+ } catch {
+ case err: PartialValueException if enablePartialResults =>
+ partialResultException =
partialResultException.orElse(Some(err.cause))
+ Some(err.partialResult)
+ case DuplicateMapKeyUtils(e) => throw e
+ case NonFatal(e) =>
+ badMapException = badMapException.orElse(Some(e))
+ skipRemainingValue(parser, mapDepth)
+ None
+ }
+ try {
+ val key = CharVarcharUtils.applyTextParseSemantics(rawKey, keyType)
+ // Re-insert so LAST_WIN follows the latest occurrence of this raw
name,
+ // matching StaxXmlParser.convertConstrainedMap.
+ lastEntries.remove(rawKey)
+ lastEntries.update(rawKey, (key, value))
+ } catch {
+ case NonFatal(e) =>
+ badMapException = badMapException.orElse(Some(e))
+ }
+ }
+
+ // Dedup runs on the drained object before any stored conversion failure is
+ // rethrown. A normalization collision therefore surfaces as
DUPLICATED_MAP_KEY
+ // in preference to EXCEED_LIMIT_LENGTH (oversize key) or a
value-conversion
+ // error. LAST_WIN resolves the collision and then reports the stored
failure
+ // (PERMISSIVE keeps the partial map via PartialMapDataResultException).
+ val mapData = DuplicateMapKeyUtils.buildConstrainedMap(
+ lastEntries, keyType, valueType)
+ val firstError = badMapException.orElse(partialResultException)
+ if (firstError.isEmpty) {
+ mapData
+ } else {
+ throw PartialMapDataResultException(mapData, firstError.get)
Review Comment:
Fixed on the current head (`c492d9f` / `30d6247`). When
`enablePartialResults` is false, `convertArray` and `convertOrdinaryMap` unwrap
`PartialValueException` to `err.cause` so an inner `MapData` cannot be
serialized as the root value. `ARRAY<MAP<CHAR(2), INT>>` and `MAP<STRING,
MAP<CHAR(2), INT>>` now assert `null` / `MALFORMED_RECORD_IN_PARSING` in that
configuration.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala:
##########
@@ -617,27 +620,86 @@ class JacksonParser(
}
}
+ /**
+ * Finish a JSON value that failed conversion. `skipChildren` only works at
+ * START_OBJECT / START_ARRAY; if the converter already consumed tokens
inside a nested
+ * object, leftover FIELD_NAMEs would otherwise leak as keys of the
enclosing map.
+ *
+ * The `_` branch is for a converter that stopped on a scalar inside a nested
+ * value. The loop advances until Jackson's nesting depth returns to this
map,
+ * which is the nested value's END_OBJECT / END_ARRAY (or EOF if truncated).
+ * A null parsing context is treated as "keep draining" so leftover tokens
+ * cannot be misread as the next map key.
+ */
+ private def skipRemainingValue(parser: JsonParser, mapDepth: Int): Unit = {
+ parser.getCurrentToken match {
+ case START_OBJECT | START_ARRAY =>
Review Comment:
Fixed: after `skipChildren()` on an inner `START_OBJECT` / `START_ARRAY`,
`skipRemainingValue` keeps advancing until Jackson's nesting depth returns to
the owning map. Tests `nestedObjectStartFailureQuery` (`STRUCT<x INT>` vs
nested object) and `nestedArrayStartFailureQuery` (`ARRAY<INT>` vs nested
array) assert `map_keys` is only `b `.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala:
##########
@@ -659,6 +721,63 @@ class JacksonParser(
}
}
+ /**
+ * CHAR/VARCHAR JSON maps apply assignment to object names, then
+ * `spark.sql.mapKeyDedupPolicy` to collisions created by that normalization.
+ * Exact repeated names stay historical last-wins.
+ */
+ private def convertConstrainedMap(
+ parser: JsonParser,
+ fieldConverter: ValueConverter,
+ keyType: DataType,
+ valueType: DataType,
+ mapDepth: Int): MapData = {
+ val lastEntries =
+ mutable.LinkedHashMap.empty[UTF8String, (UTF8String, Option[Any])]
+ var partialResultException: Option[Throwable] = None
+ var badMapException: Option[Throwable] = None
+
+ while (nextUntil(parser, JsonToken.END_OBJECT)) {
+ val rawKey = UTF8String.fromString(parser.currentName)
+ val value = try {
+ Some(fieldConverter.apply(parser))
+ } catch {
+ case err: PartialValueException if enablePartialResults =>
+ partialResultException =
partialResultException.orElse(Some(err.cause))
+ Some(err.partialResult)
+ case DuplicateMapKeyUtils(e) => throw e
+ case NonFatal(e) =>
Review Comment:
Fixed: `convertConstrainedMap` rethrows `SparkUpgradeException` before the
`NonFatal` drain. `SPARK-59722: JSON constrained maps preserve
SparkUpgradeException` covers `from_json` and the JSON datasource with
`MAP<CHAR(2), TIMESTAMP>` and `legacy.timeParserPolicy=exception`.
##########
docs/sql-migration-guide.md:
##########
@@ -28,6 +28,7 @@ license: |
## Upgrading from Spark SQL 4.3 to 4.4
+- Since Spark 4.4, `from_json` and the JSON datasource apply CHAR/VARCHAR
assignment to map keys when `spark.sql.charVarchar.standardSemantics.enabled`
is true. Distinct object names that become the same key after CHAR padding or
VARCHAR trailing-space trimming (`"a"` vs `"a "` as `CHAR(2)`) honor
`spark.sql.mapKeyDedupPolicy`. Exact repeated names stay last-wins. A
CHAR/VARCHAR map is always scanned to its closing brace so that policy can run,
even when `spark.sql.json.enablePartialResults` is false (ordinary STRING maps
still fail at the first bad value in that configuration). When both a
normalization collision and a conversion failure occur, `DUPLICATED_MAP_KEY` is
raised first; with `LAST_WIN`, PERMISSIVE mode keeps the partial map.
Independently, when `spark.sql.json.enablePartialResults` is true (the
default), a STRING-keyed JSON map whose nested value fails conversion omits
that entry and keeps later keys instead of nulling the whole map; a map whose
every entry fails become
s empty rather than null.
Review Comment:
The 4.4 note now distinguishes the two `convertOrdinaryMap` branches: a
drained failure with no partial value omits the pair (and an all-failed map is
empty), while `PartialValueException` retains the key with its typed partial
value.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala:
##########
@@ -811,6 +932,9 @@ class JacksonParser(
def handleFailure[R](elementStart: Option[JsonToken])(operation: => R): R
= {
try operation catch {
case e: SparkUpgradeException => fail(e)
+ case DuplicateMapKeyUtils(e) =>
Review Comment:
Added a multiline top-level-array datasource case with
`spark.sql.json.streamMultilineTopLevelArray=true`: colliding `MAP<CHAR(2),
INT>` raises `DUPLICATED_MAP_KEY` at collect, and LAST_WIN keeps `a ` -> 2 then
`b ` -> 3.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala:
##########
@@ -659,6 +721,63 @@ class JacksonParser(
}
}
+ /**
+ * CHAR/VARCHAR JSON maps apply assignment to object names, then
+ * `spark.sql.mapKeyDedupPolicy` to collisions created by that normalization.
+ * Exact repeated names stay historical last-wins.
+ */
+ private def convertConstrainedMap(
+ parser: JsonParser,
+ fieldConverter: ValueConverter,
+ keyType: DataType,
+ valueType: DataType,
+ mapDepth: Int): MapData = {
+ val lastEntries =
+ mutable.LinkedHashMap.empty[UTF8String, (UTF8String, Option[Any])]
+ var partialResultException: Option[Throwable] = None
+ var badMapException: Option[Throwable] = None
+
+ while (nextUntil(parser, JsonToken.END_OBJECT)) {
+ val rawKey = UTF8String.fromString(parser.currentName)
+ val value = try {
+ Some(fieldConverter.apply(parser))
+ } catch {
+ case err: PartialValueException if enablePartialResults =>
+ partialResultException =
partialResultException.orElse(Some(err.cause))
+ Some(err.partialResult)
+ case DuplicateMapKeyUtils(e) => throw e
+ case NonFatal(e) =>
+ badMapException = badMapException.orElse(Some(e))
+ skipRemainingValue(parser, mapDepth)
+ None
+ }
+ try {
+ val key = CharVarcharUtils.applyTextParseSemantics(rawKey, keyType)
+ // Re-insert so LAST_WIN follows the latest occurrence of this raw
name,
+ // matching StaxXmlParser.convertConstrainedMap.
+ lastEntries.remove(rawKey)
Review Comment:
`{"a":1,"a":"bad"}` yields an empty map and `{"a":"bad","a":2}` keeps `a `
-> 2, under both EXCEPTION and LAST_WIN. Successful exact-repeat last-wins and
distinct-name collision cases are unchanged.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
