srielau commented on code in PR #59167:
URL: https://github.com/apache/spark/pull/59167#discussion_r4157687418
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala:
##########
@@ -659,6 +708,59 @@ 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)
+ lastEntries.update(rawKey, (key, value))
+ } catch {
+ case NonFatal(e) =>
+ badMapException = badMapException.orElse(Some(e))
+ }
+ }
+
+ val mapData = DuplicateMapKeyUtils.buildConstrainedMap(
+ lastEntries, keyType, valueType)
+
+ // Ordinary value or key conversion failures invalidate the whole map.
Delay throwing
+ // until the closing brace has been consumed and constrained-key dedup has
run.
+ badMapException.foreach(throw _)
Review Comment:
Good catch. `badMapException` is now wrapped in
`PartialMapDataResultException` together with `partialResultException`, so
PERMISSIVE restores the drained map instead of dropping it. FAILFAST still
surfaces `MALFORMED_RECORD_IN_PARSING.WITHOUT_SUGGESTION` /
`EXCEED_LIMIT_LENGTH`. LAST_WIN tests now expect the partial maps (empty map
for VARCHAR overflow, last-wins after a stored value error, and so on).
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala:
##########
@@ -659,6 +708,59 @@ 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)
+ lastEntries.update(rawKey, (key, value))
+ } catch {
+ case NonFatal(e) =>
+ badMapException = badMapException.orElse(Some(e))
+ }
+ }
+
+ val mapData = DuplicateMapKeyUtils.buildConstrainedMap(
Review Comment:
This is intentional: the object is fully drained, then `mapKeyDedupPolicy`
runs, then any stored conversion error is rethrown. A comment on
`convertConstrainedMap` now states that `DUPLICATED_MAP_KEY` is preferred over
`EXCEED_LIMIT_LENGTH` and value-conversion errors. Combined cases are covered
by `overflowAfterCollisionQuery` (`a`/`a ` + oversize `abc`) and
`badValueBeforeDuplicateQuery` / `malformedValueBeforeDuplicateQuery` (value
failure + CHAR padding collision). EXCEPTION still raises `DUPLICATED_MAP_KEY`;
LAST_WIN then reports the stored failure (PERMISSIVE keeps the partial map).
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala:
##########
@@ -617,12 +620,58 @@ 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.
+ */
+ private def skipRemainingValue(parser: JsonParser, mapDepth: Int): Unit = {
+ parser.getCurrentToken match {
+ case START_OBJECT | START_ARRAY =>
+ parser.skipChildren()
+ case END_OBJECT | END_ARRAY | null =>
+ // The value converter already consumed through its boundary.
+ case _ =>
+ // Still inside a nested value: skip until Jackson returns to this map.
+ var token = parser.getCurrentToken
+ while (token != null &&
+ parser.getParsingContext != null &&
+ parser.getParsingContext.getNestingDepth > mapDepth) {
+ token = parser.nextToken()
+ }
+ }
+ }
+
/**
* Parse an object as a Map, preserving all fields.
*/
private def convertMap(
parser: JsonParser,
- fieldConverter: ValueConverter): MapData = {
+ fieldConverter: ValueConverter,
+ keyType: DataType,
+ valueType: DataType): MapData = {
+ val constrainedKeys =
+ keyType.isInstanceOf[CharType] || keyType.isInstanceOf[VarcharType]
+ // CHAR/VARCHAR maps always drain to END_OBJECT so mapKeyDedupPolicy is
applied
+ // before a stored length error. Partial results still control STRING maps.
+ val drainErrors = constrainedKeys || enablePartialResults
Review Comment:
Also intentional: CHAR/VARCHAR maps have to reach `END_OBJECT` so
`mapKeyDedupPolicy` can run and leftover nested `FIELD_NAME`s cannot leak as
outer keys. `convertMap` now comments that `jsonEnablePartialResults=false`
still drains these maps, unlike STRING maps. The 4.4 SQL migration guide
documents the same, and `SPARK-59722: JSON CHAR map value failures preserve
parser position and error class` asserts
`MALFORMED_RECORD_IN_PARSING.WITHOUT_SUGGESTION` under FAILFAST with
`enablePartialResults` both true and false.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala:
##########
@@ -617,12 +620,58 @@ 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.
+ */
+ private def skipRemainingValue(parser: JsonParser, mapDepth: Int): Unit = {
+ parser.getCurrentToken match {
+ case START_OBJECT | START_ARRAY =>
+ parser.skipChildren()
+ case END_OBJECT | END_ARRAY | null =>
+ // The value converter already consumed through its boundary.
+ case _ =>
Review Comment:
Documented the `_` branch invariant: the loop stops on the nested value's
`END_*` (or EOF if truncated). A null parsing context now keeps draining
instead of exiting early, so leftover tokens cannot be read as the next key.
The new nested-array test uses `{"a":[1,"bad",99],"b":[3]}` as `MAP<CHAR(2),
ARRAY<INT>>` and asserts `map_keys` is only `b ` (inner tokens do not leak)
with `enablePartialResults` both true and false.
--
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]
