srielau commented on code in PR #59325:
URL: https://github.com/apache/spark/pull/59325#discussion_r4238473321
##########
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:
I would like to keep `0A000`. The rejection is a static-semantics limitation
of the CHAR/VARCHAR map-key type (that type cannot represent an off-width
name), not a data-dependent runtime error, so the "unsupported feature" class
fits. You are right about the FAILFAST wrapping though: it surfaces as
`MALFORMED_RECORD_IN_PARSING.WITHOUT_SUGGESTION` (SQLSTATE `22023`) with this
error as the cause. I corrected that in the migration guide and the PR
description, and added a FAILFAST test that unwraps the cause.
##########
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:
Reworded. The message now states the rule and the remedy: "A CHAR(n) key
must be exactly n characters and a VARCHAR(n) key at most n characters; keys
are never padded or trimmed. Use a STRING map key to accept any name." The
echoed key is truncated to 128 characters to bound the message size.
##########
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:
Reworded for accuracy rather than removed - it sits alongside the existing
`charVarchar.standardSemantics` bullet, so readers enabling that flag will look
here. It now describes the actual behavior: in PERMISSIVE mode the bad record
becomes `null`; in FAILFAST mode it fails with
`UNSUPPORTED_JSON_CHAR_VARCHAR_MAP_KEY` (`0A000`) as the cause of
`MALFORMED_RECORD_IN_PARSING`; duplicate names are kept;
`spark.sql.mapKeyDedupPolicy` is not applied.
##########
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:
Added all of these: a bad key followed by a sibling (asserting the sibling
survives, in both partial-result modes), nested `MAP`/`ARRAY` of maps (valid
and failing), the multiLine top-level-array row count with a bad element, a
view created flag-on and queried flag-off, and a non-BMP key (one code point =
`CHAR(1)`, rejected by `CHAR(2)`).
##########
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:
Good point - `checkAnswer`'s map conversion would hide dropped duplicates.
The duplicate assertions now use `size(map_keys(...))` and `map_entries(...)`
so retention is checked directly instead of through a Scala `Map`.
##########
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:
Fixed. The flag-off cases now round-trip through `to_json(from_json(...))`
so they exercise the parser directly rather than the encoder, confirming
`{"a":1}` and `{"abcd":1}` are returned unchanged with the flag off.
--
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]
