This is an automated email from the ASF dual-hosted git repository.
LuciferYang pushed a commit to branch branch-4.3
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.3 by this push:
new d6209be75a6a [SPARK-58707][SQL] Do not prune a JSON schema down to the
corrupt record column
d6209be75a6a is described below
commit d6209be75a6a11d8f1a15aedeb70c042afb7d63f
Author: YangJie <[email protected]>
AuthorDate: Thu Aug 13 11:01:28 2026 +0800
[SPARK-58707][SQL] Do not prune a JSON schema down to the corrupt record
column
### What changes were proposed in this pull request?
`OptimizeCsvJsonExprs` prunes a `JsonToStructs` schema to the fields being
selected. Dropping a field stops the parser from converting it, so a malformed
value there is never detected and the corrupt record column comes back null
instead of the record text. This PR stops the two JSON branches from pruning in
that case.
The two branches need different conditions:
- The `GetStructField` branch prunes to exactly one field, so selecting the
corrupt record column there always drops the rest. It gets the same guard the
CSV branch added by SPARK-32968 already has.
- The `CreateNamedStruct` branch keeps every selected field, so it only
becomes unsafe when the selection is a strict subset of the schema. Guarding on
"corrupt column is selected" alone would give up the parse-deduplication
rewrite for an all-fields selection that drops nothing, turning one parse into
N.
`GetArrayStructFields` needs no guard: an `ArrayType` schema takes
`JsonToStructsEvaluator`'s `case other =>`, where `corruptFieldIndex` is
`None`, so corrupt-record semantics do not exist on that path.
### Why are the changes needed?
The corrupt record column silently returns null for a record that did fail
to parse. The rule requires `options.isEmpty`, which is the PERMISSIVE default,
and `spark.sql.optimizer.enableJsonExpressionOptimization` defaults to true, so
no configuration is needed to hit it:
```scala
Seq("""{"a": 1, "b": "bad"}""").toDS()
.selectExpr("from_json(value, 'a int, b int, _corrupt_record string') as
p")
.selectExpr("p._corrupt_record")
```
| spark.sql.optimizer.enableJsonExpressionOptimization | result |
|---|---|
| true | `null` |
| false | `{"a": 1, "b": "bad"}` |
The optimized plan shows the schema pruned to the corrupt column alone,
which leaves `actualSchema` empty so the parser has nothing to convert and
never raises `BadRecordException`:
```
Project [from_json(StructField(_corrupt_record,StringType,true), value#1,
...)._corrupt_record ...]
```
A type mismatch on a dropped field is needed to expose this. The existing
SPARK-33907 test uses a structurally malformed record, which fails at
tokenization whatever schema is requested, so it never depended on the pruning;
that test is now annotated to say the rule no longer fires for it.
### Does this PR introduce _any_ user-facing change?
Yes, it fixes the wrong result above. With this PR the corrupt record
column holds the record text under both settings of the config, matching the
unoptimized behavior.
### How was this patch tested?
New tests, plan-level and end-to-end:
- `OptimizeJsonExprsSuite`: two `comparePlans` cases, each with a control.
One asserts the `GetStructField` rewrite is skipped for the corrupt column
while a normal field is still pruned; the other asserts the `CreateNamedStruct`
rewrite still fires when every field is selected and is skipped when the
selection drops a field.
- `JsonFunctionsSuite`: two `checkAnswer` cases covering both branches with
the optimization on and off.
Verified by reverting each guard on the fixed tree: removing the
`GetStructField` guard fails both new catalyst tests, and weakening
`prunesCorruptRecord` to "corrupt column selected" fails the `named_struct`
case. Full `JsonFunctionsSuite` (115) and `OptimizeJsonExprsSuite` +
`OptimizeCsvExprsSuite` (36) pass.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Opus 5
Closes #57924 from LuciferYang/SPARK-json-corrupt-record.
Authored-by: YangJie <[email protected]>
Signed-off-by: yangjie01 <[email protected]>
(cherry picked from commit bdea0f8e9389b87be88d41660f7a8b7bc7f3e21b)
Signed-off-by: yangjie01 <[email protected]>
---
.../catalyst/optimizer/OptimizeCsvJsonExprs.scala | 20 +++++++++-
.../optimizer/OptimizeJsonExprsSuite.scala | 43 ++++++++++++++++++++++
.../org/apache/spark/sql/JsonFunctionsSuite.scala | 34 +++++++++++++++++
3 files changed, 95 insertions(+), 2 deletions(-)
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/OptimizeCsvJsonExprs.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/OptimizeCsvJsonExprs.scala
index 4f6c3a2da457..f04ff990251d 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/OptimizeCsvJsonExprs.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/OptimizeCsvJsonExprs.scala
@@ -44,6 +44,10 @@ import org.apache.spark.unsafe.types.UTF8String
object OptimizeCsvJsonExprs extends Rule[LogicalPlan] {
private def nameOfCorruptRecord = conf.columnNameOfCorruptRecord
+ /** Whether the field this extracts is the corrupt record column of its
`JsonToStructs`. */
+ private def selectsCorruptRecord(g: GetStructField): Boolean =
+ g.childSchema(g.ordinal).name == nameOfCorruptRecord
+
private type SimpleJsonPath = Seq[GetJsonObject.SimpleJsonPathSegment]
private type SharedJsonCandidate = (GetJsonObject, SimpleJsonPath, String)
private type SharedJsonCandidateUnit = Seq[SharedJsonCandidate]
@@ -414,9 +418,18 @@ object OptimizeCsvJsonExprs extends Rule[LogicalPlan] {
// `JsonToStructs` does not support parsing json with duplicated field
names.
val duplicateFields = c.names.map(_.toString).distinct.length !=
c.names.length
+ // Dropping a field stops the parser from converting it, so the parser
never records the
+ // row when a dropped field contains a malformed value, and the corrupt
record column comes
+ // back null. Selecting as many fields as the schema has drops nothing:
with `sameFieldName`
+ // and no duplicates the selected names are distinct names of the
schema, so the counts
+ // match only when every field is selected.
+ val fields = c.valExprs.map(_.asInstanceOf[GetStructField])
+ val prunesCorruptRecord = fields.exists(selectsCorruptRecord) &&
+ fields.length != fields.head.childSchema.length
+
// If we create struct from various fields of the same `JsonToStructs`
and we don't
// alias field names and there is no duplicated field in the struct.
- if (sameFieldName && !duplicateFields) {
+ if (sameFieldName && !duplicateFields && !prunesCorruptRecord) {
val fromJson =
jsonToStructs.head.asInstanceOf[JsonToStructs].copy(schema = c.dataType)
val nullFields = c.children.grouped(2).flatMap {
case Seq(name, value) => Seq(name, Literal(null, value.dataType))
@@ -440,12 +453,15 @@ object OptimizeCsvJsonExprs extends Rule[LogicalPlan] {
child
case g @ GetStructField(j @ JsonToStructs(schema: StructType, _, _, _),
ordinal, _)
- if schema.length > 1 && j.options.isEmpty =>
+ if schema.length > 1 && j.options.isEmpty && !selectsCorruptRecord(g)
=>
// Options here should be empty because the optimization should not be
enabled
// for some options. For example, when the parse mode is failfast it
should not
// optimize, and should force to parse the whole input JSON with
failing fast for
// an invalid input.
// To be more conservative, it does not optimize when any option is
set for now.
+ // Pruning to the corrupt record column alone is excluded as well: it
leaves the
+ // parser nothing to convert, so the parser never records the row when
a dropped
+ // field contains a malformed value, and the column comes back null.
val prunedSchema = StructType(Array(schema(ordinal)))
g.copy(child = j.copy(schema = prunedSchema), ordinal = 0)
diff --git
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/OptimizeJsonExprsSuite.scala
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/OptimizeJsonExprsSuite.scala
index 96f46344ffda..801404ba55a0 100644
---
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/OptimizeJsonExprsSuite.scala
+++
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/OptimizeJsonExprsSuite.scala
@@ -708,6 +708,49 @@ class OptimizeJsonExprsSuite extends PlanTest with
ExpressionEvalHelper {
assert(optimized != query(Map.empty), "expected the empty-options plan to
be rewritten")
}
+ test("SPARK-58707: do not prune a from_json schema down to the corrupt
record column") {
+ val schema = StructType.fromDDL("a int, b int, _corrupt_record string")
+
+ def getField(ordinal: Int): LogicalPlan = testRelation2
+ .select(GetStructField(JsonToStructs(schema, Map.empty, $"json"),
ordinal)).analyze
+
+ // Pruning to the corrupt record column alone leaves the parser nothing to
convert, so a
+ // malformed value in a dropped field would no longer populate the column.
+ comparePlans(Optimizer.execute(getField(2)), getField(2))
+
+ // Control: a non-corrupt field is still pruned, so the guard above is
what blocks it rather
+ // than some other precondition of the rule.
+ val prunedSchema = StructType.fromDDL("a int")
+ comparePlans(
+ Optimizer.execute(getField(0)),
+ testRelation2
+ .select(GetStructField(JsonToStructs(prunedSchema, Map.empty,
$"json"), 0)).analyze)
+ }
+
+ test("SPARK-58707: simplify named_struct + from_json when no field is
dropped") {
+ val schema = StructType.fromDDL("a int, _corrupt_record string")
+
+ def query(fields: (String, Int)*): LogicalPlan = testRelation2.select(
+ namedStruct(fields.flatMap { case (name, ordinal) =>
+ Seq(Literal(name), GetStructField(JsonToStructs(schema, Map.empty,
$"json"), ordinal))
+ }: _*).as("struct")).analyze
+
+ // Selecting the corrupt record column together with every other field
drops nothing, so the
+ // rewrite is safe and still collapses the repeated parses into one.
+ val nullStruct = namedStruct(
+ "a", Literal(null, IntegerType), "_corrupt_record", Literal(null,
StringType))
+ comparePlans(
+ Optimizer.execute(query("a" -> 0, "_corrupt_record" -> 1)),
+ testRelation2.select(
+ If(IsNull($"json"),
+ nullStruct,
+ KnownNotNull(JsonToStructs(schema, Map.empty,
$"json"))).as("struct")).analyze)
+
+ // Dropping `a` while selecting the corrupt record column is what makes
the column unreliable.
+ val pruning = query("_corrupt_record" -> 1)
+ comparePlans(Optimizer.execute(pruning), pruning)
+ }
+
test("SPARK-33007: simplify named_struct + from_json") {
val options = Map.empty[String, String]
val schema = StructType.fromDDL("a int, b int, c long, d string")
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/JsonFunctionsSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/JsonFunctionsSuite.scala
index 6eccbb7a6a38..53a256a18a57 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/JsonFunctionsSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/JsonFunctionsSuite.scala
@@ -1798,6 +1798,37 @@ class JsonFunctionsSuite extends SharedSparkSession {
}
}
+ test("SPARK-58707: json pruning keeps the corrupt record column populated") {
+ Seq("true", "false").foreach { enabled =>
+ withSQLConf(SQLConf.JSON_EXPRESSION_OPTIMIZATION.key -> enabled) {
+ // `b` is a type mismatch rather than a structural malformation, so it
only fails when the
+ // parser is asked for it. Pruning the schema down to the corrupt
record column alone left
+ // the parser nothing to convert, so the record was reported as clean.
+ val df = Seq("""{"a": 1, "b": "bad"}""").toDS()
+ .selectExpr("from_json(value, 'a int, b int, _corrupt_record
string') as p")
+ .selectExpr("p._corrupt_record")
+
+ checkAnswer(df, Row("""{"a": 1, "b": "bad"}"""))
+ }
+ }
+ }
+
+ test("SPARK-58707: named_struct keeps the corrupt record column populated") {
+ Seq("true", "false").foreach { enabled =>
+ withSQLConf(SQLConf.JSON_EXPRESSION_OPTIMIZATION.key -> enabled) {
+ // The two from_json calls are written inline so that CollapseProject
does not keep them
+ // behind an alias, which would stop the rule from firing at all.
+ val fromJson = "from_json(value, 'a int, b int, _corrupt_record
string')"
+ val df = Seq("""{"a": 1, "b": "bad"}""").toDS()
+ .selectExpr(
+ s"named_struct('a', $fromJson.a, '_corrupt_record',
$fromJson._corrupt_record) as s")
+ .selectExpr("s.a", "s._corrupt_record")
+
+ checkAnswer(df, Row(1, """{"a": 1, "b": "bad"}"""))
+ }
+ }
+ }
+
test("SPARK-33907: json pruning optimization with corrupt record field") {
Seq("true", "false").foreach { enabled =>
withSQLConf(SQLConf.JSON_EXPRESSION_OPTIMIZATION.key -> enabled) {
@@ -1806,6 +1837,9 @@ class JsonFunctionsSuite extends SharedSparkSession {
.add("b", IntegerType)
val badRec = """{"a" 1, "b": 11}"""
+ // Since SPARK-58707 the rule no longer prunes to the corrupt record
column, so both
+ // iterations run the same plan. The record here is structurally
malformed, which fails at
+ // tokenization whatever schema is requested, so the answer never
depended on the pruning.
val df = Seq(badRec, """{"a": 2, "b": 12}""").toDS()
.selectExpr("from_json(value, 'a int, b int, _corrupt_record
string') as parsed")
.selectExpr("parsed._corrupt_record")
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
