cloud-fan commented on code in PR #58704:
URL: https://github.com/apache/spark/pull/58704#discussion_r4016956145
##########
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:
**Non-blocking (P2):** A `PartialResultException` here is recoverable at the
array-element boundary: the current object has been consumed and later elements
remain in the parser. Closing the parser and replacing the source iterator
causes inputs such as `[{"a":"bad"},{"a":2}]` (with `a int`) to emit only the
partial/corrupt first row and silently drop the valid second row. Please
preserve the live source iterator while emitting the recovery row, then resume
with the next element; terminal structural failures should still close it.
See **Shared repair plan 1** in the review body.
##########
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:
**Nit (P3):** `CodecStreams.createInputStreamWithCloseResource` already
registers this returned stream for task-completion cleanup. Adding another
listener here retains a redundant callback for every file and closes the same
stream twice; please rely on the helper's existing ownership.
##########
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:
**Non-blocking (P2):** `createParser` can fail before `nextToken` while
detecting an unsupported encoding (for example, Jackson throws
`CharConversionException` for bytes `00 00 ff fe`). Because this call is
outside `handleFailure`, that exception escapes `FailureSafeParser` even in
PERMISSIVE or DROPMALFORMED mode. Please include parser acquisition in the same
bad-record conversion boundary as iterator advancement.
See **Shared repair plan 1** in the review body.
##########
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:
**Non-blocking (P2):** This matrix covers terminal JSON syntax failure, but
it does not enter the recoverable `PartialResultException` state or a failure
during `createParser`; existing partial-result tests use the eager parser.
Archive tests also leave the new setting disabled, so the new `readStream`
branch is never exercised. Please add enabled-path coverage for a partial
element followed by valid rows, parser-construction failure under each parse
mode, and an archive entry.
See **Shared repair plan 1** in the review body.
--
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]
