You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "akshatshenoi-db (via GitHub)" <gi...@apache.org> on 2026/08/06 05:45:01 UTC
[PR] [SPARK-58612][SQL] Recurse into archives nested inside archives [spark]
akshatshenoi-db opened a new pull request, #57814:
URL: https://github.com/apache/spark/pull/57814
### What changes were proposed in this pull request?
When an archive entry is itself an archive (`.tar`/`.tar.gz`/`.tgz`/`.zip`/`.7z`), recurse
into it and stream its entries, instead of handing the nested archive's raw bytes to the
format parser. Nested entries report the full `outer!/inner!/leaf` logical name.
- `readArchiveEntries` delegates to a depth-aware `streamEntries` that recurses when an
entry is an archive.
- A new `openNestedArchiveStream` opens a container from an already-open entry stream. Tar
reads the stream directly; zip and 7z read their index by seeking, which an entry stream
cannot do, so the entry is spilled to a local file and opened from there.
- New config `spark.sql.files.archive.reader.maxNestingDepth` (default 10) bounds the
recursion; exceeding it fails the read with `MAX_ARCHIVE_DEPTH_EXCEEDED`, guarding
against zip bombs and cyclic archives.
- `isArchiveFileName` is factored out of `isArchivePath` so an entry name is matched as a
plain string: an entry name is arbitrary, and `Path` would misparse one containing a
colon as a URI scheme.
### Why are the changes needed?
Archives commonly contain other archives, and today those inner archives are handed to the
format parser as opaque bytes, so their contents are unreadable. Recursing makes an archive
of archives behave like a directory tree of the data files it ultimately holds.
### Does this PR introduce any user-facing change?
Yes, gated by `spark.sql.files.archive.reader.enabled` (default false). With archive reading
enabled, a nested archive's entries are now read instead of its raw bytes, and the new
`spark.sql.files.archive.reader.maxNestingDepth` config (default 10) bounds recursion depth.
### How was this patch tested?
`ArchiveReadSuiteBase` gains nested-archive tests that run for every container (recursion
reads like a directory, a nested archive alongside plain entries, three levels of nesting,
empty nested, hidden entries in nested, the depth limit, and corrupt-nested under
`ignoreCorruptFiles`), plus a `writeNestedArchive` helper. `SupportsArchiveFormatSuite`
covers what only the engine sees: one archive nesting tar, zip, and 7z inside each other, a
nested archive whose name contains a colon, and the depth-limit boundary.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58612][SQL] Recurse into archives nested inside archives [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on code in PR #57814:
URL: https://github.com/apache/spark/pull/57814#discussion_r3732396350
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileScanRDD.scala:
##########
@@ -281,6 +281,10 @@ class FileScanRDD(
// Throw FileNotFoundException even if `ignoreCorruptFiles` is true
case e: FileNotFoundException if !ignoreMissingFiles => throw e
case e @ (_ : AccessControlException | _ : BlockMissingException) => throw e
+ // A resource-safety limit is not a corrupt file: skipping the rest of the archive
+ // would return partial data with no signal that the limit stopped the read.
+ case e: SparkRuntimeException
+ if e.getCondition == "MAX_ARCHIVE_DEPTH_EXCEEDED" => throw e
Review Comment:
This carve-out keeps the depth guard authoritative on the scan path, but schema inference is a separate path that still swallows it. `CSVDataSource.inferWithArchives`, `JsonDataSource`/`XmlDataSource` `skipInputOnError`, and `AvroUtils.firstArchiveEntrySchema` each catch `RuntimeException if ignoreCorruptFiles`, and `SparkRuntimeException` is a `RuntimeException` -- so with `ignoreCorruptFiles=true`, inferring a schema over a too-deeply-nested archive logs it as corrupt and infers from partial data, the exact silent-partial-data outcome this case prevents for scans. The `the depth limit still fails the read under ignoreCorruptFiles` test only exercises `read(...).collect()`, so the inference gap is untested.
Rather than repeat this `case` at all five sites, make the guard authoritative where every site already funnels: add it to `DataSourceUtils.shouldIgnoreCorruptFileException` so a `SparkThrowable` whose condition is `MAX_ARCHIVE_DEPTH_EXCEEDED` returns false. Every inference path and this `FileScanRDD` clause consult it, so this explicit `case` then becomes redundant and a future archive reader can't forget the carve-out. Please add an inference-path test alongside the scan-path one.
##########
.isaac/config.json:
##########
@@ -0,0 +1,3 @@
+{
+ "sync_reminder_last_shown": "2026-08-06"
Review Comment:
This file and `build/sbt-launch-1.12.13.jar.part` (a 0-byte partial-download artifact) look accidentally committed -- both are unrelated to the PR and neither is gitignored, so they'd ship to master. Please `git rm` both from the PR.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58612][SQL] Recurse into archives nested inside archives [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on code in PR #57814:
URL: https://github.com/apache/spark/pull/57814#discussion_r3727563135
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -3052,6 +3052,17 @@ object SQLConf {
.booleanConf
.createWithDefault(false)
+ val ARCHIVE_READER_MAX_NESTING_DEPTH =
+ buildConf("spark.sql.files.archive.reader.maxNestingDepth")
+ .doc("Maximum number of archive levels an archive reader opens. The top-level archive is " +
+ "depth 1, so 1 reads no nested archives and 2 reads one level of nesting. Recursing past " +
Review Comment:
`1 reads no nested archives` describes behavior the code never produces. At `maxDepth=1` the guard fires on the first nested archive, so the read fails instead of skipping it -- which the next line of this same doc already says. Your own test pins the failure (`ArchiveReadSuiteBase.scala:457-459`).
Worth deciding which behavior you want, because it changes whether operators have an escape hatch. If the limit is purely a zip-bomb bound, this line just needs to stop implying depth 1 is a no-recursion mode:
```suggestion
"depth 1, so 2 admits one level of nesting. Reading an archive nested deeper than " +
```
If you instead want a way to keep the pre-PR behavior for a specific read, that needs a separate knob (or a sentinel that skips nested archives rather than refusing them) -- I'd keep this config as a pure bound and not overload it. Line 3059 then reads `"this limit fails, guarding against zip bombs and cyclic archives.")`.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/SupportsArchiveFormatSuite.scala:
##########
@@ -32,14 +32,17 @@ import org.apache.commons.compress.archivers.zip.{ZipArchiveEntry, ZipArchiveOut
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.Path
-import org.apache.spark.{SparkFunSuite, SparkRuntimeException, TaskContext, TaskContextImpl}
+import org.apache.spark.{SparkRuntimeException, TaskContext, TaskContextImpl}
+import org.apache.spark.sql.QueryTest
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.test.SharedSparkSession
/**
* Unit tests for the streaming [[SupportsArchiveFormat]] engine: `isArchivePath` dispatch and
* `readArchiveEntries` (entry ordering, gzip handling, dir/dotfile skipping, lazy advance, the
* non-closing entry stream, and cleanup). Nothing here touches local disk -- entries are streams.
*/
-class SupportsArchiveFormatSuite extends SparkFunSuite {
+class SupportsArchiveFormatSuite extends QueryTest with SharedSparkSession {
Review Comment:
The class doc just above (line 43) still says `Nothing here touches local disk -- entries are streams`, which stopped being true when the nested tests started exercising the zip/7z spill branch. Worth updating that sentence to say nested zip/7z entries are spilled to a temp file.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryExecutionErrors.scala:
##########
@@ -942,6 +942,12 @@ private[sql] object QueryExecutionErrors extends QueryErrorsBase with ExecutionE
messageParameters = Map("entry" -> entry, "path" -> path))
}
+ def maxArchiveDepthExceeded(entry: String, maxDepth: Int): SparkRuntimeException = {
Review Comment:
`SparkRuntimeException` extends `RuntimeException`, and `DataSourceUtils.shouldIgnoreCorruptFileException` matches every `RuntimeException` (`DataSourceUtils.scala:237`). So with `spark.sql.files.ignoreCorruptFiles=true`, exceeding `maxNestingDepth` is logged as a corrupt file and the rest of the archive is skipped -- the query succeeds with partial data and no signal that the zip-bomb guard fired. Your own `ignoreCorruptFiles=true` nested test asserts exactly that swallow for corrupt bytes (`ArchiveReadSuiteBase.scala:470-471`).
A resource-safety limit is not a corrupt-file condition, so I'd exclude it explicitly rather than change the exception type. `FileScanRDD.scala:283` already has that carve-out (`AccessControlException`, `BlockMissingException`); adding a case ahead of the `ignoreCorruptFiles` clause that rethrows a `SparkThrowable` whose condition is `MAX_ARCHIVE_DEPTH_EXCEEDED` keeps the guard authoritative. Whichever way you go, please cover it: `ignoreCorruptFiles=true` plus a depth-limited nested archive should still fail.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/SupportsArchiveFormat.scala:
##########
@@ -255,6 +266,47 @@ object SupportsArchiveFormat {
override def close(): Unit = closeFn()
}
+ /**
+ * Opens an archive nested inside an already-open entry, reading from that entry's stream.
+ *
+ * @param name the entry's name, which selects the container by extension
+ * @param in the entry's byte stream, positioned at the entry start
+ * @param conf Hadoop configuration used to open a spilled container
+ * @return the nested archive's entries; closing it releases the container and any spilled file
+ */
+ private def openNestedArchiveStream(
+ name: String,
+ in: InputStream,
+ conf: Configuration): ArchiveEntries = {
+ val n = name.toLowerCase(Locale.ROOT)
+ if (n.endsWith(".tar") || n.endsWith(".tar.gz") || n.endsWith(".tgz")) {
+ val gzipped = n.endsWith(".gz") || n.endsWith(".tgz")
Review Comment:
`endsWith(".gz")` also matches a plain `data.gz`, which is not a tar. That name only fails to reach here because `isArchiveFileName` doesn't accept `.gz` -- an invisible coupling between the two methods. If `.gz` is ever added there, a gzipped non-tar gets handed to `TarArchiveInputStream` and fails confusingly.
```suggestion
val gzipped = n.endsWith(".tar.gz") || n.endsWith(".tgz")
```
This reaches the same decision for every name that can get here, and stands on its own. (The asymmetry with `openTarStream`'s `.tgz`-only check is fine -- that path gets `.tar.gz` decompressed by `CodecStreams` first, and this one has no such layer.)
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/SupportsArchiveFormat.scala:
##########
@@ -255,6 +266,47 @@ object SupportsArchiveFormat {
override def close(): Unit = closeFn()
}
+ /**
+ * Opens an archive nested inside an already-open entry, reading from that entry's stream.
+ *
+ * @param name the entry's name, which selects the container by extension
+ * @param in the entry's byte stream, positioned at the entry start
+ * @param conf Hadoop configuration used to open a spilled container
+ * @return the nested archive's entries; closing it releases the container and any spilled file
+ */
+ private def openNestedArchiveStream(
+ name: String,
+ in: InputStream,
+ conf: Configuration): ArchiveEntries = {
+ val n = name.toLowerCase(Locale.ROOT)
+ if (n.endsWith(".tar") || n.endsWith(".tar.gz") || n.endsWith(".tgz")) {
+ val gzipped = n.endsWith(".gz") || n.endsWith(".tgz")
+ val tar = new TarArchiveInputStream(if (gzipped) new GZIPInputStream(in) else in)
Review Comment:
`openTarStream` wraps this same construction in a try/catch and explains why in a comment (line 246-256): `GZIPInputStream` reads the gzip header in its constructor, so a corrupt member throws here. Nothing unrecoverable leaks on this path today because the entry stream belongs to the enclosing container, but the two sibling paths reading differently is the kind of asymmetry that turns into a leak after the next edit. Mirroring the guard (or a short comment saying the entry stream is not this method's to close) would settle it.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/SupportsArchiveFormat.scala:
##########
@@ -367,7 +419,28 @@ object SupportsArchiveFormat {
conf: Configuration,
ignoredPathSegmentRegex: Pattern = HadoopFSUtils.defaultIgnoredPathSegmentRegexPattern)(
parseEntry: (ArchiveEntry, InputStream) => Iterator[T]): Iterator[T] = {
- val archive = openArchiveStream(path, conf)
+ val maxDepth = SQLConf.get.getConf(SQLConf.ARCHIVE_READER_MAX_NESTING_DEPTH)
+ streamEntries(
+ openArchiveStream(path, conf), conf, "", 1, maxDepth, ignoredPathSegmentRegex)(parseEntry)
+ }
+
+ /**
+ * Streams one open `archive`, applying `parseEntry` to each non-skipped entry. An entry that is
+ * itself an archive recurses instead, bounded by `maxDepth`.
+ *
+ * @param namePrefix nested-entry name prefix ("" at the top, `inner!/` under a nested archive),
Review Comment:
The recursive call passes `s"${entry.getName}!/"` where `entry` already carries the incoming prefix, so the value is `outer!/` inside the first nested archive, never `inner!/`.
```suggestion
* @param namePrefix nested-entry name prefix ("" at the top, `outer!/` inside the first nested
* archive and `outer!/inner!/` one level deeper), composing the full
```
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/SupportsArchiveFormat.scala:
##########
@@ -401,8 +474,20 @@ object SupportsArchiveFormat {
done = true
cleanup()
} else {
+ val entry = new PrefixedArchiveEntry(next._1, namePrefix)
// Parse the entry stream; any unread remainder is skipped when the archive advances.
- currentIter = parseEntry(next._1, CloseShieldInputStream.wrap(next._2))
+ val stream = CloseShieldInputStream.wrap(next._2)
+ currentIter = if (isArchiveFileName(next._1.getName)) {
+ if (depth >= maxDepth) {
+ throw QueryExecutionErrors.maxArchiveDepthExceeded(entry.getName, maxDepth)
+ }
+ streamEntries(
+ openNestedArchiveStream(next._1.getName, stream, conf),
Review Comment:
`currentIter` can now be a nested `streamEntries` that owns a container and, on the zip/7z path, a spilled temp dir -- but `close()` at line 505-509 just does `currentIter = Iterator.empty` and then `cleanup()`, which closes only *this* level's archive. So an early close leaks the nested container and its spill dir. `advance()` already gets this right at line 464-467.
`AvroUtils.firstArchiveEntrySchema` is that caller: it takes one entry's schema and closes (`AvroUtils.scala:288-296`). It runs on the driver, so there is no task-completion listener to pick up the slack, and the spill dir then survives until the shutdown-delete hook runs.
The fix is in `close()` at line 505: match `currentIter` against `Closeable` and close it before clearing the field, exactly as `advance()` does at 464-467. A test would fit alongside the existing temp-dir leak cases in `ArchiveReadSuiteBase.scala:711-731` -- read one row from a nested zip/7z archive, close, then assert no new dir under the prefix.
##########
common/utils/src/main/resources/error/error-conditions.json:
##########
@@ -5731,6 +5731,12 @@
],
"sqlState" : "42000"
},
+ "MAX_ARCHIVE_DEPTH_EXCEEDED" : {
+ "message" : [
+ "Cannot read archive entry <entry>: it is nested more than <maxDepth> archives deep."
Review Comment:
The composed entry name (`l2.tar!/l3.tar`) is relative to the top-level archive, so on a scan over many paths this message doesn't identify the file. `CANNOT_READ_ZIP_ENTRY` (line 831) includes both, and matching it keeps the two archive conditions consistent:
```suggestion
"Cannot read archive entry <entry> in archive <path>: it is nested more than <maxDepth> archives deep."
```
That needs the path threaded to `maxArchiveDepthExceeded` -- `streamEntries` doesn't currently carry it, so this is a real (if small) plumbing change rather than a message-only edit. Non-blocking, but the error is much more actionable with it.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/ArchiveReadSuiteBase.scala:
##########
@@ -357,6 +384,99 @@ trait ArchiveReadSuiteBase extends QueryTest with SharedSparkSession {
}
}
+ // ----- nested-archive tests ------------------------------------------------
+
+ test("a nested archive is recursed into and reads like a directory of its entries") {
+ val parts = Seq(sampleDf((1, "Alice"), (2, "Bob")), sampleDf((3, "Carol")))
+ val entries = parts.zipWithIndex.map { case (p, i) => entryName(i) -> encodeFile(p) }
+ withArchiveFile() { archive =>
+ writeNestedArchive(archive, entries)
+ checkAnswer(read(archive.getCanonicalPath), parts.reduce(_ union _))
+ }
+ }
+
+ test("a nested archive alongside plain entries contributes both") {
+ val nested = sampleDf((1, "Alice"), (2, "Bob"))
+ val plain = sampleDf((3, "Carol"))
+ val ext = archiveExtensions.head
+ withArchiveFile() { archive =>
+ writeArchive(archive, Seq(
+ entryName(0) -> encodeFile(plain),
+ s"nested.$ext" -> archiveBytes(Seq(entryName(1) -> encodeFile(nested)), ext)))
+ checkAnswer(read(archive.getCanonicalPath), plain.union(nested))
+ }
+ }
+
+ test("three levels of nested archives are recursed into") {
+ val data = sampleDf((1, "Alice"), (2, "Bob"))
+ val ext = archiveExtensions.head
+ // archive -> level2.<ext> -> level3.<ext> -> the data file.
+ val level3 = archiveBytes(Seq(entryName(0) -> encodeFile(data)), ext)
+ val level2 = archiveBytes(Seq(s"level3.$ext" -> level3), ext)
+ withArchiveFile() { archive =>
+ writeArchive(archive, Seq(s"level2.$ext" -> level2))
+ checkAnswer(read(archive.getCanonicalPath), data)
+ }
+ }
+
+ test("an empty nested archive contributes no rows") {
+ withArchiveFile() { archive =>
+ writeNestedArchive(archive, Seq.empty)
+ checkAnswer(read(archive.getCanonicalPath), Seq.empty[Row])
+ }
+ }
+
+ test("hidden entries inside a nested archive are skipped") {
+ val kept = sampleDf((1, "Alice"), (2, "Bob"))
+ val ext = archiveExtensions.head
+ val innerBytes = archiveBytes(
+ Seq("_SUCCESS" -> "marker".getBytes, entryName(0) -> encodeFile(kept)), ext)
+ withArchiveFile() { archive =>
+ writeArchive(archive, Seq(s"nested.$ext" -> innerBytes))
+ checkAnswer(read(archive.getCanonicalPath), kept)
+ }
+ }
+
+ test("the depth limit bounds how deeply nested archives are read") {
+ val data = sampleDf((1, "Alice"), (2, "Bob"))
+ val ext = archiveExtensions.head
+ // Three archive levels: the top archive plus two nested ones.
+ val level3 = archiveBytes(Seq(entryName(0) -> encodeFile(data)), ext)
+ val level2 = archiveBytes(Seq(s"level3.$ext" -> level3), ext)
+ withArchiveFile() { archive =>
+ writeArchive(archive, Seq(s"level2.$ext" -> level2))
+ // 3 admits exactly these three levels.
+ withSQLConf(SQLConf.ARCHIVE_READER_MAX_NESTING_DEPTH.key -> "3") {
+ checkAnswer(read(archive.getCanonicalPath), data)
+ }
+ // 2 stops before the innermost archive is opened.
+ withSQLConf(SQLConf.ARCHIVE_READER_MAX_NESTING_DEPTH.key -> "2") {
Review Comment:
Both depth assertions here (and at line 457) accept any `SparkException`, so they don't distinguish the depth guard from an unrelated failure at the same point. `checkError` on the wrapped cause pins `MAX_ARCHIVE_DEPTH_EXCEEDED` and reads no longer than the current `intercept`.
Non-blocking, since `SupportsArchiveFormatSuite` does pin the condition and both parameters at the engine level -- this would just make the end-to-end test assert what its name claims.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org