tdcmeehan commented on code in PR #58704:
URL: https://github.com/apache/spark/pull/58704#discussion_r4061266335
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -7034,6 +7034,16 @@ object SQLConf {
.booleanConf
.createWithDefault(true)
+ val JSON_STREAM_MULTILINE_TOP_LEVEL_ARRAY =
+ buildConf("spark.sql.json.enableStreamingTopLevelArray")
+ .internal()
+ .doc("When true, multiline JSON reads stream the elements of a top-level
array one at a " +
Review Comment:
Done. The doc now states the scope: streaming applies only to reads into a
struct schema that take top-level arrays as structs, and reads using
`singleVariantColumn` or `explodeEmbeddedArray` are never streamed.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala:
##########
@@ -717,43 +746,104 @@ class JacksonParser(
}
} catch {
case e: SparkUpgradeException => throw e
- case e @ (_: RuntimeException | _: JsonProcessingException | _:
MalformedInputException) =>
- // JSON parser currently doesn't support partial results for corrupted
records.
- // For such records, all fields other than the field configured by
- // `columnNameOfCorruptRecord` are set to `null`.
- throw BadRecordException(() => recordLiteral(record), () =>
Array.empty, e)
case e: CharConversionException if options.encoding.isEmpty =>
- val msg =
- """JSON parser cannot handle a character in its input.
- |Specifying encoding as an input option explicitly might help to
resolve the issue.
- |""".stripMargin + e.getMessage
- val wrappedCharException = new CharConversionException(msg)
- wrappedCharException.initCause(e)
- throw BadRecordException(() => recordLiteral(record), () =>
Array.empty,
- wrappedCharException)
- case PartialResultException(row, cause) =>
- throw BadRecordException(
- record = () => recordLiteral(record),
- partialResults = () => Array(row),
- convertCauseForPartialResult(cause))
- case PartialResultArrayException(rows, cause) =>
- throw BadRecordException(
- record = () => recordLiteral(record),
- partialResults = () => rows,
- cause)
- // These exceptions should never be thrown outside of JacksonParser.
- // They are used for the control flow in the parser. We add them here
for completeness
- // since they also indicate a bad record.
- case PartialArrayDataResultException(arrayData, cause) =>
- throw BadRecordException(
- record = () => recordLiteral(record),
- partialResults = () => Array(InternalRow(arrayData)),
- convertCauseForPartialResult(cause))
- case PartialMapDataResultException(mapData, cause) =>
- throw BadRecordException(
- record = () => recordLiteral(record),
- partialResults = () => Array(InternalRow(mapData)),
- convertCauseForPartialResult(cause))
+ throw badRecord(e, () => recordLiteral(record))
+ case e @ (_: RuntimeException | _: JsonProcessingException | _:
MalformedInputException |
+ _: PartialResultException | _: PartialResultArrayException |
+ _: PartialArrayDataResultException | _:
PartialMapDataResultException) =>
+ throw badRecord(e, () => recordLiteral(record))
+ }
+ }
+
+ private[sql] def parseIterator[T](
+ record: T,
+ createParser: (JsonFactory, T) => JsonParser,
+ recordLiteral: T => UTF8String): Iterator[InternalRow] = {
+ val streamArray = allowArrayAsStructs && schema.isInstanceOf[StructType] &&
+ options.singleVariantColumn.isEmpty &&
options.explodeEmbeddedArray.isEmpty
+ val elementConverter = if (streamArray) makeConverter(schema) else null
Review Comment:
Done. `arrayElementConverter` is a single `private lazy val`
(`JacksonParser.scala:67`) that both the eager root converter and the streaming
iterator use, so the graph is built once per parser. Being lazy, it is not
built at all for a non-array root.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala:
##########
@@ -717,43 +746,104 @@ class JacksonParser(
}
} catch {
case e: SparkUpgradeException => throw e
- case e @ (_: RuntimeException | _: JsonProcessingException | _:
MalformedInputException) =>
- // JSON parser currently doesn't support partial results for corrupted
records.
- // For such records, all fields other than the field configured by
- // `columnNameOfCorruptRecord` are set to `null`.
- throw BadRecordException(() => recordLiteral(record), () =>
Array.empty, e)
case e: CharConversionException if options.encoding.isEmpty =>
- val msg =
- """JSON parser cannot handle a character in its input.
- |Specifying encoding as an input option explicitly might help to
resolve the issue.
- |""".stripMargin + e.getMessage
- val wrappedCharException = new CharConversionException(msg)
- wrappedCharException.initCause(e)
- throw BadRecordException(() => recordLiteral(record), () =>
Array.empty,
- wrappedCharException)
- case PartialResultException(row, cause) =>
- throw BadRecordException(
- record = () => recordLiteral(record),
- partialResults = () => Array(row),
- convertCauseForPartialResult(cause))
- case PartialResultArrayException(rows, cause) =>
- throw BadRecordException(
- record = () => recordLiteral(record),
- partialResults = () => rows,
- cause)
- // These exceptions should never be thrown outside of JacksonParser.
- // They are used for the control flow in the parser. We add them here
for completeness
- // since they also indicate a bad record.
- case PartialArrayDataResultException(arrayData, cause) =>
- throw BadRecordException(
- record = () => recordLiteral(record),
- partialResults = () => Array(InternalRow(arrayData)),
- convertCauseForPartialResult(cause))
- case PartialMapDataResultException(mapData, cause) =>
- throw BadRecordException(
- record = () => recordLiteral(record),
- partialResults = () => Array(InternalRow(mapData)),
- convertCauseForPartialResult(cause))
+ throw badRecord(e, () => recordLiteral(record))
+ case e @ (_: RuntimeException | _: JsonProcessingException | _:
MalformedInputException |
+ _: PartialResultException | _: PartialResultArrayException |
+ _: PartialArrayDataResultException | _:
PartialMapDataResultException) =>
+ throw badRecord(e, () => recordLiteral(record))
+ }
+ }
+
+ private[sql] def parseIterator[T](
+ record: T,
+ createParser: (JsonFactory, T) => JsonParser,
+ recordLiteral: T => UTF8String): Iterator[InternalRow] = {
+ val streamArray = allowArrayAsStructs && schema.isInstanceOf[StructType] &&
+ options.singleVariantColumn.isEmpty &&
options.explodeEmbeddedArray.isEmpty
+ val elementConverter = if (streamArray) makeConverter(schema) else null
+ val jsonParser = 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))
+ }
+ // Abandoning the iterator before END_ARRAY (e.g. a LIMIT) skips
finish()/fail(), so close the
+ // parser at task completion; close() is idempotent with those eager
closes.
+ Option(TaskContext.get()).foreach(_.addTaskCompletionListener[Unit](_ =>
jsonParser.close()))
Review Comment:
Done in dd76c16adef. The listener captures an `AtomicReference` cell rather
than the parser: `closeParser()` does `getAndSet(null)` and closes only what it
took, and all four eager close paths go through it alongside the listener.
Exactly one close therefore happens, and a completed read leaves nothing
reachable from the `TaskContext`. Clearing happens before the close, so a close
that throws still clears — the alternative loses the parser for good if the
clear is what fails.
Covered by `multiline top level JSON array stops retaining a closed parser`,
which counts `close()` calls through a `JsonParserDelegate` over the empty,
empty-array, valid and truncated documents and asserts `markTaskCompleted` adds
none. A stream-level counter cannot observe this: jackson's own `close()` is
idempotent, so the regression shows up only as a second *call*.
On the archive-entry coverage you asked for: one parser per entry is this
same mechanism repeated, so I have pinned the per-parser invariant instead. A
multi-entry test could only fail for the same reason this one does.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala:
##########
@@ -717,43 +746,107 @@ class JacksonParser(
}
} catch {
case e: SparkUpgradeException => throw e
- case e @ (_: RuntimeException | _: JsonProcessingException | _:
MalformedInputException) =>
- // JSON parser currently doesn't support partial results for corrupted
records.
- // For such records, all fields other than the field configured by
- // `columnNameOfCorruptRecord` are set to `null`.
- throw BadRecordException(() => recordLiteral(record), () =>
Array.empty, e)
case e: CharConversionException if options.encoding.isEmpty =>
- val msg =
- """JSON parser cannot handle a character in its input.
- |Specifying encoding as an input option explicitly might help to
resolve the issue.
- |""".stripMargin + e.getMessage
- val wrappedCharException = new CharConversionException(msg)
- wrappedCharException.initCause(e)
- throw BadRecordException(() => recordLiteral(record), () =>
Array.empty,
- wrappedCharException)
- case PartialResultException(row, cause) =>
- throw BadRecordException(
- record = () => recordLiteral(record),
- partialResults = () => Array(row),
- convertCauseForPartialResult(cause))
- case PartialResultArrayException(rows, cause) =>
- throw BadRecordException(
- record = () => recordLiteral(record),
- partialResults = () => rows,
- cause)
- // These exceptions should never be thrown outside of JacksonParser.
- // They are used for the control flow in the parser. We add them here
for completeness
- // since they also indicate a bad record.
- case PartialArrayDataResultException(arrayData, cause) =>
- throw BadRecordException(
- record = () => recordLiteral(record),
- partialResults = () => Array(InternalRow(arrayData)),
- convertCauseForPartialResult(cause))
- case PartialMapDataResultException(mapData, cause) =>
- throw BadRecordException(
- record = () => recordLiteral(record),
- partialResults = () => Array(InternalRow(mapData)),
- convertCauseForPartialResult(cause))
+ throw badRecord(e, () => recordLiteral(record))
+ case e @ (_: RuntimeException | _: JsonProcessingException | _:
MalformedInputException |
+ _: PartialResultException | _: PartialResultArrayException |
+ _: PartialArrayDataResultException | _:
PartialMapDataResultException) =>
+ throw badRecord(e, () => recordLiteral(record))
+ }
+ }
+
+ private[sql] def parseIterator[T](
+ record: T,
+ createParser: (JsonFactory, T) => JsonParser,
+ recordLiteral: T => UTF8String): Iterator[InternalRow] = {
+ val streamArray = allowArrayAsStructs && schema.isInstanceOf[StructType] &&
+ options.singleVariantColumn.isEmpty &&
options.explodeEmbeddedArray.isEmpty
+ val elementConverter = if (streamArray) makeConverter(schema) else null
+ val jsonParser = 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))
+ }
+ // Abandoning the iterator before END_ARRAY (e.g. a LIMIT) skips
finish()/fail(), so close the
+ // parser at task completion; close() is idempotent with those eager
closes.
+ Option(TaskContext.get()).foreach(_.addTaskCompletionListener[Unit](_ =>
jsonParser.close()))
+ def fail(error: Throwable): Nothing = {
+ try jsonParser.close() catch {
Review Comment:
Done in dd76c16adef, through the same cell. The truncated-array case of
`multiline top level JSON array stops retaining a closed parser` is exactly
this path: `prepare()` raises from `nextToken()`, `fail()` closes, and the test
asserts one close *before* `markTaskCompleted` and none added by it. If
`fail()` stopped closing, that first assertion would see 0 — the listener would
close it later, which is the retention this is meant to catch.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/json/JsonSuite.scala:
##########
@@ -1066,6 +1070,282 @@ 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))
Review Comment:
Done in dd76c16adef and 75561b8daae. The lifecycle grid runs over `""`,
`"[]"`, a valid two-element array and the truncated array, asserting both the
rows each yields (0, 0, 2, 1) and one `close()` in each. That covers the empty
document's close before the iterator is constructed, the empty array's close
from the first `prepare()`, and closure after normal exhaustion, with zero rows
asserted for both empty inputs.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala:
##########
@@ -717,43 +746,107 @@ class JacksonParser(
}
} catch {
case e: SparkUpgradeException => throw e
- case e @ (_: RuntimeException | _: JsonProcessingException | _:
MalformedInputException) =>
- // JSON parser currently doesn't support partial results for corrupted
records.
- // For such records, all fields other than the field configured by
- // `columnNameOfCorruptRecord` are set to `null`.
- throw BadRecordException(() => recordLiteral(record), () =>
Array.empty, e)
case e: CharConversionException if options.encoding.isEmpty =>
- val msg =
- """JSON parser cannot handle a character in its input.
- |Specifying encoding as an input option explicitly might help to
resolve the issue.
- |""".stripMargin + e.getMessage
- val wrappedCharException = new CharConversionException(msg)
- wrappedCharException.initCause(e)
- throw BadRecordException(() => recordLiteral(record), () =>
Array.empty,
- wrappedCharException)
- case PartialResultException(row, cause) =>
- throw BadRecordException(
- record = () => recordLiteral(record),
- partialResults = () => Array(row),
- convertCauseForPartialResult(cause))
- case PartialResultArrayException(rows, cause) =>
- throw BadRecordException(
- record = () => recordLiteral(record),
- partialResults = () => rows,
- cause)
- // These exceptions should never be thrown outside of JacksonParser.
- // They are used for the control flow in the parser. We add them here
for completeness
- // since they also indicate a bad record.
- case PartialArrayDataResultException(arrayData, cause) =>
- throw BadRecordException(
- record = () => recordLiteral(record),
- partialResults = () => Array(InternalRow(arrayData)),
- convertCauseForPartialResult(cause))
- case PartialMapDataResultException(mapData, cause) =>
- throw BadRecordException(
- record = () => recordLiteral(record),
- partialResults = () => Array(InternalRow(mapData)),
- convertCauseForPartialResult(cause))
+ throw badRecord(e, () => recordLiteral(record))
+ case e @ (_: RuntimeException | _: JsonProcessingException | _:
MalformedInputException |
+ _: PartialResultException | _: PartialResultArrayException |
+ _: PartialArrayDataResultException | _:
PartialMapDataResultException) =>
+ throw badRecord(e, () => recordLiteral(record))
+ }
+ }
+
+ private[sql] def parseIterator[T](
+ record: T,
+ createParser: (JsonFactory, T) => JsonParser,
+ recordLiteral: T => UTF8String): Iterator[InternalRow] = {
+ val streamArray = allowArrayAsStructs && schema.isInstanceOf[StructType] &&
+ options.singleVariantColumn.isEmpty &&
options.explodeEmbeddedArray.isEmpty
+ val elementConverter = if (streamArray) makeConverter(schema) else null
+ val jsonParser = 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))
+ }
+ // Abandoning the iterator before END_ARRAY (e.g. a LIMIT) skips
finish()/fail(), so close the
+ // parser at task completion; close() is idempotent with those eager
closes.
+ Option(TaskContext.get()).foreach(_.addTaskCompletionListener[Unit](_ =>
jsonParser.close()))
+ def fail(error: Throwable): Nothing = {
+ try jsonParser.close() catch {
+ case NonFatal(closeError) => error.addSuppressed(closeError)
+ }
+ throw badRecord(error, () => recordLiteral(record))
+ }
+ def handleFailure[T](recoverPartialResult: Boolean)(operation: => T): T = {
+ try operation catch {
+ case e: SparkUpgradeException => fail(e)
+ case e: CharConversionException if options.encoding.isEmpty => fail(e)
+ case e: PartialResultException
+ if recoverPartialResult && options.parseMode != FailFastMode =>
+ throw badRecord(e, () => recordLiteral(record)).copy(recoverable =
true)
+ case e: PartialResultException =>
+ fail(e)
+ case e @ (_: RuntimeException | _: JsonProcessingException | _:
MalformedInputException |
Review Comment:
Done in dd76c16adef. Resumability is now decided by where the parser is left
rather than by the exception alone. `resumableAfter` takes the element's start
token: a scalar start makes any `RuntimeException` recoverable, since the
parser is still on that scalar's own token; a container start still requires
`PartialResultException`, which is the only container failure that provably
drained to `END_OBJECT`. `FailFastMode` is excluded as before.
`[{"a":1},42,{"a":2}]` now yields `(1,null)`, `(null,D)`, `(2,null)` under
PERMISSIVE and `(1,null)`, `(2,null)` under DROPMALFORMED.
The parse-mode grid is parameterized over the malformed element's shape — a
bad object field, a scalar, a nested array — against one document
`[{"a":1},<malformed>,{"a":2}]`, with the nested array as the negative control
that has to stay terminal. A classification broad enough to resume from inside
a container fails there.
Two notes on the framing. First, one consequence reaches beyond the input
you cited: an all-scalar array such as `[1,2,3]` is malformed in every element
against a struct schema, so it now reports one corrupt record per element
rather than one per document. The existing SPARK-18352 multiline expectations
are updated and made configuration-dependent for that. Object elements already
behaved this way, through the partial-result path the eager reader has always
had.
Second, on "silently lose the final valid row": for that input the disabled
path loses more, not less — it yields a single `(null,D)` and drops *both*
valid rows, because the scalar failure escapes `convertArray` before any
element is accumulated. So the enabled path was already ahead of the one it
replaces. The real defect was that element granularity was promised for every
malformed element and delivered only for objects, which is what this fixes.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
