tdcmeehan opened a new pull request, #59886:
URL: https://github.com/apache/spark/pull/59886
### What changes were proposed in this pull request?
This is a follow-up to #58704 and #59138 (`enableStreamingTopLevelArray`).
It changes two things in the streaming read of a multiline top-level JSON array.
1. A malformed array element is reported with its own text as the record,
instead of the whole document.
- `PERMISSIVE` puts the element's text in the corrupt record column.
- `FAILFAST` errors quote the element.
- `JacksonParser.parseIterator` reads the element's start byte offset at
the element's first token, and its end offset when the element fails. It then
cuts the element's bytes out of the document literal that the reader already
builds once per document.
- The end offset is read after `finishToken()`, because Jackson reads a
string's text lazily. No current path fails with an unread string, since the
converter reads the text first, so no test reaches that case.
- The record stays the whole document when the failure ends the document,
which is unchanged: a syntax error inside a nested value, a failure outside any
element, a truncated array. It also stays the whole document when the parser
reports no byte offsets, which is the case for an explicit `encoding` and for
UTF-16 or UTF-32 input, and when the offsets do not fit the literal.
- The test for "the element's extent is known" is the existing resume
test without its FAILFAST condition, so when the reader resumes does not change.
2. Reading an archive entry no longer registers a task completion listener.
- `parseIterator` registered one for every streamed array, to close a
parser abandoned before the closing bracket. A listener cannot be removed, so
each one stays until the task ends, even after its document has been read.
- `readArchive` reads every entry of an archive in one task through
`readStream`, which already holds the entry's bytes in memory, so a task kept
one listener per entry.
- `parseIterator` takes a new `closeOnTaskCompletion` parameter, true by
default, and `readStream` passes false. `readFile` is unchanged.
The configuration doc and the option row in `docs/sql-data-sources-json.md`
now say what the corrupt record column holds.
### Why are the changes needed?
- With a corrupt record column, K malformed elements carried K copies of the
document. An array of 10,000 `null`s against a struct schema is about 50 KB,
and every element is malformed, so it produced 10,000 corrupt rows of about 50
KB each, about 500 MB in total. Each corrupt record is now the 4-character text
`null`.
- `FAILFAST` quoted the whole document, even where the other modes resume
past exactly that element.
- A task reading an archive kept one listener, its closure, and an emptied
cell per entry until the task ended.
### Does this PR introduce _any_ user-facing change?
No, compared to any release: SPARK-59417 is new in the unreleased 4.4.0, and
`enableStreamingTopLevelArray` defaults to false.
Compared to master, and only with streaming enabled:
- The corrupt record column of a malformed element holds the element's text
instead of the whole document.
- A `FAILFAST` error quotes the element instead of the document, in the
cases where the other modes resume.
Reading `[{"a":1},{"a":"bad"},{"a":2}]` under `PERMISSIVE` with schema `a
int, _corrupt_record string`, writing `D` for the whole document:
| | rows |
| --- | --- |
| before | `(1, null)`, `(null, D)`, `(2, null)` |
| after | `(1, null)`, `(null, {"a":"bad"})`, `(2, null)` |
This needs a backport to branch-4.x, as SPARK-59417 (#58704, backported in
#59288) and its first follow-up (#59138, cherry-picked to branch-4.x as
317f7340) had.
### How was this patch tested?
- Updated existing tests only where the corrupt record value changes, which
is where the element's extent is known:
- `multiline top level JSON array parse modes with a malformed element`
(object, number and `null` elements);
- `multiline top level JSON array reuses the corrupt record literal`,
which still deletes the file after the first malformed row, so it still shows
the literal is built once;
- `multiline top level JSON array streaming option overrides the session
config`;
- `JSON: streaming archive arrays resume after a partial element`;
- `SPARK-18352: Handle multi-line corrupt documents (PERMISSIVE)`, where
only the streaming case changes. It outer-joins the corrupt record column
against the input documents. The three records of `[1,2,3]` are now `1`, `2`
and `3`, which match no document, so the join keeps `[1,2,3]` as one more
unmatched row: 9 rows instead of 8. The valid and corrupt counts are unchanged.
- The tests where the failure ends the document still expect the whole
document, unchanged.
- Added `multiline top level JSON array cuts a malformed element's record by
byte offset`, over a string element, whitespace and newlines around an element,
multi-byte UTF-8 before and inside the element, a UTF-8 BOM right before a
malformed first element, and an explicit `encoding`. Each runs in `PERMISSIVE`
and in `FAILFAST`, with `checkError` for `FAILFAST`.
- Added `multiline top level JSON array registers a task completion listener
only for a file`. It counts the listeners a streamed read adds over an eager
read of the same input: one for a file, none for an archive entry. It compares
against the eager read because the file stream opener and the archive reader
register listeners of their own.
- Every new or changed assertion fails against master, except two that check
behavior this PR keeps:
- the explicit `encoding` cases, which expect the whole document either
way and fail if the byte-offset fallback is removed;
- the listener test's file case, which fails if `readFile` stops
registering its listener.
- Mutants, each run against the tests above:
- slicing by character instead of byte fails the multi-byte UTF-8 cases;
- `FAILFAST` quoting the document fails only the `FAILFAST` cases;
- `readStream` registering a listener, `readFile` not registering one, or
`parseIterator` ignoring the flag each fail one case of the listener test.
- Checked the byte offsets end to end with a throwaway suite, not included
here:
- an element straddling Jackson's 8,000-byte read buffer, in a plain file,
a gzip file, and a tar entry;
- comments next to elements, with `allowComments`;
- UTF-16 input, which keeps the whole document.
- Measured reading each element's start offset with `JsonBenchmark
top-level-array`, master and this change alternated twice, in a local run on
one machine (Intel Xeon 6975P-C, OpenJDK 17.0.15), not CI. The streamed
100,000-element case took 42 ms best in all four runs (422 to 424 ns per row).
The 64 KB cases moved by up to 18% between runs, in both directions, and by up
to 9% in the non-streaming case, which runs no changed code.
- Ran:
- `catalyst/testOnly *JacksonParserSuite *JsonInferSchemaSuite
*UnivocityParserSuite *JsonExpressionsSuite` 4 suites, 128 passed, 0 failed
- `sql/testOnly *JsonV1Suite *JsonV2Suite *JsonLegacyTimeParserSuite
*JsonUnsafeRowSuite *JSONTarArchiveReadSuite *JSONZipArchiveReadSuite
*JSONSevenZArchiveReadSuite *JsonFunctionsSuite` 10 suites, including the two
`ExplodeEmbeddedArrayJson*` suites the patterns also match: 1,188 passed, 0
failed, 1 ignored, the test `JsonLegacyTimeParserSuite` already excludes
- `./dev/lint-scala`: scalastyle passed. The scalafmt step covers only
`sql/api` and `sql/connect`, which this PR does not touch.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Isaac (Claude Opus 5.5)
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]
