tdcmeehan opened a new pull request, #59288:
URL: https://github.com/apache/spark/pull/59288
### What changes were proposed in this pull request?
This is a backport of #58704 to `branch-4.x`. It is a clean cherry-pick of
the merged commit `f4327af76197334dbf893a6f7ae34e2d4a096c5f`; the only
conflicts were the regenerated `JsonBenchmark*-results.txt` files, resolved to
the committed benchmark output (identical benchmark structure plus the two new
top-level-array sections).
This PR adds the Spark SQL configuration
`spark.sql.json.enableStreamingTopLevelArray` and the per-read JSON option
`enableStreamingTopLevelArray`, which overrides it. When enabled, reading a
multiline JSON file 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?
The regression coverage from #58704 is carried over unchanged: lazy
top-level array parsing; the JSON option overriding the session configuration
in both directions; 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 parameterized over one document (pinning the nested array as the shape
that must stay terminal); extended SPARK-3308 top-level-array coverage;
configuration-dependent SPARK-18352 multiline corrupt-document expectations;
the archive suites exercising the same matrix through the pre-buffered
`readStream` path; and the parser-lifetime coverage over empty, empty-array,
valid, and truncated documents counting rows and `close()` calls through a
`JsonParserDelegate`.
Backport verification: the code, test, and doc changes are byte-identical to
the merged commit (confirmed by a content-level diff against `f4327af`).
Compiled `sql/Test/compile` (catalyst + sql/core, main and test) on
`branch-4.x`. The suites above run here in CI.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Isaac (Claude Opus 4.8)
The code, tests, and docs are an unmodified cherry-pick of #58704
(generative-AI tooling for the original change is disclosed there). This
backport — the cherry-pick, the benchmark-file conflict resolution, and this
description — was prepared with Isaac.
This pull request and its description were written by Isaac.
--
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]
