This is an automated email from the ASF dual-hosted git repository.
cloud-fan pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new 4cf0be98c6f3 [SPARK-59417][SQL] Stream multiline top-level JSON arrays
4cf0be98c6f3 is described below
commit 4cf0be98c6f3d5a7cce4e433172ccf196e5393ba
Author: Timothy Meehan <[email protected]>
AuthorDate: Thu Oct 8 17:26:01 2026 +0000
[SPARK-59417][SQL] Stream multiline top-level JSON arrays
### What changes were proposed in this pull request?
This is a backport of #58704 to `branch-4.x`. It is a clean cherry-pick of
the merged commit `f4327af76197334dbf893a6f7ae34e2d4a096c5f`; the only
conflicts were the regenerated `JsonBenchmark*-results.txt` files, resolved to
the committed benchmark output (identical benchmark structure plus the two new
top-level-array sections).
This PR adds the Spark SQL configuration
`spark.sql.json.enableStreamingTopLevelArray` and the per-read JSON option
`enableStreamingTopLevelArray`, which overrides it. When enabled, reading a
multiline JSON file consumes a top-level array of structs lazily from one
Jackson parser instead of materializing the entire array before returning rows.
Both default to false. The two paths agree on valid input but differ on
malformed input, described below, so this is a choice a read makes rath [...]
It also adds an opt-in lazy path to `FailureSafeParser` for failures raised
while advancing the parser iterator, preserves rows emitted before later
structural corruption, and adds benchmark cases for both many small elements
and fewer large elements. Existing eager `FailureSafeParser` callers remain
unchanged.
A malformed element is resumed from only where the parser cannot have moved
past it: a scalar element leaves the parser on that scalar's own token, and
`convertObject` consumes through the closing brace before reporting a partial
result. Any other container failure can stop anywhere inside its element, so it
ends the document as before.
Because the streaming iterator holds a `JsonParser` for the life of the
read, it registers `close()` with the active `TaskContext` completion listener
in addition to closing eagerly when the array ends or a terminal failure
occurs. An iterator abandoned before the closing bracket (under a `LIMIT`, for
example) therefore releases the parser when the task finishes rather than at
GC. The listener has no removal API, so the parser is held in a cell that every
close path clears before clos [...]
### Why are the changes needed?
The existing parser materializes every element of a multiline top-level
JSON array in memory. Large arrays can therefore require memory proportional to
the complete document even though Spark consumes the result as an iterator.
Streaming elements bounds parser-side memory, producing the same rows as the
eager path for valid input. For malformed input it changes which unit a parse
mode applies to, which is why the path is opt-in rather than the default.
### Does this PR introduce _any_ user-facing change?
Yes, when the configuration or the JSON option is enabled. Reading a
multiline top-level JSON array no longer materializes the complete array before
producing rows, and the record a parse mode applies to becomes one array
element rather than the whole document. This follows from streaming: rows
already handed to the consumer cannot be withdrawn or retroactively stamped, so
preserving the document as the record would mean buffering to the closing
bracket.
Reading `[{"a":1},<malformed>,{"a":2}]` under `PERMISSIVE` with schema `a
int, _corrupt_record string`, writing `D` for the whole document:
| malformed element | disabled | enabled |
| --- | --- | --- |
| `{"a":"bad"}` | `(1, D)`, `(null, D)`, `(2, D)` | `(1, null)`, `(null,
D)`, `(2, null)` |
| `42` | `(null, D)` | `(1, null)`, `(null, D)`, `(2, null)` |
| `[1,2]` | `(null, D)` | `(1, null)`, `(null, D)` |
`DROPMALFORMED` drops the corrupt row of each cell, which leaves no rows at
all when disabled since every row carries the document's corruption. `FAILFAST`
raises `MALFORMED_RECORD_IN_PARSING.WITHOUT_SUGGESTION` in both columns.
The last row is the element whose failure leaves the parser at an unknown
position, so enabling the configuration keeps the rows before it and still ends
the document there. The configuration doc says so.
One consequence reaches valid documents of a shape the eager path never
supported: a top-level array whose elements are all scalars, such as `[1,2,3]`,
is malformed against a struct schema in every element, so it reports one
corrupt record per element where the whole document reported one. Object
elements already behaved that way, through the partial-result path the eager
reader has always had.
Valid input is unaffected, and both the configuration and the option
default to false.
### How was this patch tested?
The regression coverage from #58704 is carried over unchanged: lazy
top-level array parsing; the JSON option overriding the session configuration
in both directions; malformed input after partial output under PERMISSIVE,
DROPMALFORMED, and FAILFAST with the configuration both disabled and enabled,
so every cell of the table above is pinned in both columns; the three element
shapes parameterized over one document (pinning the nested array as the shape
that must stay terminal); extended [...]
Backport verification: the code, test, and doc changes are byte-identical
to the merged commit (confirmed by a content-level diff against `f4327af`).
Compiled `sql/Test/compile` (catalyst + sql/core, main and test) on
`branch-4.x`. The suites above run here in CI.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Isaac (Claude Opus 4.8)
The code, tests, and docs are an unmodified cherry-pick of #58704
(generative-AI tooling for the original change is disclosed there). This
backport — the cherry-pick, the benchmark-file conflict resolution, and this
description — was prepared with Isaac.
This pull request and its description were written by Isaac.
Closes #59288 from tdcmeehan/tim/sc-173438-stream-json-root-array-4.x.
Authored-by: Timothy Meehan <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>
---
docs/sql-data-sources-json.md | 6 +
.../spark/sql/catalyst/json/JSONOptions.scala | 5 +
.../spark/sql/catalyst/json/JacksonParser.scala | 197 ++++--
.../sql/catalyst/util/BadRecordException.scala | 4 +-
.../sql/catalyst/util/FailureSafeParser.scala | 77 ++-
.../org/apache/spark/sql/internal/SQLConf.scala | 20 +
.../benchmarks/JsonBenchmark-jdk21-results.txt | 172 ++---
.../benchmarks/JsonBenchmark-jdk25-results.txt | 172 ++---
sql/core/benchmarks/JsonBenchmark-results.txt | 172 ++---
.../datasources/json/JsonDataSource.scala | 26 +-
.../datasources/JSONArchiveReadBase.scala | 70 +-
.../execution/datasources/json/JsonBenchmark.scala | 51 ++
.../sql/execution/datasources/json/JsonSuite.scala | 729 ++++++++++++++++++---
13 files changed, 1293 insertions(+), 408 deletions(-)
diff --git a/docs/sql-data-sources-json.md b/docs/sql-data-sources-json.md
index 6ac5d690868c..aadd3c25a780 100644
--- a/docs/sql-data-sources-json.md
+++ b/docs/sql-data-sources-json.md
@@ -237,6 +237,12 @@ Data source options of JSON can be set via:
<td>Parse one record, which may span multiple lines, per file. JSON
built-in functions ignore this option.</td>
<td>read</td>
</tr>
+ <tr>
+ <td><code>enableStreamingTopLevelArray</code></td>
+ <td>(value of <code>spark.sql.json.enableStreamingTopLevelArray</code>
configuration)</td>
+ <td>When <code>multiLine</code> is enabled and a file holds a top-level
JSON array, read the array's elements one at a time instead of materializing
the whole array before returning rows. It applies to reads into a struct
schema, and has no effect on reads using <code>singleVariantColumn</code> or
<code>explodeEmbeddedArray</code>. While streaming, <code>mode</code> applies
to an individual element rather than the whole document:
<code>PERMISSIVE</code> fills the field configured by [...]
+ <td>read</td>
+ </tr>
<tr>
<td><code>allowUnquotedControlChars</code></td>
<td><code>false</code></td>
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 fba273bac265..98858076236e 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
@@ -244,6 +244,10 @@ class JSONOptions(
val useUnsafeRow: Boolean =
parameters.get(USE_UNSAFE_ROW).map(_.toBoolean).getOrElse(
SQLConf.get.getConf(SQLConf.JSON_USE_UNSAFE_ROW))
+ val streamMultilineTopLevelArray: Boolean =
+
parameters.get(ENABLE_STREAMING_TOP_LEVEL_ARRAY).map(_.toBoolean).getOrElse(
+ SQLConf.get.getConf(SQLConf.JSON_STREAM_MULTILINE_TOP_LEVEL_ARRAY))
+
/** Build a Jackson [[JsonFactory]] using JSON options. */
def buildJsonFactory(): JsonFactory = {
val streamReadConstraints = StreamReadConstraints
@@ -343,6 +347,7 @@ object JSONOptions extends DataSourceOptions {
val SINGLE_VARIANT_COLUMN =
newOption(DataSourceOptions.SINGLE_VARIANT_COLUMN)
val EXPLODE_EMBEDDED_ARRAY =
newOption(DataSourceOptions.EXPLODE_EMBEDDED_ARRAY)
val USE_UNSAFE_ROW = newOption("useUnsafeRow")
+ val ENABLE_STREAMING_TOP_LEVEL_ARRAY =
newOption("enableStreamingTopLevelArray")
// Options with alternative
val ENCODING = "encoding"
val CHARSET = "charset"
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 e1a277764d0f..e87b1d0ca2a8 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
@@ -19,6 +19,7 @@ package org.apache.spark.sql.catalyst.json
import java.io.{ByteArrayOutputStream, CharConversionException}
import java.nio.charset.MalformedInputException
+import java.util.concurrent.atomic.AtomicReference
import scala.collection.mutable
import scala.collection.mutable.ArrayBuffer
@@ -28,6 +29,7 @@ import com.fasterxml.jackson.core._
import org.apache.hadoop.fs.PositionedReadable
import org.apache.spark.SparkUpgradeException
+import org.apache.spark.TaskContext
import org.apache.spark.internal.Logging
import org.apache.spark.sql.catalyst.{InternalRow, NoopFilters, StructFilters}
import org.apache.spark.sql.catalyst.expressions._
@@ -61,6 +63,9 @@ class JacksonParser(
// `ValueConverter`s for the root schema for all fields in the schema
private val rootConverter = makeRootConverter(schema)
+ // Lazy: a schema that is not a struct never reaches either top-level-array
path.
+ private lazy val arrayElementConverter = makeConverter(schema)
+
private val factory = options.buildJsonFactory()
private lazy val timestampFormatter = TimestampFormatter(
@@ -148,7 +153,6 @@ class JacksonParser(
}
private def makeStructRootConverter(st: StructType): JsonParser =>
Iterable[InternalRow] = {
- val elementConverter = makeConverter(st)
val fieldConverters = st.map(_.dataType).map(makeConverter).toArray
val jsonFilters = if (SQLConf.get.jsonFilterPushDown) {
new JsonFilters(filters, st)
@@ -171,7 +175,8 @@ class JacksonParser(
// List([str_a_2,null], [null,str_b_3])
//
case START_ARRAY if allowArrayAsStructs =>
- val array = convertArray(parser, elementConverter, isRoot = true,
arrayAsStructs = true)
+ val array =
+ convertArray(parser, arrayElementConverter, isRoot = true,
arrayAsStructs = true)
// Here, as we support reading top level JSON arrays and take every
element
// in such an array as a row, this case is possible.
if (array.numElements() == 0) {
@@ -700,6 +705,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.
*
@@ -724,43 +757,131 @@ 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 jsonParser = try {
+ createParser(factory, record)
+ } catch {
+ case e: SparkUpgradeException => throw e
+ case e: CharConversionException if options.encoding.isEmpty =>
+ throw badRecord(e, () => recordLiteral(record))
+ case e @ (_: RuntimeException | _: JsonProcessingException | _:
MalformedInputException |
+ _: PartialResultException | _: PartialResultArrayException |
+ _: PartialArrayDataResultException | _:
PartialMapDataResultException) =>
+ throw badRecord(e, () => recordLiteral(record))
+ }
+ // An archive read builds one parser per entry inside a single task, so a
closed parser left
+ // reachable from the completion listener keeps its entry's buffer alive
for the rest of the
+ // task. Every close path takes the parser out of this cell first, so
exactly one of them closes
+ // it and the listener stops retaining it as soon as one does.
+ val openParser = new AtomicReference(jsonParser)
+ def takeParser(): Option[JsonParser] = Option(openParser.getAndSet(null))
+ // Cleanup must not fail a read that already emitted rows, nor a task
whose iterator was
+ // abandoned, so the paths that can still succeed close the way
`tryWithResource` does: quietly.
+ def closeParser(): Unit = takeParser().foreach(parser =>
Utils.closeQuietly(parser))
+ def fail(error: Throwable): Nothing = {
+ try takeParser().foreach(_.close()) catch {
+ case NonFatal(closeError) => error.addSuppressed(closeError)
+ }
+ 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
+ })
+ def handleFailure[R](elementStart: Option[JsonToken])(operation: => R): R
= {
+ try operation catch {
+ case e: SparkUpgradeException => fail(e)
+ case e: CharConversionException if options.encoding.isEmpty => fail(e)
+ case e @ (_: RuntimeException | _: JsonProcessingException | _:
MalformedInputException |
+ _: PartialResultException | _: PartialResultArrayException |
+ _: PartialArrayDataResultException | _:
PartialMapDataResultException) =>
+ if (resumableAfter(e, elementStart)) {
+ throw badRecord(e, () => recordLiteral(record)).copy(recoverable =
true)
+ } else {
+ fail(e)
+ }
+ // Anything else, such as a raw `IOException` off the underlying
stream, is not a malformed
+ // record and so propagates unwrapped -- but it still has to close
what the eager path's
+ // `tryWithResource` would have, rather than leaving it to task
completion.
+ case other: Throwable =>
+ closeParser()
+ throw other
+ }
+ }
+
+ handleFailure(elementStart = None)(jsonParser.nextToken()) match {
+ case null =>
+ closeParser()
+ Iterator.empty
+ case START_ARRAY if streamArray =>
+ // Abandoning the iterator before END_ARRAY (e.g. a LIMIT) reaches no
eager close.
+ Option(TaskContext.get()).foreach(_.addTaskCompletionListener[Unit](_
=> closeParser()))
+ new Iterator[InternalRow] {
+ private var nextRow: InternalRow = _
+ private var prepared = false
+ private var finished = 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
+ handleFailure(elementStart = None)(jsonParser.nextToken()) match {
+ case END_ARRAY => finish()
+ // Unreachable today: jackson-core signals end of input inside
an open array by
+ // throwing JsonEOFException, which handleFailure turns into a
malformed record.
+ // Kept so a backend that returns null cannot fall through with
no current token.
+ case null =>
+ fail(new JsonParseException(jsonParser, "Unexpected end of
top-level array"))
+ case elementStart =>
+ nextRow = handleFailure(Some(elementStart)) {
+ val row =
arrayElementConverter(jsonParser).asInstanceOf[InternalRow]
+ if (row == null) throw
QueryExecutionErrors.rootConverterReturnNullError()
+ row
+ }
+ prepared = true
+ }
+ }
+
+ private def finish(): Unit = {
+ finished = true
+ closeParser()
+ }
+ }
+ case _ =>
+ val rows = handleFailure(elementStart =
None)(rootConverter(jsonParser))
+ if (rows == null)
fail(QueryExecutionErrors.rootConverterReturnNullError())
+ closeParser()
+ rows.iterator
}
}
}
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/BadRecordException.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/BadRecordException.scala
index 4fa6a2275e74..54b85a6bde1b 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/BadRecordException.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/BadRecordException.scala
@@ -76,11 +76,13 @@ case class PartialResultArrayException(
* @param cause the actual exception about why the record is bad and can't be
parsed. It's better
* to use `LazyBadRecordCauseWrapper` here to delay heavy
cause construction
* until it's needed.
+ * @param recoverable whether the parser can resume at the next record after
handling this failure
*/
case class BadRecordException(
@transient record: () => UTF8String,
@transient partialResults: () => Array[InternalRow] = () =>
Array.empty[InternalRow],
- cause: Throwable) extends Exception(cause) {
+ cause: Throwable,
+ recoverable: Boolean = false) extends Exception(cause) {
override def getStackTrace(): Array[StackTraceElement] = new
Array[StackTraceElement](0)
override def fillInStackTrace(): Throwable = this
}
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 d9946d1b12ec..3c31e323491c 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
@@ -59,30 +59,63 @@ class FailureSafeParser[IN](
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)
+ }
+ }
+
+ 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 =
{
+ while (true) {
+ try {
+ return operation(delegate)
+ } catch {
+ case e: BadRecordException =>
+ val source = delegate
+ val recovery = parseFailure(e)
+ delegate = if (e.recoverable) recovery ++ source else recovery
}
+ }
+ throw new IllegalStateException("unreachable")
}
+
+ 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
+ // has 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/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 dcae5780c4cc..de2f850f9d14 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
@@ -7186,6 +7186,26 @@ object SQLConf {
.booleanConf
.createWithDefault(true)
+ val JSON_STREAM_MULTILINE_TOP_LEVEL_ARRAY =
+ buildConf("spark.sql.json.enableStreamingTopLevelArray")
+ .doc("When true, multiline JSON file reads stream the elements of a
top-level array one at " +
+ "a time instead of materializing the entire array before returning
rows. This applies " +
+ "only to reads into a struct schema that take top-level arrays as
structs, and has no " +
+ "effect on reads using the `singleVariantColumn` or
`explodeEmbeddedArray` option. " +
+ "Streaming also makes an array element, rather than the whole
document, the record " +
+ "that a parse mode applies to, since rows already emitted cannot be
withdrawn: " +
+ "PERMISSIVE fills the corrupt record column for the malformed element
only, leaving " +
+ "it null on the valid rows of the same document, and DROPMALFORMED
drops that " +
+ "element rather than the whole document. An element whose failure
leaves the parser " +
+ "at an unknown position, such as a nested value of the wrong shape,
still ends the " +
+ "document, as does a failure outside any element, such as a syntax
error between two " +
+ "elements or a missing closing bracket. It can be overwritten by the
JSON option " +
+ "`enableStreamingTopLevelArray`.")
+ .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/benchmarks/JsonBenchmark-jdk21-results.txt
b/sql/core/benchmarks/JsonBenchmark-jdk21-results.txt
index 2d49ed10a6ed..4f8b1d71af8f 100644
--- a/sql/core/benchmarks/JsonBenchmark-jdk21-results.txt
+++ b/sql/core/benchmarks/JsonBenchmark-jdk21-results.txt
@@ -3,128 +3,142 @@ Benchmark for performance of JSON parsing
================================================================================================
Preparing data for benchmarking ...
-OpenJDK 64-Bit Server VM 21.0.10+7-LTS on Linux 6.14.0-1017-azure
-AMD EPYC 7763 64-Core Processor
+OpenJDK 64-Bit Server VM 21.0.12+8-LTS on Linux 6.17.0-1022-azure
+AMD EPYC 9V74 80-Core Processor
JSON schema inferring: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
-No encoding 2505 2612
96 2.0 501.0 1.0X
-UTF-8 is set 5362 5380
16 0.9 1072.3 0.5X
+No encoding 2081 2116
47 2.4 416.2 1.0X
+UTF-8 is set 5253 5423
265 1.0 1050.6 0.4X
Preparing data for benchmarking ...
-OpenJDK 64-Bit Server VM 21.0.10+7-LTS on Linux 6.14.0-1017-azure
-AMD EPYC 7763 64-Core Processor
+OpenJDK 64-Bit Server VM 21.0.12+8-LTS on Linux 6.17.0-1022-azure
+AMD EPYC 9V74 80-Core Processor
count a short column: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
-No encoding 1952 1958
5 2.6 390.4 1.0X
-UTF-8 is set 4416 4420
5 1.1 883.3 0.4X
+No encoding 1789 1809
31 2.8 357.9 1.0X
+UTF-8 is set 4835 4838
2 1.0 967.0 0.4X
Preparing data for benchmarking ...
-OpenJDK 64-Bit Server VM 21.0.10+7-LTS on Linux 6.14.0-1017-azure
-AMD EPYC 7763 64-Core Processor
+OpenJDK 64-Bit Server VM 21.0.12+8-LTS on Linux 6.17.0-1022-azure
+AMD EPYC 9V74 80-Core Processor
count a wide column: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
-No encoding 4546 4551
5 0.2 4545.6 1.0X
-UTF-8 is set 4369 4375
10 0.2 4369.2 1.0X
+No encoding 4791 4805
14 0.2 4790.5 1.0X
+UTF-8 is set 5038 5052
15 0.2 5038.0 1.0X
Preparing data for benchmarking ...
-OpenJDK 64-Bit Server VM 21.0.10+7-LTS on Linux 6.14.0-1017-azure
-AMD EPYC 7763 64-Core Processor
+OpenJDK 64-Bit Server VM 21.0.12+8-LTS on Linux 6.17.0-1022-azure
+AMD EPYC 9V74 80-Core Processor
select wide row: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
-No encoding 10036 10149
153 0.0 200713.2 1.0X
-UTF-8 is set 10794 10832
34 0.0 215870.3 0.9X
+No encoding 10203 10286
96 0.0 204060.2 1.0X
+UTF-8 is set 10939 10969
40 0.0 218787.9 0.9X
Preparing data for benchmarking ...
-OpenJDK 64-Bit Server VM 21.0.10+7-LTS on Linux 6.14.0-1017-azure
-AMD EPYC 7763 64-Core Processor
+OpenJDK 64-Bit Server VM 21.0.12+8-LTS on Linux 6.17.0-1022-azure
+AMD EPYC 9V74 80-Core Processor
Select a subset of 10 columns: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
-Select 10 columns 1664 1665
1 0.6 1663.8 1.0X
-Select 1 column 1123 1125
3 0.9 1123.1 1.5X
+Select 10 columns 1759 1767
7 0.6 1759.1 1.0X
+Select 1 column 1157 1165
8 0.9 1156.7 1.5X
Preparing data for benchmarking ...
-OpenJDK 64-Bit Server VM 21.0.10+7-LTS on Linux 6.14.0-1017-azure
-AMD EPYC 7763 64-Core Processor
+OpenJDK 64-Bit Server VM 21.0.12+8-LTS on Linux 6.17.0-1022-azure
+AMD EPYC 9V74 80-Core Processor
creation of JSON parser per line: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
-Short column without encoding 583 586
3 1.7 583.3 1.0X
-Short column with UTF-8 1118 1124
7 0.9 1118.2 0.5X
-Wide column without encoding 5316 5335
18 0.2 5316.3 0.1X
-Wide column with UTF-8 8905 8913
7 0.1 8904.8 0.1X
+Short column without encoding 583 591
8 1.7 583.0 1.0X
+Short column with UTF-8 1288 1291
4 0.8 1288.4 0.5X
+Wide column without encoding 5623 5632
8 0.2 5622.8 0.1X
+Wide column with UTF-8 7468 7483
17 0.1 7467.5 0.1X
Preparing data for benchmarking ...
-OpenJDK 64-Bit Server VM 21.0.10+7-LTS on Linux 6.14.0-1017-azure
-AMD EPYC 7763 64-Core Processor
+OpenJDK 64-Bit Server VM 21.0.12+8-LTS on Linux 6.17.0-1022-azure
+AMD EPYC 9V74 80-Core Processor
JSON functions: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
-Text read 68 70
2 14.6 68.3 1.0X
-from_json 1085 1089
5 0.9 1084.6 0.1X
-json_tuple 1060 1061
1 0.9 1059.6 0.1X
-get_json_object wholestage off 1070 1071
2 0.9 1070.1 0.1X
-get_json_object wholestage on 990 994
4 1.0 989.6 0.1X
+Text read 66 71
5 15.2 66.0 1.0X
+from_json 1029 1034
5 1.0 1029.0 0.1X
+json_tuple 971 974
3 1.0 971.0 0.1X
+get_json_object wholestage off 986 994
11 1.0 986.0 0.1X
+get_json_object wholestage on 886 894
7 1.1 885.5 0.1X
Preparing data for benchmarking ...
-OpenJDK 64-Bit Server VM 21.0.10+7-LTS on Linux 6.14.0-1017-azure
-AMD EPYC 7763 64-Core Processor
+OpenJDK 64-Bit Server VM 21.0.12+8-LTS on Linux 6.17.0-1022-azure
+AMD EPYC 9V74 80-Core Processor
Dataset of json strings: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
-Text read 240 241
1 20.8 48.1 1.0X
-schema inferring 1831 1836
4 2.7 366.2 0.1X
-parsing 2510 2518
11 2.0 502.1 0.1X
+Text read 260 261
2 19.2 52.0 1.0X
+schema inferring 1594 1600
9 3.1 318.8 0.2X
+parsing 2212 2216
4 2.3 442.4 0.1X
Preparing data for benchmarking ...
-OpenJDK 64-Bit Server VM 21.0.10+7-LTS on Linux 6.14.0-1017-azure
-AMD EPYC 7763 64-Core Processor
+OpenJDK 64-Bit Server VM 21.0.12+8-LTS on Linux 6.17.0-1022-azure
+AMD EPYC 9V74 80-Core Processor
Json files in the per-line mode: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
-Text read 618 623
5 8.1 123.5 1.0X
-Schema inferring 2382 2384
2 2.1 476.5 0.3X
-Parsing without charset 2665 2670
5 1.9 533.0 0.2X
-Parsing with UTF-8 5166 5180
12 1.0 1033.2 0.1X
+Text read 605 612
7 8.3 121.1 1.0X
+Schema inferring 2108 2113
8 2.4 421.5 0.3X
+Parsing without charset 2673 2683
13 1.9 534.6 0.2X
+Parsing with UTF-8 5951 5955
4 0.8 1190.3 0.1X
-OpenJDK 64-Bit Server VM 21.0.10+7-LTS on Linux 6.14.0-1017-azure
-AMD EPYC 7763 64-Core Processor
+OpenJDK 64-Bit Server VM 21.0.12+8-LTS on Linux 6.17.0-1022-azure
+AMD EPYC 9V74 80-Core Processor
Write dates and timestamps: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
-Create a dataset of timestamps 105 108
3 9.5 105.5 1.0X
-to_json(timestamp) 555 557
2 1.8 555.1 0.2X
-write timestamps to files 618 626
12 1.6 617.8 0.2X
-Create a dataset of dates 112 117
4 8.9 112.2 0.9X
-to_json(date) 420 420
0 2.4 420.3 0.3X
-write dates to files 401 403
2 2.5 401.4 0.3X
-
-OpenJDK 64-Bit Server VM 21.0.10+7-LTS on Linux 6.14.0-1017-azure
-AMD EPYC 7763 64-Core Processor
+Create a dataset of timestamps 111 114
2 9.0 110.8 1.0X
+to_json(timestamp) 509 514
7 2.0 509.0 0.2X
+write timestamps to files 534 538
4 1.9 533.7 0.2X
+Create a dataset of dates 122 129
7 8.2 122.3 0.9X
+to_json(date) 366 367
1 2.7 366.4 0.3X
+write dates to files 376 383
9 2.7 376.2 0.3X
+
+OpenJDK 64-Bit Server VM 21.0.12+8-LTS on Linux 6.17.0-1022-azure
+AMD EPYC 9V74 80-Core Processor
Read dates and timestamps: Best
Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
-----------------------------------------------------------------------------------------------------------------------------------------------------
-read timestamp text from files
149 153 7 6.7 148.5 1.0X
-read timestamps from files
1052 1059 7 1.0 1051.7 0.1X
-infer timestamps from files
1976 1980 4 0.5 1976.4 0.1X
-read date text from files
154 157 3 6.5 153.5 1.0X
-read date from files
647 655 6 1.5 647.4 0.2X
-timestamp strings
146 147 0 6.8 146.5 1.0X
-parse timestamps from Dataset[String]
1226 1228 2 0.8 1226.2 0.1X
-infer timestamps from Dataset[String]
2150 2159 12 0.5 2149.5 0.1X
-date strings
198 200 2 5.0 198.3 0.7X
-parse dates from Dataset[String]
920 923 3 1.1 920.4 0.2X
-from_json(timestamp)
1749 1751 4 0.6 1748.6 0.1X
-from_json(date)
1464 1468 4 0.7 1464.3 0.1X
-infer error timestamps from Dataset[String] with default format
1358 1362 3 0.7 1358.0 0.1X
-infer error timestamps from Dataset[String] with user-provided format
1357 1360 3 0.7 1356.8 0.1X
-infer error timestamps from Dataset[String] with legacy format
1384 1389 5 0.7 1383.5 0.1X
-
-OpenJDK 64-Bit Server VM 21.0.10+7-LTS on Linux 6.14.0-1017-azure
-AMD EPYC 7763 64-Core Processor
+read timestamp text from files
160 161 1 6.2 160.1 1.0X
+read timestamps from files
1109 1117 8 0.9 1108.8 0.1X
+infer timestamps from files
2029 2030 2 0.5 2028.8 0.1X
+read date text from files
147 149 3 6.8 146.7 1.1X
+read date from files
692 701 14 1.4 691.5 0.2X
+timestamp strings
144 147 5 7.0 143.7 1.1X
+parse timestamps from Dataset[String]
1309 1312 3 0.8 1309.5 0.1X
+infer timestamps from Dataset[String]
2229 2231 3 0.4 2229.3 0.1X
+date strings
206 208 2 4.9 205.8 0.8X
+parse dates from Dataset[String]
957 960 4 1.0 956.5 0.2X
+from_json(timestamp)
1744 1749 5 0.6 1744.0 0.1X
+from_json(date)
1433 1437 4 0.7 1433.3 0.1X
+infer error timestamps from Dataset[String] with default format
1299 1302 4 0.8 1299.5 0.1X
+infer error timestamps from Dataset[String] with user-provided format
1304 1307 4 0.8 1303.9 0.1X
+infer error timestamps from Dataset[String] with legacy format
1335 1337 3 0.7 1334.7 0.1X
+
+OpenJDK 64-Bit Server VM 21.0.12+8-LTS on Linux 6.17.0-1022-azure
+AMD EPYC 9V74 80-Core Processor
Filters pushdown: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
-w/o filters 6285 6289
4 0.0 62852.4 1.0X
-pushdown disabled 6190 6193
5 0.0 61896.8 1.0X
-w/ filters 599 602
6 0.2 5989.5 10.5X
+w/o filters 5507 5528
35 0.0 55073.9 1.0X
+pushdown disabled 5300 5309
8 0.0 52999.5 1.0X
+w/ filters 744 747
3 0.1 7443.5 7.4X
-OpenJDK 64-Bit Server VM 21.0.10+7-LTS on Linux 6.14.0-1017-azure
-AMD EPYC 7763 64-Core Processor
+OpenJDK 64-Bit Server VM 21.0.12+8-LTS on Linux 6.17.0-1022-azure
+AMD EPYC 9V74 80-Core Processor
Partial JSON results: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
-parse invalid JSON 2393 2422
43 0.0 239282.1 1.0X
+parse invalid JSON 2465 2468
3 0.0 246496.7 1.0X
+
+OpenJDK 64-Bit Server VM 21.0.12+8-LTS on Linux 6.17.0-1022-azure
+AMD EPYC 9V74 80-Core Processor
+Top-level JSON array with 0-byte payloads: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
+-------------------------------------------------------------------------------------------------------------------------
+streaming enabled: false 75 82
8 1.3 752.2 1.0X
+streaming enabled: true 74 78
3 1.3 741.4 1.0X
+
+OpenJDK 64-Bit Server VM 21.0.12+8-LTS on Linux 6.17.0-1022-azure
+AMD EPYC 9V74 80-Core Processor
+Top-level JSON array with 65536-byte payloads: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
+-----------------------------------------------------------------------------------------------------------------------------
+streaming enabled: false 136 139
3 0.0 135565.5 1.0X
+streaming enabled: true 133 134
1 0.0 133383.2 1.0X
diff --git a/sql/core/benchmarks/JsonBenchmark-jdk25-results.txt
b/sql/core/benchmarks/JsonBenchmark-jdk25-results.txt
index a947a34a0ba8..e2b256f1b924 100644
--- a/sql/core/benchmarks/JsonBenchmark-jdk25-results.txt
+++ b/sql/core/benchmarks/JsonBenchmark-jdk25-results.txt
@@ -3,128 +3,142 @@ Benchmark for performance of JSON parsing
================================================================================================
Preparing data for benchmarking ...
-OpenJDK 64-Bit Server VM 25.0.2+10-LTS on Linux 6.14.0-1017-azure
-AMD EPYC 7763 64-Core Processor
+OpenJDK 64-Bit Server VM 25.0.4+7-LTS on Linux 6.17.0-1022-azure
+AMD EPYC 9V74 80-Core Processor
JSON schema inferring: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
-No encoding 2323 2386
96 2.2 464.5 1.0X
-UTF-8 is set 4970 4983
18 1.0 993.9 0.5X
+No encoding 2032 2116
98 2.5 406.4 1.0X
+UTF-8 is set 4421 4489
67 1.1 884.2 0.5X
Preparing data for benchmarking ...
-OpenJDK 64-Bit Server VM 25.0.2+10-LTS on Linux 6.14.0-1017-azure
-AMD EPYC 7763 64-Core Processor
+OpenJDK 64-Bit Server VM 25.0.4+7-LTS on Linux 6.17.0-1022-azure
+AMD EPYC 9V74 80-Core Processor
count a short column: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
-No encoding 2147 2199
67 2.3 429.4 1.0X
-UTF-8 is set 4826 4846
23 1.0 965.1 0.4X
+No encoding 1734 1744
9 2.9 346.8 1.0X
+UTF-8 is set 3939 3950
14 1.3 787.9 0.4X
Preparing data for benchmarking ...
-OpenJDK 64-Bit Server VM 25.0.2+10-LTS on Linux 6.14.0-1017-azure
-AMD EPYC 7763 64-Core Processor
+OpenJDK 64-Bit Server VM 25.0.4+7-LTS on Linux 6.17.0-1022-azure
+AMD EPYC 9V74 80-Core Processor
count a wide column: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
-No encoding 4544 4564
22 0.2 4544.2 1.0X
-UTF-8 is set 4411 4424
17 0.2 4411.0 1.0X
+No encoding 4737 4744
6 0.2 4736.7 1.0X
+UTF-8 is set 4481 4502
22 0.2 4481.0 1.1X
Preparing data for benchmarking ...
-OpenJDK 64-Bit Server VM 25.0.2+10-LTS on Linux 6.14.0-1017-azure
-AMD EPYC 7763 64-Core Processor
+OpenJDK 64-Bit Server VM 25.0.4+7-LTS on Linux 6.17.0-1022-azure
+AMD EPYC 9V74 80-Core Processor
select wide row: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
-No encoding 9487 9565
91 0.0 189740.3 1.0X
-UTF-8 is set 10224 10258
42 0.0 204476.7 0.9X
+No encoding 10142 10236
89 0.0 202833.6 1.0X
+UTF-8 is set 10400 10421
28 0.0 208006.9 1.0X
Preparing data for benchmarking ...
-OpenJDK 64-Bit Server VM 25.0.2+10-LTS on Linux 6.14.0-1017-azure
-AMD EPYC 7763 64-Core Processor
+OpenJDK 64-Bit Server VM 25.0.4+7-LTS on Linux 6.17.0-1022-azure
+AMD EPYC 9V74 80-Core Processor
Select a subset of 10 columns: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
-Select 10 columns 1678 1682
6 0.6 1677.6 1.0X
-Select 1 column 1198 1200
3 0.8 1197.5 1.4X
+Select 10 columns 1553 1557
4 0.6 1553.1 1.0X
+Select 1 column 1154 1193
34 0.9 1154.3 1.3X
Preparing data for benchmarking ...
-OpenJDK 64-Bit Server VM 25.0.2+10-LTS on Linux 6.14.0-1017-azure
-AMD EPYC 7763 64-Core Processor
+OpenJDK 64-Bit Server VM 25.0.4+7-LTS on Linux 6.17.0-1022-azure
+AMD EPYC 9V74 80-Core Processor
creation of JSON parser per line: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
-Short column without encoding 621 627
5 1.6 621.4 1.0X
-Short column with UTF-8 1178 1184
11 0.8 1177.8 0.5X
-Wide column without encoding 4988 5023
31 0.2 4987.6 0.1X
-Wide column with UTF-8 6444 6510
61 0.2 6443.6 0.1X
+Short column without encoding 470 471
2 2.1 470.2 1.0X
+Short column with UTF-8 1008 1010
2 1.0 1007.9 0.5X
+Wide column without encoding 5437 5467
34 0.2 5437.4 0.1X
+Wide column with UTF-8 7125 7134
10 0.1 7125.4 0.1X
Preparing data for benchmarking ...
-OpenJDK 64-Bit Server VM 25.0.2+10-LTS on Linux 6.14.0-1017-azure
-AMD EPYC 7763 64-Core Processor
+OpenJDK 64-Bit Server VM 25.0.4+7-LTS on Linux 6.17.0-1022-azure
+AMD EPYC 9V74 80-Core Processor
JSON functions: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
-Text read 50 51
2 19.9 50.2 1.0X
-from_json 1046 1048
2 1.0 1045.9 0.0X
-json_tuple 931 958
23 1.1 931.4 0.1X
-get_json_object wholestage off 998 1001
3 1.0 998.2 0.1X
-get_json_object wholestage on 913 915
4 1.1 912.5 0.1X
+Text read 67 71
6 15.0 66.7 1.0X
+from_json 938 956
25 1.1 938.2 0.1X
+json_tuple 899 901
1 1.1 899.2 0.1X
+get_json_object wholestage off 952 962
10 1.0 952.4 0.1X
+get_json_object wholestage on 859 862
3 1.2 859.0 0.1X
Preparing data for benchmarking ...
-OpenJDK 64-Bit Server VM 25.0.2+10-LTS on Linux 6.14.0-1017-azure
-AMD EPYC 7763 64-Core Processor
+OpenJDK 64-Bit Server VM 25.0.4+7-LTS on Linux 6.17.0-1022-azure
+AMD EPYC 9V74 80-Core Processor
Dataset of json strings: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
-Text read 210 211
1 23.8 42.0 1.0X
-schema inferring 1780 1785
7 2.8 356.0 0.1X
-parsing 2586 2592
9 1.9 517.3 0.1X
+Text read 224 231
11 22.3 44.8 1.0X
+schema inferring 1500 1503
3 3.3 300.0 0.1X
+parsing 2146 2154
12 2.3 429.2 0.1X
Preparing data for benchmarking ...
-OpenJDK 64-Bit Server VM 25.0.2+10-LTS on Linux 6.14.0-1017-azure
-AMD EPYC 7763 64-Core Processor
+OpenJDK 64-Bit Server VM 25.0.4+7-LTS on Linux 6.17.0-1022-azure
+AMD EPYC 9V74 80-Core Processor
Json files in the per-line mode: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
-Text read 574 577
4 8.7 114.8 1.0X
-Schema inferring 2344 2347
2 2.1 468.9 0.2X
-Parsing without charset 2927 2935
7 1.7 585.3 0.2X
-Parsing with UTF-8 5678 5682
4 0.9 1135.5 0.1X
+Text read 574 580
8 8.7 114.7 1.0X
+Schema inferring 1848 1854
6 2.7 369.6 0.3X
+Parsing without charset 2249 2258
9 2.2 449.9 0.3X
+Parsing with UTF-8 4644 4647
3 1.1 928.7 0.1X
-OpenJDK 64-Bit Server VM 25.0.2+10-LTS on Linux 6.14.0-1017-azure
-AMD EPYC 7763 64-Core Processor
+OpenJDK 64-Bit Server VM 25.0.4+7-LTS on Linux 6.17.0-1022-azure
+AMD EPYC 9V74 80-Core Processor
Write dates and timestamps: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
-Create a dataset of timestamps 99 103
4 10.1 99.0 1.0X
-to_json(timestamp) 533 535
3 1.9 532.7 0.2X
-write timestamps to files 589 594
4 1.7 589.2 0.2X
-Create a dataset of dates 105 107
2 9.5 104.8 0.9X
-to_json(date) 398 402
3 2.5 398.3 0.2X
-write dates to files 398 407
12 2.5 397.8 0.2X
-
-OpenJDK 64-Bit Server VM 25.0.2+10-LTS on Linux 6.14.0-1017-azure
-AMD EPYC 7763 64-Core Processor
+Create a dataset of timestamps 106 109
3 9.4 106.3 1.0X
+to_json(timestamp) 489 490
1 2.0 488.6 0.2X
+write timestamps to files 547 551
4 1.8 546.7 0.2X
+Create a dataset of dates 117 119
2 8.6 116.9 0.9X
+to_json(date) 344 347
3 2.9 344.4 0.3X
+write dates to files 367 369
2 2.7 366.8 0.3X
+
+OpenJDK 64-Bit Server VM 25.0.4+7-LTS on Linux 6.17.0-1022-azure
+AMD EPYC 9V74 80-Core Processor
Read dates and timestamps: Best
Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
-----------------------------------------------------------------------------------------------------------------------------------------------------
-read timestamp text from files
143 144 2 7.0 142.6 1.0X
-read timestamps from files
1006 1009 3 1.0 1005.7 0.1X
-infer timestamps from files
1849 1849 0 0.5 1849.1 0.1X
-read date text from files
140 140 0 7.1 140.0 1.0X
-read date from files
681 684 5 1.5 680.5 0.2X
-timestamp strings
132 134 2 7.6 131.8 1.1X
-parse timestamps from Dataset[String]
1134 1136 2 0.9 1134.1 0.1X
-infer timestamps from Dataset[String]
1935 1938 3 0.5 1934.7 0.1X
-date strings
194 198 4 5.1 194.3 0.7X
-parse dates from Dataset[String]
935 936 1 1.1 935.1 0.2X
-from_json(timestamp)
1563 1569 6 0.6 1562.8 0.1X
-from_json(date)
1372 1398 35 0.7 1371.8 0.1X
-infer error timestamps from Dataset[String] with default format
1335 1338 3 0.7 1335.2 0.1X
-infer error timestamps from Dataset[String] with user-provided format
1318 1332 16 0.8 1317.6 0.1X
-infer error timestamps from Dataset[String] with legacy format
1354 1356 1 0.7 1354.1 0.1X
-
-OpenJDK 64-Bit Server VM 25.0.2+10-LTS on Linux 6.14.0-1017-azure
-AMD EPYC 7763 64-Core Processor
+read timestamp text from files
150 154 5 6.7 149.6 1.0X
+read timestamps from files
878 884 6 1.1 878.4 0.2X
+infer timestamps from files
1684 1688 6 0.6 1684.4 0.1X
+read date text from files
136 141 6 7.3 136.2 1.1X
+read date from files
579 581 2 1.7 579.1 0.3X
+timestamp strings
136 142 8 7.4 135.9 1.1X
+parse timestamps from Dataset[String]
1046 1052 8 1.0 1046.2 0.1X
+infer timestamps from Dataset[String]
1841 1849 7 0.5 1841.0 0.1X
+date strings
197 198 1 5.1 197.4 0.8X
+parse dates from Dataset[String]
839 848 7 1.2 839.3 0.2X
+from_json(timestamp)
1527 1538 16 0.7 1526.8 0.1X
+from_json(date)
1328 1345 29 0.8 1327.9 0.1X
+infer error timestamps from Dataset[String] with default format
1186 1189 2 0.8 1186.4 0.1X
+infer error timestamps from Dataset[String] with user-provided format
1151 1152 2 0.9 1150.7 0.1X
+infer error timestamps from Dataset[String] with legacy format
1210 1213 6 0.8 1209.8 0.1X
+
+OpenJDK 64-Bit Server VM 25.0.4+7-LTS on Linux 6.17.0-1022-azure
+AMD EPYC 9V74 80-Core Processor
Filters pushdown: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
-w/o filters 5454 5462
11 0.0 54538.3 1.0X
-pushdown disabled 5418 5427
11 0.0 54175.8 1.0X
-w/ filters 584 591
6 0.2 5838.7 9.3X
+w/o filters 4727 4737
9 0.0 47265.0 1.0X
+pushdown disabled 4660 4675
14 0.0 46602.0 1.0X
+w/ filters 700 715
14 0.1 7003.7 6.7X
-OpenJDK 64-Bit Server VM 25.0.2+10-LTS on Linux 6.14.0-1017-azure
-AMD EPYC 7763 64-Core Processor
+OpenJDK 64-Bit Server VM 25.0.4+7-LTS on Linux 6.17.0-1022-azure
+AMD EPYC 9V74 80-Core Processor
Partial JSON results: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
-parse invalid JSON 2244 2247
4 0.0 224391.1 1.0X
+parse invalid JSON 2515 2532
16 0.0 251515.4 1.0X
+
+OpenJDK 64-Bit Server VM 25.0.4+7-LTS on Linux 6.17.0-1022-azure
+AMD EPYC 9V74 80-Core Processor
+Top-level JSON array with 0-byte payloads: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
+-------------------------------------------------------------------------------------------------------------------------
+streaming enabled: false 74 83
10 1.3 742.9 1.0X
+streaming enabled: true 76 77
2 1.3 760.3 1.0X
+
+OpenJDK 64-Bit Server VM 25.0.4+7-LTS on Linux 6.17.0-1022-azure
+AMD EPYC 9V74 80-Core Processor
+Top-level JSON array with 65536-byte payloads: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
+-----------------------------------------------------------------------------------------------------------------------------
+streaming enabled: false 108 111
4 0.0 107718.5 1.0X
+streaming enabled: true 107 111
5 0.0 106816.1 1.0X
diff --git a/sql/core/benchmarks/JsonBenchmark-results.txt
b/sql/core/benchmarks/JsonBenchmark-results.txt
index f6b52b4def0c..2241a4c7aed8 100644
--- a/sql/core/benchmarks/JsonBenchmark-results.txt
+++ b/sql/core/benchmarks/JsonBenchmark-results.txt
@@ -3,128 +3,142 @@ Benchmark for performance of JSON parsing
================================================================================================
Preparing data for benchmarking ...
-OpenJDK 64-Bit Server VM 17.0.18+8-LTS on Linux 6.14.0-1017-azure
-AMD EPYC 9V74 80-Core Processor
+OpenJDK 64-Bit Server VM 17.0.20+8-LTS on Linux 6.17.0-1022-azure
+AMD EPYC 7763 64-Core Processor
JSON schema inferring: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
-No encoding 1899 1983
91 2.6 379.7 1.0X
-UTF-8 is set 5121 5134
13 1.0 1024.2 0.4X
+No encoding 2438 2486
75 2.1 487.7 1.0X
+UTF-8 is set 5905 5916
12 0.8 1180.9 0.4X
Preparing data for benchmarking ...
-OpenJDK 64-Bit Server VM 17.0.18+8-LTS on Linux 6.14.0-1017-azure
-AMD EPYC 9V74 80-Core Processor
+OpenJDK 64-Bit Server VM 17.0.20+8-LTS on Linux 6.17.0-1022-azure
+AMD EPYC 7763 64-Core Processor
count a short column: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
-No encoding 1927 1963
55 2.6 385.5 1.0X
-UTF-8 is set 4451 4462
10 1.1 890.2 0.4X
+No encoding 2243 2289
51 2.2 448.6 1.0X
+UTF-8 is set 4872 4892
18 1.0 974.4 0.5X
Preparing data for benchmarking ...
-OpenJDK 64-Bit Server VM 17.0.18+8-LTS on Linux 6.14.0-1017-azure
-AMD EPYC 9V74 80-Core Processor
+OpenJDK 64-Bit Server VM 17.0.20+8-LTS on Linux 6.17.0-1022-azure
+AMD EPYC 7763 64-Core Processor
count a wide column: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
-No encoding 5230 5246
16 0.2 5230.0 1.0X
-UTF-8 is set 4768 4856
88 0.2 4767.9 1.1X
+No encoding 3307 3333
22 0.3 3307.4 1.0X
+UTF-8 is set 4700 4718
19 0.2 4699.9 0.7X
Preparing data for benchmarking ...
-OpenJDK 64-Bit Server VM 17.0.18+8-LTS on Linux 6.14.0-1017-azure
-AMD EPYC 9V74 80-Core Processor
+OpenJDK 64-Bit Server VM 17.0.20+8-LTS on Linux 6.17.0-1022-azure
+AMD EPYC 7763 64-Core Processor
select wide row: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
-No encoding 8996 9163
215 0.0 179920.8 1.0X
-UTF-8 is set 9757 9790
31 0.0 195143.4 0.9X
+No encoding 9262 9465
212 0.0 185247.3 1.0X
+UTF-8 is set 10776 10800
29 0.0 215517.2 0.9X
Preparing data for benchmarking ...
-OpenJDK 64-Bit Server VM 17.0.18+8-LTS on Linux 6.14.0-1017-azure
-AMD EPYC 9V74 80-Core Processor
+OpenJDK 64-Bit Server VM 17.0.20+8-LTS on Linux 6.17.0-1022-azure
+AMD EPYC 7763 64-Core Processor
Select a subset of 10 columns: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
-Select 10 columns 1557 1560
3 0.6 1557.2 1.0X
-Select 1 column 1184 1196
20 0.8 1183.7 1.3X
+Select 10 columns 1714 1728
13 0.6 1713.9 1.0X
+Select 1 column 1262 1266
3 0.8 1262.4 1.4X
Preparing data for benchmarking ...
-OpenJDK 64-Bit Server VM 17.0.18+8-LTS on Linux 6.14.0-1017-azure
-AMD EPYC 9V74 80-Core Processor
+OpenJDK 64-Bit Server VM 17.0.20+8-LTS on Linux 6.17.0-1022-azure
+AMD EPYC 7763 64-Core Processor
creation of JSON parser per line: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
-Short column without encoding 561 563
3 1.8 561.5 1.0X
-Short column with UTF-8 1140 1146
8 0.9 1139.9 0.5X
-Wide column without encoding 5163 5179
23 0.2 5163.2 0.1X
-Wide column with UTF-8 9810 9833
22 0.1 9810.2 0.1X
+Short column without encoding 666 669
3 1.5 665.7 1.0X
+Short column with UTF-8 1218 1221
2 0.8 1218.4 0.5X
+Wide column without encoding 6437 6459
19 0.2 6437.3 0.1X
+Wide column with UTF-8 7465 7476
11 0.1 7465.2 0.1X
Preparing data for benchmarking ...
-OpenJDK 64-Bit Server VM 17.0.18+8-LTS on Linux 6.14.0-1017-azure
-AMD EPYC 9V74 80-Core Processor
+OpenJDK 64-Bit Server VM 17.0.20+8-LTS on Linux 6.17.0-1022-azure
+AMD EPYC 7763 64-Core Processor
JSON functions: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
-Text read 62 66
4 16.2 61.7 1.0X
-from_json 969 979
9 1.0 968.7 0.1X
-json_tuple 905 908
4 1.1 905.1 0.1X
-get_json_object wholestage off 938 942
4 1.1 938.3 0.1X
-get_json_object wholestage on 835 847
18 1.2 835.0 0.1X
+Text read 59 61
2 16.9 59.0 1.0X
+from_json 1125 1143
17 0.9 1125.3 0.1X
+json_tuple 1039 1041
3 1.0 1038.7 0.1X
+get_json_object wholestage off 1054 1066
11 0.9 1054.2 0.1X
+get_json_object wholestage on 971 972
1 1.0 971.4 0.1X
Preparing data for benchmarking ...
-OpenJDK 64-Bit Server VM 17.0.18+8-LTS on Linux 6.14.0-1017-azure
-AMD EPYC 9V74 80-Core Processor
+OpenJDK 64-Bit Server VM 17.0.20+8-LTS on Linux 6.17.0-1022-azure
+AMD EPYC 7763 64-Core Processor
Dataset of json strings: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
-Text read 254 256
3 19.7 50.7 1.0X
-schema inferring 1504 1524
26 3.3 300.8 0.2X
-parsing 2391 2421
27 2.1 478.2 0.1X
+Text read 234 267
56 21.4 46.8 1.0X
+schema inferring 1908 1921
11 2.6 381.6 0.1X
+parsing 2595 2596
1 1.9 518.9 0.1X
Preparing data for benchmarking ...
-OpenJDK 64-Bit Server VM 17.0.18+8-LTS on Linux 6.14.0-1017-azure
-AMD EPYC 9V74 80-Core Processor
+OpenJDK 64-Bit Server VM 17.0.20+8-LTS on Linux 6.17.0-1022-azure
+AMD EPYC 7763 64-Core Processor
Json files in the per-line mode: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
-Text read 624 652
30 8.0 124.8 1.0X
-Schema inferring 2032 2036
5 2.5 406.5 0.3X
-Parsing without charset 2527 2529
3 2.0 505.4 0.2X
-Parsing with UTF-8 5453 5470
27 0.9 1090.6 0.1X
+Text read 577 580
4 8.7 115.4 1.0X
+Schema inferring 2454 2468
13 2.0 490.8 0.2X
+Parsing without charset 2973 2978
9 1.7 594.7 0.2X
+Parsing with UTF-8 5773 5776
2 0.9 1154.7 0.1X
-OpenJDK 64-Bit Server VM 17.0.18+8-LTS on Linux 6.14.0-1017-azure
-AMD EPYC 9V74 80-Core Processor
+OpenJDK 64-Bit Server VM 17.0.20+8-LTS on Linux 6.17.0-1022-azure
+AMD EPYC 7763 64-Core Processor
Write dates and timestamps: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
-Create a dataset of timestamps 107 110
4 9.4 106.5 1.0X
-to_json(timestamp) 576 582
9 1.7 576.3 0.2X
-write timestamps to files 623 626
4 1.6 623.2 0.2X
-Create a dataset of dates 120 123
3 8.3 120.1 0.9X
-to_json(date) 391 398
7 2.6 391.0 0.3X
-write dates to files 415 418
5 2.4 415.3 0.3X
-
-OpenJDK 64-Bit Server VM 17.0.18+8-LTS on Linux 6.14.0-1017-azure
-AMD EPYC 9V74 80-Core Processor
+Create a dataset of timestamps 103 106
3 9.7 102.7 1.0X
+to_json(timestamp) 640 641
1 1.6 640.0 0.2X
+write timestamps to files 685 692
6 1.5 685.4 0.1X
+Create a dataset of dates 120 125
4 8.3 120.4 0.9X
+to_json(date) 439 446
7 2.3 439.2 0.2X
+write dates to files 434 438
4 2.3 434.1 0.2X
+
+OpenJDK 64-Bit Server VM 17.0.20+8-LTS on Linux 6.17.0-1022-azure
+AMD EPYC 7763 64-Core Processor
Read dates and timestamps: Best
Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
-----------------------------------------------------------------------------------------------------------------------------------------------------
-read timestamp text from files
163 165 4 6.1 162.8 1.0X
-read timestamps from files
987 997 12 1.0 987.2 0.2X
-infer timestamps from files
1835 1847 20 0.5 1835.4 0.1X
-read date text from files
150 153 3 6.7 149.8 1.1X
-read date from files
621 623 2 1.6 621.4 0.3X
-timestamp strings
145 146 1 6.9 145.2 1.1X
-parse timestamps from Dataset[String]
1171 1185 21 0.9 1170.6 0.1X
-infer timestamps from Dataset[String]
2012 2015 5 0.5 2011.7 0.1X
-date strings
215 217 2 4.6 215.2 0.8X
-parse dates from Dataset[String]
927 931 5 1.1 927.4 0.2X
-from_json(timestamp)
1635 1639 6 0.6 1635.3 0.1X
-from_json(date)
1405 1412 6 0.7 1405.0 0.1X
-infer error timestamps from Dataset[String] with default format
1273 1276 5 0.8 1273.4 0.1X
-infer error timestamps from Dataset[String] with user-provided format
1257 1260 3 0.8 1256.7 0.1X
-infer error timestamps from Dataset[String] with legacy format
1258 1260 3 0.8 1258.0 0.1X
-
-OpenJDK 64-Bit Server VM 17.0.18+8-LTS on Linux 6.14.0-1017-azure
-AMD EPYC 9V74 80-Core Processor
+read timestamp text from files
148 153 8 6.7 148.3 1.0X
+read timestamps from files
1067 1068 1 0.9 1067.2 0.1X
+infer timestamps from files
2003 2013 15 0.5 2003.1 0.1X
+read date text from files
141 144 4 7.1 141.2 1.1X
+read date from files
750 752 2 1.3 750.4 0.2X
+timestamp strings
137 137 1 7.3 136.9 1.1X
+parse timestamps from Dataset[String]
1195 1201 9 0.8 1194.8 0.1X
+infer timestamps from Dataset[String]
2093 2097 4 0.5 2093.0 0.1X
+date strings
198 198 1 5.1 197.8 0.7X
+parse dates from Dataset[String]
954 956 2 1.0 953.6 0.2X
+from_json(timestamp)
1659 1677 26 0.6 1659.5 0.1X
+from_json(date)
1435 1448 19 0.7 1434.6 0.1X
+infer error timestamps from Dataset[String] with default format
1426 1435 12 0.7 1426.0 0.1X
+infer error timestamps from Dataset[String] with user-provided format
1437 1448 19 0.7 1437.5 0.1X
+infer error timestamps from Dataset[String] with legacy format
1452 1455 2 0.7 1452.1 0.1X
+
+OpenJDK 64-Bit Server VM 17.0.20+8-LTS on Linux 6.17.0-1022-azure
+AMD EPYC 7763 64-Core Processor
Filters pushdown: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
-w/o filters 4817 4833
15 0.0 48167.4 1.0X
-pushdown disabled 4772 4776
5 0.0 47721.9 1.0X
-w/ filters 695 710
16 0.1 6949.2 6.9X
+w/o filters 6016 6028
10 0.0 60158.9 1.0X
+pushdown disabled 5833 5857
22 0.0 58329.5 1.0X
+w/ filters 667 668
1 0.1 6667.9 9.0X
-OpenJDK 64-Bit Server VM 17.0.18+8-LTS on Linux 6.14.0-1017-azure
-AMD EPYC 9V74 80-Core Processor
+OpenJDK 64-Bit Server VM 17.0.20+8-LTS on Linux 6.17.0-1022-azure
+AMD EPYC 7763 64-Core Processor
Partial JSON results: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
-parse invalid JSON 2386 2451
111 0.0 238564.1 1.0X
+parse invalid JSON 2372 2465
149 0.0 237212.6 1.0X
+
+OpenJDK 64-Bit Server VM 17.0.20+8-LTS on Linux 6.17.0-1022-azure
+AMD EPYC 7763 64-Core Processor
+Top-level JSON array with 0-byte payloads: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
+-------------------------------------------------------------------------------------------------------------------------
+streaming enabled: false 77 83
6 1.3 772.6 1.0X
+streaming enabled: true 79 82
4 1.3 793.2 1.0X
+
+OpenJDK 64-Bit Server VM 17.0.20+8-LTS on Linux 6.17.0-1022-azure
+AMD EPYC 7763 64-Core Processor
+Top-level JSON array with 65536-byte payloads: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
+-----------------------------------------------------------------------------------------------------------------------------
+streaming enabled: false 160 168
13 0.0 159991.4 1.0X
+streaming enabled: true 161 162
1 0.0 161295.1 1.0X
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 d5be5c7d7272..b39c9acc33da 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
@@ -398,7 +398,7 @@ object MultiLineJsonDataSource extends JsonDataSource {
file: PartitionedFile,
parser: JacksonParser,
schema: StructType): Iterator[InternalRow] = {
- def partitionedFileString(ignored: Any): UTF8String = {
+ lazy val fileLiteral: UTF8String = {
Utils.tryWithResource {
Utils.createResourceUninterruptiblyIfInTaskThread {
CodecStreams.createInputStreamWithCloseResource(conf, file.toPath)
@@ -407,6 +407,7 @@ object MultiLineJsonDataSource extends JsonDataSource {
UTF8String.fromBytes(inputStream.readAllBytes())
}
}
+ def partitionedFileString(ignored: Any): UTF8String = fileLiteral
val streamParser = parser.options.encoding
.map(enc => CreateJacksonParser.inputStream(enc, _: JsonFactory, _:
InputStream))
.getOrElse(CreateJacksonParser.inputStream(_: JsonFactory, _:
InputStream))
@@ -417,8 +418,14 @@ object MultiLineJsonDataSource extends JsonDataSource {
schema,
parser.options.columnNameOfCorruptRecord)
- safeParser.parse(
- CodecStreams.createInputStreamWithCloseResource(conf, file.toPath))
+ val input = CodecStreams.createInputStreamWithCloseResource(conf,
file.toPath)
+ if (parser.options.streamMultilineTopLevelArray) {
+ safeParser.parseIterator(
+ input,
+ input => parser.parseIterator[InputStream](input, streamParser,
partitionedFileString))
+ } else {
+ safeParser.parse(input)
+ }
}
override protected def readStream(
@@ -428,16 +435,25 @@ object MultiLineJsonDataSource extends JsonDataSource {
// The entry is a single JSON document. Buffer its bytes so the
corrupt-record column can echo
// the whole document on a parse failure, mirroring `readFile`'s
`partitionedFileString`.
val bytes = in.readAllBytes()
+ lazy val documentLiteral: UTF8String = UTF8String.fromBytes(bytes)
val streamParser = parser.options.encoding
.map(enc => CreateJacksonParser.inputStream(enc, _: JsonFactory, _:
InputStream))
.getOrElse(CreateJacksonParser.inputStream(_: JsonFactory, _:
InputStream))
val safeParser = new FailureSafeParser[InputStream](
- input => parser.parse[InputStream](input, streamParser, _ =>
UTF8String.fromBytes(bytes)),
+ input => parser.parse[InputStream](input, streamParser, _ =>
documentLiteral),
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, _ => documentLiteral))
+ } else {
+ safeParser.parse(input)
+ }
}
}
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/JSONArchiveReadBase.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/JSONArchiveReadBase.scala
index 70de9e5e6317..854fc55faed4 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/JSONArchiveReadBase.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/JSONArchiveReadBase.scala
@@ -21,7 +21,8 @@ import java.io.File
import java.nio.charset.StandardCharsets
import java.nio.file.Files
-import org.apache.spark.sql.AnalysisException
+import org.apache.spark.SparkException
+import org.apache.spark.sql.{AnalysisException, Row}
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types.{NullType, StringType}
@@ -166,6 +167,73 @@ trait JSONArchiveReadBase extends ArchiveReadSuiteBase {
extraOptions = Map("multiLine" -> "true"))
}
+ test("JSON: streaming multi-line top-level arrays match a directory read") {
+ withSQLConf(SQLConf.JSON_STREAM_MULTILINE_TOP_LEVEL_ARRAY.key -> "true") {
+ assertArchiveMatchesDir(
+ Seq(
+ "a.json" ->
jsonBytes("""[{"id":1,"name":"Alice"},{"id":2,"name":"Bob"}]"""),
+ "b.json" -> jsonBytes("""[{"id":3,"name":"Carol"}]""")),
+ extraOptions = Map("multiLine" -> "true"))
+ // Jackson auto-detects UTF-16/32, so only a non-UTF charset shows
`encoding` is applied.
+ // scalastyle:off nonascii
+ assertArchiveMatchesDir(
+ Seq("a.json" ->
"[{\"id\":1,\"name\":\"Jos\u00e9\"},{\"id\":2,\"name\":\"Bob\"}]"
+ .getBytes(StandardCharsets.ISO_8859_1)),
+ extraOptions = Map("multiLine" -> "true", "encoding" -> "ISO-8859-1"))
+ // scalastyle:on nonascii
+ }
+ }
+
+ gridTest("JSON: streaming archive arrays resume after a partial element")(
+ Seq("PERMISSIVE", "DROPMALFORMED", "FAILFAST")) { mode =>
+ withSQLConf(SQLConf.JSON_STREAM_MULTILINE_TOP_LEVEL_ARRAY.key -> "true") {
+ val document = """[{"id":"bad","name":"Alice"},{"id":2,"name":"Bob"}]"""
+ withArchiveFile() { archive =>
+ writeArchive(archive, Seq(entryName(0) -> jsonBytes(document)))
+ val df = read(
+ archive.getCanonicalPath,
+ Map("multiLine" -> "true", "mode" -> mode),
+ s"$readSchema, _corrupt_record STRING")
+
+ mode match {
+ case "PERMISSIVE" =>
+ checkAnswer(df, Seq(Row(null, "Alice", document), Row(2, "Bob",
null)))
+ case "DROPMALFORMED" =>
+ checkAnswer(df, Row(2, "Bob", null))
+ case "FAILFAST" =>
+ val error = intercept[SparkException](df.collect())
+ assert(error.getCause.asInstanceOf[SparkException].getCondition ===
+ "MALFORMED_RECORD_IN_PARSING.WITHOUT_SUGGESTION")
+ }
+ }
+ }
+ }
+
+ gridTest("JSON: streaming archive arrays handle terminal malformed input")(
+ Seq("PERMISSIVE", "DROPMALFORMED", "FAILFAST")) { mode =>
+ withSQLConf(SQLConf.JSON_STREAM_MULTILINE_TOP_LEVEL_ARRAY.key -> "true") {
+ val document = """[{"id":1,"name":"Alice"} {"id":2,"name":"Bob"}]"""
+ withArchiveFile() { archive =>
+ writeArchive(archive, Seq(entryName(0) -> jsonBytes(document)))
+ val df = read(
+ archive.getCanonicalPath,
+ Map("multiLine" -> "true", "mode" -> mode),
+ s"$readSchema, _corrupt_record STRING")
+
+ mode match {
+ case "PERMISSIVE" =>
+ checkAnswer(df, Seq(Row(1, "Alice", null), Row(null, null,
document)))
+ case "DROPMALFORMED" =>
+ checkAnswer(df, Row(1, "Alice", null))
+ case "FAILFAST" =>
+ val error = intercept[SparkException](df.collect())
+ assert(error.getCause.asInstanceOf[SparkException].getCondition ===
+ "MALFORMED_RECORD_IN_PARSING.WITHOUT_SUGGESTION")
+ }
+ }
+ }
+ }
+
test("JSON: a malformed record in an archive entry matches a directory read
(both modes)") {
// Permissive mode (the default): a malformed record parses to nulls with
its raw text echoed
// into `_corrupt_record`. The archive path wires its own
FailureSafeParser in `readStream` --
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 94a2ccc41d30..bb779b036156 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
@@ -17,6 +17,8 @@
package org.apache.spark.sql.execution.datasources.json
import java.io.File
+import java.nio.charset.StandardCharsets
+import java.nio.file.Files
import java.time.{Instant, LocalDate}
import org.apache.spark.benchmark.Benchmark
@@ -37,6 +39,15 @@ import org.apache.spark.sql.types._
* 3. generate result:
* SPARK_GENERATE_BENCHMARK_FILES=1 build/sbt "sql/Test/runMain <this
class>"
* Results will be written to "benchmarks/JSONBenchmark-results.txt".
+ *
+ * To measure only the top-level JSON array cases, run it with the
"top-level-array"
+ * argument. It skips every other case, so it must not be combined with
+ * SPARK_GENERATE_BENCHMARK_FILES=1: the results file is truncated before
the arguments
+ * are read, so it would be left holding only the top-level array sections.
+ * 1. without sbt:
+ * bin/spark-submit --class <this class> --jars <spark core test jar>,
+ * <spark catalyst test jar> <spark sql test jar> top-level-array
+ * 2. build/sbt "sql/Test/runMain <this class> top-level-array"
* }}}
*/
object JsonBenchmark extends SqlBasedBenchmark {
@@ -578,8 +589,46 @@ object JsonBenchmark extends SqlBasedBenchmark {
benchmark.run()
}
+ private def topLevelArrayBenchmark(
+ rowsNum: Int,
+ payloadSize: Int,
+ numIters: Int): Unit = {
+ val payload = "x" * payloadSize
+ val document = (0 until rowsNum)
+ .map(i => s"""{"a":$i,"payload":"$payload"}""")
+ .mkString("[", ",", "]")
+ val schema = new StructType().add("a", IntegerType).add("payload",
StringType)
+ val benchmark = new Benchmark(
+ s"Top-level JSON array with $payloadSize-byte payloads", rowsNum, output
= output)
+
+ 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.run()
+ }
+ }
+
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)
@@ -595,6 +644,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 4cd6783e1df6..538e424a535d 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
@@ -23,28 +23,32 @@ import java.nio.file.Files
import java.sql.{Date, Timestamp}
import java.time.{Duration, Instant, LocalDate, LocalDateTime, Period, ZoneId,
ZoneOffset}
import java.util.Locale
-import java.util.concurrent.atomic.AtomicLong
+import java.util.concurrent.atomic.{AtomicInteger, AtomicLong}
-import com.fasterxml.jackson.core.JsonFactory
+import com.fasterxml.jackson.core.{JsonFactory, JsonToken}
+import com.fasterxml.jackson.core.util.JsonParserDelegate
import org.apache.hadoop.fs.{Path, PathFilter}
import org.apache.hadoop.io.SequenceFile.CompressionType
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.TaskContext
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.{BadRecordException,
CharsetProvider, DateTimeTestUtils, DateTimeUtils, HadoopCompressionCodec}
import org.apache.spark.sql.catalyst.util.HadoopCompressionCodec.GZIP
import org.apache.spark.sql.catalyst.util.TimestampNanosTestUtils
import
org.apache.spark.sql.catalyst.util.TimestampNanosTestUtils.foreachNanosPrecision
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
@@ -54,6 +58,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
@@ -1066,6 +1071,501 @@ abstract class JsonSuite
}
}
+ gridTest("SPARK-3308 Read multiline top level JSON arrays")(
+ Seq(false, true)) { enabled =>
+ withSQLConf(SQLConf.JSON_STREAM_MULTILINE_TOP_LEVEL_ARRAY.key ->
enabled.toString) {
+ 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)))
+ }
+ }
+ }
+
+ gridTest("multiline top level JSON array with singleVariantColumn is not
streamed")(
+ Seq(false, true)) { enabled =>
+ withSQLConf(SQLConf.JSON_STREAM_MULTILINE_TOP_LEVEL_ARRAY.key ->
enabled.toString) {
+ withTempPath { file =>
+ Files.write(file.toPath,
"""[{"a":1},{"a":2}]""".getBytes(StandardCharsets.UTF_8))
+ // singleVariantColumn binds the whole document to one variant, so the
streaming
+ // predicate excludes it; without that conjunct this yields one row
per element.
+ checkAnswer(
+ spark.read.option("multiLine", true).option("singleVariantColumn",
"var")
+ .json(file.getCanonicalPath).selectExpr("to_json(var)"),
+ Seq(Row("""[{"a":1},{"a":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("non-array JSON is parsed eagerly by parseIterator") {
+ 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)
+ var closed = false
+ val input = new
ByteArrayInputStream("""{"a":1}""".getBytes(StandardCharsets.UTF_8)) {
+ override def close(): Unit = {
+ closed = true
+ super.close()
+ }
+ }
+
+ val rows = parser.parseIterator[InputStream](
+ input,
+ CreateJacksonParser.inputStream(_: JsonFactory, _: InputStream),
+ stream => UTF8String.fromBytes(stream.readAllBytes()))
+
+ assert(closed)
+ assert(rows.next().getInt(0) === 1)
+ assert(!rows.hasNext)
+ }
+
+ test("non-array partial results close the parser") {
+ 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)
+ var closed = false
+ val input = new
ByteArrayInputStream("""{"a":"bad"}""".getBytes(StandardCharsets.UTF_8)) {
+ override def close(): Unit = {
+ closed = true
+ super.close()
+ }
+ }
+
+ val error = intercept[BadRecordException] {
+ parser.parseIterator[InputStream](
+ input,
+ CreateJacksonParser.inputStream(_: JsonFactory, _: InputStream),
+ stream => UTF8String.fromBytes(stream.readAllBytes()))
+ }
+
+ assert(closed)
+ assert(!error.recoverable)
+ }
+
+ gridTest("multiline top level JSON array stops retaining a closed parser")(
+ Seq("" -> 0, "[]" -> 0, """[{"a":1},{"a":2}]""" -> 2, """[{"a":1}""" ->
1)) {
+ case (document, expectedRows) =>
+ 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 closes = new AtomicInteger(0)
+ val input = new
ByteArrayInputStream(document.getBytes(StandardCharsets.UTF_8))
+ val context = TaskContext.empty()
+ TaskContext.setTaskContext(context)
+ try {
+ val rows = parser.parseIterator[InputStream](
+ input,
+ (factory: JsonFactory, stream: InputStream) =>
+ new JsonParserDelegate(CreateJacksonParser.inputStream(factory,
stream)) {
+ override def close(): Unit = {
+ closes.incrementAndGet()
+ super.close()
+ }
+ },
+ stream => UTF8String.fromBytes(stream.readAllBytes()))
+ var produced = 0
+ // The truncated document raises once the array runs out; that error is
asserted elsewhere.
+ try rows.foreach(_ => produced += 1) catch { case _: BadRecordException
=> }
+
+ assert(produced === expectedRows)
+ assert(closes.get() === 1)
+ context.markTaskCompleted(None)
+ // A second close here would mean the completion listener still held a
parser that had already
+ // been closed, which is what keeps an archive entry's buffer alive for
the rest of the task.
+ assert(closes.get() === 1)
+ } finally {
+ TaskContext.unset()
+ }
+ }
+
+ gridTest("multiline top level JSON array survives a parser close failure")(
+ Seq("exhausted", "abandoned")) { consumption =>
+ 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 closes = new AtomicInteger(0)
+ val input = new
ByteArrayInputStream("""[{"a":1},{"a":2}]""".getBytes(StandardCharsets.UTF_8))
+ val context = TaskContext.empty()
+ TaskContext.setTaskContext(context)
+ try {
+ val rows = parser.parseIterator[InputStream](
+ input,
+ (factory: JsonFactory, stream: InputStream) =>
+ new JsonParserDelegate(CreateJacksonParser.inputStream(factory,
stream)) {
+ override def close(): Unit = {
+ closes.incrementAndGet()
+ throw new IOException("close failed")
+ }
+ },
+ stream => UTF8String.fromBytes(stream.readAllBytes()))
+ // The array's end closes eagerly, an abandoned iterator closes from the
completion listener,
+ // and neither may surface the failure: rows already emitted cannot be
withdrawn.
+ if (consumption == "exhausted") {
+ assert(rows.map(_.getInt(0)).toSeq === Seq(1, 2))
+ assert(closes.get() === 1)
+ } else {
+ assert(rows.next().getInt(0) === 1)
+ assert(closes.get() === 0)
+ }
+ context.markTaskCompleted(None)
+ assert(closes.get() === 1)
+ } finally {
+ TaskContext.unset()
+ }
+ }
+
+ gridTest("multiline top level JSON array closes the parser when a raw read
fails")(
+ Seq("opening the array" -> 1, "advancing to an element" -> 2)) { case
(_, failingToken) =>
+ 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 closes = new AtomicInteger(0)
+ val tokens = new AtomicInteger(0)
+ val input = new
ByteArrayInputStream("""[{"a":1},{"a":2}]""".getBytes(StandardCharsets.UTF_8))
+ val context = TaskContext.empty()
+ TaskContext.setTaskContext(context)
+ try {
+ // A raw I/O failure is none of the types `handleFailure` names, so it
stays unwrapped rather
+ // than becoming a malformed record -- but it must still close here
rather than at task
+ // completion, or a task skipping corrupt files holds one parser and
stream per failed entry.
+ intercept[IOException] {
+ val rows = parser.parseIterator[InputStream](
+ input,
+ (factory: JsonFactory, stream: InputStream) =>
+ new JsonParserDelegate(CreateJacksonParser.inputStream(factory,
stream)) {
+ override def nextToken(): JsonToken = {
+ if (tokens.incrementAndGet() == failingToken) throw new
IOException("read failed")
+ super.nextToken()
+ }
+ override def close(): Unit = {
+ closes.incrementAndGet()
+ super.close()
+ }
+ },
+ stream => UTF8String.fromBytes(stream.readAllBytes()))
+ rows.foreach(_ => ())
+ }
+ assert(closes.get() === 1)
+ context.markTaskCompleted(None)
+ assert(closes.get() === 1)
+ } finally {
+ TaskContext.unset()
+ }
+ }
+
+ test("multiline top level JSON array keeps the parse failure primary when
the close fails") {
+ val schema = StructType(Seq(StructField("a", IntegerType)))
+ val options = new JSONOptions(
+ Map("multiLine" -> "true", "mode" -> "FAILFAST"),
SQLConf.get.sessionLocalTimeZone)
+ val parser = new JacksonParser(schema, options, allowArrayAsStructs = true)
+ val closes = new AtomicInteger(0)
+ val input = new
ByteArrayInputStream("""[{"a":1},42]""".getBytes(StandardCharsets.UTF_8))
+ val context = TaskContext.empty()
+ TaskContext.setTaskContext(context)
+ try {
+ val error = intercept[BadRecordException] {
+ val rows = parser.parseIterator[InputStream](
+ input,
+ (factory: JsonFactory, stream: InputStream) =>
+ new JsonParserDelegate(CreateJacksonParser.inputStream(factory,
stream)) {
+ override def close(): Unit = {
+ closes.incrementAndGet()
+ throw new IOException("close failed")
+ }
+ },
+ stream => UTF8String.fromBytes(stream.readAllBytes()))
+ rows.foreach(_ => ())
+ }
+ // The record is already lost here, so this close reports instead of
swallowing -- but only
+ // as a detail of the parse failure, never in place of it.
+ assert(!error.cause.isInstanceOf[IOException])
+ assert(error.cause.getSuppressed.map(_.getMessage).toSeq === Seq("close
failed"))
+ assert(closes.get() === 1)
+ context.markTaskCompleted(None)
+ assert(closes.get() === 1)
+ } finally {
+ TaskContext.unset()
+ }
+ }
+
+ test("multiline top level JSON array closes the parser on task completion") {
+ 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)
+ var closed = false
+ val input = new ByteArrayInputStream(
+ """[{"a":1},{"a":2},{"a":3}]""".getBytes(StandardCharsets.UTF_8)) {
+ override def close(): Unit = {
+ closed = true
+ super.close()
+ }
+ }
+ val context = TaskContext.empty()
+ TaskContext.setTaskContext(context)
+ try {
+ val rows = parser.parseIterator[InputStream](
+ input,
+ CreateJacksonParser.inputStream(_: JsonFactory, _: InputStream),
+ stream => UTF8String.fromBytes(stream.readAllBytes()))
+ assert(rows.next().getInt(0) === 1)
+ assert(!closed)
+ context.markTaskCompleted(None)
+ assert(closed)
+ } finally {
+ TaskContext.unset()
+ }
+ }
+
+ 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))
+ 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")
+ }
+ }
+ }
+ }
+
+ gridTest("multiline top level JSON array reports a truncated array as
malformed")(
+ Seq("PERMISSIVE", "DROPMALFORMED", "FAILFAST")) { mode =>
+ withSQLConf(SQLConf.JSON_STREAM_MULTILINE_TOP_LEVEL_ARRAY.key -> "true") {
+ withTempPath { file =>
+ // No closing bracket: jackson-core throws JsonEOFException from
nextToken() while the
+ // array context is open, so this takes the malformed-record arm, not
the null branch.
+ val document = """[{"a":1}"""
+ Files.write(file.toPath, document.getBytes(StandardCharsets.UTF_8))
+ 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))
+ assert(corruptRow.getUTF8String(1).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")
+ }
+ }
+ }
+ }
+
+ gridTest("multiline top level JSON array parse modes with a malformed
element")(
+ for {
+ element <- Seq("partial object", "scalar", "null", "nested array")
+ mode <- Seq("PERMISSIVE", "DROPMALFORMED", "FAILFAST")
+ streaming <- Seq(false, true)
+ } yield (element, mode, streaming)) { case (element, mode, streaming) =>
+ withSQLConf(SQLConf.JSON_STREAM_MULTILINE_TOP_LEVEL_ARRAY.key ->
streaming.toString) {
+ withTempPath { file =>
+ val malformed = element match {
+ case "partial object" => """{"a":"bad"}"""
+ case "scalar" => "42"
+ case "null" => "null"
+ case "nested array" => "[1,2]"
+ }
+ val document = s"""[{"a":1},$malformed,{"a":2}]"""
+ Files.write(file.toPath, document.getBytes(StandardCharsets.UTF_8))
+ val schema = StructType(Seq(
+ StructField("a", IntegerType),
+ StructField("_corrupt_record", StringType)))
+ val df = spark.read
+ .schema(schema)
+ .option("multiLine", true)
+ .option("mode", mode)
+ .json(file.getCanonicalPath)
+
+ if (mode == "FAILFAST") {
+ val error = intercept[SparkException](df.collect())
+ val malformed = error.getCause.asInstanceOf[SparkException]
+ assert(malformed.getCondition ===
"MALFORMED_RECORD_IN_PARSING.WITHOUT_SUGGESTION")
+ // Wrapping the parse failure twice empties partialResults and hides
the real cause.
+ assert(!malformed.getCause.isInstanceOf[BadRecordException])
+ } else {
+ val expected = if (streaming) {
+ // An array element is the record wherever the parser is left at
the element boundary: a
+ // partial object has been consumed through END_OBJECT, and a
scalar was never entered.
+ // A null element is a scalar too, though it is the converter's
null result rather than
+ // the converter that fails. A nested value of the wrong shape can
stop anywhere inside
+ // its element, so that one still ends the document.
+ element match {
+ case "nested array" if mode == "PERMISSIVE" => Seq(Row(1, null),
Row(null, document))
+ case "nested array" => Seq(Row(1, null))
+ case _ if mode == "PERMISSIVE" =>
+ Seq(Row(1, null), Row(null, document), Row(2, null))
+ case _ => Seq(Row(1, null), Row(2, null))
+ }
+ } else if (mode == "PERMISSIVE") {
+ // The whole document is the record, so this keeps only the rows
the eager array had
+ // accumulated when it failed, each stamped with the document.
Only a partial object
+ // accumulates any: the other three shapes abandon the array where
they fail.
+ if (element == "partial object") {
+ Seq(Row(1, document), Row(null, document), Row(2, document))
+ } else {
+ Seq(Row(null, document))
+ }
+ } else {
+ Seq.empty[Row]
+ }
+ 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) {
+ withTempPath { file =>
+ val document = """[{"a":"bad"},{"a":2}]"""
+ Files.write(file.toPath, document.getBytes(StandardCharsets.UTF_8))
+ val schema = StructType(Seq(
+ StructField("a", IntegerType),
+ StructField("_corrupt_record", StringType)))
+ val df = spark.read
+ .schema(schema)
+ .option("multiLine", true)
+ .option("enableStreamingTopLevelArray", streaming)
+ .json(file.getCanonicalPath)
+
+ val expected = if (streaming) {
+ Seq(Row(null, document), Row(2, null))
+ } else {
+ Seq(Row(null, document), Row(2, document))
+ }
+ checkAnswer(df, expected)
+ }
+ }
+ }
+
+ test("multiline top level JSON array reuses the corrupt record literal") {
+ withSQLConf(SQLConf.JSON_STREAM_MULTILINE_TOP_LEVEL_ARRAY.key -> "true") {
+ withTempPath { file =>
+ val document = """[{"a":"bad"},{"a":"also bad"},{"a":2}]"""
+ Files.write(file.toPath, document.getBytes(StandardCharsets.UTF_8))
+ 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" -> "PERMISSIVE"),
+ 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().getUTF8String(1).toString === document)
+ Files.delete(file.toPath)
+ assert(rows.next().getUTF8String(1).toString === document)
+ assert(rows.next().getInt(0) === 2)
+ assert(!rows.hasNext)
+ }
+ }
+ }
+
+ gridTest("multiline JSON parser creation failures honor parse mode")(
+ Seq("PERMISSIVE", "DROPMALFORMED", "FAILFAST")) { mode =>
+ withSQLConf(SQLConf.JSON_STREAM_MULTILINE_TOP_LEVEL_ARRAY.key -> "true") {
+ withTempPath { file =>
+ Files.write(file.toPath, Array[Byte](0, 0, 0xff.toByte, 0xfe.toByte))
+ val schema = StructType(Seq(
+ StructField("a", IntegerType),
+ StructField("_corrupt_record", StringType)))
+ val df = spark.read
+ .schema(schema)
+ .option("multiLine", true)
+ .option("mode", mode)
+ .json(file.getCanonicalPath)
+
+ mode match {
+ case "PERMISSIVE" =>
+ val rows = df.collect()
+ assert(rows.length === 1)
+ assert(rows.head.isNullAt(0))
+ assert(!rows.head.isNullAt(1))
+ case "DROPMALFORMED" =>
+ checkAnswer(df, Nil)
+ case "FAILFAST" =>
+ val error = intercept[SparkException](df.collect())
+ assert(error.getCause.asInstanceOf[SparkException].getCondition ===
+ "MALFORMED_RECORD_IN_PARSING.WITHOUT_SUGGESTION")
+ }
+ }
+ }
+ }
+
test("Corrupt records: FAILFAST mode") {
// `FAILFAST` mode should throw an exception for corrupt records.
checkError(
@@ -2008,100 +2508,117 @@ abstract class JsonSuite
}
}
- test("SPARK-18352: Handle multi-line corrupt documents (PERMISSIVE)") {
- withTempPath { dir =>
- val path = dir.getCanonicalPath
- val corruptRecordCount = additionalCorruptRecords.count().toInt
- assert(corruptRecordCount === 5)
+ gridTest("SPARK-18352: Handle multi-line corrupt documents (PERMISSIVE)")(
+ Seq(false, true)) { enabled =>
+ withSQLConf(SQLConf.JSON_STREAM_MULTILINE_TOP_LEVEL_ARRAY.key ->
enabled.toString) {
+ withTempPath { dir =>
+ val path = dir.getCanonicalPath
+ val corruptRecordCount = additionalCorruptRecords.count().toInt
+ assert(corruptRecordCount === 5)
- additionalCorruptRecords
- .toDF("value")
- // this is the minimum partition count that avoids hash collisions
- .repartition(corruptRecordCount * 4, F.hash($"value"))
- .write
- .text(path)
+ additionalCorruptRecords
+ .toDF("value")
+ // Each record must land in its own file for multiLine to read one
document per record.
+ // This count hashes the five records apart; a larger one need not
-- 21 collides.
+ .repartition(corruptRecordCount * 4, F.hash($"value"))
+ .write
+ .text(path)
+
+ val jsonDF = spark.read.option("multiLine", true).option("mode",
"PERMISSIVE").json(path)
+ // `[1,2,3]` is a top-level array of scalars. Streaming makes each
element its own record,
+ // so it reports one corrupt record per element where the whole
document reports one.
+ assert(jsonDF.count() === (if (enabled) corruptRecordCount + 2 else
corruptRecordCount))
+ assert(jsonDF.schema === new StructType()
+ .add("_corrupt_record", StringType)
+ .add("dummy", StringType))
+ val counts = jsonDF
+ .join(
+ additionalCorruptRecords.toDF("value"),
+ F.regexp_replace($"_corrupt_record", "(^\\s+|\\s+$)", "") ===
F.trim($"value"),
+ "outer")
+ .agg(
+ F.count($"dummy").as("valid"),
+ F.count($"_corrupt_record").as("corrupt"),
+ F.count("*").as("count"))
+ checkAnswer(counts, if (enabled) Row(1, 6, 8) else Row(1, 4, 6))
+ }
+ }
+ }
- val jsonDF = spark.read.option("multiLine", true).option("mode",
"PERMISSIVE").json(path)
- assert(jsonDF.count() === corruptRecordCount)
- assert(jsonDF.schema === new StructType()
- .add("_corrupt_record", StringType)
- .add("dummy", StringType))
- val counts = jsonDF
- .join(
- additionalCorruptRecords.toDF("value"),
- F.regexp_replace($"_corrupt_record", "(^\\s+|\\s+$)", "") ===
F.trim($"value"),
- "outer")
- .agg(
- F.count($"dummy").as("valid"),
- F.count($"_corrupt_record").as("corrupt"),
- F.count("*").as("count"))
- checkAnswer(counts, Row(1, 4, 6))
- }
- }
-
- test("SPARK-19641: Handle multi-line corrupt documents (DROPMALFORMED)") {
- withTempPath { dir =>
- val path = dir.getCanonicalPath
- val corruptRecordCount = additionalCorruptRecords.count().toInt
- assert(corruptRecordCount === 5)
+ gridTest("SPARK-19641: Handle multi-line corrupt documents (DROPMALFORMED)")(
+ Seq(false, true)) { enabled =>
+ withSQLConf(SQLConf.JSON_STREAM_MULTILINE_TOP_LEVEL_ARRAY.key ->
enabled.toString) {
+ withTempPath { dir =>
+ val path = dir.getCanonicalPath
+ val corruptRecordCount = additionalCorruptRecords.count().toInt
+ assert(corruptRecordCount === 5)
- additionalCorruptRecords
- .toDF("value")
- // this is the minimum partition count that avoids hash collisions
- .repartition(corruptRecordCount * 4, F.hash($"value"))
- .write
- .text(path)
+ additionalCorruptRecords
+ .toDF("value")
+ // Each record must land in its own file for multiLine to read one
document per record.
+ // This count hashes the five records apart; a larger one need not
-- 21 collides.
+ .repartition(corruptRecordCount * 4, F.hash($"value"))
+ .write
+ .text(path)
- val jsonDF = spark.read.option("multiLine", true).option("mode",
"DROPMALFORMED").json(path)
- checkAnswer(jsonDF, Seq(Row("test")))
+ val jsonDF = spark.read
+ .option("multiLine", true)
+ .option("mode", "DROPMALFORMED")
+ .json(path)
+ checkAnswer(jsonDF, Seq(Row("test")))
+ }
}
}
- test("SPARK-18352: Handle multi-line corrupt documents (FAILFAST)") {
- withTempPath { dir =>
- val path = dir.getCanonicalPath
- val corruptRecordCount = additionalCorruptRecords.count().toInt
- assert(corruptRecordCount === 5)
+ gridTest("SPARK-18352: Handle multi-line corrupt documents (FAILFAST)")(
+ Seq(false, true)) { enabled =>
+ withSQLConf(SQLConf.JSON_STREAM_MULTILINE_TOP_LEVEL_ARRAY.key ->
enabled.toString) {
+ withTempPath { dir =>
+ val path = dir.getCanonicalPath
+ val corruptRecordCount = additionalCorruptRecords.count().toInt
+ assert(corruptRecordCount === 5)
- additionalCorruptRecords
- .toDF("value")
- // this is the minimum partition count that avoids hash collisions
- .repartition(corruptRecordCount * 4, F.hash($"value"))
- .write
- .text(path)
+ additionalCorruptRecords
+ .toDF("value")
+ // Each record must land in its own file for multiLine to read one
document per record.
+ // This count hashes the five records apart; a larger one need not
-- 21 collides.
+ .repartition(corruptRecordCount * 4, F.hash($"value"))
+ .write
+ .text(path)
- val schema = new StructType().add("dummy", StringType)
+ val schema = new StructType().add("dummy", StringType)
- // `FAILFAST` mode should throw an exception for corrupt records.
- checkErrorMatchPVals(
- exception = intercept[SparkException] {
+ // `FAILFAST` mode should throw an exception for corrupt records.
+ checkErrorMatchPVals(
+ exception = intercept[SparkException] {
+ spark.read
+ .option("multiLine", true)
+ .option("mode", "FAILFAST")
+ .json(path)
+ },
+ condition = "INVALID_JSON_RECORD_TYPE",
+ parameters = Map("failFastMode" -> "FAILFAST", "invalidType" ->
"\"STRING\"|\"BIGINT\""))
+
+ val ex = intercept[SparkException] {
spark.read
.option("multiLine", true)
.option("mode", "FAILFAST")
+ .schema(schema)
.json(path)
- },
- condition = "INVALID_JSON_RECORD_TYPE",
- parameters = Map("failFastMode" -> "FAILFAST", "invalidType" ->
"\"STRING\"|\"BIGINT\""))
-
- val ex = intercept[SparkException] {
- spark.read
- .option("multiLine", true)
- .option("mode", "FAILFAST")
- .schema(schema)
- .json(path)
- .collect()
+ .collect()
+ }
+ checkErrorMatchPVals(
+ exception = ex,
+ condition = "FAILED_READ_FILE.NO_HINT",
+ parameters = Map("path" -> s".*$path.*"))
+ checkError(
+ exception = ex.getCause.asInstanceOf[SparkException],
+ condition = "MALFORMED_RECORD_IN_PARSING.WITHOUT_SUGGESTION",
+ parameters = Map(
+ "badRecord" -> "[null]",
+ "failFastMode" -> "FAILFAST")
+ )
}
- checkErrorMatchPVals(
- exception = ex,
- condition = "FAILED_READ_FILE.NO_HINT",
- parameters = Map("path" -> s".*$path.*"))
- checkError(
- exception = ex.getCause.asInstanceOf[SparkException],
- condition = "MALFORMED_RECORD_IN_PARSING.WITHOUT_SUGGESTION",
- parameters = Map(
- "badRecord" -> "[null]",
- "failFastMode" -> "FAILFAST")
- )
}
}
@@ -2404,27 +2921,30 @@ abstract class JsonSuite
checkAnswer(jsonDF, Seq(Row("Chris", "Baird")))
}
- test("SPARK-23723: specified encoding is not matched to actual encoding") {
- val fileName = "test-data/utf16LE.json"
- val schema = new StructType().add("firstName", StringType).add("lastName",
StringType)
- val inputFile = testFile(fileName)
- val exception = intercept[SparkException] {
- spark.read.schema(schema)
- .option("mode", "FAILFAST")
- .option("multiline", "true")
- .options(Map("encoding" -> "UTF-16BE"))
- .json(inputFile)
- .count()
+ gridTest("SPARK-23723: specified encoding is not matched to actual
encoding")(
+ Seq(false, true)) { enabled =>
+ withSQLConf(SQLConf.JSON_STREAM_MULTILINE_TOP_LEVEL_ARRAY.key ->
enabled.toString) {
+ val fileName = "test-data/utf16LE.json"
+ val schema = new StructType().add("firstName",
StringType).add("lastName", StringType)
+ val inputFile = testFile(fileName)
+ val exception = intercept[SparkException] {
+ spark.read.schema(schema)
+ .option("mode", "FAILFAST")
+ .option("multiline", "true")
+ .options(Map("encoding" -> "UTF-16BE"))
+ .json(inputFile)
+ .count()
+ }
+ checkErrorMatchPVals(
+ exception = exception,
+ condition = "FAILED_READ_FILE.NO_HINT",
+ parameters = Map("path" -> s".*$fileName.*"))
+ checkError(
+ exception = exception.getCause.asInstanceOf[SparkException],
+ condition = "MALFORMED_RECORD_IN_PARSING.WITHOUT_SUGGESTION",
+ parameters = Map("badRecord" -> "[empty row]", "failFastMode" ->
"FAILFAST")
+ )
}
- checkErrorMatchPVals(
- exception = exception,
- condition = "FAILED_READ_FILE.NO_HINT",
- parameters = Map("path" -> s".*$fileName.*"))
- checkError(
- exception = exception.getCause.asInstanceOf[SparkException],
- condition = "MALFORMED_RECORD_IN_PARSING.WITHOUT_SUGGESTION",
- parameters = Map("badRecord" -> "[empty row]", "failFastMode" ->
"FAILFAST")
- )
}
def checkEncoding(expectedEncoding: String, pathToJsonFiles: String,
@@ -3908,7 +4428,7 @@ abstract class JsonSuite
}
test("SPARK-40667: validate JSON Options") {
- assert(JSONOptions.getAllOptions.size == 33)
+ assert(JSONOptions.getAllOptions.size == 34)
// Please add validation on any new Json options here
assert(JSONOptions.isValidOption("samplingRatio"))
assert(JSONOptions.isValidOption("primitivesAsString"))
@@ -3941,6 +4461,7 @@ abstract class JsonSuite
assert(JSONOptions.isValidOption("singleVariantColumn"))
assert(JSONOptions.isValidOption("explodeEmbeddedArray"))
assert(JSONOptions.isValidOption("useUnsafeRow"))
+ assert(JSONOptions.isValidOption("enableStreamingTopLevelArray"))
assert(JSONOptions.isValidOption("encoding"))
assert(JSONOptions.isValidOption("charset"))
// Please add validation on any new Json options with alternative here
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
