HyukjinKwon commented on code in PR #59325:
URL: https://github.com/apache/spark/pull/59325#discussion_r4246469459
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/BaseScriptTransformationExec.scala:
##########
@@ -307,7 +310,11 @@ trait BaseScriptTransformationExec extends UnaryExecNode {
}
case _: ArrayType | _: MapType | _: StructType =>
val complexTypeFactory = JsonToStructs(attr.dataType,
- ioschema.outputSerdeProps.toMap, Literal(null),
Some(conf.sessionLocalTimeZone))
+ ioschema.outputSerdeProps.toMap, Literal(null),
Some(conf.sessionLocalTimeZone),
+ // Bind the parse-time flag so a CHAR/VARCHAR map key reaching this
fallback (standard
+ // semantics off, preserveCharVarcharTypeInfo on) keeps the
TRANSFORM-time behavior
+ // rather than picking up the executing session's flag at this
lazy-val execution.
+ charVarcharStandardSemantics = standardCharVarcharSemantics)
Review Comment:
**Nit (P3):** This argument is the only thing keeping a CHAR/VARCHAR-keyed
map output on the TRANSFORM's parse-time mode. `outputFieldWriters` is
evaluated at execution, and with standard semantics off only scalar
CHAR/VARCHAR is rejected, so a `MAP<CHAR(3), INT>` output planned with standard
semantics off and `spark.sql.preserveCharVarcharTypeInfo` on lands in this
branch. If the argument is dropped, `JsonToStructs` falls back to
`SQLConf.get`. Executed with standard semantics on, it then installs the
`CHAR(3)` key check and turns script output `{"a":1}` into a null map. Nothing
would catch that today: the `SPARK-59683` conf-change test in
`BaseScriptTransformationSuite` only uses scalar `CharType(4)` outputs, which
never reach this branch, and the new `SPARK-60108` test doesn't run TRANSFORM.
A `MAP<CHAR(3), INT>` output case beside that conf-change test would pin it.
This only concerns the no-SerDe TRANSFORM fallback; `from_json` and JSON scans
are unaffected.
**Verification:**
- **Regression:** A TRANSFORM conf-change case with a CHAR-keyed map output
keeps the parse-bound mode and fails when the fallback binding is removed.
##########
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, verified. The bullet and the PR description now say a catalog JSON
table's `MAP<CHAR(n), _>` key runs the same parser check, and that keys passing
it are rebuilt read-side with `spark.sql.mapKeyDedupPolicy`. The new
catalog-table test pins both the nulled off-width key and `DUPLICATED_MAP_KEY`.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:4241666474","thread_id":"inline:4241666474","verdict_sha256":"a6653c37cc778764bc1830228fe936f967f75b2f598caa12069aef5cbf91c002"}
-->
##########
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:
Thanks, verified. The rethrow comment now leaves containment to the
enclosing converter and no longer mentions a trimmed value.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:4241666485","thread_id":"inline:4241666485","verdict_sha256":"a6653c37cc778764bc1830228fe936f967f75b2f598caa12069aef5cbf91c002"}
-->
##########
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:
Thanks, verified. The constructor comment and the migration guide now both
say JSON file scans, including scans under persisted views, read the flag from
the executing session.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:4241666489","thread_id":"inline:4241666489","verdict_sha256":"a6653c37cc778764bc1830228fe936f967f75b2f598caa12069aef5cbf91c002"}
-->
##########
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:
Thanks, the binding is in place at both `JsonToStructs` construction sites.
One gap remains: nothing fails if the fallback-branch argument is dropped,
since the `SPARK-59683` conf-change test in `BaseScriptTransformationSuite`
only uses a scalar `CHAR(4)` output. A `MAP<CHAR(3), INT>` case there (plan
built with standard semantics off and `preserveCharVarcharTypeInfo` on,
executed with standard semantics on, script output `{"a":1}` keeping key `a`)
would pin it.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:4241666493","thread_id":"inline:4241666493","verdict_sha256":"a6653c37cc778764bc1830228fe936f967f75b2f598caa12069aef5cbf91c002"}
-->
--
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]
