This is an automated email from the ASF dual-hosted git repository.
cloud-fan pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new d3e955facc07 [SPARK-58248][SQL] Reuse single-line inference path for
JSON/XML archive inference
d3e955facc07 is described below
commit d3e955facc077197ca9217d87e04a30bc20e1408
Author: akshatshenoi-db <[email protected]>
AuthorDate: Wed Aug 5 10:47:04 2026 +0800
[SPARK-58248][SQL] Reuse single-line inference path for JSON/XML archive
inference
### What changes were proposed in this pull request?
`JsonDataSource` and `XmlDataSource` each carried a custom per-entry,
line-delimited
archive-schema-inference branch in `inferWithArchives` (added alongside
JSON/XML archive support):
for single-line mode they streamed every archive entry and loose file,
split the bytes into lines,
and fed them through a bespoke `JsonInferSchema` / `XmlInferSchema` pass.
This duplicated the
format's existing single-line inference path
(`TextInput{Json,Xml}DataSource.infer` via
`createBaseDataset` / `inferFromDataset`), which already tokenizes
line-delimited records by
delegating to the Text data source.
That duplication was only necessary because the Text data source could not
read archives. Since
SPARK-57705 added archive support to `TextFileFormat`, that is no longer
the case, so this PR
removes the custom single-line branch in both `inferWithArchives` methods.
Single-line archive
schema inference now falls through the dispatch in `inferSchema` to the
normal `infer` path — the
exact same path a directory or loose-file read takes — so it is identical
to a directory read by
construction. `inferWithArchives` moves into the multi-line data source and
is multi-line-only
(it reads whole streams / `rowTag`-tokenizes, which has no line model to
reuse).
The DSv2 archive-inference refuse guard is kept: DSv2 archive reads are
unsupported (the v2 scan
reads raw file bytes and never routes archives through `readArchive`), so
the `supportsArchiveScan`
parameter on `inferSchema` still makes the DSv2 callers (`{CSV,Json}Table`)
return `None`
(`UNABLE_TO_INFER_SCHEMA`) for an archive input, failing loudly instead of
letting the v2 scan
mis-read raw archive bytes. Only the v1 callers (`{Csv,Json}FileFormat`)
pass
`supportsArchiveScan = true`. This is unchanged behavior; the refactor
above is orthogonal to it.
### Why are the changes needed?
The custom single-line inference branch was a workaround for the Text data
source's lack of archive
support; that support now exists, so the workaround is dead code that can
diverge from the real
single-line read path. Removing it makes single-line archive inference
identical to a directory read
by construction rather than by a parallel implementation.
### Does this PR introduce _any_ user-facing change?
No. This is a behavior-preserving refactor; single-line archive inference
produces the same schema
(now via the shared path), multi-line inference is unchanged, and the DSv2
path still does not read
archives.
### How was this patch tested?
Existing archive test suites already assert that single-line archive
inference equals the directory
read (`inferredSchema(archive) == inferredSchema(dir)`), cross-entry type
widening, corrupt-entry
skip, and null-across-entries widening — these are unchanged and remain the
safety net
(`JSON{Tar,Zip}ArchiveReadSuite`, `XML{Tar,Zip}ArchiveReadSuite`), as are
the two DSv2-refuse tests
(`CSV/JSON: the DSv2 path refuses to infer a schema for an archive`). The
corrupt-entry skip now
also covers a corrupt entry reached while advancing through a multi-entry
archive during inference,
not only the first entry.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Opus 4.8)
Closes #57414 from akshatshenoi-db/archive-infer-reuse.
Authored-by: akshatshenoi-db <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>
(cherry picked from commit 65a8e236fe2f039d9763814fe7a6c72f0eab2fc6)
Signed-off-by: Wenchen Fan <[email protected]>
---
.../execution/datasources/csv/CSVDataSource.scala | 8 +-
.../execution/datasources/csv/CSVFileFormat.scala | 2 -
.../datasources/json/JsonDataSource.scala | 198 +++++++++++----------
.../execution/datasources/v2/csv/CSVTable.scala | 4 +-
.../execution/datasources/v2/json/JsonTable.scala | 2 -
.../execution/datasources/xml/XmlDataSource.scala | 168 +++++++++--------
.../datasources/ArchiveReadSuiteBase.scala | 13 ++
.../execution/datasources/CSVArchiveReadBase.scala | 7 +-
.../datasources/JSONArchiveReadBase.scala | 27 ++-
.../datasources/SevenZArchiveReadBase.scala | 8 +
.../datasources/TarArchiveTestUtils.scala | 32 +++-
.../execution/datasources/XMLArchiveReadBase.scala | 22 +++
.../datasources/ZipArchiveTestUtils.scala | 9 +
13 files changed, 307 insertions(+), 193 deletions(-)
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/csv/CSVDataSource.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/csv/CSVDataSource.scala
index 2eb0ad44f472..ddaf8ba2539c 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/csv/CSVDataSource.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/csv/CSVDataSource.scala
@@ -76,14 +76,10 @@ abstract class CSVDataSource extends Serializable with
Logging with SupportsArch
val hasArchive = parsedOptions.archiveFormatEnabled &&
inputPaths.exists(f =>
SupportsArchiveFormat.isArchivePath(f.getPath))
if (hasArchive && supportsArchiveScan) {
- // Archives (and any loose files alongside them) are inferred in a
single CSVInferSchema
- // 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 CSV.
+ // The caller's scan cannot read archives (e.g. DSv2), so refuse
rather than let it
+ // mis-read raw archive bytes as CSV.
None
} else if (inputPaths.nonEmpty) {
Some(infer(sparkSession, inputPaths, parsedOptions))
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/csv/CSVFileFormat.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/csv/CSVFileFormat.scala
index a406fbe4844d..0bfee12d1666 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/csv/CSVFileFormat.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/csv/CSVFileFormat.scala
@@ -57,8 +57,6 @@ case class CSVFileFormat() extends TextBasedFileFormat with
DataSourceRegister {
options: Map[String, String],
files: Seq[FileStatus]): Option[StructType] = {
val parsedOptions = getCsvOptions(sparkSession, options)
- // The v1 file format routes archives to `readArchive` (see
`buildReader`), so archive schema
- // inference is supported here.
CSVDataSource(parsedOptions)
.inferSchema(sparkSession, files, parsedOptions, supportsArchiveScan =
true)
}
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 a5f19ae353d4..42568a2e6b0a 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,9 +17,8 @@
package org.apache.spark.sql.execution.datasources.json
-import java.io.{ByteArrayInputStream, FileNotFoundException, InputStream,
IOException}
+import java.io.{ByteArrayInputStream, Closeable, FileNotFoundException,
InputStream, IOException}
-import scala.reflect.ClassTag
import scala.util.control.NonFatal
import com.fasterxml.jackson.core.{JsonFactory, JsonParser}
@@ -105,15 +104,9 @@ abstract class JsonDataSource extends Serializable with
Logging with SupportsArc
case None =>
val hasArchive = parsedOptions.archiveFormatEnabled &&
inputPaths.exists(f =>
SupportsArchiveFormat.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.
+ if (hasArchive && !supportsArchiveScan) {
+ // The caller's scan cannot read archives (e.g. DSv2), so refuse
rather than let it
+ // mis-read raw archive bytes as JSON.
None
} else if (inputPaths.nonEmpty) {
Some(infer(sparkSession, inputPaths, parsedOptions))
@@ -127,92 +120,6 @@ abstract class JsonDataSource extends Serializable with
Logging with SupportsArc
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 `SupportsArchiveFormat`, 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 (SupportsArchiveFormat.isArchivePath(path)) {
- SupportsArchiveFormat.readArchiveEntries(path,
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 => 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 {
@@ -344,6 +251,12 @@ object MultiLineJsonDataSource extends JsonDataSource {
sparkSession: SparkSession,
inputPaths: Seq[FileStatus],
parsedOptions: JSONOptions): StructType = {
+ val hasArchive = parsedOptions.archiveFormatEnabled &&
+ inputPaths.exists(f => SupportsArchiveFormat.isArchivePath(f.getPath))
+ if (hasArchive) {
+ return inferWithArchives(sparkSession, inputPaths, parsedOptions)
+ }
+
val json: RDD[PortableDataStream] =
JsonDataSource.createBaseRdd(sparkSession, inputPaths, parsedOptions)
val sampled: RDD[PortableDataStream] = JsonUtils.sample(json,
parsedOptions)
@@ -357,6 +270,97 @@ object MultiLineJsonDataSource extends JsonDataSource {
}
}
+ /**
+ * Infers a multi-line JSON schema when at least one input is an archive.
Each archive entry
+ * (streamed via `SupportsArchiveFormat`, never unpacked to disk) and each
loose file is one whole
+ * JSON document, and all feed a single [[JsonInferSchema]] pass -- exactly
as a directory of the
+ * same files would infer. Single-line archive inference does not come here:
the Text data source
+ * reads archives directly, so it flows through
[[TextInputJsonDataSource.infer]] like any
+ * directory read. Corrupt/missing inputs are skipped when the ignore flags
are set (see
+ * [[skipInputOnError]]).
+ */
+ private def inferWithArchives(
+ sparkSession: SparkSession,
+ inputPaths: Seq[FileStatus],
+ parsedOptions: JSONOptions): StructType = {
+ val baseRdd = JsonDataSource.createBaseRdd(sparkSession, inputPaths,
parsedOptions)
+ val encoding = parsedOptions.encoding
+ val ignoreCorruptFiles = parsedOptions.ignoreCorruptFiles
+ val ignoreMissingFiles = parsedOptions.ignoreMissingFiles
+
+ // An archive entry's stream is only valid until the shared cursor
advances, so each document
+ // must be consumed before the next is pulled; `JsonInferSchema.infer`
does.
+ val docs: RDD[InputStream] = baseRdd.flatMap { stream =>
+ val path = new Path(stream.getPath())
+ skipInputOnError(stream.getPath(), ignoreMissingFiles,
ignoreCorruptFiles) {
+ if (SupportsArchiveFormat.isArchivePath(path)) {
+ SupportsArchiveFormat.readArchiveEntries(path,
stream.getConfiguration) { (_, in) =>
+ Iterator.single(in)
+ }
+ } else {
+ Iterator.single(
+
CodecStreams.createInputStreamWithCloseResource(stream.getConfiguration, path))
+ }
+ }
+ }
+
+ SQLExecution.withSQLConfPropagated(sparkSession) {
+ val docParser: (JsonFactory, InputStream) => JsonParser = encoding
+ .map(enc => CreateJacksonParser.inputStream(enc, _: JsonFactory, _:
InputStream))
+ .getOrElse(CreateJacksonParser.inputStream(_: JsonFactory, _:
InputStream))
+ new JsonInferSchema(parsedOptions).infer[InputStream](
+ JsonUtils.sample(docs, parsedOptions), docParser, isReadFile = true)
+ }
+ }
+
+ /**
+ * Builds one input's document iterator, catching a missing/corrupt error
when the ignore flags
+ * are set. `readArchiveEntries` advances to later entries lazily, so a
corrupt later entry throws
+ * on `hasNext`, not at construction; the returned iterator catches both. A
construction failure
+ * skips the whole input; a mid-advance failure keeps the entries already
yielded and skips only
+ * the remainder of the archive.
+ */
+ private def skipInputOnError(
+ inputPath: String,
+ ignoreMissingFiles: Boolean,
+ ignoreCorruptFiles: Boolean)(
+ build: => Iterator[InputStream]): Iterator[InputStream] = {
+ def handle(e: Throwable): Iterator[InputStream] = e match {
+ case e: FileNotFoundException if ignoreMissingFiles =>
+ logWarning(log"Skipped missing input: ${MDC(PATH, inputPath)}", e)
+ Iterator.empty
+ case e: FileNotFoundException => throw e
+ case e @ (_: RuntimeException | _: IOException) if ignoreCorruptFiles =>
+ logWarning(log"Skipped the corrupted input: ${MDC(PATH, inputPath)}",
e)
+ Iterator.empty
+ case NonFatal(e) =>
+ throw QueryExecutionErrors.cannotReadFilesError(
+ e, SparkPath.fromPathString(inputPath).urlEncoded)
+ }
+
+ val underlying =
+ try build
+ catch { case NonFatal(e) => return handle(e) }
+
+ new Iterator[InputStream] {
+ private var delegate = underlying
+ override def hasNext: Boolean =
+ try delegate.hasNext
+ catch {
+ case NonFatal(e) =>
+ // A mid-advance throw reaches neither exhaustion nor close, so
close the failed
+ // iterator here to release its archive stream promptly rather
than at task completion.
+ delegate match {
+ case c: Closeable => try c.close() catch { case NonFatal(_) => }
+ case _ =>
+ }
+ delegate = handle(e)
+ delegate.hasNext
+ }
+ override def next(): InputStream = delegate.next()
+ }
+ }
+
private def dataToInputStream(dataStream: PortableDataStream): InputStream =
{
val path = new Path(dataStream.getPath())
CodecStreams.createInputStreamWithCloseResource(dataStream.getConfiguration,
path)
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/csv/CSVTable.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/csv/CSVTable.scala
index 184eb41b6c49..e7b206bd9c58 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/csv/CSVTable.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/csv/CSVTable.scala
@@ -46,9 +46,7 @@ case class CSVTable(
columnPruning = sparkSession.sessionState.conf.csvColumnPruning,
sparkSession.sessionState.conf.sessionLocalTimeZone)
- // The DSv2 reader does not route archives to `readArchive` (it calls
`readFile` directly), so
- // archive scans aren't supported here; pass supportsArchiveScan = false
so an archive input
- // keeps failing with UNABLE_TO_INFER_SCHEMA rather than having its raw
bytes parsed as CSV.
+ // The DSv2 scan cannot read archives, so refuse archive inference here.
CSVDataSource(parsedOptions)
.inferSchema(sparkSession, files, parsedOptions, supportsArchiveScan =
false)
}
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 74095e85a0f6..c635a7d06bed 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,6 @@ 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, supportsArchiveScan = false)
}
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/xml/XmlDataSource.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/xml/XmlDataSource.scala
index ef34711e041e..14116279801d 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/xml/XmlDataSource.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/xml/XmlDataSource.scala
@@ -17,7 +17,7 @@
package org.apache.spark.sql.execution.datasources.xml
-import java.io.{ByteArrayInputStream, FileNotFoundException, InputStream,
IOException}
+import java.io.{ByteArrayInputStream, Closeable, FileNotFoundException,
InputStream, IOException}
import java.nio.charset.{Charset, StandardCharsets}
import scala.util.control.NonFatal
@@ -101,15 +101,7 @@ abstract class XmlDataSource extends Serializable with
Logging with SupportsArch
parsedOptions.singleVariantColumn match {
case Some(columnName) => Some(StructType(Array(StructField(columnName,
VariantType))))
case None =>
- // When any input is a tar archive, infer over all inputs in a single
pass -- archive
- // entries are streamed (never unpacked to disk) and tokenized as XML
records alongside any
- // loose files -- so the result matches a directory read of the same
files. XML has no DSv2
- // reader, so this archive scan is always V1.
- val hasArchive = parsedOptions.archiveFormatEnabled &&
- inputPaths.exists(f =>
SupportsArchiveFormat.isArchivePath(f.getPath))
- if (hasArchive) {
- Some(inferWithArchives(sparkSession, inputPaths, parsedOptions))
- } else if (inputPaths.nonEmpty) {
+ if (inputPaths.nonEmpty) {
Some(infer(sparkSession, inputPaths, parsedOptions))
} else {
None
@@ -122,69 +114,6 @@ abstract class XmlDataSource extends Serializable with
Logging with SupportsArch
inputPaths: Seq[FileStatus],
parsedOptions: XmlOptions): StructType
- /**
- * Infers an XML schema when at least one input is a tar archive
(`.tar`/`.tar.gz`/`.tgz`). Every
- * archive entry (streamed through `SupportsArchiveFormat`, never unpacked
to disk) and every
- * loose file is tokenized into records and fed to a single
[[XmlInferSchema]] pass, exactly as a
- * directory of the same files would infer. Tokenization is per-mode so it
matches the scan:
- * multi-line splits the whole stream into `rowTag`-delimited records,
single-line treats each
- * line as a record (mirroring [[readFile]] and JSON's `inferWithArchives`).
- */
- private def inferWithArchives(
- sparkSession: SparkSession,
- inputPaths: Seq[FileStatus],
- parsedOptions: XmlOptions): StructType = {
- val baseRdd = createBaseRdd(sparkSession, inputPaths, parsedOptions)
- val ignoreCorruptFiles = parsedOptions.ignoreCorruptFiles
- val ignoreMissingFiles = parsedOptions.ignoreMissingFiles
-
- // Applies `perEntry` to each input -- an archive entry by entry
(streamed, so only one entry's
- // bytes are in flight at a time), a loose file directly -- skipping a
whole input when it is
- // corrupt/missing and the ignore flags are set.
- def perInput(perEntry: InputStream => Iterator[String]): RDD[String] =
baseRdd.flatMap {
- stream =>
- val path = new Path(stream.getPath())
- try {
- if (SupportsArchiveFormat.isArchivePath(path)) {
- SupportsArchiveFormat.readArchiveEntries(path,
stream.getConfiguration) { (_, in) =>
- perEntry(in)
- }
- } else {
- perEntry(
-
CodecStreams.createInputStreamWithCloseResource(stream.getConfiguration, path))
- }
- } catch {
- case e: FileNotFoundException if ignoreMissingFiles =>
- logWarning("Skipped missing file", e)
- Iterator.empty[String]
- case NonFatal(e) =>
- Utils.getRootCause(e) match {
- case root @ (_: AccessControlException | _:
BlockMissingException) => throw root
- case _: RuntimeException | _: IOException if ignoreCorruptFiles
=>
- logWarning("Skipped the rest of the content in the corrupted
file", e)
- Iterator.empty[String]
- case other => throw other
- }
- }
- }
-
- // Tokenize each input the way this mode's scan reads records, so the
inferred schema matches a
- // directory read: multi-line splits the whole stream into
rowTag-delimited records, single-line
- // treats each line as a record (mirroring
TextInputXmlDataSource.readFile).
- val tokenRDD: RDD[String] = if (parsedOptions.multiLine) {
- perInput(in => StaxXmlParser.tokenizeStream(in, parsedOptions))
- } else {
- val charset = parsedOptions.charset
- perInput(in => lineIterator(in, None).map { line =>
- new String(line.getBytes, 0, line.getLength, charset)
- })
- }
- SQLExecution.withSQLConfPropagated(sparkSession) {
- new XmlInferSchema(parsedOptions,
sparkSession.sessionState.conf.caseSensitiveAnalysis)
- .infer(tokenRDD)
- }
- }
-
protected def createBaseRdd(
sparkSession: SparkSession,
inputPaths: Seq[FileStatus],
@@ -358,6 +287,12 @@ object MultiLineXmlDataSource extends XmlDataSource {
inputPaths: Seq[FileStatus],
parsedOptions: XmlOptions): StructType = {
+ val hasArchive = parsedOptions.archiveFormatEnabled &&
+ inputPaths.exists(f => SupportsArchiveFormat.isArchivePath(f.getPath))
+ if (hasArchive) {
+ return inferWithArchives(sparkSession, inputPaths, parsedOptions)
+ }
+
if (!parsedOptions.useLegacyXMLParser) {
return inferOptimized(sparkSession, inputPaths, parsedOptions)
}
@@ -418,4 +353,91 @@ object MultiLineXmlDataSource extends XmlDataSource {
schema
}
}
+
+ /**
+ * Infers a multi-line XML schema when at least one input is an archive.
Every archive entry
+ * (streamed through `SupportsArchiveFormat`, never unpacked to disk) and
every loose file is
+ * tokenized into `rowTag`-delimited records and fed to a single
[[XmlInferSchema]] pass, exactly
+ * as a directory of the same files would infer. Single-line archive
inference does not come here:
+ * the Text data source reads archives directly, so it flows through
+ * [[TextInputXmlDataSource.infer]] like any directory read. Corrupt/missing
inputs are skipped
+ * when the ignore flags are set (see [[skipInputOnError]]). Uses the legacy
tokenizer because the
+ * optimized parser re-opens its input, which a single-use archive entry
stream does not support.
+ */
+ private def inferWithArchives(
+ sparkSession: SparkSession,
+ inputPaths: Seq[FileStatus],
+ parsedOptions: XmlOptions): StructType = {
+ val baseRdd = createBaseRdd(sparkSession, inputPaths, parsedOptions)
+ val ignoreCorruptFiles = parsedOptions.ignoreCorruptFiles
+ val ignoreMissingFiles = parsedOptions.ignoreMissingFiles
+
+ val tokenRDD: RDD[String] = baseRdd.flatMap { stream =>
+ val path = new Path(stream.getPath())
+ skipInputOnError(ignoreMissingFiles, ignoreCorruptFiles) {
+ if (SupportsArchiveFormat.isArchivePath(path)) {
+ SupportsArchiveFormat.readArchiveEntries(path,
stream.getConfiguration) { (_, in) =>
+ StaxXmlParser.tokenizeStream(in, parsedOptions)
+ }
+ } else {
+ StaxXmlParser.tokenizeStream(
+
CodecStreams.createInputStreamWithCloseResource(stream.getConfiguration, path),
+ parsedOptions)
+ }
+ }
+ }
+ SQLExecution.withSQLConfPropagated(sparkSession) {
+ new XmlInferSchema(parsedOptions,
sparkSession.sessionState.conf.caseSensitiveAnalysis)
+ .infer(tokenRDD)
+ }
+ }
+
+ /**
+ * Builds one input's token iterator, catching a missing/corrupt error when
the ignore flags are
+ * set. `readArchiveEntries` advances to later entries lazily, so a corrupt
later entry throws on
+ * `hasNext`, not at construction; the returned iterator catches both. A
construction failure
+ * skips the whole input; a mid-advance failure keeps the records already
yielded and skips only
+ * the remainder of the archive. Access/block errors are always rethrown
(unwrapped), matching the
+ * non-archive read path.
+ */
+ private def skipInputOnError(
+ ignoreMissingFiles: Boolean,
+ ignoreCorruptFiles: Boolean)(
+ build: => Iterator[String]): Iterator[String] = {
+ def handle(e: Throwable): Iterator[String] = e match {
+ case e: FileNotFoundException if ignoreMissingFiles =>
+ logWarning("Skipped missing file", e)
+ Iterator.empty[String]
+ case NonFatal(e) =>
+ Utils.getRootCause(e) match {
+ case root @ (_: AccessControlException | _: BlockMissingException)
=> throw root
+ case _: RuntimeException | _: IOException if ignoreCorruptFiles =>
+ logWarning("Skipped the rest of the content in the corrupted
file", e)
+ Iterator.empty[String]
+ case other => throw other
+ }
+ }
+
+ val underlying =
+ try build
+ catch { case NonFatal(e) => return handle(e) }
+
+ new Iterator[String] {
+ private var delegate = underlying
+ override def hasNext: Boolean =
+ try delegate.hasNext
+ catch {
+ case NonFatal(e) =>
+ // A mid-advance throw reaches neither exhaustion nor close, so
close the failed
+ // iterator here to release its archive stream promptly rather
than at task completion.
+ delegate match {
+ case c: Closeable => try c.close() catch { case NonFatal(_) => }
+ case _ =>
+ }
+ delegate = handle(e)
+ delegate.hasNext
+ }
+ override def next(): String = delegate.next()
+ }
+ }
}
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 ae64f5b50293..943eb637e8fa 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
@@ -105,6 +105,19 @@ trait ArchiveReadSuiteBase extends QueryTest with
SharedSparkSession {
protected def corruptEntryBytes: Array[Byte] =
s"This is not a valid $format file".getBytes("UTF-8")
+ /**
+ * Whether [[writeArchiveFailingAfterFirstEntry]] is supported, gating the
mid-advance
+ * corrupt-skip regression test. From the container trait.
+ */
+ protected def supportsMidAdvanceFailure: Boolean
+
+ /**
+ * Writes an archive at `dest` whose first entry (`firstEntry`) reads
cleanly but which then fails
+ * while advancing to a later entry. From the container trait.
+ */
+ protected def writeArchiveFailingAfterFirstEntry(
+ dest: File, firstEntry: (String, Array[Byte])): Unit
+
/** An archive extension whose reader fails on corrupt bytes (used by the
corrupt-file tests). */
protected def corruptArchiveExtension: String
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 b3ab43fcd19c..3665f857edd6 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
@@ -131,11 +131,8 @@ trait CSVArchiveReadBase extends ArchiveReadSuiteBase {
}
test("CSV: 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 and letting
the V2 scan parse the
- // raw archive bytes as CSV. Forcing csv off the V1 source list routes the
read through
- // CSVTable.
+ // Forcing csv off the V1 source list routes the archive read through the
DSv2 CSVTable, which
+ // cannot read archives and must fail with UNABLE_TO_INFER_SCHEMA, not
parse raw bytes.
withArchiveFile() { archive =>
writeArchive(archive, Seq(entryName(0) -> encodeFile(sampleDf((1,
"Alice"), (2, "Bob")))))
withSQLConf(SQLConf.USE_V1_SOURCE_LIST.key -> "") {
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
index 24a187380535..70de9e5e6317 100644
---
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
@@ -184,11 +184,30 @@ trait JSONArchiveReadBase extends ArchiveReadSuiteBase {
schema = corruptSchema)
}
+ if (supportsMidAdvanceFailure) {
+ test("JSON: multiLine inference keeps entries read before a mid-advance
failure " +
+ "(ignoreCorruptFiles)") {
+ // Entry 0 is read, then advancing to a later entry throws (not at
open). A whole-archive drop
+ // would lose entry 0's `extra`; aborting the traversal would lose the
sibling file's `later`.
+ val opts = Map("multiLine" -> "true")
+ withArchiveFile() { archive =>
+ writeArchiveFailingAfterFirstEntry(archive,
+ entryName(0) -> jsonBytes("{\n \"id\": 1,\n \"name\": \"Alice\",\n
\"extra\": 9\n}"))
+ Files.write(new File(archive.getParentFile,
s"later.$fileExtension").toPath,
+ jsonBytes("{\n \"id\": 2,\n \"name\": \"Bob\",\n \"later\":
7\n}"))
+ withSQLConf(SQLConf.IGNORE_CORRUPT_FILES.key -> "true") {
+ val schema =
inferredSchema(Seq(archive.getParentFile.getCanonicalPath), opts)
+ assert(schema.fieldNames.toSet == Set("id", "name", "extra",
"later"),
+ "expected `extra` (pre-failure entry) and `later` (sibling file)
in the inferred " +
+ s"schema after the mid-advance skip, got $schema")
+ }
+ }
+ }
+ }
+
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.
+ // Forcing json off the v1 source list routes the archive read through the
DSv2 JsonTable, which
+ // cannot read archives and must fail with UNABLE_TO_INFER_SCHEMA, not
parse raw bytes.
withArchiveFile() { archive =>
writeArchive(archive, Seq(entryName(0) -> encodeFile(sampleDf((1,
"Alice"), (2, "Bob")))))
withSQLConf(SQLConf.USE_V1_SOURCE_LIST.key -> "") {
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/SevenZArchiveReadBase.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/SevenZArchiveReadBase.scala
index efc6c280bfb6..1c77964e6a5f 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/SevenZArchiveReadBase.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/SevenZArchiveReadBase.scala
@@ -68,4 +68,12 @@ trait SevenZArchiveTestUtils {
protected def writeCorruptArchive(dest: File): Unit =
Files.write(dest.toPath, "this is not a valid 7z archive, just some random
bytes"
.getBytes(StandardCharsets.UTF_8))
+
+ // 7z validates its whole index when the archive is opened, so this helper
cannot construct a
+ // failure that occurs only while advancing to another entry.
+ protected def supportsMidAdvanceFailure: Boolean = false
+
+ protected def writeArchiveFailingAfterFirstEntry(
+ dest: File, firstEntry: (String, Array[Byte])): Unit =
+ throw new UnsupportedOperationException("7z cannot fail while advancing to
a later entry")
}
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/TarArchiveTestUtils.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/TarArchiveTestUtils.scala
index 7f1d48e8143b..a31c2b187eac 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/TarArchiveTestUtils.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/TarArchiveTestUtils.scala
@@ -17,7 +17,7 @@
package org.apache.spark.sql.execution.datasources
-import java.io.{File, FileOutputStream, OutputStream}
+import java.io.{ByteArrayOutputStream, File, FileOutputStream, OutputStream}
import java.nio.charset.StandardCharsets
import java.nio.file.Files
import java.util.Locale
@@ -62,4 +62,34 @@ trait TarArchiveTestUtils {
protected def writeCorruptArchive(dest: File): Unit =
Files.write(dest.toPath, "this is not a valid gzip-compressed tar archive"
.getBytes(StandardCharsets.UTF_8))
+
+ protected def supportsMidAdvanceFailure: Boolean = true
+
+ /**
+ * Writes a plain `.tar` (uncompressed, so the first entry survives
truncation) with `firstEntry`
+ * intact followed by a second entry whose header is cut short, so
`getNextEntry` throws while
+ * advancing to it.
+ */
+ protected def writeArchiveFailingAfterFirstEntry(
+ dest: File, firstEntry: (String, Array[Byte])): Unit = {
+ require(firstEntry._1.getBytes(StandardCharsets.UTF_8).length <= 100,
+ "the first entry's name must fit the 100-byte ustar name field so its
header is a single " +
+ s"512-byte block, otherwise the truncation offset below shifts; got
'${firstEntry._1}'")
+ val buf = new ByteArrayOutputStream()
+ val out = new TarArchiveOutputStream(buf)
+ Seq(firstEntry, ("part-later.bin", ("x" *
4096).getBytes(StandardCharsets.UTF_8))).foreach {
+ case (entryName, bytes) =>
+ val entry = new TarArchiveEntry(entryName)
+ entry.setSize(bytes.length.toLong)
+ out.putArchiveEntry(entry)
+ out.write(bytes)
+ out.closeArchiveEntry()
+ }
+ out.finish()
+ out.close()
+ // Keep the first entry (512-byte header + block-aligned body) plus half
of the second entry's
+ // 512-byte header, so `getNextEntry` hits a partial header while
advancing.
+ val firstBlocks = 512 + ((firstEntry._2.length + 511) / 512) * 512
+ Files.write(dest.toPath, buf.toByteArray.take(firstBlocks + 256))
+ }
}
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/XMLArchiveReadBase.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/XMLArchiveReadBase.scala
index c5e956d4cd38..ffa5484b3194 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/XMLArchiveReadBase.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/XMLArchiveReadBase.scala
@@ -21,6 +21,7 @@ import java.io.File
import java.nio.charset.StandardCharsets
import java.nio.file.Files
+import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types.StringType
/**
@@ -132,6 +133,27 @@ trait XMLArchiveReadBase extends ArchiveReadSuiteBase {
extraOptions = Map("multiLine" -> "true"),
schema = corruptSchema)
}
+
+ if (supportsMidAdvanceFailure) {
+ test("XML: multiLine inference keeps records read before a mid-advance
failure " +
+ "(ignoreCorruptFiles)") {
+ // Entry 0 is read, then advancing to a later entry throws (not at
open). A whole-archive drop
+ // would lose entry 0's `extra`; aborting the traversal would lose the
sibling file's `later`.
+ val opts = Map("multiLine" -> "true")
+ withArchiveFile() { archive =>
+ writeArchiveFailingAfterFirstEntry(archive, entryName(0) ->
+
xmlBytes("<rows><row><id>1</id><name>Alice</name><extra>9</extra></row></rows>"))
+ Files.write(new File(archive.getParentFile,
s"later.$fileExtension").toPath,
+
xmlBytes("<rows><row><id>2</id><name>Bob</name><later>7</later></row></rows>"))
+ withSQLConf(SQLConf.IGNORE_CORRUPT_FILES.key -> "true") {
+ val schema =
inferredSchema(Seq(archive.getParentFile.getCanonicalPath), opts)
+ assert(schema.fieldNames.toSet == Set("id", "name", "extra",
"later"),
+ "expected `extra` (pre-failure entry) and `later` (sibling file)
in the inferred " +
+ s"schema after the mid-advance skip, got $schema")
+ }
+ }
+ }
+ }
}
class XMLTarArchiveReadSuite
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/ZipArchiveTestUtils.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/ZipArchiveTestUtils.scala
index cf9e42262a7b..51282d1e856d 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/ZipArchiveTestUtils.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/ZipArchiveTestUtils.scala
@@ -57,4 +57,13 @@ trait ZipArchiveTestUtils {
protected def writeCorruptArchive(dest: File): Unit =
Files.write(dest.toPath, "this is not a valid zip archive, just some
random bytes"
.getBytes(StandardCharsets.UTF_8))
+
+ // Zip can't throw while *advancing*: a cut header reads as clean EOF, and a
cut body only throws
+ // when the entry's bytes are read -- which a lazy reader (JSON) defers
downstream. Tar is used
+ // for the mid-advance regression test instead.
+ protected def supportsMidAdvanceFailure: Boolean = false
+
+ protected def writeArchiveFailingAfterFirstEntry(
+ dest: File, firstEntry: (String, Array[Byte])): Unit =
+ throw new UnsupportedOperationException("zip cannot force a throw while
advancing to an entry")
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
