cloud-fan commented on code in PR #59167:
URL: https://github.com/apache/spark/pull/59167#discussion_r4185886769
##########
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:
**Non-blocking (P2):** When JSON partial results are disabled, this still
emits the inner constrained map as a partial carrier. An enclosing array or
ordinary map does not consume that `PartialMapDataResultException`, but
`badRecord` serializes its `MapData` as the root value. An `ARRAY<MAP<CHAR(2),
INT>>` root can then fail in `getArray(0)`, while an outer MAP root can expose
the inner map at the wrong nesting level instead of reporting the original
conversion error.
See **Shared repair plan 1** in the review body.
##########
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:
**Non-blocking (P2):** For a failure positioned on an inner `START_OBJECT`
or `START_ARRAY`, `skipChildren()` stops at that child's matching end token,
but the owning map value may still be open. Returning here lets the next
enclosing-map iteration consume tail tokens from the failed value as if they
were new entries, which can create phantom keys, retain the wrong value
fragment, or replace the original error with a secondary one.
See **Shared repair plan 1** in the review body.
##########
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:
**Nit (P3):** This sentence is broader than the implemented partial-result
boundary. `convertOrdinaryMap` retains both the key and `err.partialResult`
when a nested converter throws `PartialValueException`; for example, a failed
field in a `MAP<STRING, STRUCT<...>>` can leave a non-empty partial map. Users
therefore cannot rely on every failed nested value being omitted or every
all-failed map becoming empty.
**Verification:**
- **Inspection:** The published wording agrees with both ordinary-map catch
branches: retained key plus partial value for PartialValueException and pair
omission for a drained failure with no partial value.
##########
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:
**Nit (P3):** The remove/update sequence replaces an exact raw name's
normalized key and optional value as one state transition. Current assertions
cover exact repeats when both values convert and failed values on distinct
colliding names, but not `success -> failure` or `failure -> success` for the
same raw name. A regression could therefore resurrect an earlier successful
occurrence or discard a later successful one without changing an existing
signal.
**Verification:**
- **Regression:** Assert both success-then-failure and failure-then-success
exact-repeat orders for a constrained JSON map and compare the resulting
partial map contents.
- **Compatibility:** Keep existing successful exact-repeat last-wins and
distinct-name normalized-collision expectations unchanged.
##########
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:
**Non-blocking (P2):** `SparkUpgradeException` is `NonFatal`, so it reaches
this branch and is later wrapped in `PartialMapDataResultException`. The
record-level catches preserve only a raw upgrade exception; for a
CHAR/VARCHAR-keyed timestamp map, PERMISSIVE can therefore return a partial map
and FAILFAST can report malformed JSON instead of surfacing the required
compatibility warning.
See **Shared repair plan 1** in the review body.
##########
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:
**Nit (P3):** This new failure branch has no constrained-key test through
the streamed multiline top-level-array entry point. The added data-source case
reads a line-delimited object and does not reach this handler, so removing or
reordering this escape could make `DUPLICATED_MAP_KEY` look like a malformed
array element without any changed test detecting it.
**Verification:**
- **Regression:** Parse a multiline top-level array through the JSON data
source with a colliding constrained map and assert the duplicate-policy error
at collection time.
- **Compatibility:** Confirm the same fixture retains the documented
normalized last value under LAST_WIN if that policy is exercised.
--
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]
