You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by we...@apache.org on 2026/07/24 02:56:18 UTC
(spark) branch master updated: [SPARK-57590][SQL] Read and infer Parquet schema from archives
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 3de7e983c111 [SPARK-57590][SQL] Read and infer Parquet schema from archives
3de7e983c111 is described below
commit 3de7e983c111a52dd6fdb6be2d0b2c602a60fe69
Author: akshatshenoi-db <ak...@databricks.com>
AuthorDate: Fri Jul 24 10:55:58 2026 +0800
[SPARK-57590][SQL] Read and infer Parquet schema from archives
### What changes were proposed in this pull request?
This extends the tar/zip archive-reader feature to Parquet, continuing the series
(SPARK-57135 CSV, SPARK-57321 CSV inference, SPARK-57419 JSON, SPARK-57478 text,
SPARK-57479 XML, SPARK-57481 Avro, SPARK-57705 zip container, and SPARK-58110 which
extracted the shared `SupportsArchiveFormat` trait). With the feature enabled, an
archive (`.tar`/`.tar.gz`/`.tgz`/`.zip`) of Parquet files is read like a directory of
its entries.
Parquet is the first *random-access* archive format. The streaming formats parse each
entry straight from an `InputStream`, but Parquet needs its trailing footer, so an
archive entry cannot be streamed. This PR:
- Adds the random-access localization surface to `SupportsArchiveFormat`: the trait
gains `readLocalizedEntries` (unpack each kept entry to a temp file, one at a time,
read it, delete it) and the `archiveEntryFilter` hook; the companion gains
`copyEntryToLocalFile` and `localizeEntries` (so an executor-side inference path can
localize without a trait instance). This is the surface that was deferred from
SPARK-58110 until its first consumer landed - this is that consumer.
- `ParquetFileFormat` mixes in the trait, returns `isSplitable = false` for an archive
(the archive is one split), dispatches an archive read through `readLocalizedEntries`
reusing the existing plain per-file reader, and folds per-entry footer reads into
`readParquetFootersInParallel` so schema inference expands archives. A corrupt entry
condemns its whole archive during inference (matching the whole-file semantics of a
corrupt loose Parquet file), while a corrupt entry during a read skips the rest of
that archive as the streaming formats do.
- `input_file_name()` / `_metadata.file_path` report the archive path, not a per-entry
temp path.
The SPARK-37089 use-after-free guard is preserved: the per-task completion listener
deletes only the temp dir; `FileScanRDD` closes the per-entry reader so the vectorized
reader's off-heap column vectors are not freed while still in use.
This also fixes a stale doc reference left by SPARK-58110 in
`SupportsArchiveFormatSuite` (the comment pointed at the removed `ZipArchiveReader`'s
scaladoc).
### Why are the changes needed?
Parquet is a common archived format, and the archive-reader feature already supports the
text-oriented formats and Avro. Extending it to Parquet lets users ingest tarred/zipped
Parquet without unpacking first, and establishes the random-access path that ORC and
other footer-based formats will reuse.
### Does this PR introduce _any_ user-facing change?
Yes, gated by `spark.sql.files.archive.reader.enabled` (default `false`). When enabled,
reading a path that is a supported archive of Parquet files returns the rows of its
entries; when disabled (the default), behavior is unchanged. Loose Parquet reads are
unaffected either way.
### How was this patch tested?
Added `ParquetTarArchiveReadSuite` and `ParquetZipArchiveReadSuite` (each binding the new
`ParquetArchiveReadBase` to the tar and zip harnesses), covering vectorized and
row-based reads, schema inference (including `mergeSchema` union and type widening across
entries), `input_file_name`/`_metadata` parity with the archive path, per-entry temp-file
cleanup on an abandoned (`LIMIT`) read, and `ignoreCorruptFiles`/`ignoreMissingFiles`
behavior for both read and inference. Extended the shared `ArchiveReadSuiteBase` with an
`inferenceSamplesOneFile` hook (Parquet samples one part-file per SPARK-11500) and an
`input_file_name` parity test.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Opus 4.8)
Closes #57388 from akshatshenoi-db/archive-parquet-trait.
Authored-by: akshatshenoi-db <ak...@databricks.com>
Signed-off-by: Wenchen Fan <we...@databricks.com>
---
.../datasources/SupportsArchiveFormat.scala | 146 ++++++++++++++++-
.../datasources/parquet/ParquetFileFormat.scala | 83 +++++++++-
.../datasources/ArchiveReadSuiteBase.scala | 141 ++++++++++++++--
.../datasources/ParquetArchiveReadBase.scala | 180 +++++++++++++++++++++
.../datasources/SupportsArchiveFormatSuite.scala | 2 +-
5 files changed, 524 insertions(+), 28 deletions(-)
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/SupportsArchiveFormat.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/SupportsArchiveFormat.scala
index ffb38c52d9c5..a54e83d9e83f 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/SupportsArchiveFormat.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/SupportsArchiveFormat.scala
@@ -17,7 +17,7 @@
package org.apache.spark.sql.execution.datasources
-import java.io.{Closeable, InputStream}
+import java.io.{Closeable, File, FileOutputStream, InputStream}
import java.nio.ByteBuffer
import java.nio.channels.{NonWritableChannelException, SeekableByteChannel}
import java.util.Locale
@@ -37,9 +37,11 @@ import org.apache.hadoop.fs.{FSDataInputStream, Path}
import org.apache.hadoop.io.Text
import org.apache.hadoop.util.LineReader
-import org.apache.spark.TaskContext
+import org.apache.spark.{SparkEnv, TaskContext}
import org.apache.spark.internal.Logging
-import org.apache.spark.util.HadoopFSUtils
+import org.apache.spark.paths.SparkPath
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.util.{HadoopFSUtils, Utils}
/**
* Mixed into a file-based data source to let it read tar/zip/7z archives of its files, treating an
@@ -91,10 +93,107 @@ trait SupportsArchiveFormat extends Logging {
}
}
+ /** Which archive entries are data files of this format; others are skipped. */
+ protected def archiveEntryFilter(name: String): Boolean =
+ throw new UnsupportedOperationException(
+ s"${getClass.getName} does not support random-access archive reads")
+
+ /**
+ * Reads an archive by unpacking each entry to a temp file and applying `readEntry`, for a
+ * format that needs a complete file on disk (random access).
+ *
+ * @param file the archive as a [[PartitionedFile]]
+ * @param conf Hadoop configuration used to open the archive
+ * @param tempPrefix prefix for the per-task temp dir the entries are unpacked into
+ * @param readEntry reads one unpacked entry file into rows
+ * @return iterator of rows across all entries
+ */
+ protected def readLocalizedEntries(
+ file: PartitionedFile,
+ conf: Configuration,
+ tempPrefix: String)(
+ readEntry: PartitionedFile => Iterator[InternalRow]): Iterator[InternalRow] = {
+ val tempDir = Utils.createTempDir(Utils.getLocalDir(SparkEnv.get.conf), tempPrefix)
+ // Register cleanup before constructing `entries`, which can throw before returning an iterator
+ // that FileScanRDD could close.
+ Option(TaskContext.get()).foreach(_.addTaskCompletionListener[Unit] { _ =>
+ Utils.deleteRecursively(tempDir)
+ })
+ val entries =
+ try SupportsArchiveFormat.localizeEntries(file.toPath, conf, tempDir, archiveEntryFilter)
+ catch {
+ case NonFatal(e) =>
+ Utils.deleteRecursively(tempDir)
+ throw e
+ }
+
+ // Element type is `Object`, not `InternalRow`: a batch scan yields `ColumnarBatch`.
+ val rows = new Iterator[Object] with Closeable {
+ private var current: Iterator[Object] = Iterator.empty
+ private var currentFile: File = _
+ private var done = false
+
+ private def releaseCurrent(): Unit = {
+ current match {
+ case c: Closeable => try c.close() catch { case NonFatal(_) => }
+ case _ =>
+ }
+ current = Iterator.empty
+ if (currentFile != null) {
+ currentFile.delete()
+ currentFile = null
+ }
+ }
+
+ // Advance on `hasNext`, not `next`, so a reader reusing a mutable batch is not probed early.
+ private def advance(): Unit = {
+ while (!done && !current.hasNext) {
+ releaseCurrent()
+ if (entries.hasNext) {
+ val (_, entryFile) = entries.next()
+ currentFile = entryFile
+ current = readEntry(file.copy(
+ filePath = SparkPath.fromUri(entryFile.toURI),
+ start = 0L,
+ length = entryFile.length(),
+ fileSize = entryFile.length(),
+ modificationTime = file.modificationTime)).asInstanceOf[Iterator[Object]]
+ } else {
+ done = true
+ }
+ }
+ }
+
+ override def hasNext: Boolean = {
+ advance()
+ !done
+ }
+
+ override def next(): Object = {
+ if (!hasNext) throw new NoSuchElementException
+ current.next()
+ }
+
+ override def close(): Unit = {
+ done = true
+ releaseCurrent()
+ entries match {
+ case c: Closeable => try c.close() catch { case NonFatal(_) => }
+ case _ =>
+ }
+ }
+ }
+
+ rows.asInstanceOf[Iterator[InternalRow]]
+ }
+
}
object SupportsArchiveFormat {
+ // Basename chars unsafe in a temp file name; precompiled since it runs per entry.
+ private val unsafeEntryNameChars = Pattern.compile("[^A-Za-z0-9._-]")
+
/**
* Whether the file path is a supported archive format.
*/
@@ -258,6 +357,47 @@ object SupportsArchiveFormat {
entries
}
+ /**
+ * Copies one archive entry's bytes to a fresh temp file under localDir.
+ *
+ * @param in the entry's byte stream, positioned at the entry start
+ * @param localDir directory the temp file is created under
+ * @param entryName the entry's archive path; its sanitized basename names the temp file
+ * @return the temp [[File]] holding the entry's bytes
+ */
+ def copyEntryToLocalFile(in: InputStream, localDir: File, entryName: String): File = {
+ val rawBasename = entryName.substring(entryName.lastIndexOf('/') + 1)
+ val basename = unsafeEntryNameChars.matcher(rawBasename).replaceAll("_")
+ val local = File.createTempFile("archive-entry-", "-" + basename, localDir)
+ val out = new FileOutputStream(local)
+ try Utils.copyStream(in, out) finally out.close()
+ local
+ }
+
+ /**
+ * Materializes each entry passing `entryFilter` to a fresh temp file under `localDir`, lazily one
+ * at a time. Yields `(entryName, localFile)`; the caller owns deleting each file. On the
+ * companion so executor-side callers (a format's distributed archive inference) can use it
+ * without a trait instance.
+ *
+ * @param path the archive path
+ * @param conf Hadoop configuration used to open the archive
+ * @param localDir directory the per-entry temp files are created under
+ * @param entryFilter which entry names to keep
+ */
+ def localizeEntries(
+ path: Path,
+ conf: Configuration,
+ localDir: File,
+ entryFilter: String => Boolean): Iterator[(String, File)] =
+ readArchiveEntries(path, conf) { (name, in) =>
+ if (entryFilter(name)) {
+ Iterator.single((name, copyEntryToLocalFile(in, localDir, name)))
+ } else {
+ Iterator.empty
+ }
+ }
+
}
/**
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFileFormat.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFileFormat.scala
index 733e55a8609e..235aa428fbce 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFileFormat.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFileFormat.scala
@@ -17,7 +17,7 @@
package org.apache.spark.sql.execution.datasources.parquet
-import java.io.{Closeable, FileNotFoundException}
+import java.io.{Closeable, File, FileNotFoundException, IOException}
import java.time.ZoneId
import java.util.concurrent.atomic.AtomicBoolean
@@ -37,7 +37,7 @@ import org.apache.parquet.format.converter.ParquetMetadataConverter.SKIP_ROW_GRO
import org.apache.parquet.hadoop._
import org.apache.parquet.hadoop.util.HadoopInputFile
-import org.apache.spark.TaskContext
+import org.apache.spark.{SparkEnv, TaskContext}
import org.apache.spark.internal.Logging
import org.apache.spark.internal.LogKeys.{PATH, SCHEMA}
import org.apache.spark.sql.SparkSession
@@ -61,9 +61,14 @@ class ParquetFileFormat
extends FileFormat
with DataSourceRegister
with SessionStateHelper
+ with SupportsArchiveFormat
with Logging
with Serializable {
+ // Reads every archive entry, matching a directory scan (Parquet part-files are often
+ // extensionless).
+ override protected def archiveEntryFilter(name: String): Boolean = true
+
override def shortName(): String = "parquet"
override def toString: String = "Parquet"
@@ -113,6 +118,10 @@ class ParquetFileFormat
sparkSession: SparkSession,
options: Map[String, String],
path: Path): Boolean = {
+ val parquetOptions = new ParquetOptions(options, getSqlConf(sparkSession))
+ if (parquetOptions.archiveFormatEnabled && SupportsArchiveFormat.isArchivePath(path)) {
+ return false
+ }
true
}
@@ -204,6 +213,7 @@ class ParquetFileFormat
val parquetOptions = new ParquetOptions(options, sqlConf)
val datetimeRebaseModeInRead = parquetOptions.datetimeRebaseModeInRead
val int96RebaseModeInRead = parquetOptions.int96RebaseModeInRead
+ val archiveFormatEnabled = parquetOptions.archiveFormatEnabled
// Should always be set by FileSourceScanExec creating this.
// Check conf before checking option, to allow working around an issue by changing conf.
@@ -219,7 +229,7 @@ class ParquetFileFormat
assert(supportBatch(sparkSession, resultSchema))
}
- (file: PartitionedFile) => {
+ val readSingleFile: PartitionedFile => Iterator[InternalRow] = (file: PartitionedFile) => {
assert(file.partitionValues.numFields == partitionSchema.size)
val split = new FileSplit(file.toPath, file.start, file.length, Array.empty[String])
@@ -308,6 +318,21 @@ class ParquetFileFormat
}
}
}
+
+ // An archive is read by unpacking each Parquet entry to a local temp file and reading it with
+ // the plain reader (readSingleFile); readLocalizedEntries owns the unpack/iterate/cleanup.
+ def readArchiveFile(file: PartitionedFile): Iterator[InternalRow] =
+ readLocalizedEntries(file, broadcastedHadoopConf.value.value, "parquet-archive") {
+ entryFile => readSingleFile(entryFile)
+ }
+
+ (file: PartitionedFile) => {
+ if (archiveFormatEnabled && SupportsArchiveFormat.isArchivePath(file.toPath)) {
+ readArchiveFile(file)
+ } else {
+ readSingleFile(file)
+ }
+ }
}
// scalastyle:off argcount
@@ -526,29 +551,71 @@ object ParquetFileFormat extends Logging {
partFiles: Seq[FileStatus],
ignoreCorruptFiles: Boolean,
ignoreMissingFiles: Boolean = false): Seq[Footer] = {
+ val archiveEnabled = SQLConf.get.getConf(SQLConf.ARCHIVE_FORMAT_READER_ENABLED)
ThreadUtils.parmap(partFiles, "readingParquetFooters", 8,
preserveSparkThrowable = true) { currentFile =>
try {
// Skips row group information since we only need the schema.
// ParquetFileReader.readFooter throws RuntimeException, instead of IOException,
// when it can't read the footer.
- Some(new Footer(currentFile.getPath,
- ParquetFooterReader.readFooter(
- HadoopInputFile.fromStatus(currentFile, conf), SKIP_ROW_GROUPS)))
+ if (archiveEnabled && SupportsArchiveFormat.isArchivePath(currentFile.getPath)) {
+ // An archive is one file here; read each of its Parquet entries' footers (the archive is
+ // atomic under ignoreCorruptFiles, see readArchiveFooters).
+ readArchiveFooters(conf, currentFile)
+ } else {
+ Seq(new Footer(currentFile.getPath,
+ ParquetFooterReader.readFooter(
+ HadoopInputFile.fromStatus(currentFile, conf), SKIP_ROW_GROUPS)))
+ }
} catch {
case e: Exception if ignoreMissingFiles &&
ExceptionUtils.getThrowables(e).exists(_.isInstanceOf[FileNotFoundException]) =>
logWarning(log"Skipped missing file: ${MDC(PATH, currentFile)}", e)
- None
+ Seq.empty
case e: RuntimeException if ignoreCorruptFiles =>
logWarning(log"Skipped the footer in the corrupted file: ${MDC(PATH, currentFile)}", e)
- None
+ Seq.empty
+ // A corrupt archive's failure arrives wrapped, not as a bare RuntimeException; a missing
+ // file (FileNotFoundException) is excluded since ignoreMissingFiles governs it.
+ case e: Exception if ignoreCorruptFiles &&
+ archiveEnabled && SupportsArchiveFormat.isArchivePath(currentFile.getPath) && {
+ val root = Utils.getRootCause(e)
+ root.isInstanceOf[IOException] && !root.isInstanceOf[FileNotFoundException]
+ } =>
+ logWarning(log"Skipped the footer in the corrupted file: ${MDC(PATH, currentFile)}", e)
+ Seq.empty
case e: Exception =>
throw QueryExecutionErrors.cannotReadFooterForFileError(currentFile.getPath, e)
}
}.flatten
}
+ /** Reads every Parquet entry's footer in one archive. */
+ private def readArchiveFooters(conf: Configuration, archive: FileStatus): Seq[Footer] = {
+ val tempDir = Utils.createTempDir(Utils.getLocalDir(SparkEnv.get.conf), "parquet-archive-infer")
+ // localizeEntries eagerly opens/copies the first entry, so build it inside the try -- a corrupt
+ // archive throws there and the finally must still delete tempDir.
+ var entries: Iterator[(String, File)] = Iterator.empty
+ try {
+ entries = SupportsArchiveFormat.localizeEntries(archive.getPath, conf, tempDir, _ => true)
+ entries.map { case (_, entryFile) =>
+ try {
+ val status = new FileStatus(entryFile.length(), false, 0, 0, entryFile.lastModified(),
+ 0, null, null, null, new Path(entryFile.toURI))
+ new Footer(archive.getPath,
+ ParquetFooterReader.readFooter(HadoopInputFile.fromStatus(status, conf),
+ SKIP_ROW_GROUPS))
+ } finally entryFile.delete()
+ }.toList
+ } finally {
+ entries match {
+ case c: Closeable => c.close()
+ case _ =>
+ }
+ Utils.deleteRecursively(tempDir)
+ }
+ }
+
/**
* Figures out a merged Parquet schema with a distributed Spark job.
*
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 1d715295aeb0..ae64f5b50293 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
@@ -20,9 +20,11 @@ package org.apache.spark.sql.execution.datasources
import java.io.File
import java.nio.file.Files
+import org.apache.hadoop.fs.Path
+
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.functions.{col, input_file_name, struct}
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.test.SharedSparkSession
import org.apache.spark.sql.types.{StringType, StructType}
@@ -99,6 +101,10 @@ trait ArchiveReadSuiteBase extends QueryTest with SharedSparkSession {
/** Writes bytes that are not a readable archive at `dest` (of [[corruptArchiveExtension]]). */
protected def writeCorruptArchive(dest: File): Unit
+ /** Bytes that are not a valid data file of [[format]], used to plant a corrupt archive entry. */
+ protected def corruptEntryBytes: Array[Byte] =
+ s"This is not a valid $format file".getBytes("UTF-8")
+
/** An archive extension whose reader fails on corrupt bytes (used by the corrupt-file tests). */
protected def corruptArchiveExtension: String
@@ -149,6 +155,21 @@ trait ArchiveReadSuiteBase extends QueryTest with SharedSparkSession {
*/
protected def supportsSchemaMerge: Boolean = true
+ /**
+ * Whether default (non-merge) inference samples a single file rather than scanning every input.
+ * Parquet samples one part-file (SPARK-11500), so a corrupt/missing first-sampled archive fails
+ * to infer instead of falling through to a readable one. The streaming formats scan all inputs.
+ */
+ protected def inferenceSamplesOneFile: Boolean = false
+
+ /**
+ * Whether this format unpacks each archive entry to a local temp file before reading it (Parquet
+ * needs random access), as opposed to streaming entries. Gates the shared tests that exercise the
+ * localize path: per-entry corrupt handling during inference and read, and the differing-field
+ * read.
+ */
+ protected def localizesEntries: Boolean = false
+
/** Sample data with a nested struct column, used by the complex-type test. */
protected def complexSampleDf: DataFrame =
Seq((1, "NYC", "10001"), (2, "SF", "94105")).toDF("id", "city", "zip")
@@ -215,6 +236,20 @@ trait ArchiveReadSuiteBase extends QueryTest with SharedSparkSession {
}
}
+ test("input_file_name and _metadata.file_path report the archive path, not a per-entry path") {
+ withArchiveFile() { archive =>
+ val parts = Seq(sampleDf((1, "Alice"), (2, "Bob")), sampleDf((3, "Carol")))
+ writeArchive(archive, parts.zipWithIndex.map { case (p, i) => entryName(i) -> encodeFile(p) })
+ val df = read(archive.getCanonicalPath)
+ Seq(input_file_name(), col("_metadata.file_path")).foreach { pathCol =>
+ val paths = df.select(pathCol).distinct().collect().map(_.getString(0))
+ assert(paths.length == 1, s"expected one archive path, got: ${paths.toList}")
+ assert(new Path(paths.head) == new Path(archive.toURI),
+ s"expected the archive path ${archive.toURI}, got: ${paths.head}")
+ }
+ }
+ }
+
test("multiple entries and multiple loose files under a partitioned dir, plus an empty archive") {
withTempDir { rootDir =>
val partitionDir = new File(rootDir, "dt=2024-01-01")
@@ -327,35 +362,27 @@ trait ArchiveReadSuiteBase extends QueryTest with SharedSparkSession {
}
test("inference skips a corrupt archive among good ones (ignoreCorruptFiles)") {
+ // A format that samples one file (Parquet) only reaches the good archive when every input is
+ // scanned, i.e. under mergeSchema; otherwise the sampled corrupt archive is skipped and
+ // nothing survives (the same as a corrupt loose file).
+ val extra =
+ if (inferenceSamplesOneFile) Map("mergeSchema" -> "true") else Map.empty[String, String]
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))
+ val schema = inferredSchema(Seq(dir.getCanonicalPath), extra)
withTempDir { onlyGood =>
Files.write(new File(onlyGood, entryName(0)).toPath, encodeFile(good))
- assert(schema == inferredSchema(Seq(onlyGood.getCanonicalPath)),
+ assert(schema == inferredSchema(Seq(onlyGood.getCanonicalPath), extra),
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.
@@ -374,6 +401,22 @@ trait ArchiveReadSuiteBase extends QueryTest with SharedSparkSession {
}
}
}
+
+ // Excludes formats that sample one file (Parquet): they can't widen types across entries.
+ if (!inferenceSamplesOneFile) {
+ 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")
+ }
+ }
+ }
}
// ----- shared schema-merge tests (run when `supportsSchemaMerge`) ----------
@@ -410,6 +453,72 @@ trait ArchiveReadSuiteBase extends QueryTest with SharedSparkSession {
}
}
+ // ----- shared localize-path tests (run when `localizesEntries`) ------------
+
+ if (localizesEntries) {
+ // The merge-gated block above owns this test when a format unions by name; localize-path
+ // formats (Parquet) set `supportsSchemaMerge` off, so provide it here instead.
+ 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)))
+ }
+ }
+
+ Seq(true, false).foreach { ignoreCorrupt =>
+ test(s"ignoreCorruptFiles=$ignoreCorrupt: inference skips a corrupt entry's whole archive") {
+ // A corrupt entry condemns its whole archive during inference (no partial ingestion), so
+ // the `extra` column carried by the bad archive's valid sibling entry must not surface. A
+ // good archive is unaffected. Formats sampling one file need mergeSchema to fold every
+ // entry's footer.
+ val extra =
+ if (inferenceSamplesOneFile) Map("mergeSchema" -> "true") else Map.empty[String, String]
+ val sibling = Seq((3, "Carol", "x")).toDF("id", "name", "extra")
+ withTempDir { dir =>
+ writeArchive(new File(dir, s"good.${archiveExtensions.head}"),
+ Seq(entryName(0) -> encodeFile(sampleDf((1, "Alice"), (2, "Bob")))))
+ writeArchive(new File(dir, s"bad.${archiveExtensions.head}"), Seq(
+ entryName(0) -> encodeFile(sibling),
+ entryName(1) -> corruptEntryBytes))
+ withSQLConf(SQLConf.IGNORE_CORRUPT_FILES.key -> ignoreCorrupt.toString) {
+ if (ignoreCorrupt) {
+ assert(inferredSchema(Seq(dir.getCanonicalPath), extra).fieldNames.toSet ==
+ Set("id", "name"),
+ "the whole corrupt archive, including its valid sibling entry, should be skipped")
+ } else {
+ intercept[Exception](inferredSchema(Seq(dir.getCanonicalPath), extra))
+ }
+ }
+ }
+ }
+
+ test(s"ignoreCorruptFiles=$ignoreCorrupt: read skips the rest of an archive at a bad entry") {
+ // The read ingests entries before the corrupt one, then skips the rest of that archive (not
+ // whole-archive atomic like inference). A separate good archive is read in full.
+ val alive = sampleDf((1, "Alice"), (2, "Bob"))
+ val beforeCorrupt = sampleDf((3, "Carol"))
+ withTempDir { dir =>
+ writeArchive(new File(dir, s"good.${archiveExtensions.head}"),
+ Seq(entryName(0) -> encodeFile(alive)))
+ writeArchive(new File(dir, s"bad.${archiveExtensions.head}"), Seq(
+ entryName(0) -> encodeFile(beforeCorrupt),
+ entryName(1) -> corruptEntryBytes))
+ withSQLConf(SQLConf.IGNORE_CORRUPT_FILES.key -> ignoreCorrupt.toString) {
+ if (ignoreCorrupt) {
+ checkAnswer(read(dir.getCanonicalPath), alive.union(beforeCorrupt))
+ } else {
+ intercept[SparkException](read(dir.getCanonicalPath).collect())
+ }
+ }
+ }
+ }
+ }
+ }
+
// ----- shared complex-type test (run when `supportsComplexTypes`) ----------
if (supportsComplexTypes) {
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/ParquetArchiveReadBase.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/ParquetArchiveReadBase.scala
new file mode 100644
index 000000000000..c09f764ba3d2
--- /dev/null
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/ParquetArchiveReadBase.scala
@@ -0,0 +1,180 @@
+/*
+ * 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.file.Files
+
+import org.apache.hadoop.fs.{FileStatus, Path}
+
+import org.apache.spark.SparkException
+import org.apache.spark.sql.execution.datasources.parquet.ParquetFileFormat
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types.StructType
+import org.apache.spark.util.Utils
+
+/**
+ * Binds [[ArchiveReadSuiteBase]]'s hooks to Parquet (entries unpacked to a local file for footer
+ * random access). Parquet is self-describing, so the base's schema-inference tests run too.
+ */
+trait ParquetArchiveReadBase extends ArchiveReadSuiteBase {
+
+ import testImplicits._
+
+ override protected def format: String = "parquet"
+
+ override protected def fileExtension: String = "parquet"
+
+ override protected def readOptions: Map[String, String] = Map.empty
+
+ override protected def readSchema: String = "id INT, name STRING"
+
+ // Parquet has authoritative per-file schemas and only unions under `mergeSchema`, so it opts out
+ // of the by-name default-inference union (covered instead by the mergeSchema test below).
+ override protected def supportsSchemaMerge: Boolean = false
+
+ // Parquet samples one part-file for non-merge inference (SPARK-11500).
+ override protected def inferenceSamplesOneFile: Boolean = true
+
+ // Parquet unpacks each entry to a local temp file for footer random access.
+ override protected def localizesEntries: Boolean = true
+
+ for (vectorized <- Seq(true, false)) {
+ test(s"archive reads return the same rows with vectorized reader = $vectorized") {
+ withSQLConf(SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key -> vectorized.toString) {
+ assertArchiveMatchesDir(
+ Seq(entryName(0) -> encodeFile(sampleDf((1, "Alice"), (2, "Bob")))))
+ }
+ }
+ }
+
+ test("an abandoned read (LIMIT) over an archive returns partial rows and cleans up") {
+ def archiveTempDirs(localDir: File): Set[String] =
+ Option(localDir.listFiles()).getOrElse(Array.empty)
+ .filter(_.getName.startsWith("parquet-archive")).map(_.getName).toSet
+ withArchiveFile() { archive =>
+ val parts = (0 until 4).map(i => entryName(i) -> encodeFile(sampleDf((i, s"v$i"))))
+ writeArchive(archive, parts)
+ val localDir = new File(Utils.getLocalDir(spark.sparkContext.getConf))
+ val before = archiveTempDirs(localDir)
+ assert(read(archive.getCanonicalPath).limit(2).collect().length == 2)
+ // This read's per-entry temp dir (prefix `parquet-archive`) must be removed on task
+ // completion, so no new one survives.
+ assert((archiveTempDirs(localDir) -- before).isEmpty,
+ "the read's temp dir was not cleaned up")
+ }
+ }
+
+ test("extensionless entries are read and inferred like a directory of part-files") {
+ val data = sampleDf((1, "Alice"), (2, "Bob"))
+ withArchiveFile() { archive =>
+ writeArchive(archive, Seq("part-00000" -> encodeFile(data)))
+ checkAnswer(read(archive.getCanonicalPath), data)
+ assert(inferredSchema(Seq(archive.getCanonicalPath)).fieldNames.toSet == Set("id", "name"),
+ "an extensionless entry should be inferred like a directory of part-files")
+ }
+ }
+
+ private def parquetArchiveTempDirs(prefix: String): Set[String] = {
+ val localDir = new File(Utils.getLocalDir(spark.sparkContext.getConf))
+ Option(localDir.listFiles()).getOrElse(Array.empty)
+ .filter(_.getName.startsWith(prefix)).map(_.getName).toSet
+ }
+
+ test("a corrupt archive cleans up its read temp dir rather than leaking it") {
+ // A corrupt archive throws before the read returns an iterator, but must not leak the temp dir.
+ withArchiveFile(corruptArchiveExtension) { archive =>
+ writeCorruptArchive(archive)
+ val before = parquetArchiveTempDirs("parquet-archive")
+ intercept[SparkException](read(archive.getCanonicalPath).collect())
+ assert((parquetArchiveTempDirs("parquet-archive") -- before).isEmpty,
+ "a corrupt archive leaked its read temp dir")
+ }
+ }
+
+ test("a corrupt archive cleans up its inference temp dir rather than leaking it") {
+ // Inference localizes entries too (readArchiveFooters), on a worker without a TaskContext; a
+ // corrupt archive throws during that eager localize and must not leak parquet-archive-infer.
+ withArchiveFile(corruptArchiveExtension) { archive =>
+ writeCorruptArchive(archive)
+ val before = parquetArchiveTempDirs("parquet-archive-infer")
+ intercept[SparkException](inferredSchema(Seq(archive.getCanonicalPath)))
+ assert((parquetArchiveTempDirs("parquet-archive-infer") -- before).isEmpty,
+ "a corrupt archive leaked its inference temp dir")
+ }
+ }
+
+ test("archive inference unions differing fields across entries with mergeSchema=true") {
+ // mergeSchema=true folds every entry's footer; over an archive, one unpacked entry at a time.
+ val withName = sampleDf((1, "Alice"), (2, "Bob"))
+ val idExtra = Seq((3, 30)).toDF("id", "extra")
+ val entries = Seq(entryName(0) -> encodeFile(withName), entryName(1) -> encodeFile(idExtra))
+ val merge = Map("mergeSchema" -> "true")
+ withArchiveFile() { archive =>
+ writeArchive(archive, entries)
+ val archiveSchema = inferredSchema(Seq(archive.getCanonicalPath), merge)
+ withTempDir { dir =>
+ entries.foreach { case (n, b) => Files.write(new File(dir, n).toPath, b) }
+ assert(archiveSchema.fieldNames.toSet == Set("id", "name", "extra"),
+ s"expected the union of entry fields, got $archiveSchema")
+ assert(archiveSchema == inferredSchema(Seq(dir.getCanonicalPath), merge),
+ s"archive mergeSchema inference diverged from a directory read; got $archiveSchema")
+ }
+ }
+ }
+
+ test("inference skips a missing archive among good ones (ignoreMissingFiles)") {
+ // Exercised on ParquetFileFormat.inferSchema(files) directly: inference now runs on the
+ // executor, so a hand-built missing status reaches the archive open and throws
+ // FileNotFoundException at task time. The public read path re-lists and drops the missing path
+ // before inference, so the missing-at-open window is only reachable through this entry point. A
+ // missing archive is governed by ignoreMissingFiles, not ignoreCorruptFiles.
+ withArchiveFile() { good =>
+ writeArchive(good, Seq(entryName(0) -> encodeFile(sampleDf((1, "Alice")))))
+ val missing = new File(good.getParentFile, s"missing.${archiveExtensions.head}")
+ val archives = Seq(
+ new FileStatus(good.length(), false, 0, 0, good.lastModified(), new Path(good.toURI)),
+ new FileStatus(1, false, 0, 0, 0, new Path(missing.toURI)))
+ def infer(ignoreMissing: Boolean, ignoreCorrupt: Boolean = false): Option[StructType] =
+ new ParquetFileFormat().inferSchema(spark, readOptions ++ Map(
+ "ignoreMissingFiles" -> ignoreMissing.toString,
+ "ignoreCorruptFiles" -> ignoreCorrupt.toString,
+ "mergeSchema" -> "true"), archives)
+ assert(infer(ignoreMissing = true).exists(_.fieldNames.contains("id")),
+ "expected the surviving archive's schema")
+ intercept[Exception](infer(ignoreMissing = false))
+ intercept[Exception](infer(ignoreMissing = false, ignoreCorrupt = true))
+ }
+ }
+
+}
+
+class ParquetTarArchiveReadSuite
+ extends ArchiveReadSuiteBase
+ with ParquetArchiveReadBase
+ with TarArchiveReadBase
+
+class ParquetZipArchiveReadSuite
+ extends ArchiveReadSuiteBase
+ with ParquetArchiveReadBase
+ with ZipArchiveReadBase
+
+class ParquetSevenZArchiveReadSuite
+ extends ArchiveReadSuiteBase
+ with ParquetArchiveReadBase
+ with SevenZArchiveReadBase
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/SupportsArchiveFormatSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/SupportsArchiveFormatSuite.scala
index 4247e82a1097..5905f0ab8d60 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/SupportsArchiveFormatSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/SupportsArchiveFormatSuite.scala
@@ -87,7 +87,7 @@ class SupportsArchiveFormatSuite extends SparkFunSuite {
* bit 3 is set and the local header's crc/size fields are zeroed, so the real values live only in
* the trailing data descriptor. `ZipArchiveInputStream` cannot stream such an entry -- it has no
* size to bound the read -- so `read` throws rather than yielding truncated bytes. This is the
- * non-streamable case the zip reader documents; `ZipArchiveOutputStream` cannot produce it
+ * non-streamable case for pure-streaming zip reads; `ZipArchiveOutputStream` cannot produce it
* (it rejects an unsized STORED entry, or rewrites the header when the sink is seekable), so the
* bytes are assembled by hand.
*/
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org