srielau commented on code in PR #59325:
URL: https://github.com/apache/spark/pull/59325#discussion_r4244122922
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/jsonExpressions.scala:
##########
@@ -1873,7 +1873,14 @@ case class JsonToStructs(
options: Map[String, String],
child: Expression,
timeZoneId: Option[String] = None,
- variantAllowDuplicateKeys: Boolean =
SQLConf.get.getConf(SQLConf.VARIANT_ALLOW_DUPLICATE_KEYS))
+ variantAllowDuplicateKeys: Boolean =
SQLConf.get.getConf(SQLConf.VARIANT_ALLOW_DUPLICATE_KEYS),
+ // `spark.sql.charVarchar.standardSemantics.enabled` has PERSISTED
binding, so it is captured
+ // here at analysis time (as the default arg, like
`variantAllowDuplicateKeys`) and then
+ // threaded all the way into the parser via
`charVarcharStandardSemanticsOverride`. This keeps
+ // a view's CHAR/VARCHAR map-key semantics tied to its creation-time flag
rather than the
+ // caller's session setting. (`variantAllowDuplicateKeys` is only
captured, not threaded: the
+ // parser still reads that one live from `SQLConf.get`.)
+ charVarcharStandardSemantics: Boolean =
SQLConf.get.charVarcharStandardSemantics)
Review Comment:
Good catch - this is a real gap. `outputFieldWriters` is a lazy val
evaluated at execution, so the `JsonToStructs` default would re-read `SQLConf`
there and a CHAR/VARCHAR key reaching the non-standard fallback could flip
behavior per execution-time session. Fixed in e242ae99615 by passing
`charVarcharStandardSemantics = standardCharVarcharSemantics` (the parse-bound
value from the I/O schema) at both construction sites, so TRANSFORM keeps its
parse-time mode.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala:
##########
@@ -51,7 +51,13 @@ class JacksonParser(
schema: DataType,
val options: JSONOptions,
allowArrayAsStructs: Boolean,
- filters: Seq[Filter] = Seq.empty) extends Logging {
+ filters: Seq[Filter] = Seq.empty,
+ // `spark.sql.charVarchar.standardSemantics.enabled` has PERSISTED
binding, so for `from_json`
+ // it must be captured when the expression is analyzed (see
`JsonToStructs`) rather than read
+ // live here, otherwise a view created under the flag would skip the
CHAR/VARCHAR map-key check
+ // when queried from a session with the flag off. `None` falls back to the
live conf, which is
+ // correct for the file-based JSON data source where there is no view to
capture.
Review Comment:
You're right - "there is no view to capture" was wrong (a persisted view can
read a catalog JSON table). Reworded in e242ae99615: file-based JSON scans
build the parser with `None`, so the key check follows the executing session's
flag like other scan-time options, and a scan under a persisted view resolves
it the same way.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala:
##########
@@ -618,38 +633,65 @@ class JacksonParser(
}
/**
- * Parse an object as a Map, preserving all fields.
+ * Parse an object as a Map.
+ *
+ * JSON object names used as CHAR/VARCHAR keys are length-checked without
rewriting (see
+ * `makeMapKeyChecker`): 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, and duplicate names are kept, exactly as for STRING keys. This
differs from XML,
+ * which pads keys and then applies `mapKeyDedupPolicy`.
+ *
+ * A rejected key is treated like a failed value, regardless of
`enablePartialResults`: the
+ * cause is recorded and the parser steps past the value so the loop still
consumes the map's
+ * END_OBJECT, leaving the key unpaired. The check must not surface its
result while the parser
+ * is still on the FIELD_NAME, or an enclosing struct would misread the
map's remaining entries
+ * as its own sibling fields (SPARK-60108). STRING keys have no checker and
take the fast path.
*/
private def convertMap(
parser: JsonParser,
- fieldConverter: ValueConverter): MapData = {
+ fieldConverter: ValueConverter,
+ keyChecker: Option[UTF8String => Option[Throwable]]): MapData = {
val keys = ArrayBuffer.empty[UTF8String]
val values = ArrayBuffer.empty[Any]
var badRecordException: Option[Throwable] = None
while (nextUntil(parser, JsonToken.END_OBJECT)) {
- keys += UTF8String.fromString(parser.currentName)
- try {
- values += fieldConverter.apply(parser)
- } catch {
- case err: PartialValueException if enablePartialResults =>
- badRecordException = badRecordException.orElse(Some(err.cause))
- values += err.partialResult
- case NonFatal(e) if enablePartialResults =>
+ val key = UTF8String.fromString(parser.currentName)
+ keys += key
+ // Avoid allocating a closure per entry on the common no-checker path.
+ val keyRejection = keyChecker match {
+ case Some(check) => check(key)
+ case None => None
+ }
+ keyRejection match {
+ case Some(e) =>
badRecordException = badRecordException.orElse(Some(e))
+ parser.nextToken() // step from FIELD_NAME onto the value
parser.skipChildren()
+ case None =>
+ try {
+ values += fieldConverter.apply(parser)
+ } catch {
+ case err: PartialValueException if enablePartialResults =>
+ badRecordException = badRecordException.orElse(Some(err.cause))
+ values += err.partialResult
+ case NonFatal(e) if enablePartialResults =>
+ badRecordException = badRecordException.orElse(Some(e))
+ parser.skipChildren()
+ }
}
}
- // Value conversion can fail after the key is recorded. Do not build
MapData from
- // unpaired buffers: ArrayBasedMapData would throw a cardinality error and
hide
- // the original conversion failure (for example EXCEED_LIMIT_LENGTH).
+ // A rejected key or a failed value leaves the key recorded without a
value. Rethrow the
+ // real cause instead of building an unbalanced ArrayBasedMapData, whose
cardinality
+ // `require` would otherwise mask it. The row then becomes a bad record
(null in PERMISSIVE,
Review Comment:
Agreed, the old wording was misleading. Reworded in e242ae99615 to your
suggestion: rethrow the real cause and let the enclosing converter decide
containment (a struct nulls just this field; an array, a map, or the root
propagates it). Dropped the "(a trimmed value)" aside - the balanced partial
results that flow through `PartialMapDataResultException` come from nested
struct/array/map values, not from CHAR/VARCHAR key handling.
##########
docs/sql-migration-guide.md:
##########
@@ -28,6 +28,7 @@ license: |
## Upgrading from Spark SQL 4.3 to 4.4
+- Since Spark 4.4, when `spark.sql.charVarchar.standardSemantics.enabled` is
true, `from_json` length-checks JSON object names used as `MAP<CHAR(n), _>` or
`MAP<VARCHAR(n), _>` keys without padding or trimming them. A `CHAR(n)` key
must already be exactly `n` characters, and a `VARCHAR(n)` key must already be
at most `n` characters; duplicate names are kept, exactly as for STRING keys,
and `spark.sql.mapKeyDedupPolicy` is not applied. A key that fails the check
makes the enclosing map a bad record: in `PERMISSIVE` mode the map is set to
`null` while sibling fields are preserved (the whole `from_json` result is
`null` only when the map is the top-level type), and in `FAILFAST` mode parsing
fails with `UNSUPPORTED_JSON_CHAR_VARCHAR_MAP_KEY` (SQLSTATE `0A000`) as the
cause of `MALFORMED_RECORD_IN_PARSING`. The same parser check runs when reading
JSON files with a user-specified CHAR/VARCHAR reader schema. Reads whose
CHAR/VARCHAR keys are handled read-side instead (for example catalog
tables) are unchanged: there keys are padded, `spark.sql.mapKeyDedupPolicy` is
applied, and an over-long key raises `EXCEED_LIMIT_LENGTH`.
Review Comment:
Thanks, and thanks for the self-correction - you're right on both counts.
Under standard semantics CHAR/VARCHAR stay first-class all the way to the scan,
so a catalog JSON table with a `MAP<CHAR(n), _>` key runs the same parser-level
check: an off-width key is a bad record there too, not padded and not
`EXCEED_LIMIT_LENGTH`. The read-side difference is only for keys that pass: the
scan path rebuilds the map (so `mapKeyDedupPolicy` applies and a duplicate can
raise `DUPLICATED_MAP_KEY`), whereas `from_json` keeps duplicates. Rewrote the
migration-guide sentence accordingly in e242ae99615 and will update the PR
description to match.
--
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]
