cloud-fan commented on code in PR #58704:
URL: https://github.com/apache/spark/pull/58704#discussion_r4048017523
##########
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:
**Non-blocking (P2):** This handler also runs while eagerly converting a
non-array root. In that state a `PartialResultException` is raised before any
source iterator can be returned, so marking it recoverable bypasses `fail()`
and leaves the parser/input stream open after the recovery rows finish. A
partition with many such files can accumulate descriptors until task
completion. Please mark partial failures recoverable only inside the
top-level-array iterator and close the parser for this fallback state.
##########
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:
**Non-blocking (P2):** This proves that the enabled archive path handles
valid arrays, but it does not exercise `readStream`'s separate failure wiring
or pre-buffered corrupt-record literal. The existing malformed archive test
leaves the switch disabled, and the mode matrix in `JsonSuite` goes through
`readFile`. Please add enabled archive cases for partial-element recovery and
terminal malformed input across the parse modes, including an assertion that
PERMISSIVE returns the complete entry text in `_corrupt_record`.
##########
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:
**Non-blocking (P2):** A recoverable partial element invokes the bad-record
thunk when `_corrupt_record` is populated, and this thunk reopens and
`readAllBytes()` the complete file each time. Multiple malformed elements
therefore reread and rematerialize the same large document once per element,
defeating much of the streaming benefit. Please memoize the file literal lazily
within this `readFile` invocation and reuse it for every recovery row.
--
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]
