[SPARK-59417][SQL] Stream multiline top-level JSON arrays

[tdcmeehan]

What changes were proposed in this pull request?

This PR adds the internal Spark SQL configuration spark.sql.json.enableStreamingTopLevelArray and the per-read JSON option enableStreamingTopLevelArray, which overrides it. When enabled, multiline JSON reading consumes a top-level array of structs lazily from one Jackson parser instead of materializing the entire array before returning rows. Both default to false. The two paths agree on valid input but differ on malformed input, described below, so this is a choice a read makes rather than a switch to be flipped once.

It also adds an opt-in lazy path to FailureSafeParser for failures raised while advancing the parser iterator, preserves rows emitted before later structural corruption, and adds benchmark cases for both many small elements and fewer large elements. Existing eager FailureSafeParser callers remain unchanged.

A malformed element is resumed from only where the parser cannot have moved past it: a scalar element leaves the parser on that scalar's own token, and convertObject consumes through the closing brace before reporting a partial result. Any other container failure can stop anywhere inside its element, so it ends the document as before.

Because the streaming iterator holds a JsonParser for the life of the read, it registers close() with the active TaskContext completion listener in addition to closing eagerly when the array ends or a terminal failure occurs. An iterator abandoned before the closing bracket (under a LIMIT, for example) therefore releases the parser when the task finishes rather than at GC. The listener has no removal API, so the parser is held in a cell that every close path clears before closing; otherwise a completed read leaves a closed parser reachable from the task, keeping its buffer alive, which matters where one task builds many parsers such as an archive read. Both multiline readers also memoize the whole-document corrupt-record literal they hand to the parser, so it is built once per read instead of once per malformed element.

Why are the changes needed?

The existing parser materializes every element of a multiline top-level JSON array in memory. Large arrays can therefore require memory proportional to the complete document even though Spark consumes the result as an iterator. Streaming elements bounds parser-side memory, producing the same rows as the eager path for valid input. For malformed input it changes which unit a parse mode applies to, which is why the path is opt-in rather than the default.

Does this PR introduce any user-facing change?

Yes, when the configuration or the JSON option is enabled. Reading a multiline top-level JSON array no longer materializes the complete array before producing rows, and the record a parse mode applies to becomes one array element rather than the whole document. This follows from streaming: rows already handed to the consumer cannot be withdrawn or retroactively stamped, so preserving the document as the record would mean buffering to the closing bracket.

Reading [{"a":1},<malformed>,{"a":2}] under PERMISSIVE with schema a int, _corrupt_record string, writing D for the whole document:

malformed element	disabled	enabled
{"a":"bad"}	(1, D), (null, D), (2, D)	(1, null), (null, D), (2, null)
42	(null, D)	(1, null), (null, D), (2, null)
[1,2]	(null, D)	(1, null), (null, D)

DROPMALFORMED drops the corrupt row of each cell, which leaves no rows at all when disabled since every row carries the document's corruption. FAILFAST raises MALFORMED_RECORD_IN_PARSING.WITHOUT_SUGGESTION in both columns.

The last row is the element whose failure leaves the parser at an unknown position, so enabling the configuration keeps the rows before it and still ends the document there. The configuration doc says so.

One consequence reaches valid documents of a shape the eager path never supported: a top-level array whose elements are all scalars, such as [1,2,3], is malformed against a struct schema in every element, so it reports one corrupt record per element where the whole document reported one. Object elements already behaved that way, through the partial-result path the eager reader has always had.

Valid input is unaffected, and both the configuration and the option default to false.

How was this patch tested?

Added regression coverage for lazy top-level array parsing, for the JSON option overriding the session configuration, and for malformed input after partial output under PERMISSIVE, DROPMALFORMED, and FAILFAST with the configuration both disabled and enabled, so every cell of the table above is pinned in both columns. The three element shapes are parameterized over one document, which pins the nested array as the shape that must stay terminal, so a resumability rule that is too broad fails. Extended the existing SPARK-3308 top-level-array coverage, and made the existing SPARK-18352 multiline corrupt-document expectations configuration-dependent, which is where the all-scalar array above is pinned. The archive suites exercise the same matrix through the pre-buffered readStream path.

Two further cases cover a truncated array ([{"a":1} with no closing bracket) and an iterator abandoned mid-array, asserting the parser's stream is closed once the task completes. The truncated array reaches the malformed-record arm rather than the iterator's end-of-input branch: jackson-core throws JsonEOFException while the array context is still open, so that branch is unreachable today and is kept only so a backend that returns null cannot fall through with no current token. Parser lifetime is covered over empty, empty-array, valid, and truncated documents by counting both the rows each yields and its close() calls, through a JsonParserDelegate: exactly one close before task completion and none added by it, so a parser still retained after its close fails the test. Those four documents are also the four ways the iterator can end, so the zero-row branches are held to zero rows and to closing.

Ran:

• sql/Test/compile
• sql/testOnly *JsonV1Suite *JsonV2Suite (431 tests)
• sql/testOnly *JsonV1Suite *JsonV2Suite *JsonParsingOptionsSuite *ExplodeEmbeddedArrayJsonV1Suite *ExplodeEmbeddedArrayJsonV2Suite *JSONTarArchiveReadSuite *JSONZipArchiveReadSuite *JSONSevenZArchiveReadSuite (569 tests)
• catalyst/testOnly *JsonExpressionsSuite *JacksonParserSuite *JsonSuite (140 tests)
• sql/testOnly *JsonFunctionsSuite (115 tests)
• catalyst/scalastyle
• sql/scalastyle
• catalyst/Test/scalastyle
• sql/Test/scalastyle
• git diff --check

The focused OSS benchmark can be run with:

build/sbt "sql/Test/runMain org.apache.spark.sql.execution.datasources.json.JsonBenchmark top-level-array"

The benchmark reads multiline JSON through the data source with spark.sql.json.enableStreamingTopLevelArray disabled and enabled. On OpenJDK 17.0.15 and Intel Xeon 6975P:

• 100,000 rows with empty payloads: disabled 69 ms; enabled 42 ms.
• 1,000 rows with 64 KiB payloads: disabled 127 ms; enabled 118 ms.

Was this patch authored or co-authored using generative AI tooling?

Generated-by: OpenAI Codex (GPT-5)
Generated-by: Isaac (Claude Opus 5)

This pull request and its description were written by Isaac.

[HyukjinKwon]

Mind taking a look at https://github.com/apache/spark/pull/58704/checks?check_run_id=103066275835?

[tdcmeehan]

@HyukjinKwon done, thanks!

[cloud-fan]

Review summary

The enabled streaming path does not yet model the full parser failure state machine: it can silently lose later valid rows after recoverable element errors, bypass configured parse modes before parser creation, and lacks coverage for those states and the archive entry path. I also found one smaller duplicate cleanup registration in the multiline file reader. The flag's false default limits current exposure, so these are non-blocking findings, but the parser-state issues should be resolved before relying on the enabled path.

Findings

4 total: 0 P0, 0 P1, 3 P2, 1 P3.

Non-blocking (P2)

• Keep parsing after a recoverable partial element — sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala:776 — see inline.
• Handle parser-construction failures through the parse mode — sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala:764 — see inline.
• Cover recoverable and archive streaming paths — sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/json/JsonSuite.scala:1121 — see inline.

Nit (P3)

• Avoid registering stream cleanup twice — sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/json/JsonDataSource.scala:418 — see inline.

Shared repair plans

Shared repair plan 1

Covered findings:

• Keep parsing after a recoverable partial element — sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala:776
• Handle parser-construction failures through the parse mode — sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala:764
• Cover recoverable and archive streaming paths — sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/json/JsonSuite.scala:1121

Recommended change: Model parser acquisition and iterator advancement as separate guarded states. Convert acquisition failures to BadRecordException before returning the iterator. For a recoverable partial result after one array element is consumed, preserve the live source iterator so PERMISSIVE recovery can emit the partial/corrupt row and resume with the next element; continue to close and terminate for structural/document failures and honor the other parse modes. Add focused JSON streaming coverage for recoverable field conversion, parser-construction encoding failure, and the enabled archive-entry readStream path.

Why this works: Guard createParser with the same error-to-badRecord mapping without assuming a JsonParser exists to close. After acquisition, classify element-level PartialResultException separately from terminal tokenizer/document errors; carry recovery output without overwriting the resumable source iterator, then return to that source once recovery rows are drained. Keep terminal close, suppression, and corrupt-record materialization behavior for non-resumable failures. Exercise each state through the public multiline data-source paths, including an archive entry.

Scope: Restore eager-compatible parse-mode semantics across the streaming parser's acquisition and per-element recovery states, with complete standalone and archive regression coverage.

Compatibility: Disabled reads remain eager; enabled valid arrays remain lazy; non-array documents retain their existing eager fallback; existing corrupt-record text and parse-mode error identities remain stable.

Risks: Resuming after the wrong exception class could continue from an undefined Jackson token position. A recovery adapter could duplicate a partial row or skip the next token if hasNext and next transitions are not idempotent. Changing terminal cleanup could leak streams on syntax, construction, cancellation, or early-consumer termination.

Constraints: Resume only after failures for which the current array element has been completely consumed and parser position is defined. Preserve the disabled eager path and the off-by-default activation contract. Preserve corrupt-record text and existing FAILFAST error conversion. Do not materialize the remaining top-level array to implement recovery.

Success: In enabled PERMISSIVE mode, a recoverable schema-conversion error in one top-level-array element emits the appropriate partial/corrupt row and later valid elements exactly once. Parser-construction failures are translated through PERMISSIVE, DROPMALFORMED, and FAILFAST consistently with the disabled path. Terminal structural corruption closes the parser, preserves rows already emitted under the approved late-error contract, and does not resume from an undefined token state. Enabled standalone-file and archive-entry reads have equivalent row, corrupt-record, and error behavior for the covered modes. The streaming path remains lazy and does not materialize the complete top-level array.

[cloud-fan]

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.

[cloud-fan]

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.

[cloud-fan]

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.

[cloud-fan]

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.

[tdcmeehan]

Thanks @cloud-fan, addressed your comments

[cloud-fan]

Review summary

The latest revision addresses the previously reported iterator-resumption, parser-acquisition, and duplicate-cleanup issues. Three non-blocking problems remain: non-array partial recovery can retain file streams until task completion, repeated malformed elements can reread the complete source file once per recovery row, and the archive streaming test still covers only successful input rather than the route's distinct failure behavior.

Findings

3 total: 0 P0, 0 P1, 3 P2, 0 P3.

Non-blocking (P2)

• Close non-array parsers before returning recovery rows — sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala:785 — see inline.
• Exercise archive streaming failures, not only valid arrays — sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/JSONArchiveReadBase.scala:169 — see inline.
• Cache the corrupt-record literal once per file — sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/json/JsonDataSource.scala:421 — see inline.

Re-review status

Prior AI findings: 3 addressed, 1 still present; additional unresolved findings in this review: 2.

New attribution: 1 newly introduced, 1 late catch, 0 previously raised, 0 unattributed.

Remaining prior AI findings

• Exercise archive streaming failures, not only valid arrays — sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/JSONArchiveReadBase.scala:169

Existing discussions

• existing discussion — The current head addresses the four previously posted findings, but the revised recovery state machine leaves a distinct related cleanup defect: a partial failure in the non-array fallback is marked recoverable even though no source iterator can be resumed, so the parser is not closed promptly.
• existing discussion — The current tests add the requested partial-element, parser-construction, and successful archive cases. A related gap remains because enabled malformed archive handling still has no mode-sensitive assertion for readStream's distinct recovery and corrupt-record wiring.

[cloud-fan]

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.

[cloud-fan]

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.

[cloud-fan]

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.

[cloud-fan]

Thanks, the earlier comments are addressed. I found one related cleanup issue in the revised path: when the enabled iterator parser falls back to a non-array document and root conversion produces a partial result, it returns recovery output without closing the non-resumable parser.

[cloud-fan]

Review summary

The latest revision resolves the three findings from the previous review. I found four additional non-blocking issues: archive tasks can retain every completed entry buffer through task-completion listeners, the configuration documentation overstates which option combinations stream, the recursive element converter is built twice per input, and the zero-row and normal-exhaustion close paths lack regression assertions. None is release-blocking, but each is worth addressing before merge.

Findings

4 total: 0 P0, 0 P1, 3 P2, 1 P3.

Non-blocking (P2)

• Release completed archive-entry parsers before task completion — sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala:778 — see inline.
• Avoid constructing the recursive converter twice per input — sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala:764 — see inline.
• Cover zero-row branches and normal parser closure — sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/json/JsonSuite.scala:1077 — see inline.

Nit (P3)

• Document the option combinations excluded from streaming — sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:7040 — see inline.

Re-review status

Prior AI findings: 3 addressed, 0 still present; additional unresolved findings in this review: 4.

New attribution: 1 newly introduced, 3 late catch, 0 previously raised, 0 unattributed.

Remaining prior AI findings

No prior AI findings remain.

[cloud-fan]

Non-blocking (P2): The task-completion listener captures this JsonParser directly, and closing the parser does not clear that captured reference. On archive reads, the parser can retain the ByteArrayInputStream and complete byte array for its entry, so consuming many entries retains all completed entry buffers until task completion and can OOM despite the streaming path. Please make the fallback cleanup stop retaining a parser once its eager close path has run.

Recommended change: Replace the per-parser task-listener capture with one clearable parser-lifecycle owner used by normal completion, terminal failure, eager fallback, and task completion, and add lifecycle and archive-entry retention regression coverage.

Why this works: The task listener captures only a clearable ownership handle. Every eager close atomically consumes or clears that handle before releasing the parser, while task completion consumes the same handle and closes it only when an iterator was abandoned. Completed parsers therefore cease to be reachable from TaskContext without weakening abandonment cleanup.

Scope: Unify streaming JsonParser close ownership across eager and task-completion paths and verify that archive entries do not accumulate completed parser state.

Compatibility: Streaming remains lazy, terminal and recoverable failures retain their current parse-mode semantics, and task completion remains a cleanup backstop for early iterator abandonment.

Risks: A close route could clear ownership before the parser is actually closed, weakening cleanup after a close failure. Independent close paths could race or mask the original parse failure if they do not consume the shared owner consistently.

Constraints: Normal completion and terminal parse failures must still close eagerly. Task completion must still close the parser when a consumer abandons the iterator. Parse-mode recovery and lazy element delivery must remain unchanged.

Success: After an entry iterator finishes or fails, TaskContext no longer retains its parser or backing entry bytes. An iterator abandoned before completion still has its current parser closed at task completion. Successful, recoverable, and terminal parsing retain their current rows and parse-mode behavior.

[tdcmeehan]

Done in dd76c16. 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.

[cloud-fan]

Nit (P3): This description makes the setting sound unconditional, but the parser only selects the streaming path when neither singleVariantColumn nor explodeEmbeddedArray is active. With either option, the top-level array is still converted eagerly. Please qualify the description so it states the actual option scope.

[tdcmeehan]

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.

[cloud-fan]

Non-blocking (P2): JacksonParser construction has already called makeRootConverter(schema), whose struct branch builds the equivalent element converter. Calling makeConverter(schema) again here duplicates recursive converter construction for every file or archive entry, and the work is wasted entirely for non-array roots. Please reuse the existing converter graph or defer this construction until an array root has actually been observed.

[tdcmeehan]

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.

[cloud-fan]

Non-blocking (P2): The new enabled-path cases do not exercise either zero-row branch: an empty document closes before the streaming iterator is constructed, while an empty array closes from the iterator's first prepare call. They also do not assert synchronous closure after normal array exhaustion. Please add cases that assert both zero returned rows and parser closure for the empty inputs, plus closure after fully consuming a normal array.

[tdcmeehan]

Done in dd76c16 and 75561b8. 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.

[cloud-fan]

Review summary

The four non-blocking findings from the previous review remain open in their existing threads. This pass also found two additional P2 issues: an incompatible scalar element in the streamed top-level array terminates recovery and discards later valid rows, and the terminal-failure tests do not assert prompt parser closure. None of the six findings is release-blocking, but they are worth addressing before merge.

Findings

6 total: 0 P0, 0 P1, 5 P2, 1 P3.

Non-blocking (P2)

• Reuse the recursive element converter — sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala:764 — remaining in an existing discussion.
• Release completed archive-entry parsers before task completion — sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala:778 — remaining in an existing discussion.
• Assert prompt closure on zero-row and normal completion — sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/json/JsonSuite.scala:1077 — remaining in an existing discussion.
• Recover after malformed scalar array elements — sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala:794 — see inline.
• Assert prompt closure on terminal parser failures — sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala:780 — see inline.

Nit (P3)

• Document the option combinations excluded from streaming — sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:7039 — remaining in an existing discussion.

Re-review status

Prior AI findings: 0 addressed, 4 still present; additional unresolved findings in this review: 2.

New attribution: 0 newly introduced, 2 late catch, 0 previously raised, 0 unattributed.

Remaining prior AI findings

• Document the option combinations excluded from streaming — sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:7039
• Reuse the recursive element converter — sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala:764
• Release completed archive-entry parsers before task completion — sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala:778
• Assert prompt closure on zero-row and normal completion — sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/json/JsonSuite.scala:1077

Existing discussions

• Remaining: Document the option combinations excluded from streaming — P3 at sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:7039 — existing discussion
• Remaining: Reuse the recursive element converter — P2 at sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala:764 — existing discussion
• Remaining: Release completed archive-entry parsers before task completion — P2 at sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala:778 — existing discussion
• Remaining: Assert prompt closure on zero-row and normal completion — P2 at sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/json/JsonSuite.scala:1077 — existing discussion

[cloud-fan]

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.

[tdcmeehan]

Done in dd76c16. 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.

[cloud-fan]

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.

[tdcmeehan]

Done in dd76c16, 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.