From af30314be977bb3b7fcc01a11dc89c2e263b8b67 Mon Sep 17 00:00:00 2001 From: Tim Meehan Date: Thu, 10 Sep 2026 15:26:22 +0000 Subject: [PATCH 1/6] [SQL] Stream multiline top-level JSON arrays --- .../sql/catalyst/json/JacksonParser.scala | 153 +++++++++++++----- .../sql/catalyst/util/FailureSafeParser.scala | 67 +++++--- .../datasources/json/JsonDataSource.scala | 10 +- .../datasources/json/JsonBenchmark.scala | 38 ++++- .../datasources/json/JsonSuite.scala | 37 +++++ 5 files changed, 239 insertions(+), 66 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 ea8061b774c43..985794722295f 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 @@ -693,6 +693,34 @@ class JacksonParser( case _ => err } + private def badRecord(error: Throwable, recordLiteral: () => UTF8String): BadRecordException = + error match { + case e: SparkUpgradeException => throw e + case e: CharConversionException if options.encoding.isEmpty => + val msg = + """JSON parser cannot handle a character in its input. + |Specifying encoding as an input option explicitly might help to resolve the issue. + |""".stripMargin + e.getMessage + val wrappedCharException = new CharConversionException(msg) + wrappedCharException.initCause(e) + BadRecordException(recordLiteral, () => Array.empty, wrappedCharException) + case PartialResultException(row, cause) => + BadRecordException(recordLiteral, () => Array(row), convertCauseForPartialResult(cause)) + case PartialResultArrayException(rows, cause) => + BadRecordException(recordLiteral, () => rows, cause) + case PartialArrayDataResultException(arrayData, cause) => + BadRecordException( + recordLiteral, + () => Array(InternalRow(arrayData)), + convertCauseForPartialResult(cause)) + case PartialMapDataResultException(mapData, cause) => + BadRecordException( + recordLiteral, + () => Array(InternalRow(mapData)), + convertCauseForPartialResult(cause)) + case e => BadRecordException(recordLiteral, () => Array.empty, e) + } + /** * Parse the JSON input to the set of [[InternalRow]]s. * @@ -717,43 +745,96 @@ class JacksonParser( } } catch { case e: SparkUpgradeException => throw e - case e @ (_: RuntimeException | _: JsonProcessingException | _: MalformedInputException) => - // JSON parser currently doesn't support partial results for corrupted records. - // For such records, all fields other than the field configured by - // `columnNameOfCorruptRecord` are set to `null`. - throw BadRecordException(() => recordLiteral(record), () => Array.empty, e) case e: CharConversionException if options.encoding.isEmpty => - val msg = - """JSON parser cannot handle a character in its input. - |Specifying encoding as an input option explicitly might help to resolve the issue. - |""".stripMargin + e.getMessage - val wrappedCharException = new CharConversionException(msg) - wrappedCharException.initCause(e) - throw BadRecordException(() => recordLiteral(record), () => Array.empty, - wrappedCharException) - case PartialResultException(row, cause) => - throw BadRecordException( - record = () => recordLiteral(record), - partialResults = () => Array(row), - convertCauseForPartialResult(cause)) - case PartialResultArrayException(rows, cause) => - throw BadRecordException( - record = () => recordLiteral(record), - partialResults = () => rows, - cause) - // These exceptions should never be thrown outside of JacksonParser. - // They are used for the control flow in the parser. We add them here for completeness - // since they also indicate a bad record. - case PartialArrayDataResultException(arrayData, cause) => - throw BadRecordException( - record = () => recordLiteral(record), - partialResults = () => Array(InternalRow(arrayData)), - convertCauseForPartialResult(cause)) - case PartialMapDataResultException(mapData, cause) => - throw BadRecordException( - record = () => recordLiteral(record), - partialResults = () => Array(InternalRow(mapData)), - convertCauseForPartialResult(cause)) + throw badRecord(e, () => recordLiteral(record)) + case e @ (_: RuntimeException | _: JsonProcessingException | _: MalformedInputException | + _: PartialResultException | _: PartialResultArrayException | + _: PartialArrayDataResultException | _: PartialMapDataResultException) => + throw badRecord(e, () => recordLiteral(record)) + } + } + + private[sql] def parseIterator[T]( + record: T, + createParser: (JsonFactory, T) => JsonParser, + recordLiteral: T => UTF8String): Iterator[InternalRow] = { + val streamArray = allowArrayAsStructs && schema.isInstanceOf[StructType] && + options.singleVariantColumn.isEmpty && options.explodeEmbeddedArray.isEmpty + val elementConverter = if (streamArray) makeConverter(schema) else null + val jsonParser = createParser(factory, record) + new Iterator[InternalRow] { + private var delegate: Iterator[InternalRow] = Iterator.empty + private var nextRow: InternalRow = _ + private var prepared = false + private var finished = false + private var started = false + private var array = false + + override def hasNext: Boolean = { + prepare() + !finished + } + + override def next(): InternalRow = { + prepare() + if (finished) throw new NoSuchElementException("next on empty iterator") + prepared = false + nextRow + } + + private def prepare(): Unit = { + if (prepared || finished) return + try { + if (!started) { + started = true + val token = jsonParser.nextToken() + if (token == null) { + finish() + } else if (streamArray && token == START_ARRAY) { + array = true + } else { + val rows = rootConverter(jsonParser) + if (rows == null) throw QueryExecutionErrors.rootConverterReturnNullError() + delegate = rows.iterator + } + } + if (!finished && array) { + jsonParser.nextToken() match { + case END_ARRAY => finish() + case null => + throw new JsonParseException(jsonParser, "Unexpected end of top-level array") + case _ => + nextRow = elementConverter(jsonParser).asInstanceOf[InternalRow] + if (nextRow == null) throw QueryExecutionErrors.rootConverterReturnNullError() + prepared = true + } + } else if (!finished && delegate.hasNext) { + nextRow = delegate.next() + prepared = true + } else if (!finished) { + finish() + } + } catch { + case e: SparkUpgradeException => fail(e) + case e: CharConversionException if options.encoding.isEmpty => fail(e) + case e @ (_: RuntimeException | _: JsonProcessingException | _: MalformedInputException | + _: PartialResultException | _: PartialResultArrayException | + _: PartialArrayDataResultException | _: PartialMapDataResultException) => fail(e) + } + } + + private def finish(): Unit = { + finished = true + jsonParser.close() + } + + private def fail(error: Throwable): Nothing = { + finished = true + try jsonParser.close() catch { + case NonFatal(closeError) => error.addSuppressed(closeError) + } + throw badRecord(error, () => recordLiteral(record)) + } } } } diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/FailureSafeParser.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/FailureSafeParser.scala index d9946d1b12ec3..305fe9dd27c73 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/FailureSafeParser.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/FailureSafeParser.scala @@ -24,7 +24,7 @@ import org.apache.spark.sql.types.StructType import org.apache.spark.unsafe.types.UTF8String class FailureSafeParser[IN]( - rawParser: IN => Iterable[InternalRow], + rawParser: IN => IterableOnce[InternalRow], mode: ParseMode, schema: StructType, columnNameOfCorruptRecord: String) { @@ -56,33 +56,50 @@ class FailureSafeParser[IN]( } def parse(input: IN): Iterator[InternalRow] = { - try { + var delegate = try { rawParser.apply(input).iterator.map(row => toResultRow(Some(row), () => null)) } catch { - case e: BadRecordException => mode match { - case PermissiveMode => - val partialResults = e.partialResults() - if (partialResults.nonEmpty) { - partialResults.iterator.map(row => toResultRow(Some(row), e.record)) - } else { - Iterator(toResultRow(None, e.record)) - } - case DropMalformedMode => - Iterator.empty - case FailFastMode => - e.getCause match { - case _: JsonArraysAsStructsException => - // SPARK-42298 we recreate the exception here to make sure the error message - // have the record content. - throw QueryExecutionErrors.cannotParseJsonArraysAsStructsError(e.record().toString) - case StringAsDataTypeException(fieldName, fieldValue, dataType) => - throw QueryExecutionErrors.cannotParseStringAsDataTypeError(e.record().toString, - fieldName, fieldValue, dataType) - case causeWrapper: LazyBadRecordCauseWrapper => - throwMalformedRecordsDetectedInRecordParsingError(e, causeWrapper.cause()) - case cause => throwMalformedRecordsDetectedInRecordParsingError(e, cause) - } + case e: BadRecordException => parseFailure(e) + } + new Iterator[InternalRow] { + private def handleFailure[T](operation: Iterator[InternalRow] => T): T = { + try operation(delegate) catch { + case e: BadRecordException => + delegate = parseFailure(e) + operation(delegate) + } } + + override def hasNext: Boolean = handleFailure(_.hasNext) + + override def next(): InternalRow = handleFailure(_.next()) + } + } + + private def parseFailure(e: BadRecordException): Iterator[InternalRow] = { + mode match { + case PermissiveMode => + val partialResults = e.partialResults() + if (partialResults.nonEmpty) { + partialResults.iterator.map(row => toResultRow(Some(row), e.record)) + } else { + Iterator(toResultRow(None, e.record)) + } + case DropMalformedMode => + Iterator.empty + case FailFastMode => + e.getCause match { + case _: JsonArraysAsStructsException => + // SPARK-42298 we recreate the exception here to make sure the error message + // have the record content. + throw QueryExecutionErrors.cannotParseJsonArraysAsStructsError(e.record().toString) + case StringAsDataTypeException(fieldName, fieldValue, dataType) => + throw QueryExecutionErrors.cannotParseStringAsDataTypeError(e.record().toString, + fieldName, fieldValue, dataType) + case causeWrapper: LazyBadRecordCauseWrapper => + throwMalformedRecordsDetectedInRecordParsingError(e, causeWrapper.cause()) + case cause => throwMalformedRecordsDetectedInRecordParsingError(e, cause) + } } } diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/json/JsonDataSource.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/json/JsonDataSource.scala index 14e23a5fbdeec..442fd448a10dd 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/json/JsonDataSource.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/json/JsonDataSource.scala @@ -409,13 +409,14 @@ object MultiLineJsonDataSource extends JsonDataSource { .getOrElse(CreateJacksonParser.inputStream(_: JsonFactory, _: InputStream)) val safeParser = new FailureSafeParser[InputStream]( - input => parser.parse[InputStream](input, streamParser, partitionedFileString), + input => parser.parseIterator[InputStream](input, streamParser, partitionedFileString), parser.options.parseMode, schema, parser.options.columnNameOfCorruptRecord) - safeParser.parse( - CodecStreams.createInputStreamWithCloseResource(conf, file.toPath)) + val input = CodecStreams.createInputStreamWithCloseResource(conf, file.toPath) + Option(TaskContext.get()).foreach(_.addTaskCompletionListener[Unit](_ => input.close())) + safeParser.parse(input) } override protected def readStream( @@ -430,7 +431,8 @@ object MultiLineJsonDataSource extends JsonDataSource { .getOrElse(CreateJacksonParser.inputStream(_: JsonFactory, _: InputStream)) val safeParser = new FailureSafeParser[InputStream]( - input => parser.parse[InputStream](input, streamParser, _ => UTF8String.fromBytes(bytes)), + input => parser.parseIterator[InputStream]( + input, streamParser, _ => UTF8String.fromBytes(bytes)), parser.options.parseMode, schema, parser.options.columnNameOfCorruptRecord) diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/json/JsonBenchmark.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/json/JsonBenchmark.scala index 94a2ccc41d30b..4db02a7fb6829 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/json/JsonBenchmark.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/json/JsonBenchmark.scala @@ -16,15 +16,19 @@ */ package org.apache.spark.sql.execution.datasources.json -import java.io.File +import java.io.{ByteArrayInputStream, File, InputStream} import java.time.{Instant, LocalDate} +import com.fasterxml.jackson.core.JsonFactory + import org.apache.spark.benchmark.Benchmark import org.apache.spark.sql.{Column, Dataset, Row} +import org.apache.spark.sql.catalyst.json.{CreateJacksonParser, JacksonParser, JSONOptions} import org.apache.spark.sql.execution.benchmark.SqlBasedBenchmark import org.apache.spark.sql.functions._ import org.apache.spark.sql.internal.SQLConf import org.apache.spark.sql.types._ +import org.apache.spark.unsafe.types.UTF8String /** * The benchmarks aims to measure performance of JSON parsing when encoding is set and isn't. @@ -578,6 +582,36 @@ object JsonBenchmark extends SqlBasedBenchmark { benchmark.run() } + private def topLevelArrayBenchmark( + rowsNum: Int, + payloadSize: Int, + numIters: Int): Unit = { + val payload = "x" * payloadSize + val bytes = (0 until rowsNum) + .map(i => s"""{"a":$i,"payload":"$payload"}""") + .mkString("[", ",", "]") + .getBytes("UTF-8") + val schema = new StructType().add("a", IntegerType).add("payload", StringType) + val options = new JSONOptions(Map("multiLine" -> "true"), SQLConf.get.sessionLocalTimeZone) + val parser = new JacksonParser(schema, options, allowArrayAsStructs = true) + val createParser = CreateJacksonParser.inputStream(_: JsonFactory, _: InputStream) + val recordLiteral = (_: InputStream) => UTF8String.EMPTY_UTF8 + val benchmark = new Benchmark( + s"Top-level JSON array with $payloadSize-byte payloads", rowsNum, output = output) + + benchmark.addCase("whole document", numIters) { _ => + val input = new ByteArrayInputStream(bytes) + parser.parse(input, createParser, recordLiteral).size + } + + benchmark.addCase("streamed", numIters) { _ => + val input = new ByteArrayInputStream(bytes) + parser.parseIterator(input, createParser, recordLiteral).size + } + + benchmark.run() + } + override def runBenchmarkSuite(mainArgs: Array[String]): Unit = { val numIters = 3 runBenchmark("Benchmark for performance of JSON parsing") { @@ -595,6 +629,8 @@ object JsonBenchmark extends SqlBasedBenchmark { // TODO (SPARK-32325): Add benchmarks for filters with nested column attributes. filtersPushdownBenchmark(rowsNum = 100 * 1000, numIters) partialResultBenchmark(rowsNum = 10000, numIters) + topLevelArrayBenchmark(rowsNum = 100000, payloadSize = 0, numIters = numIters) + topLevelArrayBenchmark(rowsNum = 1000, payloadSize = 64 * 1024, numIters = numIters) } } } 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 4cd6783e1df60..b71aa644e7c17 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 @@ -54,6 +54,7 @@ import org.apache.spark.sql.types.StructType.fromDDL import org.apache.spark.sql.types.TestUDT.{MyDenseVector, MyDenseVectorUDT} import org.apache.spark.sql.util.CaseInsensitiveStringMap import org.apache.spark.tags.ExtendedSQLTest +import org.apache.spark.unsafe.types.UTF8String import org.apache.spark.util.ArrayImplicits._ import org.apache.spark.util.Utils @@ -1064,6 +1065,42 @@ abstract class JsonSuite Row("str_a_4", "str_b_4", "str_c_4") :: Nil ) } + + withTempPath { file => + Files.write(file.toPath, """[{"a":1},{"a":2}]""".getBytes(StandardCharsets.UTF_8)) + checkAnswer( + spark.read.option("multiLine", true).schema("a int").json(file.getCanonicalPath), + Seq(Row(1), Row(2))) + } + } + + test("multiline top level JSON arrays are parsed lazily") { + val schema = StructType(Seq(StructField("a", IntegerType))) + val options = new JSONOptions(Map("multiLine" -> "true"), SQLConf.get.sessionLocalTimeZone) + val parser = new JacksonParser(schema, options, allowArrayAsStructs = true) + val input = new ByteArrayInputStream( + s"""[{"a":1},{"a":2,"payload":"${"x" * 200000}"}]""".getBytes(StandardCharsets.UTF_8)) + val rows = parser.parseIterator[InputStream]( + input, + CreateJacksonParser.inputStream(_: JsonFactory, _: InputStream), + stream => UTF8String.fromBytes(stream.readAllBytes())) + + assert(rows.next().getInt(0) === 1) + assert(input.available() > 0) + } + + test("multiline top level JSON array keeps rows emitted before malformed input") { + withTempPath { file => + val document = """[{"a":1} {"a":2}]""" + Files.write(file.toPath, document.getBytes(StandardCharsets.UTF_8)) + + checkAnswer( + spark.read + .option("multiLine", true) + .schema("a int, _corrupt_record string") + .json(file.getCanonicalPath), + Seq(Row(1, null), Row(null, document))) + } } test("Corrupt records: FAILFAST mode") { From b6b8da4182f324ef9e7332b38dc1ab44490b96a4 Mon Sep 17 00:00:00 2001 From: Tim Meehan Date: Thu, 10 Sep 2026 15:42:56 +0000 Subject: [PATCH 2/6] [SQL] Gate streaming top-level JSON arrays with SQLConf --- .../spark/sql/catalyst/json/JSONOptions.scala | 3 ++ .../apache/spark/sql/internal/SQLConf.scala | 10 +++++++ .../datasources/json/JsonDataSource.scala | 17 +++++++++-- .../datasources/json/JsonSuite.scala | 30 +++++++++++-------- 4 files changed, 45 insertions(+), 15 deletions(-) diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JSONOptions.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JSONOptions.scala index c7e8a8e13116f..507b2f2aae63d 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JSONOptions.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JSONOptions.scala @@ -239,6 +239,9 @@ class JSONOptions( val useUnsafeRow: Boolean = parameters.get(USE_UNSAFE_ROW).map(_.toBoolean).getOrElse( SQLConf.get.getConf(SQLConf.JSON_USE_UNSAFE_ROW)) + val streamMultilineTopLevelArray: Boolean = + SQLConf.get.getConf(SQLConf.JSON_STREAM_MULTILINE_TOP_LEVEL_ARRAY) + /** Build a Jackson [[JsonFactory]] using JSON options. */ def buildJsonFactory(): JsonFactory = { val streamReadConstraints = StreamReadConstraints diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala index b79a5d7a3b537..1b3216a6cba3c 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala @@ -7034,6 +7034,16 @@ object SQLConf { .booleanConf .createWithDefault(true) + val JSON_STREAM_MULTILINE_TOP_LEVEL_ARRAY = + buildConf("spark.sql.json.enableStreamingTopLevelArray") + .internal() + .doc("When true, multiline JSON reads stream the elements of a top-level array one at a " + + "time instead of materializing the entire array before returning rows.") + .version("4.4.0") + .withBindingPolicy(ConfigBindingPolicy.SESSION) + .booleanConf + .createWithDefault(false) + val JSON_USE_UNSAFE_ROW = buildConf("spark.sql.json.useUnsafeRow") .doc("When set to true, use UnsafeRow to represent struct result in the JSON parser. It " + diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/json/JsonDataSource.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/json/JsonDataSource.scala index 442fd448a10dd..6b66d9bae3b3f 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/json/JsonDataSource.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/json/JsonDataSource.scala @@ -409,7 +409,13 @@ object MultiLineJsonDataSource extends JsonDataSource { .getOrElse(CreateJacksonParser.inputStream(_: JsonFactory, _: InputStream)) val safeParser = new FailureSafeParser[InputStream]( - input => parser.parseIterator[InputStream](input, streamParser, partitionedFileString), + input => { + if (parser.options.streamMultilineTopLevelArray) { + parser.parseIterator[InputStream](input, streamParser, partitionedFileString) + } else { + parser.parse[InputStream](input, streamParser, partitionedFileString) + } + }, parser.options.parseMode, schema, parser.options.columnNameOfCorruptRecord) @@ -431,8 +437,13 @@ object MultiLineJsonDataSource extends JsonDataSource { .getOrElse(CreateJacksonParser.inputStream(_: JsonFactory, _: InputStream)) val safeParser = new FailureSafeParser[InputStream]( - input => parser.parseIterator[InputStream]( - input, streamParser, _ => UTF8String.fromBytes(bytes)), + input => { + if (parser.options.streamMultilineTopLevelArray) { + parser.parseIterator[InputStream](input, streamParser, _ => UTF8String.fromBytes(bytes)) + } else { + parser.parse[InputStream](input, streamParser, _ => UTF8String.fromBytes(bytes)) + } + }, parser.options.parseMode, schema, parser.options.columnNameOfCorruptRecord) 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 b71aa644e7c17..199a834329e32 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 @@ -1068,9 +1068,13 @@ abstract class JsonSuite withTempPath { file => Files.write(file.toPath, """[{"a":1},{"a":2}]""".getBytes(StandardCharsets.UTF_8)) - checkAnswer( - spark.read.option("multiLine", true).schema("a int").json(file.getCanonicalPath), - Seq(Row(1), Row(2))) + Seq("false", "true").foreach { enabled => + withSQLConf(SQLConf.JSON_STREAM_MULTILINE_TOP_LEVEL_ARRAY.key -> enabled) { + checkAnswer( + spark.read.option("multiLine", true).schema("a int").json(file.getCanonicalPath), + Seq(Row(1), Row(2))) + } + } } } @@ -1090,16 +1094,18 @@ abstract class JsonSuite } test("multiline top level JSON array keeps rows emitted before malformed input") { - withTempPath { file => - val document = """[{"a":1} {"a":2}]""" - Files.write(file.toPath, document.getBytes(StandardCharsets.UTF_8)) + withSQLConf(SQLConf.JSON_STREAM_MULTILINE_TOP_LEVEL_ARRAY.key -> "true") { + withTempPath { file => + val document = """[{"a":1} {"a":2}]""" + Files.write(file.toPath, document.getBytes(StandardCharsets.UTF_8)) - checkAnswer( - spark.read - .option("multiLine", true) - .schema("a int, _corrupt_record string") - .json(file.getCanonicalPath), - Seq(Row(1, null), Row(null, document))) + checkAnswer( + spark.read + .option("multiLine", true) + .schema("a int, _corrupt_record string") + .json(file.getCanonicalPath), + Seq(Row(1, null), Row(null, document))) + } } } From 7a5f618918540b485f96ede86bc1189abf00d14d Mon Sep 17 00:00:00 2001 From: Tim Meehan Date: Thu, 10 Sep 2026 15:58:55 +0000 Subject: [PATCH 3/6] [SQL] Isolate lazy JSON failure handling --- .../sql/catalyst/util/FailureSafeParser.scala | 14 ++++++-- .../datasources/json/JsonDataSource.scala | 34 ++++++++++--------- 2 files changed, 30 insertions(+), 18 deletions(-) diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/FailureSafeParser.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/FailureSafeParser.scala index 305fe9dd27c73..408017ada2cb0 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/FailureSafeParser.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/FailureSafeParser.scala @@ -24,7 +24,7 @@ import org.apache.spark.sql.types.StructType import org.apache.spark.unsafe.types.UTF8String class FailureSafeParser[IN]( - rawParser: IN => IterableOnce[InternalRow], + rawParser: IN => Iterable[InternalRow], mode: ParseMode, schema: StructType, columnNameOfCorruptRecord: String) { @@ -56,11 +56,21 @@ class FailureSafeParser[IN]( } def parse(input: IN): Iterator[InternalRow] = { - var delegate = try { + try { rawParser.apply(input).iterator.map(row => toResultRow(Some(row), () => null)) } catch { case e: BadRecordException => parseFailure(e) } + } + + def parseIterator( + input: IN, + iteratorParser: IN => Iterator[InternalRow]): Iterator[InternalRow] = { + var delegate = try { + iteratorParser.apply(input).map(row => toResultRow(Some(row), () => null)) + } catch { + case e: BadRecordException => parseFailure(e) + } new Iterator[InternalRow] { private def handleFailure[T](operation: Iterator[InternalRow] => T): T = { try operation(delegate) catch { diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/json/JsonDataSource.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/json/JsonDataSource.scala index 6b66d9bae3b3f..da4d3d3104368 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/json/JsonDataSource.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/json/JsonDataSource.scala @@ -409,20 +409,20 @@ object MultiLineJsonDataSource extends JsonDataSource { .getOrElse(CreateJacksonParser.inputStream(_: JsonFactory, _: InputStream)) val safeParser = new FailureSafeParser[InputStream]( - input => { - if (parser.options.streamMultilineTopLevelArray) { - parser.parseIterator[InputStream](input, streamParser, partitionedFileString) - } else { - parser.parse[InputStream](input, streamParser, partitionedFileString) - } - }, + input => parser.parse[InputStream](input, streamParser, partitionedFileString), parser.options.parseMode, schema, parser.options.columnNameOfCorruptRecord) val input = CodecStreams.createInputStreamWithCloseResource(conf, file.toPath) Option(TaskContext.get()).foreach(_.addTaskCompletionListener[Unit](_ => input.close())) - safeParser.parse(input) + if (parser.options.streamMultilineTopLevelArray) { + safeParser.parseIterator( + input, + input => parser.parseIterator[InputStream](input, streamParser, partitionedFileString)) + } else { + safeParser.parse(input) + } } override protected def readStream( @@ -437,17 +437,19 @@ object MultiLineJsonDataSource extends JsonDataSource { .getOrElse(CreateJacksonParser.inputStream(_: JsonFactory, _: InputStream)) val safeParser = new FailureSafeParser[InputStream]( - input => { - if (parser.options.streamMultilineTopLevelArray) { - parser.parseIterator[InputStream](input, streamParser, _ => UTF8String.fromBytes(bytes)) - } else { - parser.parse[InputStream](input, streamParser, _ => UTF8String.fromBytes(bytes)) - } - }, + input => parser.parse[InputStream](input, streamParser, _ => UTF8String.fromBytes(bytes)), parser.options.parseMode, schema, parser.options.columnNameOfCorruptRecord) - safeParser.parse(new ByteArrayInputStream(bytes)) + val input = new ByteArrayInputStream(bytes) + if (parser.options.streamMultilineTopLevelArray) { + safeParser.parseIterator( + input, + input => parser.parseIterator[InputStream]( + input, streamParser, _ => UTF8String.fromBytes(bytes))) + } else { + safeParser.parse(input) + } } } From a0a668dbbe9c32c9e70c5f53f4a271b5c9211f40 Mon Sep 17 00:00:00 2001 From: Tim Meehan Date: Thu, 10 Sep 2026 18:06:20 +0000 Subject: [PATCH 4/6] [SQL] Add focused top-level JSON array benchmark runner --- .../sql/execution/datasources/json/JsonBenchmark.scala | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/json/JsonBenchmark.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/json/JsonBenchmark.scala index 4db02a7fb6829..1f3a06ffed398 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/json/JsonBenchmark.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/json/JsonBenchmark.scala @@ -614,6 +614,13 @@ object JsonBenchmark extends SqlBasedBenchmark { override def runBenchmarkSuite(mainArgs: Array[String]): Unit = { val numIters = 3 + if (mainArgs.contains("top-level-array")) { + runBenchmark("Benchmark for top-level JSON array parsing") { + topLevelArrayBenchmark(rowsNum = 100000, payloadSize = 0, numIters = numIters) + topLevelArrayBenchmark(rowsNum = 1000, payloadSize = 64 * 1024, numIters = numIters) + } + return + } runBenchmark("Benchmark for performance of JSON parsing") { schemaInferring(5 * 1000 * 1000, numIters) countShortColumn(5 * 1000 * 1000, numIters) From 94fe0585d66e62f627458fa6aebdffa6476affff Mon Sep 17 00:00:00 2001 From: Tim Meehan Date: Thu, 10 Sep 2026 20:43:27 +0000 Subject: [PATCH 5/6] [SQL] Cover lazy JSON failures in all parse modes --- .../sql/catalyst/util/FailureSafeParser.scala | 2 +- .../datasources/json/JsonSuite.scala | 46 +++++++++++++++---- 2 files changed, 38 insertions(+), 10 deletions(-) diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/FailureSafeParser.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/FailureSafeParser.scala index 408017ada2cb0..62f8a3594bbf3 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/FailureSafeParser.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/FailureSafeParser.scala @@ -101,7 +101,7 @@ class FailureSafeParser[IN]( e.getCause match { case _: JsonArraysAsStructsException => // SPARK-42298 we recreate the exception here to make sure the error message - // have the record content. + // has the record content. throw QueryExecutionErrors.cannotParseJsonArraysAsStructsError(e.record().toString) case StringAsDataTypeException(fieldName, fieldValue, dataType) => throw QueryExecutionErrors.cannotParseStringAsDataTypeError(e.record().toString, 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 199a834329e32..f82d36e0e987f 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 @@ -33,9 +33,11 @@ import org.apache.hadoop.io.compress.{CompressionCodecFactory, GzipCodec} import org.apache.spark.{SparkConf, SparkException, SparkRuntimeException, SparkUpgradeException, TestUtils} import org.apache.spark.SparkIllegalArgumentException import org.apache.spark.io.ZStdCompressionCodec +import org.apache.spark.paths.SparkPath import org.apache.spark.rdd.RDD import org.apache.spark.scheduler.{SparkListener, SparkListenerJobEnd} import org.apache.spark.sql.{functions => F, _} +import org.apache.spark.sql.catalyst.InternalRow import org.apache.spark.sql.catalyst.json._ import org.apache.spark.sql.catalyst.util.{CharsetProvider, DateTimeTestUtils, DateTimeUtils, HadoopCompressionCodec} import org.apache.spark.sql.catalyst.util.HadoopCompressionCodec.GZIP @@ -44,7 +46,7 @@ import org.apache.spark.sql.catalyst.util.TimestampNanosTestUtils.foreachNanosPr import org.apache.spark.sql.catalyst.util.TypeUtils.toSQLType import org.apache.spark.sql.errors.QueryExecutionErrors.toSQLId import org.apache.spark.sql.execution.ExternalRDD -import org.apache.spark.sql.execution.datasources.{CommonFileDataSourceSuite, DataSource, InMemoryFileIndex, NoopCache} +import org.apache.spark.sql.execution.datasources.{CommonFileDataSourceSuite, DataSource, InMemoryFileIndex, NoopCache, PartitionedFile} import org.apache.spark.sql.execution.datasources.v2.json.JsonScanBuilder import org.apache.spark.sql.internal.SQLConf import org.apache.spark.sql.test.SharedSparkSession @@ -1093,18 +1095,44 @@ abstract class JsonSuite assert(input.available() > 0) } - test("multiline top level JSON array keeps rows emitted before malformed input") { + gridTest("multiline top level JSON array keeps rows emitted before malformed input")( + Seq("PERMISSIVE", "DROPMALFORMED", "FAILFAST")) { mode => withSQLConf(SQLConf.JSON_STREAM_MULTILINE_TOP_LEVEL_ARRAY.key -> "true") { withTempPath { file => val document = """[{"a":1} {"a":2}]""" Files.write(file.toPath, document.getBytes(StandardCharsets.UTF_8)) - - checkAnswer( - spark.read - .option("multiLine", true) - .schema("a int, _corrupt_record string") - .json(file.getCanonicalPath), - Seq(Row(1, null), Row(null, document))) + val actualSchema = StructType(Seq(StructField("a", IntegerType))) + val schema = StructType(Seq( + actualSchema.head, + StructField("_corrupt_record", StringType))) + val options = new JSONOptions( + Map("multiLine" -> "true", "mode" -> mode), + SQLConf.get.sessionLocalTimeZone, + SQLConf.get.columnNameOfCorruptRecord) + val parser = new JacksonParser(actualSchema, options, allowArrayAsStructs = true) + val partitionedFile = PartitionedFile( + InternalRow.empty, + SparkPath.fromPathString(file.getCanonicalPath), + 0, + file.length()) + val rows = MultiLineJsonDataSource.readFile( + spark.sessionState.newHadoopConf(), partitionedFile, parser, schema) + + assert(rows.next().getInt(0) === 1) + mode match { + case "PERMISSIVE" => + val corruptRow = rows.next() + assert(corruptRow.isNullAt(0)) + val corruptRecord = corruptRow.getUTF8String(1) + assert(corruptRecord != null, corruptRow.toString) + assert(corruptRecord.toString === document) + assert(!rows.hasNext) + case "DROPMALFORMED" => + assert(!rows.hasNext) + case "FAILFAST" => + val error = intercept[SparkException](rows.hasNext) + assert(error.getCondition === "MALFORMED_RECORD_IN_PARSING.WITHOUT_SUGGESTION") + } } } } From 9c1cb0afd79ea4a8c55f02ce9697e8a9fcd53ff0 Mon Sep 17 00:00:00 2001 From: Tim Meehan Date: Thu, 10 Sep 2026 22:04:06 +0000 Subject: [PATCH 6/6] [SQL] Benchmark configured streaming JSON path --- .../datasources/json/JsonBenchmark.scala | 39 +++++++++---------- 1 file changed, 19 insertions(+), 20 deletions(-) diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/json/JsonBenchmark.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/json/JsonBenchmark.scala index 1f3a06ffed398..5b21082dfd82e 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/json/JsonBenchmark.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/json/JsonBenchmark.scala @@ -16,19 +16,17 @@ */ package org.apache.spark.sql.execution.datasources.json -import java.io.{ByteArrayInputStream, File, InputStream} +import java.io.File +import java.nio.charset.StandardCharsets +import java.nio.file.Files import java.time.{Instant, LocalDate} -import com.fasterxml.jackson.core.JsonFactory - import org.apache.spark.benchmark.Benchmark import org.apache.spark.sql.{Column, Dataset, Row} -import org.apache.spark.sql.catalyst.json.{CreateJacksonParser, JacksonParser, JSONOptions} import org.apache.spark.sql.execution.benchmark.SqlBasedBenchmark import org.apache.spark.sql.functions._ import org.apache.spark.sql.internal.SQLConf import org.apache.spark.sql.types._ -import org.apache.spark.unsafe.types.UTF8String /** * The benchmarks aims to measure performance of JSON parsing when encoding is set and isn't. @@ -587,29 +585,30 @@ object JsonBenchmark extends SqlBasedBenchmark { payloadSize: Int, numIters: Int): Unit = { val payload = "x" * payloadSize - val bytes = (0 until rowsNum) + val document = (0 until rowsNum) .map(i => s"""{"a":$i,"payload":"$payload"}""") .mkString("[", ",", "]") - .getBytes("UTF-8") val schema = new StructType().add("a", IntegerType).add("payload", StringType) - val options = new JSONOptions(Map("multiLine" -> "true"), SQLConf.get.sessionLocalTimeZone) - val parser = new JacksonParser(schema, options, allowArrayAsStructs = true) - val createParser = CreateJacksonParser.inputStream(_: JsonFactory, _: InputStream) - val recordLiteral = (_: InputStream) => UTF8String.EMPTY_UTF8 val benchmark = new Benchmark( s"Top-level JSON array with $payloadSize-byte payloads", rowsNum, output = output) - benchmark.addCase("whole document", numIters) { _ => - val input = new ByteArrayInputStream(bytes) - parser.parse(input, createParser, recordLiteral).size - } + withTempPath { path => + Files.write(path.toPath, document.getBytes(StandardCharsets.UTF_8)) + + Seq(false, true).foreach { enabled => + benchmark.addCase(s"streaming enabled: $enabled", numIters) { _ => + withSQLConf(SQLConf.JSON_STREAM_MULTILINE_TOP_LEVEL_ARRAY.key -> enabled.toString) { + spark.read + .option("multiLine", true) + .schema(schema) + .json(path.getCanonicalPath) + .noop() + } + } + } - benchmark.addCase("streamed", numIters) { _ => - val input = new ByteArrayInputStream(bytes) - parser.parseIterator(input, createParser, recordLiteral).size + benchmark.run() } - - benchmark.run() } override def runBenchmarkSuite(mainArgs: Array[String]): Unit = {