This is an automated email from the ASF dual-hosted git repository.
cloud-fan pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 6507faafa40d [SPARK-57419][SQL] Read and infer JSON schema from tar
archives
6507faafa40d is described below
commit 6507faafa40d76a428284effdc2dc923482d5d2a
Author: akshatshenoi-db <[email protected]>
AuthorDate: Mon Jun 15 22:36:46 2026 -0700
[SPARK-57419][SQL] Read and infer JSON schema from tar archives
### 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 dire [...]
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
Closes #56480 from akshatshenoi-db/archive-json.
Authored-by: akshatshenoi-db <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>
---
.../sql/catalyst/json/CreateJacksonParser.scala | 10 +
.../datasources/json/JsonDataSource.scala | 225 ++++++++++++++++++---
.../datasources/json/JsonFileFormat.scala | 22 +-
.../sql/execution/datasources/json/JsonUtils.scala | 7 +-
.../execution/datasources/v2/json/JsonTable.scala | 4 +-
.../datasources/ArchiveReadSuiteBase.scala | 171 ++++++++++++++++
.../execution/datasources/CSVArchiveReadBase.scala | 92 +--------
.../datasources/JSONArchiveReadBase.scala | 223 ++++++++++++++++++++
.../datasources/JSONTarArchiveReadSuite.scala | 28 +++
9 files changed, 663 insertions(+), 119 deletions(-)
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/CreateJacksonParser.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/CreateJacksonParser.scala
index 7190f6fff79f..939500d8028d 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/CreateJacksonParser.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/CreateJacksonParser.scala
@@ -90,4 +90,14 @@ object CreateJacksonParser extends Serializable {
jsonFactory.createParser(sd)
}
+
+ def bytes(jsonFactory: JsonFactory, record: Array[Byte]): JsonParser = {
+ jsonFactory.createParser(record, 0, record.length)
+ }
+
+ def bytes(enc: String, jsonFactory: JsonFactory, record: Array[Byte]):
JsonParser = {
+ val sd = getStreamDecoder(enc, record, record.length)
+
+ jsonFactory.createParser(sd)
+ }
}
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/json/JsonDataSource.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/json/JsonDataSource.scala
index aaa5af478dbb..5f2891985564 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/json/JsonDataSource.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/json/JsonDataSource.scala
@@ -17,7 +17,10 @@
package org.apache.spark.sql.execution.datasources.json
-import java.io.InputStream
+import java.io.{ByteArrayInputStream, FileNotFoundException, InputStream,
IOException}
+
+import scala.reflect.ClassTag
+import scala.util.control.NonFatal
import com.fasterxml.jackson.core.{JsonFactory, JsonParser}
import org.apache.hadoop.conf.Configuration
@@ -28,12 +31,16 @@ import org.apache.hadoop.mapreduce.lib.input.FileInputFormat
import org.apache.spark.TaskContext
import org.apache.spark.input.{PortableDataStream, StreamInputFormat}
+import org.apache.spark.internal.Logging
+import org.apache.spark.internal.LogKeys.PATH
+import org.apache.spark.paths.SparkPath
import org.apache.spark.rdd.{BinaryFileRDD, RDD}
import org.apache.spark.sql.{Dataset, Encoders, SparkSession}
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.json.{CreateJacksonParser, JacksonParser,
JsonInferSchema, JSONOptions}
import org.apache.spark.sql.catalyst.util.FailureSafeParser
import org.apache.spark.sql.classic.ClassicConversions.castToImpl
+import org.apache.spark.sql.errors.QueryExecutionErrors
import org.apache.spark.sql.execution.SQLExecution
import org.apache.spark.sql.execution.datasources._
import org.apache.spark.sql.execution.datasources.text.TextFileFormat
@@ -44,7 +51,7 @@ import org.apache.spark.util.Utils
/**
* Common functions for parsing JSON files
*/
-abstract class JsonDataSource extends Serializable {
+abstract class JsonDataSource extends Serializable with Logging {
def isSplitable: Boolean
/**
@@ -56,14 +63,56 @@ abstract class JsonDataSource extends Serializable {
parser: JacksonParser,
schema: StructType): Iterator[InternalRow]
+ /**
+ * Parse a single already-open [[InputStream]] -- one decompressed archive
entry -- into 0 or more
+ * [[InternalRow]] instances, the same way this mode reads a standalone
file: line by line for
+ * [[TextInputJsonDataSource]], as one whole document for
[[MultiLineJsonDataSource]]. Used only
+ * by [[readArchive]]; the stream is not closed here.
+ */
+ protected def readStream(
+ in: InputStream,
+ parser: JacksonParser,
+ schema: StructType): Iterator[InternalRow]
+
+ /**
+ * Streams a tar archive (`.tar`/`.tar.gz`/`.tgz`) entry by entry through
the JSON parser without
+ * unpacking it to disk. The whole archive is a single split (see
`JsonFileFormat.isSplitable`);
+ * each entry's bytes are parsed exactly like a standalone JSON file via
[[readStream]], so this
+ * is mode-agnostic (line-delimited and multi-line both flow through
`readStream`) and a single
+ * `parser` serves every entry -- unlike CSV there is no per-entry header to
rebuild. Kept apart
+ * from [[readFile]] because only the V1 `JsonFileFormat` read path supports
archives; the V2 data
+ * source calls [[readFile]] directly and is intentionally left untouched.
+ */
+ def readArchive(
+ conf: Configuration,
+ file: PartitionedFile,
+ parser: JacksonParser,
+ schema: StructType): Iterator[InternalRow] =
+ ArchiveReader(file.toPath).readEntries(conf) { (_, in) =>
+ readStream(in, parser, schema)
+ }
+
final def inferSchema(
sparkSession: SparkSession,
inputPaths: Seq[FileStatus],
- parsedOptions: JSONOptions): Option[StructType] = {
+ parsedOptions: JSONOptions,
+ supportsArchiveScan: Boolean): Option[StructType] = {
parsedOptions.singleVariantColumn match {
case Some(columnName) => Some(StructType(Array(StructField(columnName,
VariantType))))
case None =>
- if (inputPaths.nonEmpty) {
+ val hasArchive = parsedOptions.archiveFormatEnabled &&
+ inputPaths.exists(f => ArchiveReader.isArchivePath(f.getPath))
+ if (hasArchive && supportsArchiveScan) {
+ // Archives (and any loose files alongside them) are inferred in a
single JsonInferSchema
+ // pass over all inputs -- archive entries are streamed, never
unpacked -- so the result
+ // matches what the scan returns for the same files.
+ Some(inferWithArchives(sparkSession, inputPaths, parsedOptions))
+ } else if (hasArchive) {
+ // The caller's scan path cannot read archives (e.g. the DSv2
reader), so refuse to infer
+ // a schema when any input is an archive: returning None raises
UNABLE_TO_INFER_SCHEMA,
+ // which fails loudly instead of letting the scan parse raw archive
bytes as JSON.
+ None
+ } else if (inputPaths.nonEmpty) {
Some(infer(sparkSession, inputPaths, parsedOptions))
} else {
None
@@ -75,6 +124,89 @@ 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 when it is corrupt/missing and the ignore flags are set.
The entry/file stream
+ // is consumed lazily by `perEntry`, never buffered whole; mirrors CSV's
`inferWithArchives`.
+ //
+ // An archive entry's stream is a `CloseShieldInputStream` view over the
one shared
+ // `TarArchiveInputStream` cursor, valid only until `readEntries` advances
to the next entry
+ // (`getNextEntry` skips the prior entry's unread bytes). A consumer that
emits the raw stream
+ // (the multiLine path below) must therefore fully consume each element
before the iterator
+ // advances; `JsonInferSchema.infer` does, parsing each record before
pulling the next.
+ // Buffering, look-ahead, or parallelizing the per-partition consumption
would read from an
+ // advanced cursor and infer a wrong schema. The line-delimited path
sidesteps this by
+ // materializing each record (`copyBytes()`).
+ 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))
+ val docParser: (JsonFactory, InputStream) => JsonParser = encoding
+ .map(enc => CreateJacksonParser.inputStream(enc, _: JsonFactory, _:
InputStream))
+ .getOrElse(CreateJacksonParser.inputStream(_: JsonFactory, _:
InputStream))
+ inferSchema.infer[InputStream](
+ JsonUtils.sample(docs, parsedOptions), docParser, isReadFile = true)
+ } else {
+ // Line-delimited: each line is a record, copied off the reused line
buffer and parsed from
+ // its bytes (`CreateJacksonParser.bytes`, matching
TextInputJsonDataSource).
+ val lines = perInput(in => ArchiveReader.lineIterator(in,
lineSeparator).map(_.copyBytes()))
+ val lineParser: (JsonFactory, Array[Byte]) => JsonParser = encoding
+ .map(enc => CreateJacksonParser.bytes(enc, _: JsonFactory, _:
Array[Byte]))
+ .getOrElse(CreateJacksonParser.bytes(_: JsonFactory, _: Array[Byte]))
+ inferSchema.infer[Array[Byte]](
+ JsonUtils.sample(lines, parsedOptions), lineParser, isReadFile =
false)
+ }
+ }
+ }
}
object JsonDataSource {
@@ -85,6 +217,31 @@ object JsonDataSource {
TextInputJsonDataSource
}
}
+
+ /**
+ * One `PortableDataStream` per input file (the whole file, never split),
shared by the multiLine
+ * schema-inference path and the archive inference path.
+ */
+ private[json] def createBaseRdd(
+ sparkSession: SparkSession,
+ inputPaths: Seq[FileStatus],
+ parsedOptions: JSONOptions): RDD[PortableDataStream] = {
+ val paths = inputPaths.map(_.getPath)
+ val job =
Job.getInstance(sparkSession.sessionState.newHadoopConfWithOptions(
+ parsedOptions.parameters))
+ val conf = job.getConfiguration
+ val name = paths.mkString(",")
+ FileInputFormat.setInputPaths(job, paths: _*)
+ new BinaryFileRDD(
+ sparkSession.sparkContext,
+ classOf[StreamInputFormat],
+ classOf[String],
+ classOf[PortableDataStream],
+ conf,
+ sparkSession.sparkContext.defaultMinPartitions)
+ .setName(s"JsonFile: $name")
+ .values
+ }
}
object TextInputJsonDataSource extends JsonDataSource {
@@ -149,6 +306,22 @@ object TextInputJsonDataSource extends JsonDataSource {
linesReader.flatMap(safeParser.parse)
}
+ override protected def readStream(
+ in: InputStream,
+ parser: JacksonParser,
+ schema: StructType): Iterator[InternalRow] = {
+ val textParser = parser.options.encoding
+ .map(enc => CreateJacksonParser.text(enc, _: JsonFactory, _: Text))
+ .getOrElse(CreateJacksonParser.text(_: JsonFactory, _: Text))
+
+ val safeParser = new FailureSafeParser[Text](
+ input => parser.parse(input, textParser, textToUTF8String),
+ parser.options.parseMode,
+ schema,
+ parser.options.columnNameOfCorruptRecord)
+ ArchiveReader.lineIterator(in,
parser.options.lineSeparatorInRead).flatMap(safeParser.parse)
+ }
+
private def textToUTF8String(value: Text): UTF8String = {
UTF8String.fromBytes(value.getBytes, 0, value.getLength)
}
@@ -163,7 +336,8 @@ object MultiLineJsonDataSource extends JsonDataSource {
sparkSession: SparkSession,
inputPaths: Seq[FileStatus],
parsedOptions: JSONOptions): StructType = {
- val json: RDD[PortableDataStream] = createBaseRdd(sparkSession,
inputPaths, parsedOptions)
+ val json: RDD[PortableDataStream] =
+ JsonDataSource.createBaseRdd(sparkSession, inputPaths, parsedOptions)
val sampled: RDD[PortableDataStream] = JsonUtils.sample(json,
parsedOptions)
val parser = parsedOptions.encoding
.map(enc => createParser(enc, _: JsonFactory, _: PortableDataStream))
@@ -175,27 +349,6 @@ object MultiLineJsonDataSource extends JsonDataSource {
}
}
- private def createBaseRdd(
- sparkSession: SparkSession,
- inputPaths: Seq[FileStatus],
- parsedOptions: JSONOptions): RDD[PortableDataStream] = {
- val paths = inputPaths.map(_.getPath)
- val job =
Job.getInstance(sparkSession.sessionState.newHadoopConfWithOptions(
- parsedOptions.parameters))
- val conf = job.getConfiguration
- val name = paths.mkString(",")
- FileInputFormat.setInputPaths(job, paths: _*)
- new BinaryFileRDD(
- sparkSession.sparkContext,
- classOf[StreamInputFormat],
- classOf[String],
- classOf[PortableDataStream],
- conf,
- sparkSession.sparkContext.defaultMinPartitions)
- .setName(s"JsonFile: $name")
- .values
- }
-
private def dataToInputStream(dataStream: PortableDataStream): InputStream =
{
val path = new Path(dataStream.getPath())
CodecStreams.createInputStreamWithCloseResource(dataStream.getConfiguration,
path)
@@ -237,4 +390,24 @@ object MultiLineJsonDataSource extends JsonDataSource {
safeParser.parse(
CodecStreams.createInputStreamWithCloseResource(conf, file.toPath))
}
+
+ override protected def readStream(
+ in: InputStream,
+ parser: JacksonParser,
+ schema: StructType): Iterator[InternalRow] = {
+ // The entry is a single JSON document. Buffer its bytes so the
corrupt-record column can echo
+ // the whole document on a parse failure, mirroring `readFile`'s
`partitionedFileString`.
+ val bytes = in.readAllBytes()
+ val streamParser = parser.options.encoding
+ .map(enc => CreateJacksonParser.inputStream(enc, _: JsonFactory, _:
InputStream))
+ .getOrElse(CreateJacksonParser.inputStream(_: JsonFactory, _:
InputStream))
+
+ val safeParser = new FailureSafeParser[InputStream](
+ input => parser.parse[InputStream](input, streamParser, _ =>
UTF8String.fromBytes(bytes)),
+ parser.options.parseMode,
+ schema,
+ parser.options.columnNameOfCorruptRecord)
+
+ safeParser.parse(new ByteArrayInputStream(bytes))
+ }
}
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/json/JsonFileFormat.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/json/JsonFileFormat.scala
index 331230c4446b..2ddce6286729 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/json/JsonFileFormat.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/json/JsonFileFormat.scala
@@ -40,6 +40,10 @@ case class JsonFileFormat() extends TextBasedFileFormat with
DataSourceRegister
options: Map[String, String],
path: Path): Boolean = {
val parsedOptions = getJsonOptions(sparkSession, options)
+ if (parsedOptions.archiveFormatEnabled &&
ArchiveReader.isArchivePath(path)) {
+ // A tar archive is read as one sequential stream (entry by entry), so
it is never split.
+ return false
+ }
JsonDataSource(parsedOptions).isSplitable &&
super.isSplitable(sparkSession, options, path)
}
@@ -48,7 +52,8 @@ case class JsonFileFormat() extends TextBasedFileFormat with
DataSourceRegister
options: Map[String, String],
files: Seq[FileStatus]): Option[StructType] = {
val parsedOptions = getJsonOptions(sparkSession, options)
- JsonDataSource(parsedOptions).inferSchema(sparkSession, files,
parsedOptions)
+ JsonDataSource(parsedOptions)
+ .inferSchema(sparkSession, files, parsedOptions, supportsArchiveScan =
true)
}
override def prepareWrite(
@@ -102,11 +107,16 @@ case class JsonFileFormat() extends TextBasedFileFormat
with DataSourceRegister
parsedOptions,
allowArrayAsStructs = true,
filters)
- JsonDataSource(parsedOptions).readFile(
- broadcastedHadoopConf.value.value,
- file,
- parser,
- requiredSchema)
+ if (parsedOptions.archiveFormatEnabled &&
ArchiveReader.isArchivePath(file.toPath)) {
+ JsonDataSource(parsedOptions).readArchive(
+ broadcastedHadoopConf.value.value, file, parser, requiredSchema)
+ } else {
+ JsonDataSource(parsedOptions).readFile(
+ broadcastedHadoopConf.value.value,
+ file,
+ parser,
+ requiredSchema)
+ }
}
}
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/json/JsonUtils.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/json/JsonUtils.scala
index 43fc69a52286..6a68a8d03525 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/json/JsonUtils.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/json/JsonUtils.scala
@@ -18,7 +18,6 @@
package org.apache.spark.sql.execution.datasources.json
import org.apache.spark.SparkException
-import org.apache.spark.input.PortableDataStream
import org.apache.spark.rdd.RDD
import
org.apache.spark.sql.catalyst.analysis.TypeCheckResult.{DataTypeMismatch,
TypeCheckSuccess}
import org.apache.spark.sql.catalyst.expressions.ExprUtils
@@ -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
paths (`RDD[InputStream]`
+ * for multi-line, `RDD[Array[Byte]]` for line-delimited) share one sampler.
*/
- def sample(json: RDD[PortableDataStream], options: JSONOptions):
RDD[PortableDataStream] = {
+ def sample[T](json: RDD[T], options: JSONOptions): RDD[T] = {
require(options.samplingRatio > 0,
s"samplingRatio (${options.samplingRatio}) should be greater than 0")
if (options.samplingRatio > 0.99) {
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/json/JsonTable.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/json/JsonTable.scala
index dc3729a119b1..2edc5702e0b8 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/json/JsonTable.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/json/JsonTable.scala
@@ -45,8 +45,10 @@ case class JsonTable(
options.asScala.toMap,
sparkSession.sessionState.conf.sessionLocalTimeZone,
sparkSession.sessionState.conf.columnNameOfCorruptRecord)
+ // The DSv2 reader calls `readFile` directly and cannot read archives, so
refuse to infer a
+ // schema for archive inputs (supportsArchiveScan = false) rather than
mis-reading raw bytes.
JsonDataSource(parsedOptions).inferSchema(
- sparkSession, files, parsedOptions)
+ sparkSession, files, parsedOptions, supportsArchiveScan = false)
}
override def newWriteBuilder(info: LogicalWriteInfo): WriteBuilder = {
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/ArchiveReadSuiteBase.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/ArchiveReadSuiteBase.scala
index 903adb9f98bb..b9bec4fd756a 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/ArchiveReadSuiteBase.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/ArchiveReadSuiteBase.scala
@@ -22,8 +22,10 @@ import java.nio.file.Files
import org.apache.spark.{SparkConf, SparkException}
import org.apache.spark.sql.{DataFrame, QueryTest, Row}
+import org.apache.spark.sql.functions.{col, struct}
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.test.SharedSparkSession
+import org.apache.spark.sql.types.{StringType, StructType}
import org.apache.spark.util.Utils
/**
@@ -104,6 +106,51 @@ trait ArchiveReadSuiteBase extends QueryTest with
SharedSparkSession {
schema: String = readSchema): DataFrame =
spark.read.format(format).options(readOptions ++
extraOptions).schema(schema).load(path)
+ /**
+ * 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]]; a format that does
not infer overrides this
+ * to false.
+ */
+ protected def supportsSchemaInference: Boolean = true
+
+ /** 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; JSON keeps it, CSV and text override it to
false.
+ */
+ protected def supportsComplexTypes: Boolean = true
+
+ /**
+ * Whether inputs with different field sets are unioned by field name -- the
self-describing,
+ * field-name-keyed model of JSON (and Avro/Parquet/XML), as opposed to
CSV's positional,
+ * first-header-keyed model. Gates the shared differing-field read/inference
tests; CSV and text
+ * override it to false.
+ */
+ protected def supportsSchemaMerge: Boolean = true
+
+ /** 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 does.
+ */
+ protected def inferredSchema(
+ paths: Seq[String],
+ extraOptions: Map[String, String] = Map.empty): StructType =
+ spark.read.options(readOptions ++ inferenceOptions ++ extraOptions)
+ .format(format).load(paths: _*).schema
+
/**
* Writes `entries` both into an archive and as loose files in a directory,
then asserts the
* archive read produces exactly the same rows as the directory read.
@@ -231,4 +278,128 @@ trait ArchiveReadSuiteBase extends QueryTest with
SharedSparkSession {
}
}
}
+
+ // ----- shared schema-inference tests (run when `supportsSchemaInference`)
--
+
+ if (supportsSchemaInference) {
+ test("archive infers the same schema as a directory of the same files") {
+ val entries = Seq(sampleDf((1, "Alice"), (2, "Bob")), sampleDf((3,
"Carol")))
+ .zipWithIndex.map { case (p, i) => entryName(i) -> encodeFile(p) }
+ withArchiveFile() { archive =>
+ writeArchive(archive, entries)
+ val archiveSchema = inferredSchema(Seq(archive.getCanonicalPath))
+ withTempDir { dir =>
+ entries.foreach { case (n, b) => Files.write(new File(dir,
n).toPath, b) }
+ assert(archiveSchema == inferredSchema(Seq(dir.getCanonicalPath)),
+ s"inference parity broken; archive=$archiveSchema")
+ }
+ }
+ }
+
+ test("all archive formats infer the same schema") {
+ val entries = Seq(sampleDf((1, "Alice"), (2, "Bob")), sampleDf((3,
"Carol")))
+ .zipWithIndex.map { case (p, i) => entryName(i) -> encodeFile(p) }
+ val schemas = archiveExtensions.map { ext =>
+ withArchiveFile(ext) { archive =>
+ writeArchive(archive, entries)
+ inferredSchema(Seq(archive.getCanonicalPath))
+ }
+ }
+ assert(schemas.distinct.size == 1,
+ s"archive formats inferred different schemas:
${archiveExtensions.zip(schemas)}")
+ }
+
+ test("inference skips a corrupt archive among good ones
(ignoreCorruptFiles)") {
+ withTempDir { dir =>
+ val good = sampleDf((1, "Alice"), (2, "Bob"))
+ writeArchive(new File(dir, s"good.${archiveExtensions.head}"),
+ Seq(entryName(0) -> encodeFile(good)))
+ writeCorruptArchive(new File(dir, s"bad.$corruptArchiveExtension"))
+ withSQLConf(SQLConf.IGNORE_CORRUPT_FILES.key -> "true") {
+ val schema = inferredSchema(Seq(dir.getCanonicalPath))
+ withTempDir { onlyGood =>
+ Files.write(new File(onlyGood, entryName(0)).toPath,
encodeFile(good))
+ assert(schema == inferredSchema(Seq(onlyGood.getCanonicalPath)),
+ s"corrupt archive not skipped during inference; got $schema")
+ }
+ }
+ }
+ }
+
+ test("archive inference widens a column's type across entries like a
directory") {
+ // The column is integral in the first entry and string in the second;
inference over all
+ // entries widens the merged type to string, exactly as a directory read
would.
+ withArchiveFile() { archive =>
+ writeArchive(archive, Seq(
+ entryName(0) -> encodeFile(Seq(1, 2).toDF("c")),
+ entryName(1) -> encodeFile(Seq("x").toDF("c"))))
+ val schema = inferredSchema(Seq(archive.getCanonicalPath))
+ assert(schema.length == 1 && schema.head.dataType == StringType,
+ s"expected the column widened to string across entries, got $schema")
+ }
+ }
+
+ test("inference merges archive entries with loose files in the same
directory") {
+ // An archive entry and a loose file with the same schema infer one
combined schema, matching
+ // a directory read of the same files.
+ val inArchive = sampleDf((1, "Alice"), (2, "Bob"))
+ val loose = sampleDf((3, "Carol"))
+ withTempDir { dir =>
+ writeArchive(new File(dir, s"data.${archiveExtensions.head}"),
+ Seq(entryName(0) -> encodeFile(inArchive)))
+ Files.write(new File(dir, s"loose.$fileExtension").toPath,
encodeFile(loose))
+ val schema = inferredSchema(Seq(dir.getCanonicalPath))
+ withTempDir { looseDir =>
+ Files.write(new File(looseDir, entryName(0)).toPath,
encodeFile(inArchive))
+ Files.write(new File(looseDir, s"loose.$fileExtension").toPath,
encodeFile(loose))
+ assert(schema == inferredSchema(Seq(looseDir.getCanonicalPath)),
+ s"mixed archive+loose inference diverged from a directory read;
got $schema")
+ }
+ }
+ }
+ }
+
+ // ----- shared schema-merge tests (run when `supportsSchemaMerge`)
----------
+
+ if (supportsSchemaMerge) {
+ test("archive entries with differing fields read like a directory") {
+ // One entry carries an extra field the other lacks; read under a schema
covering both, the
+ // missing field reads back null -- exactly as a directory read of the
same files does.
+ val withName = sampleDf((1, "Alice"), (2, "Bob"))
+ val idOnly = Seq(3).toDF("id")
+ assertArchiveMatchesDir(
+ Seq(entryName(0) -> encodeFile(withName), entryName(1) ->
encodeFile(idOnly)))
+ }
+
+ test("inference unions differing fields across archive entries and loose
files") {
+ // The archive entry has fields (id, name); the loose file has (id,
extra). A field-name-keyed
+ // format unions them by name -- exactly as a directory read of the same
files does.
+ val inArchive = sampleDf((1, "Alice"), (2, "Bob"))
+ val loose = Seq((3, 30)).toDF("id", "extra")
+ withTempDir { dir =>
+ writeArchive(new File(dir, s"data.${archiveExtensions.head}"),
+ Seq(entryName(0) -> encodeFile(inArchive)))
+ Files.write(new File(dir, s"loose.$fileExtension").toPath,
encodeFile(loose))
+ val schema = inferredSchema(Seq(dir.getCanonicalPath))
+ assert(schema.fieldNames.toSet == Set("id", "name", "extra"),
+ s"expected the union of the entry and loose fields, got $schema")
+ withTempDir { looseDir =>
+ Files.write(new File(looseDir, entryName(0)).toPath,
encodeFile(inArchive))
+ Files.write(new File(looseDir, s"loose.$fileExtension").toPath,
encodeFile(loose))
+ assert(schema == inferredSchema(Seq(looseDir.getCanonicalPath)),
+ s"differing-field inference diverged from a directory read; got
$schema")
+ }
+ }
+ }
+ }
+
+ // ----- shared complex-type test (run when `supportsComplexTypes`)
----------
+
+ if (supportsComplexTypes) {
+ test("archive round-trips nested struct types like a directory") {
+ assertArchiveMatchesDir(
+ Seq(entryName(0) -> encodeFile(complexSampleDf)),
+ schema = complexReadSchema)
+ }
+ }
}
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/CSVArchiveReadBase.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/CSVArchiveReadBase.scala
index e8076cfb9427..a18d950fb0f0 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/CSVArchiveReadBase.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/CSVArchiveReadBase.scala
@@ -44,6 +44,15 @@ trait CSVArchiveReadBase extends ArchiveReadSuiteBase {
override protected def readSchema: String = "id INT, name STRING"
+ // CSV infers its schema from row content (supportsSchemaInference defaults
true); inference is
+ // triggered by the `inferSchema` option. CSV is positional/header-keyed and
cannot represent
+ // nested types, so it opts out of the schema-merge and complex-type tests.
+ override protected def inferenceOptions: Map[String, String] =
Map("inferSchema" -> "true")
+
+ override protected def supportsComplexTypes: Boolean = false
+
+ override protected def supportsSchemaMerge: Boolean = false
+
override protected def encodeFile(
df: DataFrame,
writeOptions: Map[String, String]): Array[Byte] = {
@@ -65,93 +74,10 @@ trait CSVArchiveReadBase extends ArchiveReadSuiteBase {
/** Raw CSV bytes, for tests that need precise control over the row layout.
*/
protected def csvBytes(s: String): Array[Byte] =
s.getBytes(StandardCharsets.UTF_8)
- test("CSV: archive infers the same schema as a directory of the same files")
{
- val entries = Seq(sampleDf((1, "Alice"), (2, "Bob")), sampleDf((3,
"Carol")))
- .zipWithIndex.map { case (p, i) => entryName(i) -> encodeFile(p) }
- withArchiveFile() { archive =>
- writeArchive(archive, entries)
- val archiveSchema =
spark.read.options(readOptions).option("inferSchema", "true")
- .format(format).load(archive.getCanonicalPath).schema
- withTempDir { dir =>
- entries.foreach { case (n, b) => Files.write(new File(dir, n).toPath,
b) }
- val dirSchema = spark.read.options(readOptions).option("inferSchema",
"true")
- .format(format).load(dir.getCanonicalPath).schema
- assert(archiveSchema == dirSchema,
- s"inference parity broken; archive=$archiveSchema dir=$dirSchema")
- }
- }
- }
-
- test("CSV: all archive formats infer the same schema") {
- val entries = Seq(sampleDf((1, "Alice"), (2, "Bob")), sampleDf((3,
"Carol")))
- .zipWithIndex.map { case (p, i) => entryName(i) -> encodeFile(p) }
- val schemas = archiveExtensions.map { ext =>
- withArchiveFile(ext) { archive =>
- writeArchive(archive, entries)
- spark.read.options(readOptions).option("inferSchema", "true")
- .format(format).load(archive.getCanonicalPath).schema
- }
- }
- assert(schemas.distinct.size == 1,
- s"archive formats inferred different schemas:
${archiveExtensions.zip(schemas)}")
- }
-
/** CSV bytes for `rows`, prefixed with a `cols` header line when [[header]]
is set. */
private def csvEntry(cols: String, rows: String*): Array[Byte] =
csvBytes((if (header) cols +: rows else rows).mkString("", "\n", "\n"))
- test("CSV: inference skips a corrupt archive among good ones
(ignoreCorruptFiles)") {
- withTempDir { dir =>
- val good = sampleDf((1, "Alice"), (2, "Bob"))
- writeArchive(new File(dir, s"good.${archiveExtensions.head}"),
- Seq(entryName(0) -> encodeFile(good)))
- writeCorruptArchive(new File(dir, s"bad.$corruptArchiveExtension"))
- withSQLConf(SQLConf.IGNORE_CORRUPT_FILES.key -> "true") {
- val schema = spark.read.options(readOptions).option("inferSchema",
"true")
- .format(format).load(dir.getCanonicalPath).schema
- withTempDir { onlyGood =>
- Files.write(new File(onlyGood, entryName(0)).toPath,
encodeFile(good))
- val expected = spark.read.options(readOptions).option("inferSchema",
"true")
- .format(format).load(onlyGood.getCanonicalPath).schema
- assert(schema == expected,
- s"corrupt archive not skipped during inference; got $schema, want
$expected")
- }
- }
- }
- }
-
- test("CSV: inference widens a column's type across archive entries") {
- withArchiveFile() { archive =>
- writeArchive(archive, Seq(
- entryName(0) -> csvEntry("c", "1", "2"),
- entryName(1) -> csvEntry("c", "x")))
- val schema = spark.read.options(readOptions).option("inferSchema",
"true")
- .format(format).load(archive.getCanonicalPath).schema
- assert(schema.length == 1 && schema.head.dataType == StringType,
- s"expected the column widened to string across entries, got $schema")
- }
- }
-
- test("CSV: inference merges archive entries with loose files in the same
directory") {
- withTempDir { dir =>
- val inArchive = sampleDf((1, "Alice"), (2, "Bob"))
- val loose = sampleDf((3, "Carol"))
- writeArchive(new File(dir, s"data.${archiveExtensions.head}"),
- Seq(entryName(0) -> encodeFile(inArchive)))
- Files.write(new File(dir, s"loose.$fileExtension").toPath,
encodeFile(loose))
- val schema = spark.read.options(readOptions).option("inferSchema",
"true")
- .format(format).load(dir.getCanonicalPath).schema
- withTempDir { looseDir =>
- Files.write(new File(looseDir, entryName(0)).toPath,
encodeFile(inArchive))
- Files.write(new File(looseDir, s"loose.$fileExtension").toPath,
encodeFile(loose))
- val expected = spark.read.options(readOptions).option("inferSchema",
"true")
- .format(format).load(looseDir.getCanonicalPath).schema
- assert(schema == expected,
- s"mixed archive+loose inference diverged from directory; got
$schema, want $expected")
- }
- }
- }
-
test("CSV: a column empty in the archive but typed in a loose file is not
collapsed to string") {
// One inference pass over all inputs keeps the empty column NullType
until the end, so it
// widens with the loose file's Int. Merging two already-finished schemas
would have collapsed
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/JSONArchiveReadBase.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/JSONArchiveReadBase.scala
new file mode 100644
index 000000000000..fdad5df4b12d
--- /dev/null
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/JSONArchiveReadBase.scala
@@ -0,0 +1,223 @@
+/*
+ * 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<Container>ArchiveReadSuite` 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"
+
+ 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)
+
+ 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: multi-line documents match a directory read") {
+ assertArchiveMatchesDir(
+ Seq(
+ "a.json" -> jsonBytes("{\n \"id\": 1,\n \"name\": \"Alice\"\n}"),
+ "b.json" -> jsonBytes("{\n \"id\": 2,\n \"name\": \"Bob\"\n}")),
+ extraOptions = Map("multiLine" -> "true"))
+ }
+
+ test("JSON: a malformed record in an archive entry matches a directory read
(both modes)") {
+ // Permissive mode (the default): a malformed record parses to nulls with
its raw text echoed
+ // into `_corrupt_record`. The archive path wires its own
FailureSafeParser in `readStream` --
+ // and in multiLine it echoes the entry's pre-buffered bytes rather than
re-reading the file --
+ // so assert the corrupt-record column matches a directory read of the
same files in both the
+ // line-delimited and whole-document modes.
+ val corruptSchema = s"$readSchema, _corrupt_record STRING"
+ // Line-delimited: a good record, then a malformed one on the next line.
+ assertArchiveMatchesDir(
+ Seq("a.json" -> jsonBytes("{\"id\":1,\"name\":\"Alice\"}\n{ not valid
json\n")),
+ schema = corruptSchema)
+ // multiLine: the whole entry is one malformed document.
+ assertArchiveMatchesDir(
+ Seq("a.json" -> jsonBytes("{ not valid json")),
+ extraOptions = Map("multiLine" -> "true"),
+ schema = corruptSchema)
+ }
+
+ test("JSON: the DSv2 path refuses to infer a schema for an archive
(UNABLE_TO_INFER_SCHEMA)") {
+ // Archive scanning is wired into the v1 file source only, so the DSv2
reader cannot read
+ // archives. On the v2 path inference must keep returning None for an
archive input -- raising
+ // UNABLE_TO_INFER_SCHEMA -- rather than inferring a schema the v2 scan
would then mis-read as
+ // raw archive bytes. Forcing json off the v1 source list routes the read
through JsonTable.
+ withArchiveFile() { archive =>
+ writeArchive(archive, Seq(entryName(0) -> encodeFile(sampleDf((1,
"Alice"), (2, "Bob")))))
+ withSQLConf(SQLConf.USE_V1_SOURCE_LIST.key -> "") {
+ val e = intercept[AnalysisException] {
+
spark.read.options(readOptions).format(format).load(archive.getCanonicalPath)
+ }
+ assert(e.getCondition == "UNABLE_TO_INFER_SCHEMA",
+ s"expected UNABLE_TO_INFER_SCHEMA on the DSv2 path, got " +
+ s"${e.getCondition}: ${e.getMessage}")
+ }
+ }
+ }
+}
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/JSONTarArchiveReadSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/JSONTarArchiveReadSuite.scala
new file mode 100644
index 000000000000..5a703a321c59
--- /dev/null
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/JSONTarArchiveReadSuite.scala
@@ -0,0 +1,28 @@
+/*
+ * 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
+
+/**
+ * Reads of JSON files packed in tar archives (`.tar`/`.tar.gz`/`.tgz`): the
shared archive tests
+ * from [[ArchiveReadSuiteBase]] plus the JSON-specific ones from
[[JSONArchiveReadBase]], run over
+ * tar containers via [[TarArchiveReadBase]].
+ */
+class JSONTarArchiveReadSuite
+ extends ArchiveReadSuiteBase
+ with JSONArchiveReadBase
+ with TarArchiveReadBase
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
