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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
