cloud-fan commented on code in PR #59167:
URL: https://github.com/apache/spark/pull/59167#discussion_r4226583756
##########
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:
**Non-blocking (P2):** With preserveCharVarcharTypeInfo=true and standard
semantics=false, physicalSchema has already erased CHAR/VARCHAR but this gate
makes assignment a no-op. The same supported JSON queries that previously
padded CHAR and rejected overflow now return raw unchecked strings, even though
the PR declares user-visible changes only for standard semantics. Please keep
the declared schema/converter behavior in preserve-only mode or supply an
equivalent assignment plan.
**Verification:**
- **Compatibility:** Cover the preserve-only, standard-only, both-enabled,
and both-disabled configuration matrix for CHAR padding, overflow, constrained
maps, and datasource filter behavior.
- **Regression:** Show that preserve-only from_json and datasource reads no
longer return raw unchecked strings.
##########
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:
**Non-blocking (P2):** Post-parse assignment is added only for the
row-shaped PartialResultException branch. The array-of-rows, ArrayData, and
MapData partial branches below still return physical STRING containers
directly, so PERMISSIVE can expose unpadded or overlength CHAR/VARCHAR values
and raw constrained map keys. Every partial carrier needs the same logical
assignment boundary before it becomes a SQL value.
See **Shared repair plan 1** in the review body.
##########
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:
**Non-blocking (P2):** This disables every pushed predicate whenever any
selected field contains CHAR/VARCHAR, including a predicate on an independent
INT key. For selective wide JSON scans that moves rejected rows from early
streaming rejection to full decode, materialization, assignment, and residual
filtering; the checked-in benchmark for this boundary is about 7x-9x slower.
Please retain pushdown for predicates whose referenced values are unchanged by
assignment and keep only assignment-sensitive predicates residual.
See **Shared repair plan 2** in the review body.
##########
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:
**Non-blocking (P2):** At this point the object element has been fully
consumed, but an assignment-time overflow is an ordinary RuntimeException.
resumableAfter only accepts PartialResultException for START_OBJECT, so
PERMISSIVE emits recovery for this element and closes the parser, dropping
later valid elements. Track the completed element boundary so post-assignment
failures can resume without broadening recovery for failures still inside a
container.
**Verification:**
- **Regression:** For a multiline top-level array containing an overflowing
CHAR/VARCHAR element followed by a valid element, assert PERMISSIVE and
DROPMALFORMED preserve the later element and FAILFAST still raises.
- **Compatibility:** Retain current recovery behavior for scalar failures,
parser-side partial objects, duplicate constrained keys, and failures
positioned inside unfinished containers.
##########
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:
**Nit (P3):** For a struct root, the changed test expects a non-null struct
with all fields null, not a null from_json value. Since IS NULL distinguishes
these results, please qualify the PERMISSIVE statement by root shape rather
than saying it always returns null for the whole parsed value.
**Verification:**
- **Inspection:** Compare each documented root shape with its executable
assertion and ensure IS NULL observability is described accurately.
##########
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:
**Non-blocking (P2):** This catch turns a failed logical assignment into a
successful raw partial row. With a parser-side sibling mismatch plus an
overlength CHAR/VARCHAR or normalized map-key collision, PERMISSIVE returns the
physical STRING value instead of applying whole-value error handling; a
swallowed duplicate can also leave MapFromArrays state dirty for the next
record. Assignment failures here need to preserve the declared overflow and
duplicate-key policies rather than fall back to row.
See **Shared repair plan 1** in the review body.
##########
sql/api/src/main/scala/org/apache/spark/sql/catalyst/util/SparkCharVarcharUtils.scala:
##########
@@ -70,6 +70,28 @@ trait SparkCharVarcharUtils {
case v: VarcharType if !SqlApiConf.get.charVarcharFirstClassTypes =>
v.toStringType
case _ => dt
}
+
+ /**
+ * Always replaces CharType/VarcharType with unbounded StringType, including
when first-class
+ * CHAR/VARCHAR types are enabled. Used for STRING-on-the-wire I/O such as
JSON parsing and
Review Comment:
**Nit (P3):** Hive TRANSFORM does not call this helper: its output path uses
ScriptTransformationIOSchema.toUnboundedStringType and its own restorer. Please
describe the actual JSON caller or the generic transformation contract without
implying that changes to this helper cover the Hive implementation.
**Verification:**
- **Inspection:** Resolve every named consumer in the revised comments to an
actual call site, and verify HiveScriptTransformationExec remains accurately
described by its own utilities.
##########
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:
**Nit (P3):** These new datasource cases exercise assignment but not
predicate evaluation. Please add a fixture where a CHAR/VARCHAR field's raw and
assigned values produce distinguishable predicate results, plus an independent
primitive predicate, so the suite proves assignment-sensitive filters remain
residual while safe filters can still be pushed in both reader paths.
See **Shared repair plan 2** in the review body.
--
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]
