You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by we...@apache.org on 2026/10/21 02:22:12 UTC
(spark) branch master updated: [SPARK-59417][SQL][FOLLOWUP] Stop resuming a streamed JSON array inside a malformed element
This is an automated email from the ASF dual-hosted git repository.
cloud-fan 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 8e14957c2936 [SPARK-59417][SQL][FOLLOWUP] Stop resuming a streamed JSON array inside a malformed element
8e14957c2936 is described below
commit 8e14957c293652030fb81747ffc09ed374bed0ee
Author: Timothy Meehan <ti...@timdmeehan.com>
AuthorDate: Wed Oct 21 02:21:56 2026 +0000
[SPARK-59417][SQL][FOLLOWUP] Stop resuming a streamed JSON array inside a malformed element
### What changes were proposed in this pull request?
This is a follow-up to #58704. With `spark.sql.json.enableStreamingTopLevelArray` or the `enableStreamingTopLevelArray` option enabled, `JacksonParser.parseIterator` resumes after a malformed array element only where the parser cannot have moved past it. For an object element it took a `PartialResultException` as proof, since `convertObject` normally consumes through the element's `END_OBJECT` before raising one.
A syntax error inside a nested object breaks that premise. The nested `convertObject` fails while reading its next token, the element's `convertObject` catches the error, and its loop then stops at the nested object's `END_OBJECT`, still inside the element. The iterator resumed there, read one of the element's fields as the next element, and returned that field's object value as a row.
`resumableAfter` now also requires the parser to be back in the top-level array. Any other position ends the document, which is what the configuration doc already says for an element whose failure leaves the parser at an unknown position.
### Why are the changes needed?
Reading `[{"a":1},{"s":{"x":1 2},"t":{"a":9}},{"a":2}]` (a missing comma inside `s`) with schema `a int, s struct<x: int>, _corrupt_record string` and streaming enabled returned rows that are not in the data. Writing `D` for the whole document:
| mode | before | after |
| --- | --- | --- |
| PERMISSIVE | `(1, null, null)`, `(null, null, D)`, `(9, null, null)`, `(null, null, D)`, `(2, null, null)` | `(1, null, null)`, `(null, null, D)` |
| DROPMALFORMED | `(1, null, null)`, `(9, null, null)`, `(2, null, null)` | `(1, null, null)` |
`(9, null, null)` is `t`'s value rather than an array element, and it carries no corrupt record. The failure needs `spark.sql.json.enablePartialResults`, which defaults to true. With it disabled, the element's error propagates and already ended the document.
### Does this PR introduce _any_ user-facing change?
No, compared to released versions: SPARK-59417 is only in master. Within master, and only with streaming enabled (it defaults to false), a malformed element that leaves the parser inside itself now ends the document instead of producing rows from its fields.
### How was this patch tested?
Added `multiline top level JSON array ends the document at a syntax error in a nested object` over PERMISSIVE and DROPMALFORMED, with the malformed object in a struct field and in an array of structs. Both fail without the fix. The array-of-structs case also fails if the guard only checks that the parser is in an array, not that the array is the top-level one.
Ran:
- `catalyst/scalastyle`
- `sql/Test/scalastyle`
- `catalyst/testOnly *JacksonParserSuite`
- `sql/testOnly *JsonV1Suite *JsonV2Suite *JSONTarArchiveReadSuite *JSONZipArchiveReadSuite *JSONSevenZArchiveReadSuite` (582 tests)
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Isaac (Claude Opus 5.5)
This pull request and its description were written by Isaac.
Closes #59138 from tdcmeehan/tim/sc-173438-streaming-resume-boundary-oss.
Authored-by: Timothy Meehan <ti...@timdmeehan.com>
Signed-off-by: Wenchen Fan <we...@databricks.com>
---
.../spark/sql/catalyst/json/JacksonParser.scala | 22 +++++++-------
.../sql/execution/datasources/json/JsonSuite.scala | 34 ++++++++++++++++++++++
2 files changed, 46 insertions(+), 10 deletions(-)
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala
index 9f418d0da100..6596b813c5cf 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala
@@ -937,16 +937,18 @@ class JacksonParser(
}
throw badRecord(error, () => recordLiteral(record))
}
- // Resuming is only safe where the parser cannot have moved past the element that failed. A
- // scalar element leaves it on that scalar's own token, and `convertObject` consumes through
- // END_OBJECT before raising `PartialResultException`; any other container failure can have
- // stopped anywhere inside the element.
- def resumableAfter(error: Throwable, elementStart: Option[JsonToken]): Boolean =
- options.parseMode != FailFastMode && (elementStart match {
- case Some(START_OBJECT) | Some(START_ARRAY) => error.isInstanceOf[PartialResultException]
- case Some(_) => error.isInstanceOf[RuntimeException]
- case None => false
- })
+ // Resuming is only safe where the failed element left the parser back in the top-level array.
+ // A syntax error inside a nested object ends `convertObject` at that object's END_OBJECT, not
+ // the element's, so a `PartialResultException` alone does not prove the element was consumed.
+ def resumableAfter(error: Throwable, elementStart: Option[JsonToken]): Boolean = {
+ val context = jsonParser.getParsingContext
+ options.parseMode != FailFastMode && context.inArray && context.getParent.inRoot &&
+ (elementStart match {
+ case Some(START_OBJECT) | Some(START_ARRAY) => error.isInstanceOf[PartialResultException]
+ case Some(_) => error.isInstanceOf[RuntimeException]
+ case None => false
+ })
+ }
def handleFailure[R](elementStart: Option[JsonToken])(operation: => R): R = {
try operation catch {
case e: SparkUpgradeException => fail(e)
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/json/JsonSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/json/JsonSuite.scala
index a561ec255cf9..e6e7d7eac0b8 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/json/JsonSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/json/JsonSuite.scala
@@ -1479,6 +1479,40 @@ abstract class JsonSuite
}
}
+ gridTest("multiline top level JSON array ends the document at a syntax error in a nested object")(
+ for {
+ field <- Seq("struct", "array of structs")
+ mode <- Seq("PERMISSIVE", "DROPMALFORMED")
+ } yield (field, mode)) { case (field, mode) =>
+ withSQLConf(
+ SQLConf.JSON_STREAM_MULTILINE_TOP_LEVEL_ARRAY.key -> "true",
+ SQLConf.JSON_ENABLE_PARTIAL_RESULTS.key -> "true") {
+ withTempPath { file =>
+ val (document, fieldSchema) = field match {
+ // Resuming inside the second element would return `t`'s object as a row.
+ case "struct" =>
+ ("""[{"a":1},{"s":{"x":1 2},"t":{"a":9}},{"a":2}]""", "s STRUCT<x: INT>")
+ // Resuming inside `arr` would return its second struct as a row.
+ case "array of structs" =>
+ ("""[{"a":1},{"arr":[{"x":1 2},{"a":5}]},{"a":2}]""", "arr ARRAY<STRUCT<x: INT>>")
+ }
+ Files.write(file.toPath, document.getBytes(StandardCharsets.UTF_8))
+ val df = spark.read
+ .schema(s"a INT, $fieldSchema, _corrupt_record STRING")
+ .option("multiLine", true)
+ .option("mode", mode)
+ .json(file.getCanonicalPath)
+
+ val expected = if (mode == "PERMISSIVE") {
+ Seq(Row(1, null, null), Row(null, null, document))
+ } else {
+ Seq(Row(1, null, null))
+ }
+ checkAnswer(df, expected)
+ }
+ }
+ }
+
gridTest("multiline top level JSON array streaming option overrides the session config")(
Seq(false, true)) { streaming =>
withSQLConf(SQLConf.JSON_STREAM_MULTILINE_TOP_LEVEL_ARRAY.key -> (!streaming).toString) {
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org