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 cde68b671a57 [SPARK-57479][SQL] Read and infer XML schema from tar
archives
cde68b671a57 is described below
commit cde68b671a570fb157c7f1bbfebc1b70fcbeb9be
Author: akshatshenoi-db <[email protected]>
AuthorDate: Thu Jun 18 16:55:19 2026 -0700
[SPARK-57479][SQL] Read and infer XML 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`), SPARK-57321 added CSV schema inference, SPARK-57419
extended both to JSON, and SPARK-57478 to text, all gated by
`spark.sql.files.archive.reader.enabled`. This extends the same capability to
the XML data source.
When the flag is enabled, the V1 XML 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
XML file. `readArchive` is overridden per data source to mirror `readFile`:
single-line entries are split into lines and run through a `FailureSafeParser`
(so they get the same per-record corrupt-record handling as a non-archive
read), while multi-line entries are pars [...]
This also adjusts the `readArchive` entry point (JSON and XML) to take a
parser factory and build a fresh parser for each archive entry -- matching the
per-file parser of a non-archive read -- rather than sharing one parser across
all entries.
### Why are the changes needed?
To let XML ingestion read tar archives without unpacking them to disk,
matching the CSV, JSON, and text behavior already in Spark.
### Does this PR introduce _any_ user-facing change?
Yes. With `spark.sql.files.archive.reader.enabled=true` (default false),
the XML data source can read and infer schemas from `.tar`/`.tar.gz`/`.tgz`
files.
### How was this patch tested?
New `XMLTarArchiveReadSuite` (mixing `XMLArchiveReadBase` with the shared
`ArchiveReadSuiteBase` and `TarArchiveReadBase`), exercising the shared archive
read/inference/complex-type tests plus XML-specific tests: multi-line records,
attributes, and single-pass null-field widening against a loose file.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code
Closes #56572 from akshatshenoi-db/archive-xml.
Authored-by: akshatshenoi-db <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>
(cherry picked from commit 1c20625c4cf158bbb0f98269b41b925d1b10aacf)
Signed-off-by: Wenchen Fan <[email protected]>
---
.../datasources/json/JsonDataSource.scala | 15 +-
.../datasources/json/JsonFileFormat.scala | 6 +-
.../execution/datasources/xml/XmlDataSource.scala | 193 ++++++++++++++++++---
.../execution/datasources/xml/XmlFileFormat.scala | 27 ++-
.../execution/datasources/XMLArchiveReadBase.scala | 155 +++++++++++++++++
.../datasources/XMLTarArchiveReadSuite.scala | 28 +++
6 files changed, 384 insertions(+), 40 deletions(-)
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 5f2891985564..8c69a4af2826 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
@@ -78,18 +78,21 @@ abstract class JsonDataSource extends Serializable with
Logging {
* 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.
+ * is mode-agnostic (line-delimited and multi-line both flow through
`readStream`). Each entry is
+ * parsed with its own parser -- matching the per-file parser of a
non-archive read -- and 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.
+ *
+ * @param parser builds a fresh JSON parser for each entry.
*/
def readArchive(
conf: Configuration,
file: PartitionedFile,
- parser: JacksonParser,
+ parser: () => JacksonParser,
schema: StructType): Iterator[InternalRow] =
ArchiveReader(file.toPath).readEntries(conf) { (_, in) =>
- readStream(in, parser, schema)
+ readStream(in, parser(), schema)
}
final def inferSchema(
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 2d3424847ff8..09883c0786fa 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
@@ -102,19 +102,19 @@ case class JsonFileFormat() extends TextBasedFileFormat
with DataSourceRegister
}
(file: PartitionedFile) => {
- val parser = new JacksonParser(
+ def parser() = new JacksonParser(
actualSchema,
parsedOptions,
allowArrayAsStructs = true,
filters)
if (parsedOptions.archiveFormatEnabled &&
ArchiveReader.isArchivePath(file.toPath)) {
JsonDataSource(parsedOptions).readArchive(
- broadcastedHadoopConf.value.value, file, parser, requiredSchema)
+ broadcastedHadoopConf.value.value, file, () => parser(),
requiredSchema)
} else {
JsonDataSource(parsedOptions).readFile(
broadcastedHadoopConf.value.value,
file,
- parser,
+ parser(),
requiredSchema)
}
}
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 9dbca57e2ae9..0618a4c51da1 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.{FileNotFoundException, IOException}
+import java.io.{ByteArrayInputStream, FileNotFoundException, InputStream,
IOException}
import java.nio.charset.{Charset, StandardCharsets}
import scala.util.control.NonFatal
@@ -59,6 +59,38 @@ abstract class XmlDataSource extends Serializable with
Logging {
parser: StaxXmlParser,
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
+ * [[TextInputXmlDataSource]], as one whole document for
[[MultiLineXmlDataSource]]. Used only by
+ * [[readArchive]]; the stream is not closed here.
+ */
+ protected def readStream(
+ in: InputStream,
+ parser: StaxXmlParser,
+ schema: StructType): Iterator[InternalRow]
+
+ /**
+ * Streams a tar archive (`.tar`/`.tar.gz`/`.tgz`) entry by entry through
the XML parser without
+ * unpacking it to disk. The whole archive is a single split (see
`XmlFileFormat.isSplitable`);
+ * each entry's bytes are parsed exactly like a standalone XML file via
[[readStream]], which each
+ * mode overrides (single-line splits into lines, multi-line parses the
whole entry). Each entry
+ * is parsed with its own parser -- matching the per-file parser of a
non-archive read.
+ *
+ * Kept separate from [[readFile]] (rather than dispatched inside it)
because only the V1
+ * `XmlFileFormat` read path supports archives; XML has no DSv2 reader.
+ *
+ * @param parser builds a fresh XML parser for each entry.
+ */
+ def readArchive(
+ conf: Configuration,
+ file: PartitionedFile,
+ parser: () => StaxXmlParser,
+ schema: StructType): Iterator[InternalRow] =
+ ArchiveReader(file.toPath).readEntries(conf) { (_, in) =>
+ readStream(in, parser(), schema)
+ }
+
/**
* Infers the schema from `inputPaths` files.
*/
@@ -69,7 +101,15 @@ abstract class XmlDataSource extends Serializable with
Logging {
parsedOptions.singleVariantColumn match {
case Some(columnName) => Some(StructType(Array(StructField(columnName,
VariantType))))
case None =>
- if (inputPaths.nonEmpty) {
+ // 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 => ArchiveReader.isArchivePath(f.getPath))
+ if (hasArchive) {
+ Some(inferWithArchives(sparkSession, inputPaths, parsedOptions))
+ } else if (inputPaths.nonEmpty) {
Some(infer(sparkSession, inputPaths, parsedOptions))
} else {
None
@@ -81,6 +121,90 @@ abstract class XmlDataSource extends Serializable with
Logging {
sparkSession: SparkSession,
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 `ArchiveReader`, 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
this mode's 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 (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("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 => ArchiveReader.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],
+ options: XmlOptions): RDD[PortableDataStream] = {
+ val paths = inputPaths.map(_.getPath)
+ val name = paths.mkString(",")
+ val job =
Job.getInstance(sparkSession.sessionState.newHadoopConfWithOptions(
+ options.parameters))
+ FileInputFormat.setInputPaths(job, paths: _*)
+ val conf = job.getConfiguration
+
+ val rdd = new BinaryFileRDD(
+ sparkSession.sparkContext,
+ classOf[StreamInputFormat],
+ classOf[String],
+ classOf[PortableDataStream],
+ conf,
+ sparkSession.sparkContext.defaultMinPartitions)
+
+ // Only returns `PortableDataStream`s without paths.
+ rdd.setName(s"XMLFile: $name").values
+ }
}
object XmlDataSource extends Logging {
@@ -120,6 +244,27 @@ object TextInputXmlDataSource extends XmlDataSource {
lines.flatMap(safeParser.parse)
}
+ /**
+ * Mirrors [[readFile]] for an archive entry: split it into lines and run
each line through a
+ * [[FailureSafeParser]], so a single-line archive entry gets the same
per-record corrupt-record
+ * handling as a non-archive single-line read. (Whole-stream parsing, as the
multi-line override
+ * uses, would bypass that handling for single-line input.)
+ */
+ override protected def readStream(
+ in: InputStream,
+ parser: StaxXmlParser,
+ schema: StructType): Iterator[InternalRow] = {
+ val lines = ArchiveReader.lineIterator(in, None).map { line =>
+ new String(line.getBytes, 0, line.getLength, parser.options.charset)
+ }
+ val safeParser = new FailureSafeParser[String](
+ input => parser.parse(input),
+ parser.options.parseMode,
+ schema,
+ parser.options.columnNameOfCorruptRecord)
+ lines.flatMap(safeParser.parse)
+ }
+
override def infer(
sparkSession: SparkSession,
inputPaths: Seq[FileStatus],
@@ -185,6 +330,27 @@ object MultiLineXmlDataSource extends XmlDataSource {
}
}
+ /**
+ * Parses an archive entry as a single XML document, mirroring [[readFile]]:
the optimized parser
+ * re-reads its input (to echo the corrupt-record text on a parse failure),
which a single-use
+ * entry stream cannot do, so the entry's bytes are buffered and re-opened
over; the legacy parser
+ * reads the entry stream directly. Buffering one whole entry in memory is
an intended trade-off
+ * here -- the optimized parser requires a re-readable input, so a single
very large XML document
+ * packed in an archive is materialized in full (a non-archive read streams
from and re-opens the
+ * file instead). Entries are still read one at a time, so archive size
itself stays bounded.
+ */
+ override protected def readStream(
+ in: InputStream,
+ parser: StaxXmlParser,
+ schema: StructType): Iterator[InternalRow] = {
+ if (parser.options.useLegacyXMLParser) {
+ parser.parseStream(in, schema)
+ } else {
+ val bytes = in.readAllBytes()
+ parser.parseStreamOptimized(() => new ByteArrayInputStream(bytes),
schema)
+ }
+ }
+
override def infer(
sparkSession: SparkSession,
inputPaths: Seq[FileStatus],
@@ -250,27 +416,4 @@ object MultiLineXmlDataSource extends XmlDataSource {
schema
}
}
-
- private def createBaseRdd(
- sparkSession: SparkSession,
- inputPaths: Seq[FileStatus],
- options: XmlOptions): RDD[PortableDataStream] = {
- val paths = inputPaths.map(_.getPath)
- val name = paths.mkString(",")
- val job =
Job.getInstance(sparkSession.sessionState.newHadoopConfWithOptions(
- options.parameters))
- FileInputFormat.setInputPaths(job, paths: _*)
- val conf = job.getConfiguration
-
- val rdd = new BinaryFileRDD(
- sparkSession.sparkContext,
- classOf[StreamInputFormat],
- classOf[String],
- classOf[PortableDataStream],
- conf,
- sparkSession.sparkContext.defaultMinPartitions)
-
- // Only returns `PortableDataStream`s without paths.
- rdd.setName(s"XMLFile: $name").values
- }
}
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/xml/XmlFileFormat.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/xml/XmlFileFormat.scala
index 1ce253e76f45..ff4e57570a1d 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/xml/XmlFileFormat.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/xml/XmlFileFormat.scala
@@ -51,6 +51,10 @@ case class XmlFileFormat() extends TextBasedFileFormat with
DataSourceRegister {
options: Map[String, String],
path: Path): Boolean = {
val xmlOptions = getXmlOptions(sparkSession, options)
+ if (xmlOptions.archiveFormatEnabled && ArchiveReader.isArchivePath(path)) {
+ // A tar archive is read as one sequential stream (entry by entry), so
it is never split.
+ return false
+ }
XmlDataSource(xmlOptions).isSplitable && super.isSplitable(sparkSession,
options, path)
}
@@ -116,14 +120,25 @@ case class XmlFileFormat() extends TextBasedFileFormat
with DataSourceRegister {
}
(file: PartitionedFile) => {
- val parser = new StaxXmlParser(
+ def parser() = new StaxXmlParser(
actualRequiredSchema,
xmlOptions)
- XmlDataSource(xmlOptions).readFile(
- broadcastedHadoopConf.value.value,
- file,
- parser,
- requiredSchema)
+ // A tar archive (always a single split, see `isSplitable`) is streamed
entry by entry when
+ // archive reads are enabled; otherwise the file is parsed directly. XML
has no DSv2 reader,
+ // so this dispatch lives here rather than inside the shared `readFile`.
+ if (xmlOptions.archiveFormatEnabled &&
ArchiveReader.isArchivePath(file.toPath)) {
+ XmlDataSource(xmlOptions).readArchive(
+ broadcastedHadoopConf.value.value,
+ file,
+ () => parser(),
+ requiredSchema)
+ } else {
+ XmlDataSource(xmlOptions).readFile(
+ broadcastedHadoopConf.value.value,
+ file,
+ parser(),
+ requiredSchema)
+ }
}
}
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
new file mode 100644
index 000000000000..ef6a27a99e25
--- /dev/null
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/XMLArchiveReadBase.scala
@@ -0,0 +1,155 @@
+/*
+ * 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.DataFrame
+import org.apache.spark.sql.types.StringType
+import org.apache.spark.util.Utils
+
+/**
+ * Binds [[ArchiveReadSuiteBase]]'s file-format hooks to XML. XML opts into
the shared
+ * schema-inference and complex-type tests (see
`supportsSchemaInference`/`supportsComplexTypes`),
+ * and adds the XML-specific tests with no format-agnostic analogue:
multi-line records,
+ * element-union across entries, and attributes. Records are delimited by a
`rowTag` (here `row`),
+ * so a single `rowTag` is used for both writing and reading. Reusable across
archive formats: a
+ * `XML<Container>ArchiveReadSuite` mixes this in alongside the archive-format
trait.
+ */
+trait XMLArchiveReadBase extends ArchiveReadSuiteBase {
+
+ private val rowTag = "row"
+
+ override protected def format: String = "xml"
+
+ override protected def fileExtension: String = "xml"
+
+ override protected def readOptions: Map[String, String] = Map("rowTag" ->
rowTag)
+
+ override protected def readSchema: String = "id INT, name STRING"
+
+ // XML infers from record content, unions fields across inputs by name, and
represents nested
+ // elements as structs, so it keeps all three `supports*` defaults
(inference, schema-merge,
+ // complex types) and runs the full shared test set. Inference needs no
trigger option, so
+ // `inferenceOptions` keeps its empty default.
+
+ 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("xml")
+ .options(Map("rowTag" -> rowTag) ++ 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 XML bytes, for tests that need precise control over the record
layout. */
+ protected def xmlBytes(s: String): Array[Byte] =
s.getBytes(StandardCharsets.UTF_8)
+
+ // ----- XML-specific tests
--------------------------------------------------
+
+ test("XML: records spanning multiple lines match a directory read") {
+ assertArchiveMatchesDir(
+ Seq(
+ "a.xml" -> xmlBytes(
+ "<rows>\n <row>\n <id>1</id>\n <name>Alice</name>\n
</row>\n</rows>\n"),
+ "b.xml" -> xmlBytes(
+ "<rows>\n <row>\n <id>2</id>\n <name>Bob</name>\n
</row>\n</rows>\n")))
+ }
+
+ test("XML: attributes match a directory read") {
+ assertArchiveMatchesDir(
+ Seq(
+ "a.xml" -> xmlBytes("<rows><row
id=\"1\"><name>Alice</name></row></rows>"),
+ "b.xml" -> xmlBytes("<rows><row
id=\"2\"><name>Bob</name></row></rows>")),
+ schema = "_id INT, name STRING")
+ }
+
+ test("XML: inference widens a null archive field against a typed loose file
like a directory") {
+ // `c` is empty (NullType) in the archive entry and an integer in the
loose file. A single
+ // inference pass widens `c` to the integer type, exactly as a directory
read does. Inferring
+ // the archive and the loose file separately would canonicalize the
archive's `c` to string
+ // first, then merge to string -- diverging from the directory read.
+ val inArchive = xmlBytes("<rows><row><id>1</id><c></c></row></rows>")
+ val loose = xmlBytes("<rows><row><id>2</id><c>5</c></row></rows>")
+ withTempDir { dir =>
+ 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.find(_.name == "c").exists(_.dataType != StringType),
+ s"expected `c` to widen to its real type, not collapse to string; got
$schema")
+ withTempDir { looseDir =>
+ Files.write(new File(looseDir, entryName(0)).toPath, inArchive)
+ Files.write(new File(looseDir, s"loose.$fileExtension").toPath, loose)
+ assert(schema == inferredSchema(Seq(looseDir.getCanonicalPath)),
+ s"archive+loose inference diverged from a directory read; got
$schema")
+ }
+ }
+ }
+
+ test("XML: single-line mode reads and infers an archive like a directory") {
+ // multiLine=false: each line is one record in both the scan and
inference, matching a
+ // non-archive single-line read. (The default multiLine=true is covered by
the tests above.)
+ val opts = Map("multiLine" -> "false")
+ val entries = Seq(
+ entryName(0) -> xmlBytes(
+
"<row><id>1</id><name>Alice</name></row>\n<row><id>2</id><name>Bob</name></row>\n"),
+ entryName(1) -> xmlBytes("<row><id>3</id><name>Carol</name></row>\n"))
+ assertArchiveMatchesDir(entries, extraOptions = opts)
+ withTempDir { dir =>
+ writeArchive(new File(dir, s"data.${archiveExtensions.head}"), entries)
+ val archiveSchema = inferredSchema(Seq(dir.getCanonicalPath), opts)
+ withTempDir { looseDir =>
+ entries.foreach { case (n, b) => Files.write(new File(looseDir,
n).toPath, b) }
+ assert(archiveSchema == inferredSchema(Seq(looseDir.getCanonicalPath),
opts),
+ s"single-line archive inference diverged from a directory read; got
$archiveSchema")
+ }
+ }
+ }
+
+ test("XML: 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 single-line archive path wires its own
FailureSafeParser in
+ // `readStream`, and the multi-line path buffers the entry's bytes to echo
the corrupt record --
+ // so assert the corrupt-record column matches a directory read of the
same files in both the
+ // single-line and whole-document modes.
+ val corruptSchema = s"$readSchema, _corrupt_record STRING"
+ // Single-line: a good record, then a malformed one on the next line.
+ assertArchiveMatchesDir(
+ Seq(entryName(0) -> xmlBytes(
+ "<row><id>1</id><name>Alice</name></row>\n<row><id>2</id><name>\n")),
+ extraOptions = Map("multiLine" -> "false"),
+ schema = corruptSchema)
+ // multiLine: the whole entry is one malformed document (unclosed element).
+ assertArchiveMatchesDir(
+ Seq(entryName(0) -> xmlBytes("<row><id>1</id><name>Alice</name>")),
+ extraOptions = Map("multiLine" -> "true"),
+ schema = corruptSchema)
+ }
+}
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/XMLTarArchiveReadSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/XMLTarArchiveReadSuite.scala
new file mode 100644
index 000000000000..ec4a7047d600
--- /dev/null
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/XMLTarArchiveReadSuite.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 XML files packed in tar archives (`.tar`/`.tar.gz`/`.tgz`): the
shared archive tests
+ * from [[ArchiveReadSuiteBase]] plus the XML-specific ones from
[[XMLArchiveReadBase]], run over
+ * tar containers via [[TarArchiveReadBase]].
+ */
+class XMLTarArchiveReadSuite
+ extends ArchiveReadSuiteBase
+ with XMLArchiveReadBase
+ with TarArchiveReadBase
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
