dtenedor commented on code in PR #59325:
URL: https://github.com/apache/spark/pull/59325#discussion_r4244655501
##########
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:
Two sentences here still don't match the code:
- "the whole `from_json` result is `null` only when the map is the top-level
type": `convertArray` and an enclosing `convertMap` don't contain the failure,
so a map inside an array or another map nulls that whole container. For
example, `from_json('[{"abc":1},{"a":2}]', 'ARRAY<MAP<CHAR(3), INT>>')` returns
`NULL`. Wording like "the nearest enclosing struct field becomes `null`; if the
map is inside an array or another map, that whole array or map is `null`" would
match.
- "A JSON file scan with no persisted view resolves the flag from the
executing session" suggests scans under a persisted view behave differently,
but per the `JacksonParser` comment they also use the executing session's flag.
Something like "JSON file scans, including scans inside persisted views, read
the flag from the executing session" would be accurate.
The PR description has the same two sentences.
##########
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:
After merging master this is stale: with SPARK-60103, XML CHAR/VARCHAR map
keys are also length-checked without padding, and repeated names last-win with
`mapKeyDedupPolicy` ignored. The test comment at `CharVarcharTestSuite.scala`
L3452 ("contrast XML, which pads keys and applies the policy") has the same
problem. Maybe state the rule that now holds for both: CHAR/VARCHAR keys handle
duplicates the same way as STRING keys in the same 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:
Nit: nothing calls `assertDuplicateMapKey` or `assertDuplicateMapKeyError`
anymore, and master removed them, so these look like a merge leftover and can
be dropped.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala:
##########
@@ -619,25 +622,59 @@ class JacksonParser(
/**
* Parse an object as a Map, preserving all fields.
+ *
+ * JSON object names used as CHAR/VARCHAR keys are length-checked without
rewriting:
+ * 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.
+ * Repeated names that are binary-equal last-win for CHAR/VARCHAR keys
(collation is
+ * not consulted, matching how STRING keys are compared here); STRING keys
keep every
+ * pair. This differs from XML, which pads keys and then applies
mapKeyDedupPolicy.
Review Comment:
Thanks, keep-all matches STRING keys and the `size`/`map_entries` assertions
pin it. One follow-up after the merge: the scaladoc's comparison with XML is
stale since SPARK-60103 (inline comment).
##########
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:
Two gaps remain in the new tests:
- These nested shapes are only checked in FAILFAST. A PERMISSIVE assertion
here would pin the containment the migration guide describes; for example,
`from_json('[{"abc":1},{"a":2}]', 'ARRAY<MAP<CHAR(3), INT>>')` returns `NULL`,
dropping the valid first element.
- Nothing exercises the scan behavior the migration guide now documents: a
catalog JSON table with a `MAP<CHAR(n), _>` column rejecting an off-width key,
and a duplicate name on a scan raising `DUPLICATED_MAP_KEY` under the default
policy.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala:
##########
@@ -659,6 +696,33 @@ class JacksonParser(
}
}
+ /**
+ * A CHAR/VARCHAR map key under standard semantics is length-checked (not
padded or
+ * trimmed) and deduplicated last-win on exact names. This is the single
predicate both
+ * `convertMap` and `convertJsonMapKey` consult so the two cannot drift
apart.
+ */
+ private def isLengthCheckedKeyType(keyType: DataType): Boolean =
+ charVarcharStandardSemantics && (keyType match {
+ case _: CharType | _: VarcharType => true
+ case _ => false
+ })
+
+ private def convertJsonMapKey(rawName: String, keyType: DataType):
UTF8String = {
+ val key = UTF8String.fromString(rawName)
+ if (!isLengthCheckedKeyType(keyType)) {
+ key
+ } else {
+ keyType match {
+ case c: CharType =>
+ CharVarcharCodegenUtils.charTypeJsonMapKeyCheck(key, c.length)
+ case v: VarcharType =>
+ CharVarcharCodegenUtils.varcharTypeJsonMapKeyCheck(key, v.length)
+ case _ =>
+ key
Review Comment:
Thanks, verified.
##########
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:
Thanks, the new message reads much better, and the 128-character cap is
tested. Optional follow-up: `UNSUPPORTED_XML_CHAR_VARCHAR_MAP_KEY` still has
the old "cannot be padded or trimmed" wording and doesn't cap the key, so
aligning the two would keep them consistent.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryExecutionErrors.scala:
##########
@@ -2739,6 +2739,19 @@ private[sql] object QueryExecutionErrors extends
QueryErrorsBase with ExecutionE
)
}
+ // A SparkRuntimeException (not SparkSQLFeatureNotSupportedException)
despite the 0A000
+ // SQLSTATE: restricting CHAR/VARCHAR JSON map keys is a limitation, but the
error is
+ // raised while parsing a row and must flow through Jackson's throw/catch
parse-mode
+ // model (PERMISSIVE wraps it as a bad record, FAILFAST surfaces it).
Review Comment:
Fair enough. `UNSUPPORTED_XML_CHAR_VARCHAR_MAP_KEY` also uses `0A000`, so
keeping the two consistent makes sense, and the FAILFAST wording is accurate
now. Not blocking.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala:
##########
@@ -619,25 +622,59 @@ class JacksonParser(
/**
* Parse an object as a Map, preserving all fields.
+ *
+ * JSON object names used as CHAR/VARCHAR keys are length-checked without
rewriting:
+ * 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.
+ * Repeated names that are binary-equal last-win for CHAR/VARCHAR keys
(collation is
+ * not consulted, matching how STRING keys are compared here); STRING keys
keep every
+ * pair. This differs from XML, which pads keys and then applies
mapKeyDedupPolicy.
*/
private def convertMap(
parser: JsonParser,
- fieldConverter: ValueConverter): MapData = {
+ fieldConverter: ValueConverter,
+ keyType: DataType): MapData = {
val keys = ArrayBuffer.empty[UTF8String]
val values = ArrayBuffer.empty[Any]
var badRecordException: Option[Throwable] = None
+ // CHAR/VARCHAR keys last-win on exact (binary) duplicate names. This maps
a key to
+ // the index of its value in `values` for O(1) overwrite; it stays null
(and the
+ // buffers keep every pair, as STRING maps always have) when last-win does
not apply.
+ // It indexes `values` (not `keys`) so it stays valid even after the
NonFatal arm
+ // appends a dangling key, leaving the two buffers unbalanced.
+ val lastWinIndex: mutable.HashMap[UTF8String, Int] =
+ if (isLengthCheckedKeyType(keyType)) mutable.HashMap.empty[UTF8String,
Int] else null
while (nextUntil(parser, JsonToken.END_OBJECT)) {
- keys += UTF8String.fromString(parser.currentName)
+ val key = convertJsonMapKey(parser.currentName, keyType)
Review Comment:
Thanks, verified: the rejected-key branch now consumes the value, and the
new sibling, object/array-value, and streaming tests cover it. One related path
is still open: the checker also runs when the loop resumes inside an array
after an element failure, where `currentName` is `null`. I left an inline
comment on the new code.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala:
##########
@@ -107,6 +107,9 @@ class JacksonParser(
private val enablePartialResults = SQLConf.get.jsonEnablePartialResults
+ // CHAR/VARCHAR JSON map keys are length-checked without pad or trim under
this flag.
+ private val charVarcharStandardSemantics =
SQLConf.get.charVarcharStandardSemantics
Review Comment:
Thanks, verified for `from_json`: `JsonToStructs` captures the flag at
analysis, and the view test covers the created-on/queried-off direction.
Keeping file scans session-bound like other scan-time options seems reasonable;
I left a wording suggestion on the migration guide about scans under persisted
views.
##########
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, JSON object names used as `MAP<CHAR(n), _>` or `MAP<VARCHAR(n), _>` keys
in `from_json` and the JSON datasource are length-checked without padding or
trimming. A `CHAR(n)` key must already be exactly `n` characters, and a
`VARCHAR(n)` key must already be at most `n` characters. Mismatched keys fail
with `UNSUPPORTED_JSON_CHAR_VARCHAR_MAP_KEY` (SQLSTATE `0A000`) and follow the
JSON parse mode (`PERMISSIVE` or `FAILFAST`). Exact repeated member names
last-win; `spark.sql.mapKeyDedupPolicy` is not applied. With the flag off, CHAR
keys are still padded and VARCHAR overflow still uses `EXCEED_LIMIT_LENGTH`.
Review Comment:
Keeping it next to the other standard-semantics bullets makes sense. Two
sentences in the current text still don't match the code; I left an inline
comment on it.
##########
sql/catalyst/src/main/java/org/apache/spark/sql/catalyst/util/CharVarcharCodegenUtils.java:
##########
@@ -111,4 +113,28 @@ public static UTF8String charTypeReadSideCheck(UTF8String
inputStr, int limit) {
public static UTF8String varcharTypeReadSideCheck(UTF8String inputStr, int
limit) {
return varcharTypeWriteSideCheck(inputStr, limit);
}
+
+ /**
+ * JSON object names used as MAP<CHAR(n), _> keys cannot be padded or
trimmed.
+ * The name must already be exactly n characters.
+ */
+ public static UTF8String charTypeJsonMapKeyCheck(UTF8String inputStr, int
limit) {
+ if (inputStr.numChars() != limit) {
+ throw QueryExecutionErrors.unsupportedJsonCharVarcharMapKey(
+ inputStr, CharType.apply(limit));
Review Comment:
Thanks, verified, including the test for the collated type in the message.
##########
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:
This runs the checker on every loop iteration, including ones where the
parser isn't on a `FIELD_NAME`. That happens after a failure inside an array
value: `convertArray` only catches `PartialValueException`, so an element
failure escapes with the parser still inside the array, and this loop resumes
on the next array element. There `parser.currentName` is `null`,
`UTF8String.fromString(null)` returns `null`, and `checkCharJsonMapKey` calls
`numChars()` on it.
With standard semantics on, schema `m MAP<CHAR(1), ARRAY<INT>>, tail INT`,
PERMISSIVE (replayed with the same standalone Jackson harness, now mirroring
this version of `convertMap`):
- `{"m":{"k":["x",2]},"tail":1}` gives `{m: null, tail: null}`, and the
FAILFAST cause is a `NullPointerException`. With the flag off it gives `{m:
null, tail: 1}` with `CANNOT_PARSE_JSON_FIELD` as the cause.
- The same happens when the array holds CHAR-keyed maps and one of them has
a rejected key, e.g. `{"m":{"k":[{"a":1},{"abc":2}]},"tail":1}` with values
`ARRAY<MAP<CHAR(3), INT>>`.
- With `enableStreamingTopLevelArray`, that bad element followed by a good
one yields 4 rows instead of 2.
An overflowing `ARRAY<VARCHAR(n)>` element triggers it too, since
`EXCEED_LIMIT_LENGTH` isn't a partial result. Only checking while the parser is
on a field name fixes all three cases in the harness and leaves the existing
tests' scenarios unchanged:
```scala
val keyRejection = keyChecker match {
case Some(check) if parser.currentToken == JsonToken.FIELD_NAME =>
check(key)
case _ => None
}
```
A regression test with the first input in both partial-results modes would
cover it.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala:
##########
@@ -619,25 +622,59 @@ class JacksonParser(
/**
* Parse an object as a Map, preserving all fields.
+ *
+ * JSON object names used as CHAR/VARCHAR keys are length-checked without
rewriting:
+ * 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.
+ * Repeated names that are binary-equal last-win for CHAR/VARCHAR keys
(collation is
+ * not consulted, matching how STRING keys are compared here); STRING keys
keep every
+ * pair. This differs from XML, which pads keys and then applies
mapKeyDedupPolicy.
*/
private def convertMap(
parser: JsonParser,
- fieldConverter: ValueConverter): MapData = {
+ fieldConverter: ValueConverter,
+ keyType: DataType): MapData = {
val keys = ArrayBuffer.empty[UTF8String]
val values = ArrayBuffer.empty[Any]
var badRecordException: Option[Throwable] = None
+ // CHAR/VARCHAR keys last-win on exact (binary) duplicate names. This maps
a key to
+ // the index of its value in `values` for O(1) overwrite; it stays null
(and the
+ // buffers keep every pair, as STRING maps always have) when last-win does
not apply.
+ // It indexes `values` (not `keys`) so it stays valid even after the
NonFatal arm
+ // appends a dangling key, leaving the two buffers unbalanced.
+ val lastWinIndex: mutable.HashMap[UTF8String, Int] =
+ if (isLengthCheckedKeyType(keyType)) mutable.HashMap.empty[UTF8String,
Int] else null
Review Comment:
Agreed, moot now.
##########
sql/core/src/test/scala/org/apache/spark/sql/CharVarcharTestSuite.scala:
##########
@@ -3072,6 +3094,113 @@ 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") {
+ 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)))
+ checkAnswer(
+ sql("""SELECT map_entries(from_json('{"ab": 1, "ab ": 2}',
'MAP<VARCHAR(3), INT>'))"""),
+ Row(Seq(Row("ab", 1), Row("ab ", 2))))
+ checkAnswer(
+ sql("""SELECT size(from_json('{"abc": 1, "abc": 2}', 'MAP<CHAR(3),
INT>'))"""),
+ Row(1))
+ checkAnswer(
+ sql("""SELECT from_json('{"abc": 1, "abc": 2}', 'MAP<CHAR(3),
INT>')"""),
+ Row(Map("abc" -> 2)))
+ checkAnswer(
+ sql("""SELECT from_json('{"outer": {"xy ": 1}}',
+ | 'MAP<STRING, MAP<CHAR(3), INT>>')""".stripMargin),
+ Row(Map("outer" -> Map("xy " -> 1))))
+
+ // Too-short CHAR, CHAR overflow, and VARCHAR overflow are not
EXCEED_LIMIT_LENGTH.
+ 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)")
+
+ withTempPath { path =>
+ Seq("""{"m":{"ab ":1}}""", """{"m":{"a":1}}""").toDS()
+ .repartition(1)
+ .write.text(path.getCanonicalPath)
+ val schema = "m MAP<CHAR(3), INT>"
+ val permissive = spark.read.schema(schema).json(path.getCanonicalPath)
+ checkAnswer(permissive, Seq(Row(Map("ab " -> 1)), Row(null)))
+ val failFast = spark.read.option("mode", "FAILFAST").schema(schema)
+ .json(path.getCanonicalPath)
+ assertUnsupportedJsonMapKeyError(failFast.collect(), key = "a",
dataTypeSql = "CHAR(3)")
+ }
+
+ withSQLConf(
+ SQLConf.MAP_KEY_DEDUP_POLICY.key ->
SQLConf.MapKeyDedupPolicy.EXCEPTION.toString) {
+ checkAnswer(
+ sql("""SELECT from_json('{"abc": 1, "abc": 2}', 'MAP<VARCHAR(3),
INT>')"""),
+ Row(Map("abc" -> 2)))
+ }
+
+ // A last-win duplicate whose winning (second) value overflows nulls the
whole
+ // record and keeps EXCEED_LIMIT_LENGTH, matching the non-duplicate
case, instead
+ // of silently keeping the superseded first value. Holds for both
partial modes.
+ Seq(true, false).foreach { partial =>
+ withSQLConf(SQLConf.JSON_ENABLE_PARTIAL_RESULTS.key ->
partial.toString) {
+ checkAnswer(
+ sql("""SELECT from_json('{"abc": "x", "abc": "abcdef"}',
+ | 'MAP<CHAR(3), CHAR(2)>')""".stripMargin),
+ Row(null))
+ assertParseExceedLimit(
+ """SELECT from_json('{"abc": "x", "abc": "abcdef"}',
+ | 'MAP<CHAR(3), CHAR(2)>',
+ | map('mode', 'FAILFAST'))""".stripMargin,
+ expectedLimit = "2")
+ }
+ }
+
+ // Collation is not consulted for last-win: binary-distinct names are
both kept,
+ // unlike XML which pads keys and then applies the dedup policy.
+ withSQLConf(SQLConf.ALLOW_COLLATIONS_IN_MAP_KEYS.key -> "true") {
+ checkAnswer(
+ sql("""SELECT map_entries(from_json('{"ab": 1, "AB": 2}',
+ | 'MAP<VARCHAR(2) COLLATE UTF8_LCASE, INT>'))""".stripMargin),
+ Row(Seq(Row("ab", 1), Row("AB", 2))))
+ }
+ }
+
+ // Flag off is unchanged: write-side pad/trim still apply to keys, and the
new
+ // UNSUPPORTED_JSON_CHAR_VARCHAR_MAP_KEY check is not used.
+ withSQLConf(
+ SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "false",
+ SQLConf.PRESERVE_CHAR_VARCHAR_TYPE_INFO.key -> "true") {
+ checkAnswer(
+ sql("""SELECT from_json('{"a": 1}', 'MAP<CHAR(3), INT>')"""),
+ Row(Map("a " -> 1)))
Review Comment:
Thanks, verified: `to_json` now observes exactly what the parser produced.
##########
sql/core/src/test/scala/org/apache/spark/sql/CharVarcharTestSuite.scala:
##########
@@ -3072,6 +3094,113 @@ 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") {
+ 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)))
+ checkAnswer(
+ sql("""SELECT map_entries(from_json('{"ab": 1, "ab ": 2}',
'MAP<VARCHAR(3), INT>'))"""),
+ Row(Seq(Row("ab", 1), Row("ab ", 2))))
+ checkAnswer(
+ sql("""SELECT size(from_json('{"abc": 1, "abc": 2}', 'MAP<CHAR(3),
INT>'))"""),
+ Row(1))
+ checkAnswer(
+ sql("""SELECT from_json('{"abc": 1, "abc": 2}', 'MAP<CHAR(3),
INT>')"""),
+ Row(Map("abc" -> 2)))
+ checkAnswer(
+ sql("""SELECT from_json('{"outer": {"xy ": 1}}',
+ | 'MAP<STRING, MAP<CHAR(3), INT>>')""".stripMargin),
+ Row(Map("outer" -> Map("xy " -> 1))))
+
+ // Too-short CHAR, CHAR overflow, and VARCHAR overflow are not
EXCEED_LIMIT_LENGTH.
+ 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)")
+
+ withTempPath { path =>
+ Seq("""{"m":{"ab ":1}}""", """{"m":{"a":1}}""").toDS()
+ .repartition(1)
+ .write.text(path.getCanonicalPath)
+ val schema = "m MAP<CHAR(3), INT>"
+ val permissive = spark.read.schema(schema).json(path.getCanonicalPath)
+ checkAnswer(permissive, Seq(Row(Map("ab " -> 1)), Row(null)))
+ val failFast = spark.read.option("mode", "FAILFAST").schema(schema)
+ .json(path.getCanonicalPath)
+ assertUnsupportedJsonMapKeyError(failFast.collect(), key = "a",
dataTypeSql = "CHAR(3)")
+ }
+
+ withSQLConf(
+ SQLConf.MAP_KEY_DEDUP_POLICY.key ->
SQLConf.MapKeyDedupPolicy.EXCEPTION.toString) {
+ checkAnswer(
+ sql("""SELECT from_json('{"abc": 1, "abc": 2}', 'MAP<VARCHAR(3),
INT>')"""),
+ Row(Map("abc" -> 2)))
Review Comment:
Thanks, verified.
##########
sql/core/src/test/scala/org/apache/spark/sql/CharVarcharTestSuite.scala:
##########
@@ -3072,6 +3094,113 @@ class BasicCharVarcharTestSuite extends
SharedSparkSession {
}
}
+ test("SPARK-60108: JSON CHAR/VARCHAR map keys are not padded or trimmed") {
Review Comment:
Thanks for adding all of these. Two gaps remain; I noted them inline next to
the nested-container assertions.
--
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]
