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 b3a5089e5222 [SPARK-58394][SQL] Use ZipFile instead of
ZipArchiveInputStream for zip archives
b3a5089e5222 is described below
commit b3a5089e52225ba8206e9bd9faae029c21f99136
Author: akshatshenoi-db <[email protected]>
AuthorDate: Tue Aug 4 20:14:08 2026 +0800
[SPARK-58394][SQL] Use ZipFile instead of ZipArchiveInputStream for zip
archives
### What changes were proposed in this pull request?
commons-compress recommends preferring the random-access `ZipFile` over the
streaming `ZipArchiveInputStream`: the streaming reader parses local file
headers sequentially (so it can return entries absent from the central
directory, duplicate names, and incomplete metadata) and cannot handle STORED
entries sized only by a trailing data descriptor. `ZipFile` reads the central
directory first, giving correct metadata and reliable extraction across all
compression methods.
This reuses the seekable-channel abstraction already built for 7z
(`HadoopSeekableByteChannel`): `openArchiveStream` now returns an
`Iterator[(ArchiveEntry, InputStream)]`, letting each container expose its
natural shape -- tar/7z advance one forward cursor, while zip serves each
entry's own `getInputStream` from the central directory.
`ZipFile#canReadEntryData` flags encrypted / unsupported-method entries with a
clear `CANNOT_READ_ZIP_ENTRY` error. The tar and 7z paths are unchanged [...]
This is part of the archive-reader series (SPARK-57135 CSV, SPARK-57321
CSV-infer, SPARK-57419 JSON, SPARK-57478 text, SPARK-57479 XML, SPARK-57481
Avro, SPARK-57705 zip container).
### Why are the changes needed?
The streaming reader rejects valid zips -- notably STORED entries whose
size lives only in a trailing data descriptor -- and exposes
incomplete/duplicated entry metadata. Reading the central directory fixes both,
and the per-entry byte offsets it exposes are a prerequisite for future zip
parallelization.
### Does this PR introduce _any_ user-facing change?
Yes, gated by `spark.sql.files.archive.reader.enabled` (default false): zip
archives that previously failed to read (e.g. a STORED entry sized by a
trailing data descriptor) now read successfully.
### How was this patch tested?
Updated `SupportsArchiveFormatSuite`: the previously-non-streamable
STORED+data-descriptor zip now reads successfully (assertion flipped); added
duplicate-name, encrypted-entry (asserting `CANNOT_READ_ZIP_ENTRY` via
`checkError`), and corrupt-archive cases. The existing tar and 7z cases are
unchanged.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code
Closes #57588 from akshatshenoi-db/archive-zipfile.
Authored-by: akshatshenoi-db <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>
(cherry picked from commit 548b828781d3f0ea4d3392a378f846da8d43d1dd)
Signed-off-by: Wenchen Fan <[email protected]>
---
.../src/main/resources/error/error-conditions.json | 6 +
.../spark/sql/errors/QueryExecutionErrors.scala | 6 +
.../datasources/SupportsArchiveFormat.scala | 177 ++++++++++++++-------
.../datasources/BinaryFileArchiveReadBase.scala | 16 --
.../datasources/SupportsArchiveFormatSuite.scala | 127 ++++++++++++---
5 files changed, 241 insertions(+), 91 deletions(-)
diff --git a/common/utils/src/main/resources/error/error-conditions.json
b/common/utils/src/main/resources/error/error-conditions.json
index 69dfa0cc68cb..dd43b76ca403 100644
--- a/common/utils/src/main/resources/error/error-conditions.json
+++ b/common/utils/src/main/resources/error/error-conditions.json
@@ -819,6 +819,12 @@
],
"sqlState" : "22007"
},
+ "CANNOT_READ_ZIP_ENTRY" : {
+ "message" : [
+ "Cannot read zip entry <entry> in archive <path>: encrypted or an
unsupported compression method."
+ ],
+ "sqlState" : "KD003"
+ },
"CANNOT_RECOGNIZE_HIVE_TYPE" : {
"message" : [
"Cannot recognize hive type string: <fieldType>, column: <fieldName>.
The specified data type for the field cannot be recognized by Spark SQL. Please
check the data type of the specified field and ensure that it is a valid Spark
SQL data type. Refer to the Spark SQL documentation for a list of valid data
types and their format. If the data type is correct, please ensure that you are
using a supported version of Spark SQL."
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryExecutionErrors.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryExecutionErrors.scala
index c6bd11ae7167..2ef972a9b20a 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryExecutionErrors.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryExecutionErrors.scala
@@ -936,6 +936,12 @@ private[sql] object QueryExecutionErrors extends
QueryErrorsBase with ExecutionE
cause = e)
}
+ def cannotReadZipEntry(entry: String, path: String): SparkRuntimeException =
{
+ new SparkRuntimeException(
+ errorClass = "CANNOT_READ_ZIP_ENTRY",
+ messageParameters = Map("entry" -> entry, "path" -> path))
+ }
+
def cannotCreateColumnarReaderError(): Throwable = {
new SparkException(
errorClass = "_LEGACY_ERROR_TEMP_2065",
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 1c6c63582039..ff25f957c29d 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
@@ -24,12 +24,13 @@ import java.util.Locale
import java.util.regex.Pattern
import java.util.zip.GZIPInputStream
+import scala.jdk.CollectionConverters._
import scala.util.control.NonFatal
-import org.apache.commons.compress.archivers.{ArchiveEntry, ArchiveInputStream}
-import org.apache.commons.compress.archivers.sevenz.{SevenZArchiveEntry,
SevenZFile}
+import org.apache.commons.compress.archivers.ArchiveEntry
+import org.apache.commons.compress.archivers.sevenz.SevenZFile
import org.apache.commons.compress.archivers.tar.TarArchiveInputStream
-import org.apache.commons.compress.archivers.zip.ZipArchiveInputStream
+import org.apache.commons.compress.archivers.zip.ZipFile
import org.apache.commons.io.ByteOrderMark
import org.apache.commons.io.input.{BOMInputStream, CloseShieldInputStream}
import org.apache.hadoop.conf.Configuration
@@ -41,6 +42,7 @@ import org.apache.spark.{SparkEnv, TaskContext}
import org.apache.spark.internal.Logging
import org.apache.spark.paths.SparkPath
import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.errors.QueryExecutionErrors
import org.apache.spark.util.{HadoopFSUtils, Utils}
/**
@@ -203,28 +205,20 @@ object SupportsArchiveFormat {
name.endsWith(".zip") || name.endsWith(".7z")
}
+ /** An archive's entries as lazy `(entry, stream)` pairs; closing releases
the container. */
+ private type ArchiveEntries = Iterator[(ArchiveEntry, InputStream)] with
Closeable
+
/**
- * Opens the archive at `path` as a commons-compress stream, selecting the
container by extension.
+ * Opens the archive at `path`, selecting the container by extension and
exposing its entries as
+ * `(entry, stream)` pairs.
*/
- private def openArchiveStream(
- path: Path,
- conf: Configuration): ArchiveInputStream[_ <: ArchiveEntry] = {
+ private def openArchiveStream(path: Path, conf: Configuration):
ArchiveEntries = {
val name = path.getName.toLowerCase(Locale.ROOT)
name match {
case n if n.endsWith(".tar") || n.endsWith(".tar.gz") ||
n.endsWith(".tgz") =>
- val base = CodecStreams.createInputStreamWithCloseResource(conf, path)
- try {
- // GZIPInputStream reads the gzip header in its constructor, so a
corrupt archive can
- // throw here -- after `base` is already open -- and `base` must not
leak.
- val tarBytes = if (n.endsWith(".tgz")) new GZIPInputStream(base)
else base
- new TarArchiveInputStream(tarBytes)
- } catch {
- case NonFatal(e) =>
- try base.close() catch { case NonFatal(_) => }
- throw e
- }
+ openTarStream(path, conf)
case n if n.endsWith(".zip") =>
- new
ZipArchiveInputStream(CodecStreams.createInputStreamWithCloseResource(conf,
path))
+ openZipStream(path, conf)
case n if n.endsWith(".7z") =>
openSevenZStream(path, conf)
case _ =>
@@ -233,25 +227,111 @@ object SupportsArchiveFormat {
}
}
- /**
- * Opens a `.7z` archive by seeking.
- *
- * @param path the archive path
- * @param conf Hadoop configuration used to open the archive
- * @return the archive's entries as an [[ArchiveInputStream]] cursor
- */
- private def openSevenZStream(
- path: Path,
- conf: Configuration): ArchiveInputStream[_ <: ArchiveEntry] = {
+ /** Opens a `.tar`/`.tar.gz`/`.tgz` archive, streaming its entries through
one forward cursor. */
+ private def openTarStream(path: Path, conf: Configuration): ArchiveEntries =
{
+ val gzipped = path.getName.toLowerCase(Locale.ROOT).endsWith(".tgz")
+ val base = CodecStreams.createInputStreamWithCloseResource(conf, path)
+ try {
+ // GZIPInputStream reads the gzip header in its constructor, so a
corrupt archive can throw
+ // here -- after `base` is already open -- and `base` must not leak.
+ val tar = new TarArchiveInputStream(if (gzipped) new
GZIPInputStream(base) else base)
+ val entries = Iterator.continually(tar.getNextEntry).takeWhile(_ != null)
+ .map((_, tar: InputStream))
+ closeable(entries, () => tar.close())
+ } catch {
+ case NonFatal(e) =>
+ try base.close() catch { case NonFatal(_) => }
+ throw e
+ }
+ }
+
+ /** Pairs an entry iterator with the resource it reads from, closed when the
caller is done. */
+ private def closeable(
+ entries: Iterator[(ArchiveEntry, InputStream)],
+ closeFn: () => Unit): ArchiveEntries =
+ new Iterator[(ArchiveEntry, InputStream)] with Closeable {
+ override def hasNext: Boolean = entries.hasNext
+ override def next(): (ArchiveEntry, InputStream) = entries.next()
+ override def close(): Unit = closeFn()
+ }
+
+ /** Opens a `.7z` archive by seeking. */
+ private def openSevenZStream(path: Path, conf: Configuration):
ArchiveEntries = {
+ val fs = path.getFileSystem(conf)
+ val length = fs.getFileStatus(path).getLen
+ var channel: SeekableByteChannel = null
+ var sevenZ: SevenZFile = null
+ try {
+ channel = new HadoopSeekableByteChannel(fs.open(path), length)
+ sevenZ = SevenZFile.builder().setSeekableByteChannel(channel).get()
+ // SevenZFile is a forward cursor: one stream serves whichever entry
getNextEntry selects.
+ val entryStream = new SevenZEntryInputStream(sevenZ)
+ val entries = Iterator.continually(sevenZ.getNextEntry).takeWhile(_ !=
null)
+ .map((_, entryStream))
+ closeable(entries, () => {
+ try {
+ sevenZ.close()
+ } finally {
+ channel.close()
+ }
+ })
+ } catch {
+ case NonFatal(e) =>
+ Utils.closeQuietly(sevenZ)
+ Utils.closeQuietly(channel)
+ throw e
+ }
+ }
+
+ /** Opens a `.zip` archive by seeking, reading the central directory first.
*/
+ private def openZipStream(path: Path, conf: Configuration): ArchiveEntries =
{
val fs = path.getFileSystem(conf)
- val in = fs.open(path)
+ val length = fs.getFileStatus(path).getLen
+ var channel: SeekableByteChannel = null
+ var zipFile: ZipFile = null
try {
- val channel = new HadoopSeekableByteChannel(in,
fs.getFileStatus(path).getLen)
- new SevenZArchiveInputStream(
- SevenZFile.builder().setSeekableByteChannel(channel).get(), channel)
+ channel = new HadoopSeekableByteChannel(fs.open(path), length)
+ zipFile = ZipFile.builder().setSeekableByteChannel(channel).get()
+ var current: InputStream = null
+ def closeCurrentStream(): Unit = if (current != null) {
+ try current.close() catch { case NonFatal(_) => }
+ current = null
+ }
+ // Open each entry's stream lazily, on first read. An entry the engine
skips (directory or
+ // dotfile) is never read, so it is never opened and its readability
never checked -- else a
+ // skipped-but-encrypted entry would throw CANNOT_READ_ZIP_ENTRY, unlike
the streaming reader
+ // ZipFile replaced. Opening an entry releases the previous one's
inflater.
+ val entries = zipFile.getEntries.asScala.map { entry =>
+ val stream = new InputStream {
+ private var delegate: InputStream = _
+ private def open(): InputStream = {
+ if (delegate == null) {
+ closeCurrentStream()
+ if (!zipFile.canReadEntryData(entry)) {
+ throw QueryExecutionErrors.cannotReadZipEntry(entry.getName,
path.toString)
+ }
+ delegate = zipFile.getInputStream(entry)
+ current = delegate
+ }
+ delegate
+ }
+ override def read(): Int = open().read()
+ override def read(b: Array[Byte], off: Int, len: Int): Int =
open().read(b, off, len)
+ }
+ (entry: ArchiveEntry, stream)
+ }
+ closeable(entries, () => {
+ try {
+ closeCurrentStream()
+ zipFile.close()
+ } finally {
+ channel.close()
+ }
+ })
} catch {
case NonFatal(e) =>
- try in.close() catch { case NonFatal(_) => }
+ Utils.closeQuietly(zipFile)
+ Utils.closeQuietly(channel)
throw e
}
}
@@ -312,17 +392,17 @@ object SupportsArchiveFormat {
case c: Closeable => try c.close() catch { case NonFatal(_) => }
case _ =>
}
- var entry = archive.getNextEntry
- while (entry != null && shouldSkipEntry(entry,
ignoredPathSegmentRegex)) {
- entry = archive.getNextEntry
+ var next: (ArchiveEntry, InputStream) = null
+ while (next == null && archive.hasNext) {
+ val entry = archive.next()
+ if (!shouldSkipEntry(entry._1, ignoredPathSegmentRegex)) next =
entry
}
- if (entry == null) {
+ if (next == null) {
done = true
cleanup()
} else {
- // CloseShieldInputStream ignores close(), so a parser closing its
input does not close
- // the archive; any unread remainder is skipped by getNextEntry()
when advancing.
- currentIter = parseEntry(entry,
CloseShieldInputStream.wrap(archive))
+ // Parse the entry stream; any unread remainder is skipped when
the archive advances.
+ currentIter = parseEntry(next._1,
CloseShieldInputStream.wrap(next._2))
}
}
}
@@ -447,18 +527,3 @@ private class SevenZEntryInputStream(sevenZ: SevenZFile)
extends InputStream {
override def read(): Int = sevenZ.read()
override def read(b: Array[Byte], off: Int, len: Int): Int = sevenZ.read(b,
off, len)
}
-
-/**
- * Adapts a [[SevenZFile]] to the [[ArchiveInputStream]] cursor the engine
consumes. Closing it
- * closes both the `SevenZFile` and the channel it reads from, which
`SevenZFile` does not own.
- *
- * @param sevenZ the 7z file to adapt
- * @param channel the channel `sevenZ` reads from, closed alongside it
- */
-private class SevenZArchiveInputStream(sevenZ: SevenZFile, channel:
SeekableByteChannel)
- extends ArchiveInputStream[SevenZArchiveEntry](new
SevenZEntryInputStream(sevenZ), "UTF-8") {
-
- override def getNextEntry(): SevenZArchiveEntry = sevenZ.getNextEntry
-
- override def close(): Unit = try sevenZ.close() finally channel.close()
-}
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/BinaryFileArchiveReadBase.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/BinaryFileArchiveReadBase.scala
index bcbfb89d1241..c115f9bafa53 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/BinaryFileArchiveReadBase.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/BinaryFileArchiveReadBase.scala
@@ -50,14 +50,6 @@ trait BinaryFileArchiveReadBase extends QueryTest with
SharedSparkSession {
/** Extension of the archive [[writeCorruptArchive]] produces (corruption is
format-specific). */
protected def corruptArchiveExtension: String
- /**
- * Whether the container reports each entry's size up front. Streaming zip
- * (`ZipArchiveInputStream`) does not -- an entry sized only by a trailing
data descriptor reads
- * back as `-1` -- so the per-entry `length` tests are skipped for zip until
it moves to
- * `ZipFile`. tar and 7z report real sizes.
- */
- protected def entrySizeKnown: Boolean = true
-
override def sparkConf: SparkConf =
super.sparkConf.set(SQLConf.ARCHIVE_FORMAT_READER_ENABLED.key, "true")
@@ -98,7 +90,6 @@ trait BinaryFileArchiveReadBase extends QueryTest with
SharedSparkSession {
}
test("wholeFile=false sources path and length from each entry, modtime from
the parent") {
- assume(entrySizeKnown)
withArchiveFile() { archive =>
writeArchive(archive, Seq("a.bin" -> bytes("aaa"), "b.bin" ->
bytes("bbbb")))
val rows = read(archive.getCanonicalPath, Map("wholeFile" -> "false"))
@@ -136,7 +127,6 @@ trait BinaryFileArchiveReadBase extends QueryTest with
SharedSparkSession {
}
test("wholeFile=false enforces SOURCES_BINARY_FILE_MAX_LENGTH per entry") {
- assume(entrySizeKnown)
withArchiveFile() { archive =>
writeArchive(archive, Seq("big.bin" -> bytes("0123456789")))
withSQLConf(SQLConf.SOURCES_BINARY_FILE_MAX_LENGTH.key -> "4") {
@@ -149,7 +139,6 @@ trait BinaryFileArchiveReadBase extends QueryTest with
SharedSparkSession {
}
test("wholeFile=false honors length filter pushdown against each entry") {
- assume(entrySizeKnown)
withArchiveFile() { archive =>
writeArchive(archive, Seq("a.bin" -> bytes("aaa"), "b.bin" ->
bytes("bbbb")))
// Entry lengths are 3 and 4; the filter selects per entry, not against
the archive size.
@@ -220,11 +209,6 @@ class BinaryFileTarArchiveReadSuite extends
BinaryFileArchiveReadBase with TarAr
class BinaryFileZipArchiveReadSuite extends BinaryFileArchiveReadBase with
ZipArchiveTestUtils {
override protected def corruptArchiveExtension: String = "zip"
-
- // Streaming `ZipArchiveInputStream` cannot report an entry's size before
reading it, so the
- // per-entry `length` tests are skipped for zip. Remove this override once
zip reads move to
- // `ZipFile`, which exposes entry sizes from the central directory.
- override protected def entrySizeKnown: Boolean = false
}
class BinaryFileSevenZArchiveReadSuite
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 b63b9ac31031..0c362afd31d8 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
@@ -19,6 +19,7 @@ package org.apache.spark.sql.execution.datasources
import java.io.{ByteArrayOutputStream, Closeable, File, FileOutputStream,
InputStream, OutputStream}
import java.nio.charset.StandardCharsets
+import java.nio.file.Files
import java.util.Properties
import java.util.regex.Pattern
import java.util.zip.GZIPOutputStream
@@ -31,7 +32,7 @@ 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, TaskContext, TaskContextImpl}
+import org.apache.spark.{SparkFunSuite, SparkRuntimeException, TaskContext,
TaskContextImpl}
/**
* Unit tests for the streaming [[SupportsArchiveFormat]] engine:
`isArchivePath` dispatch and
@@ -82,25 +83,29 @@ class SupportsArchiveFormatSuite extends SparkFunSuite {
} finally out.close()
}
+ /** Appends `v` as an unsigned 2-byte little-endian integer -- one of ZIP's
header fields. */
+ private def u16(out: ByteArrayOutputStream, v: Int): Unit = {
+ out.write(v & 0xFF); out.write((v >>> 8) & 0xFF)
+ }
+
+ /** Appends `v` as an unsigned 4-byte little-endian integer -- one of ZIP's
header fields. */
+ private def u32(out: ByteArrayOutputStream, v: Long): Unit = {
+ out.write((v & 0xFF).toInt); out.write(((v >>> 8) & 0xFF).toInt)
+ out.write(((v >>> 16) & 0xFF).toInt); out.write(((v >>> 24) & 0xFF).toInt)
+ }
+
/**
* Writes a zip with one STORED (uncompressed) entry that uses a data
descriptor: general-purpose
* 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 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.
+ * the trailing data descriptor.
*/
private def writeStoredEntryWithDataDescriptor(file: File, name: String,
body: String): Unit = {
val nameBytes = name.getBytes(StandardCharsets.UTF_8)
val data = body.getBytes(StandardCharsets.UTF_8)
val crc = { val c = new java.util.zip.CRC32(); c.update(data); c.getValue }
val out = new ByteArrayOutputStream()
- def u16(v: Int): Unit = { out.write(v & 0xFF); out.write((v >>> 8) & 0xFF)
}
- def u32(v: Long): Unit = {
- out.write((v & 0xFF).toInt); out.write(((v >>> 8) & 0xFF).toInt)
- out.write(((v >>> 16) & 0xFF).toInt); out.write(((v >>> 24) &
0xFF).toInt)
- }
+ def u16(v: Int): Unit = this.u16(out, v)
+ def u32(v: Long): Unit = this.u32(out, v)
// Local file header: GP bit 3 set (data descriptor), STORED method, sizes
zeroed here.
val localHeaderOffset = out.size()
u32(0x04034b50L); u16(10); u16(0x0008); u16(0); u16(0); u16(0)
@@ -120,7 +125,52 @@ class SupportsArchiveFormatSuite extends SparkFunSuite {
u32(0x06054b50L); u16(0); u16(0); u16(1); u16(1)
u32(cdSize.toLong); u32(cdOffset.toLong); u16(0)
val fos = new FileOutputStream(file)
- try fos.write(out.toByteArray) finally fos.close()
+ try {
+ fos.write(out.toByteArray)
+ } finally {
+ fos.close()
+ }
+ }
+
+ /**
+ * Writes a zip whose single STORED entry sets the encryption flag
(general-purpose bit 0), so
+ * `ZipFile#canReadEntryData` returns false. Assembled by hand: no writer we
depend on emits an
+ * encryption-flagged entry (commons-compress's `ZipArchiveOutputStream`
cannot). The bytes are
+ * not actually encrypted -- only the flag is set, which alone drives the
read path under test.
+ * ZIP is a sequence of little-endian records, each led by a 4-byte
signature.
+ */
+ private def writeEncryptedEntry(file: File, name: String, body: String):
Unit = {
+ val nameBytes = name.getBytes(StandardCharsets.UTF_8)
+ val data = body.getBytes(StandardCharsets.UTF_8)
+ val crc = { val c = new java.util.zip.CRC32(); c.update(data); c.getValue }
+ val out = new ByteArrayOutputStream()
+ def u16(v: Int): Unit = this.u16(out, v)
+ def u32(v: Long): Unit = this.u32(out, v)
+ val size = data.length.toLong
+ // Local file header (sig 0x04034b50): version, GP flags = 0x0001 (bit 0 =
encrypted), STORED
+ // method (0), mod time/date, crc, compressed & uncompressed sizes, name &
extra lengths.
+ val localHeaderOffset = out.size()
+ u32(0x04034b50L); u16(10); u16(0x0001); u16(0); u16(0); u16(0)
+ u32(crc); u32(size); u32(size)
+ u16(nameBytes.length); u16(0)
+ out.write(nameBytes); out.write(data)
+ // Central directory record (sig 0x02014b50): mirrors the header, same
0x0001 encrypted flag,
+ // and points back at the local header offset.
+ val cdOffset = out.size()
+ u32(0x02014b50L); u16(20); u16(10); u16(0x0001); u16(0); u16(0); u16(0)
+ u32(crc); u32(size); u32(size)
+ u16(nameBytes.length); u16(0); u16(0); u16(0); u16(0); u32(0);
u32(localHeaderOffset.toLong)
+ out.write(nameBytes)
+ val cdSize = out.size() - cdOffset
+ // End of central directory (sig 0x06054b50): 1 entry, directory size and
offset.
+ u32(0x06054b50L); u16(0); u16(0); u16(1); u16(1)
+ u32(cdSize.toLong); u32(cdOffset.toLong); u16(0)
+ val fos = new FileOutputStream(file)
+ try {
+ fos.write(out.toByteArray)
+ } finally {
+ fos.close()
+ }
}
/** Write a 7z archive, used to verify the `.7z` archive path. */
@@ -359,8 +409,6 @@ class SupportsArchiveFormatSuite extends SparkFunSuite {
}
// ----- zip ----------------------------------------------------------------
- // The streaming engine is shared with tar (only stream-opening differs), so
these cases focus on
- // the `.zip` dispatch and the `ZipArchiveInputStream` container behaving
like the tar path.
test("readArchiveEntries: empty zip yields empty iterator") {
withTempDir { dir =>
@@ -453,15 +501,56 @@ class SupportsArchiveFormatSuite extends SparkFunSuite {
}
}
- test("readArchiveEntries: a non-streamable zip entry fails loudly, not with
garbled bytes") {
+ test("readArchiveEntries: a STORED zip entry sized only by a data descriptor
reads correctly") {
withTempDir { dir =>
val zip = new File(dir, "stored-dd.zip")
writeStoredEntryWithDataDescriptor(zip, "a.csv", "hello")
- // A stored entry sized only by a trailing data descriptor is the
documented non-streamable
- // case: ZipArchiveInputStream throws on read instead of returning
truncated/garbled bytes.
+ assert(collect(zip) == Seq("a.csv" -> "hello"))
+ }
+ }
+
+ test("readArchiveEntries: corrupt zip bytes fail loudly") {
+ withTempDir { dir =>
+ val zip = new File(dir, "corrupt.zip")
+ Files.write(zip.toPath,
+ "this is not a valid zip archive, just some random bytes"
+ .getBytes(StandardCharsets.UTF_8))
+ // Not a ZIP (no end-of-central-directory signature), so ZipFile
construction fails fast
+ // rather than reporting an empty archive. The ZipException surfaces
wrapped in IOException.
val ex = intercept[java.io.IOException](collect(zip))
- assert(ex.getMessage != null && ex.getMessage.contains("data
descriptor"),
- s"expected a clear unsupported-feature error, got $ex")
+ assert(ex.getCause.isInstanceOf[java.util.zip.ZipException])
+ assert(ex.getCause.getMessage.contains("Archive is not a ZIP archive"))
+ }
+ }
+
+ test("readArchiveEntries: zip with duplicate entry names yields both
entries") {
+ withTempDir { dir =>
+ val zip = new File(dir, "dupes.zip")
+ writeZip(zip, Seq(textEntry("a.csv", "first"), textEntry("a.csv",
"second")))
+ // ZipFile reads every central-directory record, so both duplicate-named
entries surface.
+ assert(collect(zip) == Seq("a.csv" -> "first", "a.csv" -> "second"))
+ }
+ }
+
+ test("readArchiveEntries: encrypted zip entry fails with a clear unsupported
error") {
+ withTempDir { dir =>
+ val zip = new File(dir, "encrypted.zip")
+ writeEncryptedEntry(zip, "a.csv", "hello")
+ val ex = intercept[SparkRuntimeException](collect(zip))
+ checkError(
+ exception = ex,
+ condition = "CANNOT_READ_ZIP_ENTRY",
+ parameters = Map("entry" -> "a.csv", "path" -> new
Path(zip.toURI).toString))
+ }
+ }
+
+ test("readArchiveEntries: an encrypted zip entry that is skipped does not
throw") {
+ withTempDir { dir =>
+ val zip = new File(dir, "encrypted-dotfile.zip")
+ // The encrypted entry has a dotfile name the engine filters out, so it
is never read; its
+ // readability must not be checked (else it would throw
CANNOT_READ_ZIP_ENTRY on a skip).
+ writeEncryptedEntry(zip, "._skipped.csv", "secret")
+ assert(collect(zip).isEmpty)
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
