srielau commented on code in PR #59167:
URL: https://github.com/apache/spark/pull/59167#discussion_r4226803709
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala:
##########
@@ -60,11 +60,50 @@ class JacksonParser(
// to a value in a field for `InternalRow`.
private type ValueConverter = JsonParser => AnyRef
+ // CHAR/VARCHAR is STRING-on-the-wire in JSON. Parse as unbounded STRING,
then assign.
+ private val physicalSchema: DataType =
+ CharVarcharUtils.replaceCharVarcharWithStringAlways(schema)
+
// `ValueConverter`s for the root schema for all fields in the schema
- private val rootConverter = makeRootConverter(schema)
+ private val rootConverter = makeRootConverter(physicalSchema)
// Lazy: a schema that is not a struct never reaches either top-level-array
path.
- private lazy val arrayElementConverter = makeConverter(schema)
+ private lazy val arrayElementConverter = makeConverter(physicalSchema)
+
+ private val assignExpr: Expression =
+ CharVarcharUtils.assignAfterParse(
+ BoundReference(0, physicalSchema, nullable = true),
+ schema)
+
+ private val needsAssignment =
+ SQLConf.get.charVarcharStandardSemantics &&
+ CharVarcharUtils.hasCharVarchar(schema)
+
+ /**
+ * Pad CHAR / length-check VARCHAR / rewrite CHAR map keys on a row parsed
as STRING.
+ * `DUPLICATED_MAP_KEY` from key assignment is left unwrapped so it fails
the query
+ * even under PERMISSIVE, matching XML constrained maps.
+ */
+ private def applyAssignment(row: InternalRow): InternalRow = {
+ if (!needsAssignment) {
+ row
+ } else {
+ schema match {
+ case _: StructType =>
+ assignExpr.eval(InternalRow(row)).asInstanceOf[InternalRow]
+ case _ =>
+ InternalRow(assignExpr.eval(InternalRow(row.get(0, physicalSchema))))
+ }
+ }
+ }
+
+ private def tryAssignPartial(row: InternalRow): InternalRow = {
+ try {
+ applyAssignment(row)
+ } catch {
+ case NonFatal(_) => row
Review Comment:
Fixed. Assignment on a partial row no longer falls back to the physical
STRING payload. Overflow and other assignment failures produce an empty partial
(PERMISSIVE whole-value null / FAILFAST raises). `DUPLICATED_MAP_KEY` is still
rethrown as query-fatal.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala:
##########
@@ -717,7 +758,10 @@ class JacksonParser(
wrappedCharException.initCause(e)
BadRecordException(recordLiteral, () => Array.empty,
wrappedCharException)
case PartialResultException(row, cause) =>
- BadRecordException(recordLiteral, () => Array(row),
convertCauseForPartialResult(cause))
+ BadRecordException(
+ recordLiteral,
+ () => Array(tryAssignPartial(row)),
Review Comment:
Fixed. `PartialResultException`, `PartialResultArrayException`,
`PartialArrayDataResultException`, and `PartialMapDataResultException` all run
through the same assignment helper before they become a SQL value.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala:
##########
@@ -60,11 +60,50 @@ class JacksonParser(
// to a value in a field for `InternalRow`.
private type ValueConverter = JsonParser => AnyRef
+ // CHAR/VARCHAR is STRING-on-the-wire in JSON. Parse as unbounded STRING,
then assign.
+ private val physicalSchema: DataType =
+ CharVarcharUtils.replaceCharVarcharWithStringAlways(schema)
+
// `ValueConverter`s for the root schema for all fields in the schema
- private val rootConverter = makeRootConverter(schema)
+ private val rootConverter = makeRootConverter(physicalSchema)
// Lazy: a schema that is not a struct never reaches either top-level-array
path.
- private lazy val arrayElementConverter = makeConverter(schema)
+ private lazy val arrayElementConverter = makeConverter(physicalSchema)
+
+ private val assignExpr: Expression =
+ CharVarcharUtils.assignAfterParse(
+ BoundReference(0, physicalSchema, nullable = true),
+ schema)
+
+ private val needsAssignment =
+ SQLConf.get.charVarcharStandardSemantics &&
Review Comment:
Leaving this as-is on purpose. The baseline for this PR is pre-CHAR/VARCHAR
JSON, where CHAR/VARCHAR was STRING. Assignment is only for
`spark.sql.charVarchar.standardSemantics.enabled=true`. With the flag off
(including preserve-only), JSON stays STRING-on-the-wire: no parser pad and no
`EXCEED_LIMIT_LENGTH`.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala:
##########
@@ -866,7 +915,7 @@ class JacksonParser(
nextRow = handleFailure(Some(elementStart)) {
val row =
arrayElementConverter(jsonParser).asInstanceOf[InternalRow]
if (row == null) throw
QueryExecutionErrors.rootConverterReturnNullError()
- row
+ applyAssignment(row)
Review Comment:
Fixed. After the converter has consumed the element, assignment overflow is
raised as `PartialResultException` so `resumableAfter` can keep later top-level
array elements. PERMISSIVE/DROPMALFORMED keep the following valid element;
FAILFAST still raises. Covered in `SPARK-59722: streamed JSON array resumes
after assignment overflow` for V1 and V2.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala:
##########
@@ -154,7 +193,8 @@ class JacksonParser(
private def makeStructRootConverter(st: StructType): JsonParser =>
Iterable[InternalRow] = {
val fieldConverters = st.map(_.dataType).map(makeConverter).toArray
- val jsonFilters = if (SQLConf.get.jsonFilterPushDown) {
+ val jsonFilters = if (SQLConf.get.jsonFilterPushDown &&
+ !CharVarcharUtils.hasCharVarchar(schema)) {
Review Comment:
Fixed. The schema-wide `JsonFilters` veto is gone. Under standard semantics,
only predicates that reference a CHAR/VARCHAR field stay residual; independent
predicates (e.g. an INT key) still push. With the flag off, pushdown is
unchanged from STRING.
##########
sql/core/src/test/scala/org/apache/spark/sql/CharVarcharTestSuite.scala:
##########
@@ -2987,6 +3006,125 @@ class BasicCharVarcharTestSuite extends
SharedSparkSession {
}
}
+ test("SPARK-59722: JSON CHAR/VARCHAR assignment runs after STRING parse") {
+ withSQLConf(SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "true") {
+ // Scalar pad still applies, via stringLengthCheck on the parsed STRING
value.
+ checkAnswer(
+ sql("""SELECT from_json('{"a": "str"}', 'a CHAR(5)')"""),
+ Row(Row("str ")))
+ checkAnswer(
+ sql("""SELECT from_json('{"ab": 1}', 'MAP<CHAR(4), INT>')"""),
+ Row(Map("ab " -> 1)))
+
+ val collision = """SELECT from_json('{"a":1,"a ":2}', 'MAP<CHAR(2),
INT>')"""
+ val exactRepeat = """SELECT from_json('{"a":1,"a":2}', 'MAP<CHAR(2),
INT>')"""
+ val varcharCollision =
+ """SELECT from_json('{"ab":1,"ab ":2}', 'MAP<VARCHAR(2), INT>')"""
+ val nestedCollision =
+ """SELECT from_json(
+ | '{"outer":{"a":1,"a ":2}}',
+ | 'MAP<STRING, MAP<CHAR(2), INT>>')""".stripMargin
+ val overflow = """SELECT from_json('{"abc":1}', 'MAP<VARCHAR(2),
INT>')"""
+ val overflowFailfast =
+ """SELECT from_json(
+ | '{"abc":1}',
+ | 'MAP<VARCHAR(2), INT>',
+ | map('mode', 'FAILFAST'))""".stripMargin
+ val overflowWithSibling =
+ """SELECT from_json(
+ | '{"m":{"k":"abcdef"},"tail":1}',
+ | 'm MAP<STRING, CHAR(2)>, tail INT')""".stripMargin
+
+ // Pad/trim collisions and exact JSON member repeats both use
mapKeyDedupPolicy.
+ assertDuplicateMapKey(collision)
+ assertDuplicateMapKey(exactRepeat)
+ assertDuplicateMapKey(varcharCollision, expectedKey = "ab")
+ assertDuplicateMapKey(nestedCollision)
+ checkAnswer(sql(overflow), Row(null))
+ assertParseExceedLimit(overflowFailfast, expectedLimit = "2")
+ // Assignment failure is whole-value: PERMISSIVE does not keep `tail`.
+ checkAnswer(sql(overflowWithSibling), Row(Row(null, null)))
+ assertParseExceedLimit(
+ """SELECT from_json(
+ | '{"m":{"k":"abcdef"},"tail":1}',
+ | 'm MAP<STRING, CHAR(2)>, tail INT',
+ | map('mode', 'FAILFAST'))""".stripMargin,
+ expectedLimit = "2")
+
+ withTempPath { path =>
+ Seq("""{"m":{"a":1,"a
":2}}""").toDS().write.text(path.getCanonicalPath)
+ def readJsonMap(): DataFrame = spark.read
Review Comment:
Added `SPARK-59722: JSON datasource CHAR filters stay residual`. It uses `id
INT, c CHAR(4)` so raw `"ab"` and assigned `"ab "` differ, plus an independent
`id` predicate, for V1 and V2 and with JSON filter pushdown on and off.
##########
docs/sql-migration-guide.md:
##########
@@ -29,6 +29,7 @@ license: |
## Upgrading from Spark SQL 4.3 to 4.4
- Since Spark 4.4, when `spark.sql.preserveCharVarcharTypeInfo` is true and
`spark.sql.charVarchar.standardSemantics.enabled` is false, ORC reads that
apply a CHAR/VARCHAR schema over STRING storage return the stored values
without ORC truncation, matching Parquet. Previously the ORC reader requested
`char(n)`/`varchar(n)` and truncated STRING-stored values to `n`. Read-side
length checks (`EXCEED_LIMIT_LENGTH`) apply only when
`spark.sql.charVarchar.standardSemantics.enabled` is true.
+- Since Spark 4.4, when `spark.sql.charVarchar.standardSemantics.enabled` is
true, `from_json` and the JSON datasource parse CHAR/VARCHAR fields as
unbounded STRING and then apply write-side assignment (`CHAR` pad,
`EXCEED_LIMIT_LENGTH` on overflow), including nested values and map keys.
Distinct object names that become the same key after CHAR padding or VARCHAR
trailing-space trimming honor `spark.sql.mapKeyDedupPolicy`, as do exact
repeated member names in a CHAR/VARCHAR-keyed map. An assignment failure is a
parse-mode error for the whole parsed value (PERMISSIVE returns null; FAILFAST
raises). JSON type-conversion partial results
(`spark.sql.json.enablePartialResults`) are unchanged for non-CHAR mismatches.
Review Comment:
Updated. PERMISSIVE `from_json` of a STRUCT is a struct whose fields are
null (`IS NULL` is false). PERMISSIVE non-struct roots and datasource rows
return null.
--
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]
