[SPARK-57419][SQL] Read and infer JSON schema from tar archives

[akshatshenoi-db]

What changes were proposed in this pull request?

SPARK-57135 added reading CSV files packed in tar archives (.tar/.tar.gz/.tgz) and SPARK-57321 added schema inference for them, both gated by spark.sql.files.archive.reader.enabled. This extends the same capability to the JSON data source.

When the flag is enabled, the V1 JSON data source reads a tar archive as if it were a directory of its entries: each entry is streamed through ArchiveReader (never unpacked to disk) and parsed exactly like a standalone JSON file, for both line-delimited and multi-line JSON (JsonDataSource.readArchive/readStream). Schema inference reads every archive entry together with any loose files in a single JsonInferSchema pass (inferWithArchives), so the inferred schema matches a directory read of the same files. The whole archive is one non-splittable unit (JsonFileFormat.isSplitable returns false), and a corrupt/missing archive is skipped as a unit under ignoreCorruptFiles/ignoreMissingFiles. The DSv2 reader cannot read archives, so JsonTable passes supportsArchiveScan = false and refuses to infer a schema for archive inputs (raising UNABLE_TO_INFER_SCHEMA).

Unlike CSV, JSON needs no per-entry header handling (records are self-describing, so one parser serves every entry) and no mergeSchema-style branching (JsonInferSchema already merges record types by field name across all inputs, so one pass is itself the union).

This also unifies the archive test suites: the format-agnostic inference and complex-type tests are hoisted into ArchiveReadSuiteBase behind capability hooks (supportsSchemaInference, supportsComplexTypes), so CSV, JSON, and future archive formats share them instead of each duplicating them.

Why are the changes needed?

To let JSON ingestion read tar archives without unpacking them to disk, matching the CSV behavior already in Spark.

Does this PR introduce any user-facing change?

Yes. With spark.sql.files.archive.reader.enabled=true (default false), the JSON data source can read and infer schemas from .tar/.tar.gz/.tgz files.

How was this patch tested?

New JSONTarArchiveReadSuite (mixing JSONArchiveReadBase with the shared ArchiveReadSuiteBase and TarArchiveReadBase), plus the hoisted shared inference and complex-type tests now also exercised by the CSV suites.

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

Generated-by: Claude Code

[cloud-fan]

1 blocking, 2 non-blocking, 1 nits.
Faithful port of the CSV archive pattern with nicely unified shared tests; one narrow parity bug in no-encoding multiLine inference should be fixed before merge.

Correctness (2)

• JsonDataSource.scala:159: no-encoding inference parses strictly as UTF-8 while the scan and directory reads auto-detect — UTF-16 multiLine docs infer corrupt with archives present — see inline
• JsonDataSource.scala:148: comment claims records are copied but fromBytes wraps the reused line buffer — see inline

Suggestions (1)

• JSONArchiveReadBase.scala:193: no test covers a malformed record inside an archive entry (_corrupt_record parity) — see inline

Nits: 1 minor item (see inline comments).

[cloud-fan]

When encoding is unset, this parses records strictly as UTF-8: CreateJacksonParser.utf8String wraps an InputStreamReader with StandardCharsets.UTF_8. But the archive scan path (MultiLineJsonDataSource.readStream -> CreateJacksonParser.inputStream) and a directory read's inference are byte-based, so Jackson auto-detects UTF-16/UTF-32 there. A multiLine UTF-16 document with no encoding option therefore infers a corrupt-record-only schema whenever an archive is among the inputs, but the real schema from a directory read of the same files — and loose files alongside the archive are affected too, since every input routes through this pass. The new encoding test only covers the explicit-encoding case.

I'd carry records as Array[Byte] and choose the parser the way CreateJacksonParser.text/internalRow do: factory.createParser(bytes, 0, len) when encoding is empty (byte-based, auto-detects, matching the scan), getStreamDecoder when set. That also subsumes the buffer-copy issue flagged at line 148.

[cloud-fan]

This comment overstates what the no-encoding branch does: UTF8String.fromBytes(bytes, 0, length) wraps the array without copying, and line.getBytes is lineIterator's single reused Text buffer. It works today only because sampling and inference consume each record before the next readLine overwrites the buffer — any future buffering step downstream (a cache, batched sampling) would silently corrupt records. Copying for real (line.copyBytes()) is cheap next to the parse cost and makes the comment true.

[cloud-fan]

Both readStream implementations wire new FailureSafeParsers, and the multiLine corrupt-record echo differs from readFile's (pre-buffered bytes vs. file re-read), but no test puts a malformed JSON record inside an archive entry. A small assertArchiveMatchesDir case with one bad record would pin down _corrupt_record/permissive-mode parity with a directory read for both modes.

[cloud-fan]

Only JSON overrides this — no Avro/Parquet/XML archive traits exist yet.

Suggested change

	* complex-type round-trip test; CSV and text leave it false, JSON/Avro/Parquet/XML override true.
	* complex-type round-trip test; CSV and text leave it false, JSON overrides it to true.

[cloud-fan]

4 addressed, 0 remaining, 2 new. (2 = 2 newly introduced, 0 late catches.)
The fix commit resolves every round-1 finding exactly as recommended; one narrow charset-parity edge and a stale doc line remain.

Correctness (1)

• JsonDataSource.scala:185: multiLine + an encoding outside CharsetProvider's allow-list fails archive inference with INVALID_PARAMETER_VALUE.CHARSET while the scan and a directory read's inference accept it — see inline

Nits: 1 minor item (see inline comments).

Verification

Traced all charset modes through the new Array[Byte] pipeline against the scan and loose-file inference paths: no-encoding line-delimited and multiLine now auto-detect like the scan (UTF-16 pinned by the new test); explicit allow-listed encodings share the scan's decoder machinery; explicit non-allow-listed line-delimited matches the scan's rejection. The only divergent cell is the multiLine non-allow-listed case flagged inline. Also verified Text.copyBytes() returns a fresh exact-length array, closing the round-1 buffer-reuse hazard.

[cloud-fan]

One charset cell still diverges from the scan: multiLine with an encoding outside CharsetProvider's allow-list (e.g. windows-1252, which JSONOptionsInRead allows in multiLine mode). bytes(enc, ...) decodes via CharsetProvider.newDecoder, which rejects such charsets unless spark.sql.legacy.javaCharsets is set, and with isReadFile=true the exception propagates out of JsonInferSchema.infer — so archive inference fails with INVALID_PARAMETER_VALUE.CHARSET while the multiLine scan (CreateJacksonParser.inputStream, a raw InputStreamReader) and a directory read's inference accept the same files. Line-delimited is unaffected: its scan path uses the same getStreamDecoder.

I'd build the multiLine parser as CreateJacksonParser.inputStream(enc, factory, new ByteArrayInputStream(record)) so each mode matches its own scan path exactly. Non-blocking: the failure is loud and the charset combination is rare.

[cloud-fan]

Went stale with the switch to raw bytes:

Suggested change

	* multiLine path (`RDD[PortableDataStream]`) and the archive inference path (`RDD[UTF8String]`)
	* multiLine path (`RDD[PortableDataStream]`) and the archive inference path (`RDD[Array[Byte]]`)

[cloud-fan]

2 addressed, 0 remaining, 3 new. (3 = 1 newly introduced, 2 late catches — my misses from earlier rounds.)
The fix resolves both round-2 findings exactly as recommended; the charset matrix is now fully scan-parity. What remains is one test-strengthening suggestion and two doc nits.

Suggestions (1)

• JSONArchiveReadBase.scala:218: the windows-1252 fixture is pure ASCII, so the test can't detect the encoding being silently ignored (only it being rejected) — see inline

Nits: 2 minor items (see inline comments).

Verification

Traced the new per-mode recordParser against each scan path: multiLine now uses the same CreateJacksonParser.inputStream calls as MultiLineJsonDataSource (plain InputStreamReader for an explicit encoding, byte auto-detect otherwise), closing the round-2 windows-1252 divergence, and line-delimited keeps the bytes/stream-decoder path that already matched. JsonInferSchema.infer is invoked with isReadFile = multiLine, matching each mode's loose-file inference, so error classification is also identical between archive and directory inference.

[cloud-fan]

The fixture is pure ASCII, which windows-1252 and UTF-8 encode identically — so if the explicit encoding were silently ignored (the auto-detect branch), the test would still pass. A non-ASCII windows-1252 byte (0xE9) is malformed standalone UTF-8, so it makes that regression observable: the archive side would infer a corrupt-record-only schema and fail the parity assert.

Suggested change

	// No "name" field: the schema's "name" column must read back as null for this entry.
	val bytes = "{\n \"id\": 1,\n \"name\": \"Jos\u00e9\"\n}".getBytes("windows-1252")

[cloud-fan]

Reads as an incomplete comparison — the sibling comments spell out the verb ("as a directory read does"). The line is at 99 chars, hence the rewrap.

Suggested change

	* `extraOptions`). Loading several paths reads them as one fileset, exactly as a directory read.
	* `extraOptions`). Loading several paths reads them as one fileset, exactly as a directory
	* read does.

[cloud-fan]

JSON<Archive>Read doesn't expand to the real suite name — the tar suite is JSONTarArchiveReadSuite, and TarArchiveReadBase documents the <format>TarArchiveReadSuite pattern.

Suggested change

	* `JSON<Archive>Read` suite mixes this in alongside the archive-format trait.
	* `JSON<Container>ArchiveReadSuite` mixes this in alongside the archive-format trait.

[cloud-fan]

3 addressed, 0 remaining, 2 new. (2 = 2 newly introduced, 0 late catches.)
All three round-3 items are resolved and the charset matrix stays scan-parity. The two new items are on the round-4 streaming change: one non-blocking design note and one doc nit.

Design / architecture (1)

• JsonDataSource.scala:183: multi-line inference now hands infer an RDD[InputStream] of live views over the shared tar cursor; correct only because infer fully consumes each before advancing. Undocumented invariant the line-delimited and CSV paths don't carry (both materialize) — recommend documenting it. See inline.

Nits: 1 minor item (see inline comments).

Verification

Traced the multi-line streaming path: JsonInferSchema.infer consumes each entry stream via flatMap { tryWithResource(parse) }.reduceOption, fully parsing each document before the readEntries iterator advances the tar cursor (which skips unread bytes), and inferWithArchives is its only consumer — so the live CloseShieldInputStream views stay valid. RDD[InputStream] never crosses a serialization boundary (one narrow-dependency stage). Charset parity holds: inference and the scan both build the parser via CreateJacksonParser.inputStream. Default sampling is a no-op; under <0.99 the sampler's filter discards a dropped stream and the next advance skips it at the tar level.

[cloud-fan]

These RDD[InputStream] elements are live CloseShieldInputStream views over the single shared TarArchiveInputStream cursor — each valid only until readEntries advances to the next entry (getNextEntry skips the prior entry's unread bytes). This is correct today only because JsonInferSchema.infer fully consumes each stream (flatMap { tryWithResource(parse) }.reduceOption) before pulling the next, and inferWithArchives is its only consumer.

The line-delimited sibling (RDD[Array[Byte]]) and the CSV analogue (RDD[Array[String]]) both materialize their records, so they don't carry this constraint. If infer's per-partition consumption ever changes to buffer / look ahead / parallelize, this path would silently read from an advanced cursor and infer a wrong schema (no exception; the one-entry fixtures here can't observe it). Non-blocking, and the streaming is a worthwhile memory optimization — but worth a line in the perInput comment documenting that the consumer must fully consume each stream before the iterator advances.

[cloud-fan]

Repeated "input" reads awkwardly.

Suggested change

	// a whole input on corrupt/missing input when the ignore flags are set. The entry/file stream
	// a whole input when it is corrupt/missing and the ignore flags are set. The entry/file stream

[cloud-fan]

thanks, merging to master/4.x!