srielau commented on code in PR #59325:
URL: https://github.com/apache/spark/pull/59325#discussion_r4244815380
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala:
##########
@@ -618,38 +634,66 @@ 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
+ }
Review Comment:
Great catch, and thanks for the harness and the exact fix. Applied in
0e894f626aa: the check now runs only while `parser.currentToken ==
JsonToken.FIELD_NAME`, so after a non-partial failure inside an array/map value
the loop no longer runs the check on a null `currentName`.
Regression test added for `{"m":{"k":["x",2]},"tail":1}` with `m
MAP<CHAR(1), ARRAY<INT>>, tail INT` in both partial-results modes. Since `k` is
a valid CHAR(1) the key check is a no-op, so the test asserts the result
matches the STRING-keyed baseline - which rules out both the NPE and the
extra-row/lost-sibling divergence. On this Spark path the STRING baseline nulls
the whole struct for the non-partial array failure, and CHAR(1) now matches it
exactly.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala:
##########
@@ -618,38 +634,66 @@ 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`.
Review Comment:
Done in 0e894f626aa. Reworded the scaladoc (and the test comment at the old
L3452) to state the rule that now holds for both formats after SPARK-60103:
CHAR/VARCHAR keys handle duplicates the same way STRING keys do in each format
- JSON keeps every pair, XML keeps the last.
##########
sql/core/src/test/scala/org/apache/spark/sql/CharVarcharTestSuite.scala:
##########
@@ -1066,6 +1088,26 @@ class BasicCharVarcharTestSuite extends
SharedSparkSession {
parameters = Map("key" -> s"'$key'", "dataType" ->
s""""$dataTypeSql""""))
}
+ private def assertDuplicateMapKey(query: String, expectedKey: String = "a
"): Unit = {
+ assertDuplicateMapKeyError(sql(query).collect(), expectedKey)
+ }
+
+ private def assertDuplicateMapKeyError(body: => Any, expectedKey: String =
"a "): Unit = {
+ val error = intercept[Exception] { body }
+ val duplicateError = Iterator.iterate[Throwable](error)(_.getCause)
+ .takeWhile(_ != null)
+ .collectFirst {
+ case e: SparkRuntimeException if e.getCondition ==
"DUPLICATED_MAP_KEY" => e
+ }
+ .getOrElse(fail("expected DUPLICATED_MAP_KEY cause", error))
+ checkError(
+ exception = duplicateError,
+ condition = "DUPLICATED_MAP_KEY",
+ parameters = Map(
+ "key" -> expectedKey,
+ "mapKeyDedupPolicy" -> "\"spark.sql.mapKeyDedupPolicy\""))
+ }
Review Comment:
Right - those were a merge leftover. diff3 split my
`assertDuplicateMapKey`/`assertDuplicateMapKeyError` while resolving the
conflict with your SPARK-60103 changes and I restored them, but nothing
references them anymore. Removed both in 0e894f626aa.
##########
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-level check runs for
any JSON scan whose reader schema carries a CHAR/VARCHAR map key, including
catalog JSON tables, since under standard semantics those keys stay C
HAR/VARCHAR all the way to the scan: an off-width key becomes a bad record
there too, rather than being padded or raising `EXCEED_LIMIT_LENGTH`. Unlike
`from_json`, keys that pass the check are then rebuilt read-side, so on the
scan path `spark.sql.mapKeyDedupPolicy` does apply and a duplicate name can
raise `DUPLICATED_MAP_KEY`. A JSON file scan with no persisted view resolves
the flag from the executing session.
Review Comment:
Fixed both sentences in 0e894f626aa (and the matching PR description):
- PERMISSIVE containment now reads "the nearest enclosing struct field
becomes `null` (its siblings are preserved), and if the map is inside an array
or another map, that whole array or map becomes `null`". The new array-of-map
test pins `from_json('[{"abc":1},{"a":2}]', 'ARRAY<MAP<CHAR(3), INT>>')`
returning `NULL`.
- The scan sentence now reads "JSON file scans, including scans inside
persisted views, read the flag from the executing session."
##########
sql/core/src/test/scala/org/apache/spark/sql/CharVarcharTestSuite.scala:
##########
@@ -3213,6 +3255,255 @@ class BasicCharVarcharTestSuite extends
SharedSparkSession {
}
}
+ test("SPARK-60108: JSON CHAR/VARCHAR map keys are not padded or trimmed") {
+ withSQLConf(SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "true") {
+ // Exact CHAR and within-limit VARCHAR names are accepted as-is: the
trailing space in
+ // "ab " is preserved for CHAR(3) and not trimmed for VARCHAR(3).
+ checkAnswer(
+ sql("""SELECT from_json('{"ab ": 1}', 'MAP<CHAR(3), INT>')"""),
+ Row(Map("ab " -> 1)))
+ checkAnswer(
+ sql("""SELECT from_json('{"ab": 1}', 'MAP<VARCHAR(3), INT>')"""),
+ Row(Map("ab" -> 1)))
+ // Binary-distinct names ("ab" vs "ab ") are both kept.
+ checkAnswer(
+ sql("""SELECT map_entries(from_json('{"ab": 1, "ab ": 2}',
'MAP<VARCHAR(3), INT>'))"""),
+ Row(Seq(Row("ab", 1), Row("ab ", 2))))
+ // Duplicate names are kept, exactly as for STRING keys;
mapKeyDedupPolicy is not applied.
+ // checkAnswer alone would collapse the duplicates when it builds a
Scala Map, so pin the
+ // behavior with size and map_entries.
+ checkAnswer(
+ sql("""SELECT size(from_json('{"abc": 1, "abc": 2}', 'MAP<CHAR(3),
INT>'))"""),
+ Row(2))
+ checkAnswer(
+ sql("""SELECT map_entries(from_json('{"abc": 1, "abc": 2}',
'MAP<CHAR(3), INT>'))"""),
+ Row(Seq(Row("abc", 1), Row("abc", 2))))
+ // Length counts characters (code points), not UTF-16 units or bytes:
one non-BMP code
+ // point is exactly CHAR(1) and is kept verbatim, and the same name is
rejected by CHAR(2).
+ // scalastyle:off nonascii (the \u escape is decoded before the nonascii
check runs)
+ val nonBmp = "\uD83D\uDE00" // U+1F600, one code point stored as a
UTF-16 surrogate pair
+ // scalastyle:on nonascii
+ checkAnswer(
+ sql(s"""SELECT map_keys(from_json('{"$nonBmp": 1}', 'MAP<CHAR(1),
INT>'))[0]"""),
+ Row(nonBmp))
+ checkAnswer(sql(s"""SELECT from_json('{"$nonBmp": 1}', 'MAP<CHAR(2),
INT>')"""), Row(null))
+
+ // Length-0 boundary: CHAR(0)/VARCHAR(0) accept only the empty-string
name, and any
+ // non-empty name is rejected (the empty key is still kept as a real map
entry).
+ checkAnswer(
+ sql("""SELECT from_json('{"": 1}', 'MAP<CHAR(0), INT>')"""),
+ Row(Map("" -> 1)))
+ checkAnswer(
+ sql("""SELECT from_json('{"": 1}', 'MAP<VARCHAR(0), INT>')"""),
+ Row(Map("" -> 1)))
+ checkAnswer(sql("""SELECT from_json('{"a": 1}', 'MAP<CHAR(0),
INT>')"""), Row(null))
+ checkAnswer(sql("""SELECT from_json('{"a": 1}', 'MAP<VARCHAR(0),
INT>')"""), Row(null))
+ assertUnsupportedJsonMapKey(
+ """SELECT from_json('{"a": 1}', 'MAP<CHAR(0), INT>', map('mode',
'FAILFAST'))""",
+ key = "a",
+ dataTypeSql = "CHAR(0)")
+ assertUnsupportedJsonMapKey(
+ """SELECT from_json('{"a": 1}', 'MAP<VARCHAR(0), INT>', map('mode',
'FAILFAST'))""",
+ key = "a",
+ dataTypeSql = "VARCHAR(0)")
+
+ // A key longer than the 128-character echo cap is truncated with a
trailing "..." in the
+ // error so a huge JSON object name does not bloat the message.
+ val longKey = "a" * 200
+ assertUnsupportedJsonMapKey(
+ s"""SELECT from_json('{"$longKey": 1}', 'MAP<CHAR(3), INT>',
map('mode', 'FAILFAST'))""",
+ key = "a" * 128 + "...",
+ dataTypeSql = "CHAR(3)")
+
+ // Valid keys inside nested containers.
+ checkAnswer(
+ sql("""SELECT from_json('{"outer": {"xy ": 1}}',
+ | 'MAP<STRING, MAP<CHAR(3), INT>>')""".stripMargin),
+ Row(Map("outer" -> Map("xy " -> 1))))
+ checkAnswer(
+ sql("""SELECT from_json('[{"ab ": 1}]', 'ARRAY<MAP<CHAR(3),
INT>>')"""),
+ Row(Seq(Map("ab " -> 1))))
+
+ // A rejected key makes the row a bad record: null in PERMISSIVE (not
EXCEED_LIMIT_LENGTH),
+ // and in FAILFAST the UNSUPPORTED_JSON_CHAR_VARCHAR_MAP_KEY cause is
surfaced.
+ checkAnswer(sql("""SELECT from_json('{"a": 1}', 'MAP<CHAR(3),
INT>')"""), Row(null))
+ checkAnswer(sql("""SELECT from_json('{"abcd": 1}', 'MAP<CHAR(3),
INT>')"""), Row(null))
+ checkAnswer(sql("""SELECT from_json('{"abcd": 1}', 'MAP<VARCHAR(3),
INT>')"""), Row(null))
+ checkAnswer(sql("""SELECT from_json('{"ab ": 1}', 'MAP<VARCHAR(2),
INT>')"""), Row(null))
+ assertUnsupportedJsonMapKey(
+ """SELECT from_json('{"a": 1}', 'MAP<CHAR(3), INT>', map('mode',
'FAILFAST'))""",
+ key = "a",
+ dataTypeSql = "CHAR(3)")
+ assertUnsupportedJsonMapKey(
+ """SELECT from_json('{"abcd": 1}', 'MAP<CHAR(3), INT>', map('mode',
'FAILFAST'))""",
+ key = "abcd",
+ dataTypeSql = "CHAR(3)")
+ assertUnsupportedJsonMapKey(
+ """SELECT from_json('{"abcd": 1}', 'MAP<VARCHAR(3), INT>', map('mode',
'FAILFAST'))""",
+ key = "abcd",
+ dataTypeSql = "VARCHAR(3)")
+ // A rejected key inside a nested container is surfaced too (not
swallowed).
+ assertUnsupportedJsonMapKey(
+ """SELECT from_json('{"outer": {"a": 1}}',
+ | 'MAP<STRING, MAP<CHAR(3), INT>>', map('mode',
'FAILFAST'))""".stripMargin,
+ key = "a",
+ dataTypeSql = "CHAR(3)")
+ assertUnsupportedJsonMapKey(
+ """SELECT from_json('[{"a": 1}]',
+ | 'ARRAY<MAP<CHAR(3), INT>>', map('mode',
'FAILFAST'))""".stripMargin,
+ key = "a",
+ dataTypeSql = "CHAR(3)")
Review Comment:
Both added in 0e894f626aa:
- A PERMISSIVE assertion for `from_json('[{"abc":1},{"a":2}]',
'ARRAY<MAP<CHAR(3), INT>>')` returning `NULL` (dropping the valid first
element).
- A catalog JSON table scan: a `MAP<CHAR(3), INT>` column nulls the row with
an off-width key under the default PERMISSIVE mode, and a duplicate name
(`{"abc":1,"abc":2}`) raises `DUPLICATED_MAP_KEY` on the read-side rebuild
under the default EXCEPTION policy. This confirms the migration-guide claim end
to end.
##########
common/utils/src/main/resources/error/error-conditions.json:
##########
@@ -9605,6 +9605,13 @@
],
"sqlState" : "0A000"
},
+ "UNSUPPORTED_JSON_CHAR_VARCHAR_MAP_KEY" : {
+ "message" : [
+ "JSON map keys of CHAR or VARCHAR cannot be padded or trimmed.",
+ "The key <key> is not valid for type <dataType>."
Review Comment:
Good point on consistency. I've left `UNSUPPORTED_XML_CHAR_VARCHAR_MAP_KEY`
alone in this PR to keep it scoped to SPARK-60108 - its message and key cap are
SPARK-60103's surface, which just merged. Happy to align both messages and add
the 128-character cap to the XML path in a small follow-up (or here if you'd
prefer). Flagging it so it isn't lost.
--
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]
