HyukjinKwon commented on code in PR #59325:
URL: https://github.com/apache/spark/pull/59325#discussion_r4241666474
##########
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:
**Non-blocking (P2):** The last sentence here says that reads whose
CHAR/VARCHAR keys are handled read-side, "for example catalog tables", are
unchanged: keys are padded, `spark.sql.mapKeyDedupPolicy` is applied, and long
keys raise `EXCEED_LIMIT_LENGTH`. That is not what happens under standard
semantics. The sentence comes from my earlier comment, which had this wrong.
With `spark.sql.charVarchar.standardSemantics.enabled=true`,
`charVarcharFirstClassTypes` is true, so `replaceCharVarcharWithString` keeps
`CharType`. For a catalog JSON table, `SessionCatalog.getTableMetadata` and
`LogicalRelation.apply` keep `MAP<CHAR(3), INT>`,
`DataSourceStrategy.readDataSourceTable` passes it as the reader schema, and
`JsonFileFormat` builds `JacksonParser` over it. `makeMapKeyChecker` then
installs the new check:
```sql
CREATE TABLE t (m MAP<CHAR(3), INT>) USING json;
-- data rows: {"m":{"a":1}} and {"m":{"abcd":1}}
SELECT m FROM t;
-- PERMISSIVE: m is NULL for both rows
-- FAILFAST: UNSUPPORTED_JSON_CHAR_VARCHAR_MAP_KEY
```
Neither row gets the documented `'a '` or `EXCEED_LIMIT_LENGTH`. Keys that
pass the parser are then rebuilt by the read-side `MapFromArrays`, for catalog
tables and user-specified reader schemas alike. So `{"m":{"abc":1,"abc":2}}`
still raises `DUPLICATED_MAP_KEY` under the default policy, unlike `from_json`.
Could this sentence, and the matching PR-description paragraph, say that JSON
scans with CHAR/VARCHAR map keys run the same parser check and still apply
`mapKeyDedupPolicy`? If catalog reads should skip the parser check instead,
that would need a code change.
See **Shared repair plan 1** in the review body.
##########
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:
**Nit (P3):** "there is no view to capture" does not hold for file scans. A
persisted view can read a catalog JSON table, and under standard semantics that
table keeps `MAP<CHAR(n), _>` all the way to the scan. The view body (table
resolution plus `ApplyCharTypePadding`) is analyzed under the view's flag. But
`JsonFileFormat` and `JsonPartitionReaderFactory` build this parser with
`None`, so the key check reads the querying session's flag at execution:
```sql
-- created with the flag on
CREATE TABLE t (m MAP<CHAR(3), INT>) USING json; -- data row: {"m":{"a":1}}
CREATE VIEW v AS SELECT m FROM t;
SELECT * FROM v;
-- queried with the flag on: m is NULL
-- queried with the flag off: m = {'a ' -> 1}
```
In the flag-off session the parser installs no checker and accepts `'a'`,
and the view's read-side rewrite pads it. The same persisted view returns
different data depending on the caller. Other JSON scan options already follow
the session, so documenting this may be enough. Could this comment, and the
guide's datasource sentence, say that the file-scan check uses the executing
session's flag, rather than that no view can be involved?
**Verification:**
- **Behavior:** A persisted view over a JSON table with a MAP<CHAR(3), INT>
column, created under standard semantics, produces the documented results for
an off-width key when queried from flag-on and flag-off sessions.
- **Inspection:** The JacksonParser comment and the migration-guide sentence
describe the same session-bound file-scan behavior the test asserts.
##########
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:
**Nit (P3):** This says the row then becomes a bad record, `null` in
`PERMISSIVE` mode. In fact the rethrow leaves containment to the enclosing
converter. When the map is a struct field, `convertObject` absorbs it and nulls
only that field, which is what the new test expects: `Row(Row(null, 1))` for
`{"m":{"a":1},"tail":1}`. An enclosing array or map, or the root, propagates it
further. "(a trimmed value)" on the next line also has no counterpart in the
code. The balanced partial results that reach `PartialMapDataResultException`
come from nested struct, array, or map values throwing `PartialValueException`.
CHAR/VARCHAR value handling returns the padded or trimmed string without
throwing. Maybe something like: "Rethrow the real cause and let the enclosing
converter decide containment: a struct nulls just this field, while an array, a
map, or the root propagates it. Balanced partial results from nested struct,
array, or map values still flow through PartialMapDataResultException b
elow."
**Verification:**
- **Inspection:** The comment matches convertObject/convertArray handling
and the PartialValueException producers, consistent with the
sibling-preservation test expecting Row(Row(null, 1)).
##########
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:
**Nit (P3):** This default is evaluated wherever `JsonToStructs` is
constructed, and one caller constructs it at execution.
`BaseScriptTransformationExec.outputFieldWriters` is a lazy val. Its
non-standard complex-type branch (`case _: ArrayType | _: MapType | _:
StructType`) builds `JsonToStructs(attr.dataType,
ioschema.outputSerdeProps.toMap, Literal(null),
Some(conf.sessionLocalTimeZone))` without this argument, so it picks up
`SQLConf.get` at execution. That file says the mode is "Bound at parse into the
I/O schema ... Do not re-read SQLConf".
Take a no-SerDe `TRANSFORM ... AS (m MAP<CHAR(3), INT>)` parsed with
standard semantics off and `spark.sql.preserveCharVarcharTypeInfo` on. That
happens in a persisted view, or in a DataFrame whose conf changes before the
action. If it then executes with the flag on, it gets the CHAR(3) key check,
and a script row `{"a":1}` becomes a null map, although the PR description says
TRANSFORM is unchanged. Passing `charVarcharStandardSemantics =
standardCharVarcharSemantics` at that construction would keep TRANSFORM on the
mode it bound at parse.
**Verification:**
- **Regression:** A no-SerDe TRANSFORM with MAP<CHAR(3), INT> output, parsed
with standard semantics off and preserveCharVarcharTypeInfo on, then executed
with standard semantics on, keeps an off-width script output key instead of
nulling the map.
- **Compatibility:** TRANSFORM output under standard semantics (the standard
branch) is unchanged.
##########
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. A mismatched key turns
the row into a bad record: in `PERMISSIVE` mode it is dropped (the output is
`null`), while in `FAILFAST` mode parsing fails with
`UNSUPPORTED_JSON_CHAR_VARCHAR_MAP_KEY` (SQLSTATE `0A000`) as the cause of
`MALFORMED_RECORD_IN_PARSING`. Duplicate names are kept, exactly as for STRING
keys; `spark.sql.mapKeyDedupPolicy` is not applied.
Review Comment:
Thanks for rescoping this; the `from_json` part now reads correctly. I need
to correct my earlier comment, though. Under
`spark.sql.charVarchar.standardSemantics.enabled`, CHAR/VARCHAR are first-class
types, so `SessionCatalog.getTableMetadata` and `LogicalRelation.apply` keep
`CharType` map keys, and a catalog JSON table's reader schema also reaches
`makeMapKeyChecker`. Catalog-table reads are therefore not unchanged: an
off-width key becomes a bad record from the new parser check before any
read-side padding, and an over-long key fails with
`UNSUPPORTED_JSON_CHAR_VARCHAR_MAP_KEY`, not `EXCEED_LIMIT_LENGTH`. Keys that
pass the parser still go through the read-side `MapFromArrays` rebuild, for
catalog tables and user-specified reader schemas alike, so duplicates still hit
`mapKeyDedupPolicy` there. Could the last two sentences of the bullet (and the
PR description) say that, or should catalog reads skip the parser check?
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:4240217530","thread_id":"inline:4240217530","verdict_sha256":"f60c3d302b334b9fcd643ab2faf97690f335c375ab5bdf6b356e1dcb80985720"}
-->
##########
sql/core/src/test/scala/org/apache/spark/sql/CharVarcharTestSuite.scala:
##########
@@ -3072,6 +3094,229 @@ 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)")
+
+ // The DataFrame functions.from_json path constructs JsonToStructs the
same way and must
+ // behave identically: accepted key kept, rejected key null in
PERMISSIVE and surfaced in
+ // FAILFAST.
+ checkAnswer(
+ spark.range(1).select(
+ functions.from_json(
+ functions.lit("""{"ab ": 1}"""), "MAP<CHAR(3), INT>",
Map.empty[String, String])),
+ Row(Map("ab " -> 1)))
+ checkAnswer(
+ spark.range(1).select(
+ functions.from_json(
+ functions.lit("""{"a": 1}"""), "MAP<CHAR(3), INT>",
Map.empty[String, String])),
+ Row(null))
+ assertUnsupportedJsonMapKeyError(
+ spark.range(1).select(
+ functions.from_json(
+ functions.lit("""{"a": 1}"""),
+ "MAP<CHAR(3), INT>",
+ Map("mode" -> "FAILFAST"))).collect(),
+ key = "a",
+ dataTypeSql = "CHAR(3)")
+
+ // SPARK-60108: a rejected key must not corrupt a sibling field of an
enclosing struct.
+ // The map becomes null but the following `tail` field is still parsed,
and an inner name
+ // ("tail") is not mistaken for the outer field. This holds in both
partial-results modes.
+ Seq(true, false).foreach { partial =>
Review Comment:
Thanks, verified. The new object- and array-valued cases run in both
partial-results modes and assert that `tail` survives, so dropping the
`skipChildren()` after the rejected key would now fail them.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:4240217531","thread_id":"inline:4240217531","verdict_sha256":"f60c3d302b334b9fcd643ab2faf97690f335c375ab5bdf6b356e1dcb80985720"}
-->
--
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]
