tdcmeehan commented on code in PR #58704:
URL: https://github.com/apache/spark/pull/58704#discussion_r4066297643
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala:
##########
@@ -717,43 +745,86 @@ class JacksonParser(
}
} catch {
case e: SparkUpgradeException => throw e
- case e @ (_: RuntimeException | _: JsonProcessingException | _:
MalformedInputException) =>
- // JSON parser currently doesn't support partial results for corrupted
records.
- // For such records, all fields other than the field configured by
- // `columnNameOfCorruptRecord` are set to `null`.
- throw BadRecordException(() => recordLiteral(record), () =>
Array.empty, e)
case e: CharConversionException if options.encoding.isEmpty =>
- val msg =
- """JSON parser cannot handle a character in its input.
- |Specifying encoding as an input option explicitly might help to
resolve the issue.
- |""".stripMargin + e.getMessage
- val wrappedCharException = new CharConversionException(msg)
- wrappedCharException.initCause(e)
- throw BadRecordException(() => recordLiteral(record), () =>
Array.empty,
- wrappedCharException)
- case PartialResultException(row, cause) =>
- throw BadRecordException(
- record = () => recordLiteral(record),
- partialResults = () => Array(row),
- convertCauseForPartialResult(cause))
- case PartialResultArrayException(rows, cause) =>
- throw BadRecordException(
- record = () => recordLiteral(record),
- partialResults = () => rows,
- cause)
- // These exceptions should never be thrown outside of JacksonParser.
- // They are used for the control flow in the parser. We add them here
for completeness
- // since they also indicate a bad record.
- case PartialArrayDataResultException(arrayData, cause) =>
- throw BadRecordException(
- record = () => recordLiteral(record),
- partialResults = () => Array(InternalRow(arrayData)),
- convertCauseForPartialResult(cause))
- case PartialMapDataResultException(mapData, cause) =>
- throw BadRecordException(
- record = () => recordLiteral(record),
- partialResults = () => Array(InternalRow(mapData)),
- convertCauseForPartialResult(cause))
+ throw badRecord(e, () => recordLiteral(record))
+ case e @ (_: RuntimeException | _: JsonProcessingException | _:
MalformedInputException |
+ _: PartialResultException | _: PartialResultArrayException |
+ _: PartialArrayDataResultException | _:
PartialMapDataResultException) =>
+ throw badRecord(e, () => recordLiteral(record))
+ }
+ }
+
+ private[sql] def parseIterator[T](
+ record: T,
+ createParser: (JsonFactory, T) => JsonParser,
+ recordLiteral: T => UTF8String): Iterator[InternalRow] = {
+ val streamArray = allowArrayAsStructs && schema.isInstanceOf[StructType] &&
+ options.singleVariantColumn.isEmpty &&
options.explodeEmbeddedArray.isEmpty
+ val elementConverter = if (streamArray) makeConverter(schema) else null
+ val jsonParser = createParser(factory, record)
+ def fail(error: Throwable): Nothing = {
+ try jsonParser.close() catch {
+ case NonFatal(closeError) => error.addSuppressed(closeError)
+ }
+ throw badRecord(error, () => recordLiteral(record))
+ }
+ def handleFailure[T](operation: => T): T = {
+ try operation catch {
+ case e: SparkUpgradeException => fail(e)
+ case e: CharConversionException if options.encoding.isEmpty => fail(e)
+ case e @ (_: RuntimeException | _: JsonProcessingException | _:
MalformedInputException |
+ _: PartialResultException | _: PartialResultArrayException |
Review Comment:
Done in 488871bc3b8, generalized in dd76c16adef. A recoverable element
failure no longer replaces the source iterator: `handleFailure` marks the
`BadRecordException` recoverable and `FailureSafeParser` splices the recovery
row ahead of the surviving source -- `delegate = if (e.recoverable) recovery ++
source else recovery` (`FailureSafeParser.scala:83`) -- so later valid elements
still come through. Terminal structural failures still go through `fail()`,
which closes the parser first.
Resumability is now decided by where the parser was left rather than by the
exception alone: `resumableAfter` (`JacksonParser.scala:797-802`) requires
`PartialResultException` for a container start, since `convertObject` raises it
only after consuming through `END_OBJECT`. The `partial object` row of the
parse-mode grid (`JsonSuite.scala:1309-1320`) pins it on
`[{"a":1},{"a":"bad"},{"a":2}]` -- `(1,null)`, `(null,D)`, `(2,null)` under
PERMISSIVE and `(1,null)`, `(2,null)` under DROPMALFORMED -- the shape you
cited plus the preceding valid element.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/json/JsonDataSource.scala:
##########
@@ -414,8 +414,15 @@ object MultiLineJsonDataSource extends JsonDataSource {
schema,
parser.options.columnNameOfCorruptRecord)
- safeParser.parse(
- CodecStreams.createInputStreamWithCloseResource(conf, file.toPath))
+ val input = CodecStreams.createInputStreamWithCloseResource(conf,
file.toPath)
+ Option(TaskContext.get()).foreach(_.addTaskCompletionListener[Unit](_ =>
input.close()))
Review Comment:
Done in 488871bc3b8 -- the extra `addTaskCompletionListener[Unit](_ =>
input.close())` is gone. At head the only completion listener in this file
closes a `linesReader` (`JsonDataSource.scala:214`), and the four
`CodecStreams.createInputStreamWithCloseResource` call sites (316, 381, 401,
418) add none, so the helper owns those streams alone.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala:
##########
@@ -717,43 +745,86 @@ class JacksonParser(
}
} catch {
case e: SparkUpgradeException => throw e
- case e @ (_: RuntimeException | _: JsonProcessingException | _:
MalformedInputException) =>
- // JSON parser currently doesn't support partial results for corrupted
records.
- // For such records, all fields other than the field configured by
- // `columnNameOfCorruptRecord` are set to `null`.
- throw BadRecordException(() => recordLiteral(record), () =>
Array.empty, e)
case e: CharConversionException if options.encoding.isEmpty =>
- val msg =
- """JSON parser cannot handle a character in its input.
- |Specifying encoding as an input option explicitly might help to
resolve the issue.
- |""".stripMargin + e.getMessage
- val wrappedCharException = new CharConversionException(msg)
- wrappedCharException.initCause(e)
- throw BadRecordException(() => recordLiteral(record), () =>
Array.empty,
- wrappedCharException)
- case PartialResultException(row, cause) =>
- throw BadRecordException(
- record = () => recordLiteral(record),
- partialResults = () => Array(row),
- convertCauseForPartialResult(cause))
- case PartialResultArrayException(rows, cause) =>
- throw BadRecordException(
- record = () => recordLiteral(record),
- partialResults = () => rows,
- cause)
- // These exceptions should never be thrown outside of JacksonParser.
- // They are used for the control flow in the parser. We add them here
for completeness
- // since they also indicate a bad record.
- case PartialArrayDataResultException(arrayData, cause) =>
- throw BadRecordException(
- record = () => recordLiteral(record),
- partialResults = () => Array(InternalRow(arrayData)),
- convertCauseForPartialResult(cause))
- case PartialMapDataResultException(mapData, cause) =>
- throw BadRecordException(
- record = () => recordLiteral(record),
- partialResults = () => Array(InternalRow(mapData)),
- convertCauseForPartialResult(cause))
+ throw badRecord(e, () => recordLiteral(record))
+ case e @ (_: RuntimeException | _: JsonProcessingException | _:
MalformedInputException |
+ _: PartialResultException | _: PartialResultArrayException |
+ _: PartialArrayDataResultException | _:
PartialMapDataResultException) =>
+ throw badRecord(e, () => recordLiteral(record))
+ }
+ }
+
+ private[sql] def parseIterator[T](
+ record: T,
+ createParser: (JsonFactory, T) => JsonParser,
+ recordLiteral: T => UTF8String): Iterator[InternalRow] = {
+ val streamArray = allowArrayAsStructs && schema.isInstanceOf[StructType] &&
+ options.singleVariantColumn.isEmpty &&
options.explodeEmbeddedArray.isEmpty
+ val elementConverter = if (streamArray) makeConverter(schema) else null
+ val jsonParser = createParser(factory, record)
Review Comment:
Done in 488871bc3b8. `createParser` is now inside `parseIterator`'s own try
(`JacksonParser.scala:768-778`), which catches `CharConversionException` when
`options.encoding` is empty along with the `RuntimeException` /
`JsonProcessingException` / `MalformedInputException` / `Partial*` family and
rethrows `badRecord(...)`, so an acquisition failure reaches
`FailureSafeParser` as a `BadRecordException` and honours the parse mode.
`SparkUpgradeException` is still rethrown as-is. Covered by `multiline JSON
parser creation failures honor parse mode` (`JsonSuite.scala:1420`), which
drives the `00 00 ff fe` encoding case through PERMISSIVE, DROPMALFORMED and
FAILFAST.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/json/JsonSuite.scala:
##########
@@ -1066,6 +1069,97 @@ 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)))
+ }
+ }
+ }
+
+ 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)
+ }
+
+ gridTest("multiline top level JSON array keeps rows emitted before malformed
input")(
Review Comment:
Done -- all three gaps.
`multiline top level JSON array parse modes with a malformed element`
(`JsonSuite.scala:1309-1320`) runs `[{"a":1},<malformed>,{"a":2}]` over three
element shapes -- a partial object, a scalar, a nested array -- times
PERMISSIVE/DROPMALFORMED/FAILFAST, times the setting off and on, so the enabled
path's element granularity is asserted against the disabled path's document
granularity in the same case. `multiline JSON parser creation failures honor
parse mode` (line 1420) covers parser construction across the three modes. Both
from 488871bc3b8, extended in dd76c16adef and 75561b8daae.
Archive coverage with the setting enabled came in 5999f6b5974: `JSON:
streaming archive arrays resume after a partial element` and `JSON: streaming
archive arrays handle terminal malformed input`
(`JSONArchiveReadBase.scala:180`, `:205`), both under
`withSQLConf(JSON_STREAM_MULTILINE_TOP_LEVEL_ARRAY -> "true")`.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala:
##########
@@ -717,43 +745,100 @@ 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))
+ }
+ def fail(error: Throwable): Nothing = {
+ try jsonParser.close() catch {
+ case NonFatal(closeError) => error.addSuppressed(closeError)
+ }
+ throw badRecord(error, () => recordLiteral(record))
+ }
+ def handleFailure[T](operation: => T): T = {
+ try operation catch {
+ case e: SparkUpgradeException => fail(e)
+ case e: CharConversionException if options.encoding.isEmpty => fail(e)
+ case e: PartialResultException if options.parseMode != FailFastMode =>
Review Comment:
Done in 5999f6b5974, tightened in dd76c16adef. The non-array root is
converted with `handleFailure(elementStart = None)` (`JacksonParser.scala:865`)
and `resumableAfter` returns `false` for `None` unconditionally (`:801`), so
that state always routes to `fail()`, which closes the parser before throwing.
Recoverability is reachable only from inside the top-level-array iterator,
where an element start token exists, so the retention is excluded by
construction rather than by case analysis.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/JSONArchiveReadBase.scala:
##########
@@ -166,6 +166,16 @@ trait JSONArchiveReadBase extends ArchiveReadSuiteBase {
extraOptions = Map("multiLine" -> "true"))
}
+ test("JSON: streaming multi-line top-level arrays match a directory read") {
Review Comment:
Done in 5999f6b5974. Two enabled archive gridTests over
PERMISSIVE/DROPMALFORMED/FAILFAST: `JSON: streaming archive arrays resume after
a partial element` (`JSONArchiveReadBase.scala:180`) and `JSON: streaming
archive arrays handle terminal malformed input` (`:205`). Both go through
`readStream`, so its own `FailureSafeParser` wiring and pre-buffered literal
are what get exercised, not `readFile`'s. The PERMISSIVE case asserts the
complete entry text in `_corrupt_record`: `checkAnswer(df, Seq(Row(1, "Alice",
null), Row(null, null, document)))`, where `document` is the whole entry.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/json/JsonDataSource.scala:
##########
@@ -414,8 +414,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))
Review Comment:
Done in 5999f6b5974, with the archive side in 5072e171a5c. `fileLiteral` is
a `lazy val` inside `readFile` (`JsonDataSource.scala:398-404`), so the open
and `readAllBytes()` happen at most once per invocation however many elements
recover; `partitionedFileString` just returns it. The archive path's
`documentLiteral` (`:435`) is a `lazy val` over bytes already in hand.
--
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]
