sarutak commented on code in PR #59167:
URL: https://github.com/apache/spark/pull/59167#discussion_r4155497313
##########
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:
`badMapException.foreach(throw _)` throws the **raw** exception (not wrapped
in `PartialMapDataResultException`). Tracing the raw exception through `parse`:
- `parse`'s catch matches it under the broad `_: RuntimeException | ... | _:
PartialMapDataResultException` case and calls `badRecord(e, ...)`.
- In `badRecord`, there is a dedicated `case
PartialMapDataResultException(mapData, cause) => BadRecordException(..., () =>
Array(InternalRow(mapData)), ...)` branch that **restores the partial map**. A
raw exception instead falls to the final `case e => BadRecordException(..., ()
=> Array.empty, e)`, i.e. **no partial result**.
So for a CHAR/VARCHAR map where one entry's value fails conversion,
PERMISSIVE mode yields an empty/`_corrupt_record` row, whereas the ordinary
STRING-map path (`convertOrdinaryMap`, which wraps in
`PartialMapDataResultException`) would preserve the partial map. This is a
silent, user-visible behavioral asymmetry driven purely by the key type.
Suggestion: route `badMapException` through `PartialMapDataResultException`
as well, so it lands on the existing `badRecord` branch, e.g.:
```scala
val firstError = badMapException.orElse(partialResultException)
if (firstError.isEmpty) mapData
else throw PartialMapDataResultException(mapData, firstError.get)
```
If the intent is that a value-conversion failure must be strictly
unrecoverable here (unlike STRING maps), please state that intent explicitly
and add a test fixing the resulting exception type under PERMISSIVE.
##########
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:
Because dedup runs first, a `DUPLICATED_MAP_KEY` is surfaced in preference
to a value-conversion error, and also in preference to a key
`EXCEED_LIMIT_LENGTH` raised by `applyTextParseSemantics` (which is captured
into `badMapException` via the `catch NonFatal` around `lastEntries.update`).
From a user's point of view, a per-key constraint violation (key too long) or a
broken value may be the more fundamental cause, so leading with
`DUPLICATED_MAP_KEY` can obscure it.
This may well be intentional ("normalize all keys before deciding
collisions"), but the precedence should be made explicit in a code comment and
pinned by tests for the combined cases (e.g. "key normalization collision" +
"value type mismatch", and "oversize key" + "collision").
##########
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:
For a CHAR/VARCHAR key, errors are drained regardless of
`enablePartialResults`, whereas an ordinary STRING map with
`enablePartialResults=false` propagates the first `NonFatal` immediately. So
the same malformed input throws a different exception, depending solely on
whether the key type is CHAR/VARCHAR vs STRING. The failure also surfaces at a
different time, after a full scan and dedup rather than at the first bad entry.
This is understandable, but it is a user-visible behavior change. Please:
- add a code comment on the intent, and
- document it in the SQL migration guide, and
- add a test asserting the final exception for "`enablePartialResults=false`
+ CHAR map + value-conversion failure" (assert the thrown type, not just that
it throws, so the test isn't vacuous).
##########
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:
For a scalar value failure, the current token sits at the scalar at `depth
== mapDepth`, so the loop body never runs (fine). The loop only does work when
a converter stopped on a scalar *inside* a nested value. In that case the loop
advances until depth returns to `mapDepth`; the stop happens when the nested
container's `END_*` is read. This relies on the implicit invariant that "when
depth first equals `mapDepth` again, the current token is an `END_OBJECT`/
`END_ARRAY`." If that invariant were ever violated (depth observed `==
mapDepth` while sitting on a sibling token), the outer `while (nextUntil(...
END_OBJECT))` would double-advance and skip/mis-align a sibling entry, a silent
corruption.
Also, `getParsingContext == null` is used as a loop-exit guard; if that
happened mid-value (e.g. a truncated/malformed JSON), draining would stop early
and leftover tokens could be mis-read as the next key.
Suggestions:
- Add a comment (or a debug `assert`) documenting that on exit the parser
sits
at `depth == mapDepth` on an `END_*` token.
- Add a negative test with a truncated/malformed nested value that actually
exercises this `_` branch, asserting key count/contents (not just the final
state) so "inner field does not leak as an outer key" is verified.
--
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]
