[SPARK-58612][SQL] Recurse into archives nested inside archives

[akshatshenoi-db]

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

[HyukjinKwon]

3 blocking, 4 non-blocking, 2 nits.
The recursion design is sound and lands in the right place, but the early-close path abandons nested containers and their spill dirs, and the depth guard is silently swallowed by ignoreCorruptFiles.

Design / architecture (1)

• sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:3058: The config doc's first sentence describes behavior no value of the config produces: with maxNestingDepth=1, reading an archive that contains a nested archive fails with MAX_ARCHIVE_DEPTH_EXCEEDED rather than reading no nested archives. The next sentence contradicts the first by saying recursion past the limit fails the read. -- see inline

Nits: 2 minor items (see inline comments).

Correctness (2)

• sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/SupportsArchiveFormat.scala:485: streamEntries.close() drops currentIter without closing it, so closing the iterator early abandons the nested container it holds -- and for zip/7z its spilled temp dir. Avro header inference is exactly that caller, on the driver, where no task-completion listener compensates. -- see inline
• sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryExecutionErrors.scala:945: The depth guard is silently swallowed under ignoreCorruptFiles: SparkRuntimeException is a RuntimeException, which shouldIgnoreCorruptFileException treats as ignorable, so hitting the zip-bomb bound returns partial results with only a warning. -- see inline

Suggestions (4)

• common/utils/src/main/resources/error/error-conditions.json:5736: MAX_ARCHIVE_DEPTH_EXCEEDED names the entry but not the archive it came from, unlike its sibling CANNOT_READ_ZIP_ENTRY, so on a multi-path scan the message does not say which file to look at. -- see inline
• sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/SupportsArchiveFormat.scala:284: The nested tar branch omits the try/catch that openTarStream has for exactly this construction, where GZIPInputStream reads the gzip header in its constructor. -- see inline
• sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/SupportsArchiveFormat.scala:283: The nested gzip test uses endsWith(".gz"), which is only safe because isArchiveFileName happens not to accept .gz; testing the two tar suffixes directly removes that hidden coupling. -- see inline
• sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/ArchiveReadSuiteBase.scala:453: The end-to-end depth-limit test only intercepts SparkException, so it would pass on any failure at that point, including the pre-PR mis-parse it is meant to rule out. -- see inline

Verification

Traced the two resource paths against master. advance() closes an exhausted currentIter
before opening the next entry, so a nested container is released as iteration walks past it, and
each level's closeFn deletes exactly the spill dir that level created; live spilled bytes are
therefore bounded by depth, not by how many sibling nested archives an archive holds. close() is
the asymmetric path: it sets currentIter = Iterator.empty without closing it, which was
resource-neutral on master (a parse iterator over a close-shielded stream) but now drops a whole
container.

For the guard: maxArchiveDepthExceeded returns SparkRuntimeException, which extends
RuntimeException, and DataSourceUtils.shouldIgnoreCorruptFileException matches every
RuntimeException, so FileScanRDD swallows the depth failure whenever
spark.sql.files.ignoreCorruptFiles=true -- confirmed against the PR's own corrupt-nested test,
which asserts exactly that swallow for a corrupt nested archive.

Also confirmed the non-nested path is unchanged: with an empty prefix PrefixedArchiveEntry.getName
returns the identical string, and SQLConf.get on the executor is the established idiom in this
package (HadoopFileLinesReader, FileFormatDataWriter), so the new config reaches tasks.

PR description suggestions

• Document: what happens when the depth limit is hit under spark.sql.files.ignoreCorruptFiles=true. The description presents the limit as a guard that "fails the read", but a SparkRuntimeException is ignorable, so with that flag on the read returns partial results instead.
• Document: that nested zip/7z entries are spilled to a temp file under the executor local dir. The description mentions the spill as an implementation detail of opening the container, but it is also a new disk-space and cleanup consideration for operators enabling the feature.