You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by ya...@apache.org on 2026/08/13 03:01:45 UTC
(spark) branch master updated: [SPARK-58707][SQL] Do not prune a JSON schema down to the corrupt record column
This is an automated email from the ASF dual-hosted git repository.
LuciferYang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new bdea0f8e9389 [SPARK-58707][SQL] Do not prune a JSON schema down to the corrupt record column
bdea0f8e9389 is described below
commit bdea0f8e9389b87be88d41660f7a8b7bc7f3e21b
Author: YangJie <ya...@baidu.com>
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 <ya...@baidu.com>
Signed-off-by: yangjie01 <ya...@baidu.com>
---
.../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 2bf18ba469dc..c0ae3fc157f7 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: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org