cloud-fan commented on code in PR #58704:
URL: https://github.com/apache/spark/pull/58704#discussion_r4057250910
##########
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:
**Non-blocking (P2):** An incompatible scalar element reaches this terminal
RuntimeException arm even though its token is already at a safe top-level
array-element boundary. FailureSafeParser then replaces the source iterator
with only the recovery result, so PERMISSIVE and DROPMALFORMED reads of an
input such as [{"a":1},42,{"a":2}] silently lose the final valid row. Please
classify this scalar-boundary failure as recoverable while keeping nested
object and array mismatches terminal.
**Recommended change:** Classify only scalar element-type conversion
failures that remain at the current top-level element boundary as recoverable
in non-FAILFAST modes, and cover scalars before, between, and after valid
objects across parse modes and archive/file paths.
**Why this works:** Inspect the current element token and error category at
the array-loop owner. Convert a safe scalar mismatch into a recoverable
BadRecordException so FailureSafeParser emits or drops that element and resumes
the preserved iterator; keep nested containers and structurally ambiguous
failures terminal.
**Scope:** Complete element-boundary recovery without broadening recovery to
undefined Jackson token positions.
**Compatibility:** Existing partial-object recovery, terminal
malformed-document handling, corrupt-record text, and the disabled eager path
remain unchanged.
**Risks:** An overly broad classification could resume from inside a
malformed nested container and duplicate or misparse later tokens. An overly
narrow classification could continue dropping later valid objects after other
safely consumed scalar forms.
**Constraints:** Resume only when parser position is known to be at a
complete top-level element boundary. FAILFAST must still fail immediately.
Terminal structural corruption must still close and stop the iterator.
**Success:** In PERMISSIVE mode, an incompatible scalar produces one
recovery row and later valid objects exactly once. In DROPMALFORMED mode, the
incompatible scalar is omitted and later valid objects remain. In FAILFAST
mode, the first incompatible scalar still raises the established
malformed-record error. Nested or structurally ambiguous failures remain
terminal and close the parser.
##########
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:
**Non-blocking (P2):** The malformed and truncated-array tests do not
observe this terminal cleanup side effect. If fail() stopped closing the
parser, they would still return the same rows and errors while retaining the
input until task completion. Please add a regression assertion that terminal
advancement failures close the parser and underlying input synchronously.
**Recommended change:** Introduce one clearable parser-lifecycle owner
captured by TaskContext, route every eager close through it, and add
resource-sensitive JSON lifecycle coverage for empty input, empty arrays,
normal exhaustion, terminal failure, and abandonment, including repeated
archive entries.
**Why this works:** The task listener captures only a small owner that
atomically consumes its current parser. Normal completion, terminal failure,
and eager fallback use the same close-and-clear operation so a completed parser
becomes unreachable immediately; task completion consumes and closes only a
parser still owned because its iterator was abandoned.
**Scope:** Unify streaming parser ownership and verify every lifecycle
transition without changing row or parse-mode semantics.
**Compatibility:** The enabled path remains lazy; disabled and excluded
option paths remain eager; recovery and terminal error semantics do not change.
**Risks:** Clearing ownership before close completes could lose the only
cleanup backstop after a close failure. Independent close routes could race or
mask the original parse exception if they do not share one consume operation.
**Constraints:** Normal exhaustion and terminal failures must still close
synchronously. Task completion must still close an iterator abandoned before
END_ARRAY. Existing recoverable-element and parse-mode row behavior must remain
unchanged.
**Success:** After normal, zero-row, eager-fallback, or terminal completion,
TaskContext no longer retains the completed parser. An abandoned iterator still
closes its current parser when the task completes. Repeated completed archive
entries do not accumulate parser or entry-buffer ownership until task
completion. Rows, corrupt-record values, and parse-mode errors remain unchanged.
--
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]
