You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "akshatshenoi-db (via GitHub)" <gi...@apache.org> on 2026/06/12 18:58:14 UTC
[PR] [SPARK-57419][SQL] Read and infer JSON schema from tar archives [spark]
akshatshenoi-db opened a new pull request, #56480:
URL: https://github.com/apache/spark/pull/56480
### 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
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57419][SQL] Read and infer JSON schema from tar archives [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #56480:
URL: https://github.com/apache/spark/pull/56480#discussion_r3406719629
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/ArchiveReadSuiteBase.scala:
##########
@@ -104,6 +106,47 @@ trait ArchiveReadSuiteBase extends QueryTest with SharedSparkSession {
schema: String = readSchema): DataFrame =
spark.read.format(format).options(readOptions ++ extraOptions).schema(schema).load(path)
+ // ----- schema-inference / complex-type capability hooks (default off) ------
+ //
+ // A format opts into the shared inference / complex-type tests below by overriding the
+ // corresponding `supports*` hook to true; a format that supports neither (e.g. text) leaves both
+ // false and runs only the read tests.
+
+ /**
+ * Whether this format infers its read schema from the textual content of the files (CSV, JSON,
+ * XML), as opposed to carrying an embedded schema (Avro, Parquet) or none (text). Gates the
+ * shared schema-inference tests. A format that needs an option to trigger inference (e.g. CSV's
+ * `inferSchema`) also overrides [[inferenceOptions]].
+ */
+ protected def supportsSchemaInference: Boolean = false
+
+ /** Extra options that trigger/control inference for [[supportsSchemaInference]] formats. */
+ protected def inferenceOptions: Map[String, String] = Map.empty
+
+ /**
+ * Whether this format can represent nested/complex types (struct/array/map). Gates the shared
+ * complex-type round-trip test; CSV and text leave it false, JSON overrides it to true.
+ */
+ protected def supportsComplexTypes: Boolean = false
+
+ /** Sample data with a nested struct column, used by the complex-type test. */
+ protected def complexSampleDf: DataFrame =
+ Seq((1, "NYC", "10001"), (2, "SF", "94105")).toDF("id", "city", "zip")
+ .select(col("id"), struct(col("city"), col("zip")).as("addr"))
+
+ /** Read schema matching [[complexSampleDf]]. */
+ protected def complexReadSchema: String = "id INT, addr STRUCT<city: STRING, zip: STRING>"
+
+ /**
+ * Schema [[format]] infers from `paths` under [[readOptions]] ++ [[inferenceOptions]] (plus
+ * `extraOptions`). Loading several paths reads them as one fileset, exactly as a directory read.
Review Comment:
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.
```suggestion
* `extraOptions`). Loading several paths reads them as one fileset, exactly as a directory
* read does.
```
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/JSONArchiveReadBase.scala:
##########
@@ -0,0 +1,265 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution.datasources
+
+import java.io.File
+import java.nio.charset.StandardCharsets
+import java.nio.file.Files
+
+import org.apache.spark.sql.{AnalysisException, DataFrame}
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types.{NullType, StringType}
+import org.apache.spark.util.Utils
+
+/**
+ * Binds [[ArchiveReadSuiteBase]]'s file-format hooks to JSON. JSON opts into the shared
+ * schema-inference and complex-type tests (see `supportsSchemaInference`/`supportsComplexTypes`),
+ * and adds the JSON-specific tests with no format-agnostic analogue: NullType canonicalization,
+ * field-union/null-in-loose merging, and multi-line documents. Reusable across archive formats: a
+ * `JSON<Archive>Read` suite mixes this in alongside the archive-format trait.
+ */
+trait JSONArchiveReadBase extends ArchiveReadSuiteBase {
+
+ override protected def format: String = "json"
+
+ override protected def fileExtension: String = "json"
+
+ // JSON records are self-describing, so no per-file read options are required.
+ override protected def readOptions: Map[String, String] = Map.empty
+
+ override protected def readSchema: String = "id INT, name STRING"
+
+ // JSON infers its schema from record content and represents nested structs, so it opts into the
+ // shared inference and complex-type tests. Inference needs no trigger option, so the inherited
+ // `inferenceOptions` keeps its empty default.
+ override protected def supportsSchemaInference: Boolean = true
+
+ override protected def supportsComplexTypes: Boolean = true
+
+ override protected def encodeFile(
+ df: DataFrame,
+ writeOptions: Map[String, String]): Array[Byte] = {
+ val dir = Utils.createTempDir(namePrefix = "archive-test-encode")
+ try {
+ df.coalesce(1).write.format("json")
+ .options(writeOptions)
+ .mode("overwrite").save(dir.getCanonicalPath)
+ val parts = dir.listFiles().filter { f =>
+ f.isFile && !f.getName.startsWith("_") && !f.getName.startsWith(".") &&
+ !f.getName.endsWith(".crc")
+ }
+ assert(parts.length == 1,
+ s"expected exactly one data file, got: ${parts.map(_.getName).toList}")
+ Files.readAllBytes(parts.head.toPath)
+ } finally Utils.deleteRecursively(dir)
+ }
+
+ /** Raw JSON bytes, for tests that need precise control over the record layout. */
+ protected def jsonBytes(s: String): Array[Byte] = s.getBytes(StandardCharsets.UTF_8)
+
+ // ----- JSON-specific schema inference --------------------------------------
+ // The format-agnostic parity/widening/corrupt-skip and complex-type tests run from
+ // ArchiveReadSuiteBase (gated by the `supports*` hooks above); the tests below assert
+ // JSON-specific inference behavior -- NullType canonicalization and field-union merging -- that
+ // has no format-agnostic analogue. They use the shared `inferredSchema` helper from the base.
+
+ test("JSON: inference merges archive entries with loose files in the same directory") {
+ withTempDir { dir =>
+ val inArchive = jsonBytes("{\"id\":1,\"name\":\"Alice\"}\n{\"id\":2,\"name\":\"Bob\"}\n")
+ // The loose file drops `name` (which the archive entry has) and adds `age` (which it lacks);
+ // JSON's single inference pass unions both by field name, exactly as a directory read does.
+ val loose = jsonBytes("{\"id\":3,\"age\":30}\n")
+ writeArchive(new File(dir, s"data.${archiveExtensions.head}"), Seq(entryName(0) -> inArchive))
+ Files.write(new File(dir, s"loose.$fileExtension").toPath, loose)
+ val schema = inferredSchema(Seq(dir.getCanonicalPath))
+ assert(schema.fieldNames.toSet == Set("id", "name", "age"),
+ s"expected the union of the entry and loose fields, got $schema")
+ withTempDir { looseDir =>
+ Files.write(new File(looseDir, entryName(0)).toPath, inArchive)
+ Files.write(new File(looseDir, s"loose.$fileExtension").toPath, loose)
+ val expected = inferredSchema(Seq(looseDir.getCanonicalPath))
+ assert(schema == expected,
+ s"mixed archive+loose inference diverged from directory; got $schema, want $expected")
+ }
+ }
+ }
+
+ test("JSON: a column null across all archive entries infers as string") {
+ // Field `v` is null in every record across both entries, so each per-record type is NullType.
+ // The single inference pass canonicalizes the surviving NullType to StringType at the end (a
+ // valid schema, no NullType), matching a directory read of the same files.
+ val entries = Seq(
+ entryName(0) -> jsonBytes("{\"k\":1,\"v\":null}\n{\"k\":2,\"v\":null}\n"),
+ entryName(1) -> jsonBytes("{\"k\":3,\"v\":null}\n"))
+ withArchiveFile() { archive =>
+ writeArchive(archive, entries)
+ val archiveSchema = inferredSchema(Seq(archive.getCanonicalPath))
+ assert(archiveSchema.forall(_.dataType != NullType),
+ s"expected no NullType columns after canonicalization, got $archiveSchema")
+ assert(archiveSchema.find(_.name == "v").exists(_.dataType == StringType),
+ s"expected the all-null column to canonicalize to string, got $archiveSchema")
+ withTempDir { dir =>
+ entries.foreach { case (n, b) => Files.write(new File(dir, n).toPath, b) }
+ val dirSchema = inferredSchema(Seq(dir.getCanonicalPath))
+ assert(archiveSchema == dirSchema,
+ s"all-null column inference diverged from a directory; " +
+ s"archive=$archiveSchema dir=$dirSchema")
+ }
+ }
+ }
+
+ test("JSON: multiline inference merges archive entries with loose files") {
+ // multiLine + mixed inputs: inference reads every archive entry and loose file as one whole
+ // document and infers a single schema, matching a directory read of the same files.
+ val opts = Map("multiLine" -> "true")
+ withTempDir { dir =>
+ writeArchive(new File(dir, s"data.${archiveExtensions.head}"),
+ Seq(entryName(0) -> jsonBytes("{\n \"id\": 1,\n \"name\": \"Alice\"\n}")))
+ Files.write(new File(dir, s"loose.$fileExtension").toPath,
+ jsonBytes("{\n \"id\": 2,\n \"age\": 30\n}"))
+ val schema = inferredSchema(Seq(dir.getCanonicalPath), opts)
+ withTempDir { looseDir =>
+ Files.write(new File(looseDir, entryName(0)).toPath,
+ jsonBytes("{\n \"id\": 1,\n \"name\": \"Alice\"\n}"))
+ Files.write(new File(looseDir, s"loose.$fileExtension").toPath,
+ jsonBytes("{\n \"id\": 2,\n \"age\": 30\n}"))
+ val dirSchema = inferredSchema(Seq(looseDir.getCanonicalPath), opts)
+ assert(schema == dirSchema,
+ s"multiline mixed inference diverged from a directory read; got $schema want $dirSchema")
+ }
+ }
+ }
+
+ test("JSON: a field typed in the archive but null in a loose file is not collapsed to string") {
+ // One inference pass over all inputs keeps the loose file's null `v` as NullType until the end,
+ // so it widens with the archive's int. Inferring the loose file alone and merging two finished
+ // schemas would have canonicalized `v` to string first and yielded string here -- diverging
+ // from a directory read of the same files.
+ val archived = jsonBytes("{\"k\":1,\"v\":10}\n{\"k\":2,\"v\":20}\n")
+ val loose = jsonBytes("{\"k\":3,\"v\":null}\n")
+ withTempDir { dir =>
+ writeArchive(new File(dir, s"data.${archiveExtensions.head}"), Seq(entryName(0) -> archived))
+ Files.write(new File(dir, s"loose.$fileExtension").toPath, loose)
+ val schema = inferredSchema(Seq(dir.getCanonicalPath))
+ withTempDir { looseDir =>
+ Files.write(new File(looseDir, entryName(0)).toPath, archived)
+ Files.write(new File(looseDir, s"loose.$fileExtension").toPath, loose)
+ assert(schema == inferredSchema(Seq(looseDir.getCanonicalPath)),
+ s"null-in-loose field diverged from a directory read; got $schema")
+ }
+ assert(schema.find(_.name == "v").exists(_.dataType != StringType),
+ s"field null in the loose file should widen with the archive int, not collapse: $schema")
+ }
+ }
+
+ test("JSON: archive inference honors the encoding option like a directory") {
+ // A non-UTF-8 (UTF-16) archive: inference must decode each entry with `encoding`, exactly as
+ // the scan and a directory read do. Reading the bytes as UTF-8 would mis-parse them, diverging
+ // from a directory read of the same files. multiLine makes the whole document one record, so
+ // line-separator handling does not enter the test.
+ val opts = Map("encoding" -> "UTF-16", "multiLine" -> "true")
+ val bytes = "{\n \"id\": 1,\n \"name\": \"Alice\"\n}".getBytes(StandardCharsets.UTF_16)
+ withTempDir { dir =>
+ writeArchive(new File(dir, s"data.${archiveExtensions.head}"), Seq(entryName(0) -> bytes))
+ val schema = inferredSchema(Seq(dir.getCanonicalPath), opts)
+ withTempDir { looseDir =>
+ Files.write(new File(looseDir, entryName(0)).toPath, bytes)
+ val dirSchema = inferredSchema(Seq(looseDir.getCanonicalPath), opts)
+ assert(schema == dirSchema,
+ s"encoding inference diverged from a directory read; archive=$schema dir=$dirSchema")
+ }
+ assert(schema.fieldNames.toSet == Set("id", "name"),
+ s"expected id/name decoded from UTF-16, got $schema")
+ }
+ }
+
+ test("JSON: archive inference auto-detects a non-UTF-8 charset with no encoding option") {
+ // With no `encoding` set, inference parses each record from its raw bytes, so Jackson
+ // auto-detects the charset (here UTF-16, carrying a BOM) exactly as the scan and a directory
+ // read do. Forcing the bytes through UTF-8 would mis-parse them into a corrupt-record-only
+ // schema, diverging from a directory read of the same files.
+ val opts = Map("multiLine" -> "true")
+ val bytes = "{\n \"id\": 1,\n \"name\": \"Alice\"\n}".getBytes(StandardCharsets.UTF_16)
+ withTempDir { dir =>
+ writeArchive(new File(dir, s"data.${archiveExtensions.head}"), Seq(entryName(0) -> bytes))
+ val schema = inferredSchema(Seq(dir.getCanonicalPath), opts)
+ withTempDir { looseDir =>
+ Files.write(new File(looseDir, entryName(0)).toPath, bytes)
+ val dirSchema = inferredSchema(Seq(looseDir.getCanonicalPath), opts)
+ assert(schema == dirSchema,
+ s"auto-detected inference diverged from a directory read; archive=$schema dir=$dirSchema")
+ }
+ assert(schema.fieldNames.toSet == Set("id", "name"),
+ s"expected id/name auto-detected from UTF-16, got $schema")
+ }
+ }
+
+ // ----- JSON-specific read tests --------------------------------------------
+
+ test("JSON: entries with differing fields union like a directory") {
+ assertArchiveMatchesDir(
+ Seq(
+ "a.json" -> jsonBytes("{\"id\":1,\"name\":\"Alice\"}\n{\"id\":2,\"name\":\"Bob\"}\n"),
+ // No "name" field: the schema's "name" column must read back as null for this entry.
Review Comment:
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.
```suggestion
val bytes = "{\n \"id\": 1,\n \"name\": \"Jos\u00e9\"\n}".getBytes("windows-1252")
```
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57419][SQL] Read and infer JSON schema from tar archives [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on PR #56480:
URL: https://github.com/apache/spark/pull/56480#issuecomment-4715328182
thanks, merging to master/4.x!
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57419][SQL] Read and infer JSON schema from tar archives [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #56480:
URL: https://github.com/apache/spark/pull/56480#discussion_r3406577526
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/json/JsonDataSource.scala:
##########
@@ -75,6 +123,72 @@ abstract class JsonDataSource extends Serializable {
sparkSession: SparkSession,
inputPaths: Seq[FileStatus],
parsedOptions: JSONOptions): StructType
+
+ /**
+ * Infers a JSON schema when at least one input is a tar archive. Every archive entry (streamed
+ * via `ArchiveReader`, never unpacked to disk) and every loose file is read as JSON records --
+ * each line is a record, or the whole input is one document in multi-line mode -- and all of them
+ * feed a single [[JsonInferSchema]] pass, exactly as a directory of the same files would infer.
+ * Because [[JsonInferSchema]] already merges every record's type by field name across all inputs,
+ * one pass is itself the union: a field empty in one input but typed in another widens to the
+ * real type, and a `NullType` field survives to the single final canonicalization rather than
+ * being collapsed per-input. A corrupt/missing input is skipped as a unit (a whole archive or a
+ * whole file) when `ignoreCorruptFiles`/`ignoreMissingFiles` are set.
+ */
+ private def inferWithArchives(
+ sparkSession: SparkSession,
+ inputPaths: Seq[FileStatus],
+ parsedOptions: JSONOptions): StructType = {
+ val streams = JsonDataSource.createBaseRdd(sparkSession, inputPaths, parsedOptions)
+ val multiLine = parsedOptions.multiLine
+ val lineSeparator = parsedOptions.lineSeparatorInRead
+ val encoding = parsedOptions.encoding
+ val ignoreCorruptFiles = parsedOptions.ignoreCorruptFiles
+ val ignoreMissingFiles = parsedOptions.ignoreMissingFiles
+ // Each input is streamed lazily, carrying each record as its raw bytes in a fresh array (the
+ // line reader reuses one buffer, so a line is copied off it). The bytes are parsed downstream
+ // exactly as the scan parses a file: a byte-array parser auto-detects the charset when no
+ // `encoding` is set (so UTF-16/UTF-32 is read correctly, not just UTF-8), and a stream decoder
+ // applies the explicit `encoding` when one is given. An archive is read entry by entry; a loose
+ // file is read directly -- both yield the same record units, so all inputs feed one pass.
+ val records: RDD[Array[Byte]] = streams.flatMap { stream =>
+ val path = new Path(stream.getPath())
+ def recordsOf(in: InputStream): Iterator[Array[Byte]] =
+ if (multiLine) {
+ Iterator.single(in.readAllBytes())
+ } else {
+ ArchiveReader.lineIterator(in, lineSeparator).map(_.copyBytes())
+ }
+ try {
+ if (ArchiveReader.isArchivePath(path)) {
+ ArchiveReader(path).readEntries(stream.getConfiguration)((_, in) => recordsOf(in))
+ } else {
+ recordsOf(CodecStreams.createInputStreamWithCloseResource(stream.getConfiguration, path))
+ }
+ } catch {
+ case e: FileNotFoundException if ignoreMissingFiles =>
+ logWarning(log"Skipped missing input: ${MDC(PATH, stream.getPath())}", e)
+ Iterator.empty
+ case e: FileNotFoundException => throw e
+ case e @ (_: RuntimeException | _: IOException) if ignoreCorruptFiles =>
+ logWarning(log"Skipped the corrupted input: ${MDC(PATH, stream.getPath())}", e)
+ Iterator.empty
+ case NonFatal(e) =>
+ throw QueryExecutionErrors.cannotReadFilesError(
+ e, SparkPath.fromPathString(stream.getPath()).urlEncoded)
+ }
+ }
+ // Honor `samplingRatio` like the loose-file infer paths, so an archive samples its records like
+ // a directory read rather than always reading every one.
+ val sampled = JsonUtils.sample(records, parsedOptions)
+ val recordParser = encoding
+ .map(enc => CreateJacksonParser.bytes(enc, _: JsonFactory, _: Array[Byte]))
Review Comment:
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.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/json/JsonUtils.scala:
##########
@@ -42,9 +41,11 @@ object JsonUtils {
}
/**
- * Sample JSON RDD as configured by `samplingRatio`.
+ * Sample a JSON record RDD as configured by `samplingRatio`. Generic over the record type so the
+ * multiLine path (`RDD[PortableDataStream]`) and the archive inference path (`RDD[UTF8String]`)
Review Comment:
Went stale with the switch to raw bytes:
```suggestion
* multiLine path (`RDD[PortableDataStream]`) and the archive inference path (`RDD[Array[Byte]]`)
```
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57419][SQL] Read and infer JSON schema from tar archives [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #56480:
URL: https://github.com/apache/spark/pull/56480#discussion_r3406719632
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/JSONArchiveReadBase.scala:
##########
@@ -0,0 +1,265 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution.datasources
+
+import java.io.File
+import java.nio.charset.StandardCharsets
+import java.nio.file.Files
+
+import org.apache.spark.sql.{AnalysisException, DataFrame}
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types.{NullType, StringType}
+import org.apache.spark.util.Utils
+
+/**
+ * Binds [[ArchiveReadSuiteBase]]'s file-format hooks to JSON. JSON opts into the shared
+ * schema-inference and complex-type tests (see `supportsSchemaInference`/`supportsComplexTypes`),
+ * and adds the JSON-specific tests with no format-agnostic analogue: NullType canonicalization,
+ * field-union/null-in-loose merging, and multi-line documents. Reusable across archive formats: a
+ * `JSON<Archive>Read` suite mixes this in alongside the archive-format trait.
Review Comment:
`JSON<Archive>Read` doesn't expand to the real suite name — the tar suite is `JSONTarArchiveReadSuite`, and `TarArchiveReadBase` documents the `<format>TarArchiveReadSuite` pattern.
```suggestion
* `JSON<Container>ArchiveReadSuite` mixes this in alongside the archive-format trait.
```
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57419][SQL] Read and infer JSON schema from tar archives [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #56480:
URL: https://github.com/apache/spark/pull/56480#discussion_r3415881663
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/json/JsonDataSource.scala:
##########
@@ -75,6 +124,80 @@ abstract class JsonDataSource extends Serializable {
sparkSession: SparkSession,
inputPaths: Seq[FileStatus],
parsedOptions: JSONOptions): StructType
+
+ /**
+ * Infers a JSON schema when at least one input is a tar archive. Every archive entry (streamed
+ * via `ArchiveReader`, never unpacked to disk) and every loose file is read as JSON records --
+ * each line is a record, or the whole input is one document in multi-line mode -- and all of them
+ * feed a single [[JsonInferSchema]] pass, exactly as a directory of the same files would infer.
+ * Because [[JsonInferSchema]] already merges every record's type by field name across all inputs,
+ * one pass is itself the union: a field empty in one input but typed in another widens to the
+ * real type, and a `NullType` field survives to the single final canonicalization rather than
+ * being collapsed per-input. A corrupt/missing input is skipped as a unit (a whole archive or a
+ * whole file) when `ignoreCorruptFiles`/`ignoreMissingFiles` are set.
+ */
+ private def inferWithArchives(
+ sparkSession: SparkSession,
+ inputPaths: Seq[FileStatus],
+ parsedOptions: JSONOptions): StructType = {
+ val baseRdd = JsonDataSource.createBaseRdd(sparkSession, inputPaths, parsedOptions)
+ val multiLine = parsedOptions.multiLine
+ val lineSeparator = parsedOptions.lineSeparatorInRead
+ val encoding = parsedOptions.encoding
+ val ignoreCorruptFiles = parsedOptions.ignoreCorruptFiles
+ val ignoreMissingFiles = parsedOptions.ignoreMissingFiles
+
+ // Applies `perEntry` to each input -- once per archive entry, once for a loose file -- skipping
+ // a whole input on corrupt/missing input when the ignore flags are set. The entry/file stream
Review Comment:
Repeated "input" reads awkwardly.
```suggestion
// a whole input when it is corrupt/missing and the ignore flags are set. The entry/file stream
```
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/json/JsonDataSource.scala:
##########
@@ -75,6 +124,80 @@ abstract class JsonDataSource extends Serializable {
sparkSession: SparkSession,
inputPaths: Seq[FileStatus],
parsedOptions: JSONOptions): StructType
+
+ /**
+ * Infers a JSON schema when at least one input is a tar archive. Every archive entry (streamed
+ * via `ArchiveReader`, never unpacked to disk) and every loose file is read as JSON records --
+ * each line is a record, or the whole input is one document in multi-line mode -- and all of them
+ * feed a single [[JsonInferSchema]] pass, exactly as a directory of the same files would infer.
+ * Because [[JsonInferSchema]] already merges every record's type by field name across all inputs,
+ * one pass is itself the union: a field empty in one input but typed in another widens to the
+ * real type, and a `NullType` field survives to the single final canonicalization rather than
+ * being collapsed per-input. A corrupt/missing input is skipped as a unit (a whole archive or a
+ * whole file) when `ignoreCorruptFiles`/`ignoreMissingFiles` are set.
+ */
+ private def inferWithArchives(
+ sparkSession: SparkSession,
+ inputPaths: Seq[FileStatus],
+ parsedOptions: JSONOptions): StructType = {
+ val baseRdd = JsonDataSource.createBaseRdd(sparkSession, inputPaths, parsedOptions)
+ val multiLine = parsedOptions.multiLine
+ val lineSeparator = parsedOptions.lineSeparatorInRead
+ val encoding = parsedOptions.encoding
+ val ignoreCorruptFiles = parsedOptions.ignoreCorruptFiles
+ val ignoreMissingFiles = parsedOptions.ignoreMissingFiles
+
+ // Applies `perEntry` to each input -- once per archive entry, once for a loose file -- skipping
+ // a whole input on corrupt/missing input when the ignore flags are set. The entry/file stream
+ // is consumed lazily by `perEntry`, never buffered whole; mirrors CSV's `inferWithArchives`.
+ def perInput[T: ClassTag](perEntry: InputStream => Iterator[T]): RDD[T] = baseRdd.flatMap {
+ stream =>
+ val path = new Path(stream.getPath())
+ try {
+ if (ArchiveReader.isArchivePath(path)) {
+ ArchiveReader(path).readEntries(stream.getConfiguration) { (_, in) => perEntry(in) }
+ } else {
+ perEntry(
+ CodecStreams.createInputStreamWithCloseResource(stream.getConfiguration, path))
+ }
+ } catch {
+ case e: FileNotFoundException if ignoreMissingFiles =>
+ logWarning(log"Skipped missing input: ${MDC(PATH, stream.getPath())}", e)
+ Iterator.empty
+ case e: FileNotFoundException => throw e
+ case e @ (_: RuntimeException | _: IOException) if ignoreCorruptFiles =>
+ logWarning(log"Skipped the corrupted input: ${MDC(PATH, stream.getPath())}", e)
+ Iterator.empty
+ case NonFatal(e) =>
+ throw QueryExecutionErrors.cannotReadFilesError(
+ e, SparkPath.fromPathString(stream.getPath()).urlEncoded)
+ }
+ }
+
+ SQLExecution.withSQLConfPropagated(sparkSession) {
+ val inferSchema = new JsonInferSchema(parsedOptions)
+ if (multiLine) {
+ // Each input/entry is one JSON document: hand its stream straight to the parser
+ // (`CreateJacksonParser.inputStream`, matching MultiLineJsonDataSource and its charset
+ // auto-detect) so the document is parsed incrementally rather than buffered.
+ val docs = perInput(in => Iterator.single(in))
Review Comment:
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.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57419][SQL] Read and infer JSON schema from tar archives [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan closed pull request #56480: [SPARK-57419][SQL] Read and infer JSON schema from tar archives
URL: https://github.com/apache/spark/pull/56480
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57419][SQL] Read and infer JSON schema from tar archives [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #56480:
URL: https://github.com/apache/spark/pull/56480#discussion_r3406471065
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/json/JsonDataSource.scala:
##########
@@ -75,6 +123,79 @@ abstract class JsonDataSource extends Serializable {
sparkSession: SparkSession,
inputPaths: Seq[FileStatus],
parsedOptions: JSONOptions): StructType
+
+ /**
+ * Infers a JSON schema when at least one input is a tar archive. Every archive entry (streamed
+ * via `ArchiveReader`, never unpacked to disk) and every loose file is read as JSON records --
+ * each line is a record, or the whole input is one document in multi-line mode -- and all of them
+ * feed a single [[JsonInferSchema]] pass, exactly as a directory of the same files would infer.
+ * Because [[JsonInferSchema]] already merges every record's type by field name across all inputs,
+ * one pass is itself the union: a field empty in one input but typed in another widens to the
+ * real type, and a `NullType` field survives to the single final canonicalization rather than
+ * being collapsed per-input. A corrupt/missing input is skipped as a unit (a whole archive or a
+ * whole file) when `ignoreCorruptFiles`/`ignoreMissingFiles` are set.
+ */
+ private def inferWithArchives(
+ sparkSession: SparkSession,
+ inputPaths: Seq[FileStatus],
+ parsedOptions: JSONOptions): StructType = {
+ val streams = JsonDataSource.createBaseRdd(sparkSession, inputPaths, parsedOptions)
+ val multiLine = parsedOptions.multiLine
+ val lineSeparator = parsedOptions.lineSeparatorInRead
+ val encoding = parsedOptions.encoding
+ val ignoreCorruptFiles = parsedOptions.ignoreCorruptFiles
+ val ignoreMissingFiles = parsedOptions.ignoreMissingFiles
+ // Each input is streamed lazily; records are copied into fresh `UTF8String`s (the line reader
+ // reuses one buffer) so they stay valid after the stream advances. An archive is read entry by
+ // entry; a loose file is read directly -- both yield the same record units, so all inputs feed
+ // one inference pass.
+ val records: RDD[UTF8String] = streams.flatMap { stream =>
+ val path = new Path(stream.getPath())
+ // Decode each record with the configured `encoding` (re-encoding to UTF-8) so archive
+ // inference matches the scan and a directory read. With no `encoding`, the raw bytes are
+ // kept and parsed as UTF-8 by `CreateJacksonParser.utf8String`.
+ def toRecord(bytes: Array[Byte], length: Int): UTF8String = encoding match {
+ case Some(enc) => UTF8String.fromString(new String(bytes, 0, length, enc))
+ case None => UTF8String.fromBytes(bytes, 0, length)
Review Comment:
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.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/json/JsonDataSource.scala:
##########
@@ -75,6 +123,79 @@ abstract class JsonDataSource extends Serializable {
sparkSession: SparkSession,
inputPaths: Seq[FileStatus],
parsedOptions: JSONOptions): StructType
+
+ /**
+ * Infers a JSON schema when at least one input is a tar archive. Every archive entry (streamed
+ * via `ArchiveReader`, never unpacked to disk) and every loose file is read as JSON records --
+ * each line is a record, or the whole input is one document in multi-line mode -- and all of them
+ * feed a single [[JsonInferSchema]] pass, exactly as a directory of the same files would infer.
+ * Because [[JsonInferSchema]] already merges every record's type by field name across all inputs,
+ * one pass is itself the union: a field empty in one input but typed in another widens to the
+ * real type, and a `NullType` field survives to the single final canonicalization rather than
+ * being collapsed per-input. A corrupt/missing input is skipped as a unit (a whole archive or a
+ * whole file) when `ignoreCorruptFiles`/`ignoreMissingFiles` are set.
+ */
+ private def inferWithArchives(
+ sparkSession: SparkSession,
+ inputPaths: Seq[FileStatus],
+ parsedOptions: JSONOptions): StructType = {
+ val streams = JsonDataSource.createBaseRdd(sparkSession, inputPaths, parsedOptions)
+ val multiLine = parsedOptions.multiLine
+ val lineSeparator = parsedOptions.lineSeparatorInRead
+ val encoding = parsedOptions.encoding
+ val ignoreCorruptFiles = parsedOptions.ignoreCorruptFiles
+ val ignoreMissingFiles = parsedOptions.ignoreMissingFiles
+ // Each input is streamed lazily; records are copied into fresh `UTF8String`s (the line reader
Review Comment:
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.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/JSONArchiveReadBase.scala:
##########
@@ -0,0 +1,226 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution.datasources
+
+import java.io.File
+import java.nio.charset.StandardCharsets
+import java.nio.file.Files
+
+import org.apache.spark.sql.{AnalysisException, DataFrame}
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types.{NullType, StringType}
+import org.apache.spark.util.Utils
+
+/**
+ * Binds [[ArchiveReadSuiteBase]]'s file-format hooks to JSON. JSON opts into the shared
+ * schema-inference and complex-type tests (see `supportsSchemaInference`/`supportsComplexTypes`),
+ * and adds the JSON-specific tests with no format-agnostic analogue: NullType canonicalization,
+ * field-union/null-in-loose merging, and multi-line documents. Reusable across archive formats: a
+ * `JSON<Archive>Read` suite mixes this in alongside the archive-format trait.
+ */
+trait JSONArchiveReadBase extends ArchiveReadSuiteBase {
+
+ override protected def format: String = "json"
+
+ override protected def fileExtension: String = "json"
+
+ // JSON records are self-describing, so no per-file read options are required.
+ override protected def readOptions: Map[String, String] = Map.empty
+
+ override protected def readSchema: String = "id INT, name STRING"
+
+ // JSON infers its schema from record content and represents nested structs, so it opts into the
+ // shared inference and complex-type tests. Inference needs no trigger option, so the inherited
+ // `inferenceOptions` keeps its empty default.
+ override protected def supportsSchemaInference: Boolean = true
+
+ override protected def supportsComplexTypes: Boolean = true
+
+ override protected def encodeFile(
+ df: DataFrame,
+ writeOptions: Map[String, String]): Array[Byte] = {
+ val dir = Utils.createTempDir(namePrefix = "archive-test-encode")
+ try {
+ df.coalesce(1).write.format("json")
+ .options(writeOptions)
+ .mode("overwrite").save(dir.getCanonicalPath)
+ val parts = dir.listFiles().filter { f =>
+ f.isFile && !f.getName.startsWith("_") && !f.getName.startsWith(".") &&
+ !f.getName.endsWith(".crc")
+ }
+ assert(parts.length == 1,
+ s"expected exactly one data file, got: ${parts.map(_.getName).toList}")
+ Files.readAllBytes(parts.head.toPath)
+ } finally Utils.deleteRecursively(dir)
+ }
+
+ /** Raw JSON bytes, for tests that need precise control over the record layout. */
+ protected def jsonBytes(s: String): Array[Byte] = s.getBytes(StandardCharsets.UTF_8)
+
+ // ----- JSON-specific schema inference --------------------------------------
+ // The format-agnostic parity/widening/corrupt-skip and complex-type tests run from
+ // ArchiveReadSuiteBase (gated by the `supports*` hooks above); the tests below assert
+ // JSON-specific inference behavior -- NullType canonicalization and field-union merging -- that
+ // has no format-agnostic analogue. They use the shared `inferredSchema` helper from the base.
+
+ test("JSON: inference merges archive entries with loose files in the same directory") {
+ withTempDir { dir =>
+ val inArchive = jsonBytes("{\"id\":1,\"name\":\"Alice\"}\n{\"id\":2,\"name\":\"Bob\"}\n")
+ // The loose file drops `name` (which the archive entry has) and adds `age` (which it lacks);
+ // JSON's single inference pass unions both by field name, exactly as a directory read does.
+ val loose = jsonBytes("{\"id\":3,\"age\":30}\n")
+ writeArchive(new File(dir, s"data.${archiveExtensions.head}"), Seq(entryName(0) -> inArchive))
+ Files.write(new File(dir, s"loose.$fileExtension").toPath, loose)
+ val schema = inferredSchema(Seq(dir.getCanonicalPath))
+ assert(schema.fieldNames.toSet == Set("id", "name", "age"),
+ s"expected the union of the entry and loose fields, got $schema")
+ withTempDir { looseDir =>
+ Files.write(new File(looseDir, entryName(0)).toPath, inArchive)
+ Files.write(new File(looseDir, s"loose.$fileExtension").toPath, loose)
+ val expected = inferredSchema(Seq(looseDir.getCanonicalPath))
+ assert(schema == expected,
+ s"mixed archive+loose inference diverged from directory; got $schema, want $expected")
+ }
+ }
+ }
+
+ test("JSON: a column null across all archive entries infers as string") {
+ // Field `v` is null in every record across both entries, so each per-record type is NullType.
+ // The single inference pass canonicalizes the surviving NullType to StringType at the end (a
+ // valid schema, no NullType), matching a directory read of the same files.
+ val entries = Seq(
+ entryName(0) -> jsonBytes("{\"k\":1,\"v\":null}\n{\"k\":2,\"v\":null}\n"),
+ entryName(1) -> jsonBytes("{\"k\":3,\"v\":null}\n"))
+ withArchiveFile() { archive =>
+ writeArchive(archive, entries)
+ val archiveSchema = inferredSchema(Seq(archive.getCanonicalPath))
+ assert(archiveSchema.forall(_.dataType != NullType),
+ s"expected no NullType columns after canonicalization, got $archiveSchema")
+ assert(archiveSchema.find(_.name == "v").exists(_.dataType == StringType),
+ s"expected the all-null column to canonicalize to string, got $archiveSchema")
+ withTempDir { dir =>
+ entries.foreach { case (n, b) => Files.write(new File(dir, n).toPath, b) }
+ val dirSchema = inferredSchema(Seq(dir.getCanonicalPath))
+ assert(archiveSchema == dirSchema,
+ s"all-null column inference diverged from a directory; " +
+ s"archive=$archiveSchema dir=$dirSchema")
+ }
+ }
+ }
+
+ test("JSON: multiline inference merges archive entries with loose files") {
+ // multiLine + mixed inputs: inference reads every archive entry and loose file as one whole
+ // document and infers a single schema, matching a directory read of the same files.
+ val opts = Map("multiLine" -> "true")
+ withTempDir { dir =>
+ writeArchive(new File(dir, s"data.${archiveExtensions.head}"),
+ Seq(entryName(0) -> jsonBytes("{\n \"id\": 1,\n \"name\": \"Alice\"\n}")))
+ Files.write(new File(dir, s"loose.$fileExtension").toPath,
+ jsonBytes("{\n \"id\": 2,\n \"age\": 30\n}"))
+ val schema = inferredSchema(Seq(dir.getCanonicalPath), opts)
+ withTempDir { looseDir =>
+ Files.write(new File(looseDir, entryName(0)).toPath,
+ jsonBytes("{\n \"id\": 1,\n \"name\": \"Alice\"\n}"))
+ Files.write(new File(looseDir, s"loose.$fileExtension").toPath,
+ jsonBytes("{\n \"id\": 2,\n \"age\": 30\n}"))
+ val dirSchema = inferredSchema(Seq(looseDir.getCanonicalPath), opts)
+ assert(schema == dirSchema,
+ s"multiline mixed inference diverged from a directory read; got $schema want $dirSchema")
+ }
+ }
+ }
+
+ test("JSON: a field typed in the archive but null in a loose file is not collapsed to string") {
+ // One inference pass over all inputs keeps the loose file's null `v` as NullType until the end,
+ // so it widens with the archive's int. Inferring the loose file alone and merging two finished
+ // schemas would have canonicalized `v` to string first and yielded string here -- diverging
+ // from a directory read of the same files.
+ val archived = jsonBytes("{\"k\":1,\"v\":10}\n{\"k\":2,\"v\":20}\n")
+ val loose = jsonBytes("{\"k\":3,\"v\":null}\n")
+ withTempDir { dir =>
+ writeArchive(new File(dir, s"data.${archiveExtensions.head}"), Seq(entryName(0) -> archived))
+ Files.write(new File(dir, s"loose.$fileExtension").toPath, loose)
+ val schema = inferredSchema(Seq(dir.getCanonicalPath))
+ withTempDir { looseDir =>
+ Files.write(new File(looseDir, entryName(0)).toPath, archived)
+ Files.write(new File(looseDir, s"loose.$fileExtension").toPath, loose)
+ assert(schema == inferredSchema(Seq(looseDir.getCanonicalPath)),
+ s"null-in-loose field diverged from a directory read; got $schema")
+ }
+ assert(schema.find(_.name == "v").exists(_.dataType != StringType),
+ s"field null in the loose file should widen with the archive int, not collapse: $schema")
+ }
+ }
+
+ test("JSON: archive inference honors the encoding option like a directory") {
+ // A non-UTF-8 (UTF-16) archive: inference must decode each entry with `encoding`, exactly as
+ // the scan and a directory read do. Reading the bytes as UTF-8 would mis-parse them, diverging
+ // from a directory read of the same files. multiLine makes the whole document one record, so
+ // line-separator handling does not enter the test.
+ val opts = Map("encoding" -> "UTF-16", "multiLine" -> "true")
+ val bytes = "{\n \"id\": 1,\n \"name\": \"Alice\"\n}".getBytes(StandardCharsets.UTF_16)
+ withTempDir { dir =>
+ writeArchive(new File(dir, s"data.${archiveExtensions.head}"), Seq(entryName(0) -> bytes))
+ val schema = inferredSchema(Seq(dir.getCanonicalPath), opts)
+ withTempDir { looseDir =>
+ Files.write(new File(looseDir, entryName(0)).toPath, bytes)
+ val dirSchema = inferredSchema(Seq(looseDir.getCanonicalPath), opts)
+ assert(schema == dirSchema,
+ s"encoding inference diverged from a directory read; archive=$schema dir=$dirSchema")
+ }
+ assert(schema.fieldNames.toSet == Set("id", "name"),
+ s"expected id/name decoded from UTF-16, got $schema")
+ }
+ }
+
+ // ----- JSON-specific read tests --------------------------------------------
+
+ test("JSON: entries with differing fields union like a directory") {
Review Comment:
Both `readStream` implementations wire new `FailureSafeParser`s, 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.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/ArchiveReadSuiteBase.scala:
##########
@@ -104,6 +106,47 @@ trait ArchiveReadSuiteBase extends QueryTest with SharedSparkSession {
schema: String = readSchema): DataFrame =
spark.read.format(format).options(readOptions ++ extraOptions).schema(schema).load(path)
+ // ----- schema-inference / complex-type capability hooks (default off) ------
+ //
+ // A format opts into the shared inference / complex-type tests below by overriding the
+ // corresponding `supports*` hook to true; a format that supports neither (e.g. text) leaves both
+ // false and runs only the read tests.
+
+ /**
+ * Whether this format infers its read schema from the textual content of the files (CSV, JSON,
+ * XML), as opposed to carrying an embedded schema (Avro, Parquet) or none (text). Gates the
+ * shared schema-inference tests. A format that needs an option to trigger inference (e.g. CSV's
+ * `inferSchema`) also overrides [[inferenceOptions]].
+ */
+ protected def supportsSchemaInference: Boolean = false
+
+ /** Extra options that trigger/control inference for [[supportsSchemaInference]] formats. */
+ protected def inferenceOptions: Map[String, String] = Map.empty
+
+ /**
+ * Whether this format can represent nested/complex types (struct/array/map). Gates the shared
+ * complex-type round-trip test; CSV and text leave it false, JSON/Avro/Parquet/XML override true.
Review Comment:
Only JSON overrides this — no Avro/Parquet/XML archive traits exist yet.
```suggestion
* complex-type round-trip test; CSV and text leave it false, JSON overrides it to true.
```
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org