From 5030b82f6ab579c3947c147a5b6f151850d86f9a Mon Sep 17 00:00:00 2001 From: Akshat Shenoi Date: Sat, 25 Jul 2026 03:50:08 +0000 Subject: [PATCH 1/2] [SPARK-58382][SQL] binaryFile archive support via wholeFile option Let format("binaryFile") read tar/zip/7z archives, gated by a new `wholeFile` option (default true) and the existing spark.sql.files.archive.reader.enabled flag. Extends the archive-reader series (SPARK-57135 CSV, SPARK-57419 JSON, SPARK-57478 text, SPARK-57479 XML, SPARK-57481 Avro). - wholeFile=true (default): the whole archive is one record -- unchanged non-archive behavior. - wholeFile=false: one row per inner entry; `content` is the entry's unpacked bytes, `path` is `!/`, `length` is the entry's size, and `modificationTime` is the parent archive's (entry timestamps are optional). BinaryFileFormat mixes in SupportsArchiveFormat and, on the wholeFile=false archive path, reads each entry via readArchiveEntries, building one row per entry from a synthetic per-entry FileStatus through the shared `parse`. The readArchiveEntries callback now receives the ArchiveEntry (not just its name) so consumers can read the entry's size; the CSV and localizeEntries callers are updated accordingly. New BinaryFileArchiveReadBase with BinaryFileTarArchiveReadSuite / BinaryFileZipArchiveReadSuite / BinaryFileSevenZArchiveReadSuite. The per-entry `length` tests are skipped for zip: streaming ZipArchiveInputStream reports an entry sized by a trailing data descriptor as -1, so they are gated on `entrySizeKnown` and re-enable when zip reads move to ZipFile. Generated-by: Claude Code --- .../datasources/SupportsArchiveFormat.scala | 11 +- .../binaryfile/BinaryFileFormat.scala | 86 +++++-- .../datasources/csv/CSVDataSource.scala | 4 +- .../BinaryFileArchiveReadBase.scala | 234 ++++++++++++++++++ .../SupportsArchiveFormatSuite.scala | 32 +-- 5 files changed, 320 insertions(+), 47 deletions(-) create mode 100644 sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/BinaryFileArchiveReadBase.scala 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 a54e83d9e83fc..1c6c635820393 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 @@ -272,21 +272,21 @@ object SupportsArchiveFormat { /** * Streams the archive at `path` entry by entry, applying `parseEntry` to each non-skipped - * `(name, stream)` and concatenating the results. + * `(entry, stream)` and concatenating the results. * * @param path the archive path * @param conf Hadoop configuration used to open the archive * @param ignoredPathSegmentRegex per-segment filter for entries to skip (defaults to the * `InMemoryFileIndex` filter); pass a custom one to match a * loose-file scan - * @param parseEntry turns one entry's `(name, stream)` into an iterator of results + * @param parseEntry turns one entry's `(entry, stream)` into an iterator of results * @return the concatenated results across kept entries, lazily one entry at a time */ def readArchiveEntries[T]( path: Path, conf: Configuration, ignoredPathSegmentRegex: Pattern = HadoopFSUtils.defaultIgnoredPathSegmentRegexPattern)( - parseEntry: (String, InputStream) => Iterator[T]): Iterator[T] = { + parseEntry: (ArchiveEntry, InputStream) => Iterator[T]): Iterator[T] = { val archive = openArchiveStream(path, conf) var closed = false @@ -322,7 +322,7 @@ object SupportsArchiveFormat { } 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.getName, CloseShieldInputStream.wrap(archive)) + currentIter = parseEntry(entry, CloseShieldInputStream.wrap(archive)) } } } @@ -390,7 +390,8 @@ object SupportsArchiveFormat { conf: Configuration, localDir: File, entryFilter: String => Boolean): Iterator[(String, File)] = - readArchiveEntries(path, conf) { (name, in) => + readArchiveEntries(path, conf) { (entry, in) => + val name = entry.getName if (entryFilter(name)) { Iterator.single((name, copyEntryToLocalFile(in, localDir, name))) } else { diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/binaryfile/BinaryFileFormat.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/binaryfile/BinaryFileFormat.scala index 19b67d0c53900..dd430444bdd51 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/binaryfile/BinaryFileFormat.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/binaryfile/BinaryFileFormat.scala @@ -17,6 +17,7 @@ package org.apache.spark.sql.execution.datasources.binaryfile +import java.io.InputStream import java.sql.Timestamp import com.google.common.io.Closeables @@ -25,13 +26,15 @@ import org.apache.hadoop.fs.{FileStatus, Path} import org.apache.hadoop.mapreduce.Job import org.apache.spark.sql.SparkSession +import org.apache.spark.sql.catalyst.FileSourceOptions import org.apache.spark.sql.catalyst.InternalRow import org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowWriter import org.apache.spark.sql.catalyst.types.DataTypeUtils -import org.apache.spark.sql.catalyst.util.DateTimeUtils +import org.apache.spark.sql.catalyst.util.{CaseInsensitiveMap, DateTimeUtils} import org.apache.spark.sql.errors.QueryExecutionErrors -import org.apache.spark.sql.execution.datasources.{FileFormat, OutputWriterFactory, PartitionedFile} +import org.apache.spark.sql.execution.datasources.{FileFormat, OutputWriterFactory, PartitionedFile, SupportsArchiveFormat} import org.apache.spark.sql.internal.SessionStateHelper +import org.apache.spark.sql.internal.SQLConf import org.apache.spark.sql.internal.SQLConf.SOURCES_BINARY_FILE_MAX_LENGTH import org.apache.spark.sql.sources.{And, DataSourceRegister, EqualTo, Filter, GreaterThan, GreaterThanOrEqual, LessThan, LessThanOrEqual, Not, Or} import org.apache.spark.sql.types._ @@ -57,6 +60,7 @@ import org.apache.spark.util.SerializableConfiguration * }}} */ case class BinaryFileFormat() extends FileFormat + with SupportsArchiveFormat with DataSourceRegister with SessionStateHelper { import BinaryFileFormat._ @@ -104,34 +108,29 @@ case class BinaryFileFormat() extends FileFormat val filterFuncs = filters.flatMap(filter => createFilterFunction(filter)) val maxLength = getSqlConf(sparkSession).getConf(SOURCES_BINARY_FILE_MAX_LENGTH) + // `wholeFile=false` reads an archive as one row per inner entry; the default keeps the whole + // archive as a single record. + val caseInsensitiveOptions = CaseInsensitiveMap(options) + val archiveReadEnabled = !caseInsensitiveOptions.get(WHOLE_FILE).forall(_.toBoolean) && + getSqlConf(sparkSession).getConf(SQLConf.ARCHIVE_FORMAT_READER_ENABLED) + file: PartitionedFile => { val path = file.toPath val fs = path.getFileSystem(broadcastedHadoopConf.value.value) val status = fs.getFileStatus(path) - if (filterFuncs.forall(_.apply(status))) { - val writer = new UnsafeRowWriter(requiredSchema.length) - writer.resetRowWriter() - requiredSchema.fieldNames.zipWithIndex.foreach { - case (PATH, i) => writer.write(i, UTF8String.fromString(status.getPath.toString)) - case (LENGTH, i) => writer.write(i, status.getLen) - case (MODIFICATION_TIME, i) => - writer.write(i, DateTimeUtils.millisToMicros(status.getModificationTime)) - case (CONTENT, i) => - if (status.getLen > maxLength) { - throw QueryExecutionErrors.fileLengthExceedsMaxLengthError(status, maxLength) - } - val stream = fs.open(status.getPath) - try { - writer.write(i, stream.readAllBytes()) - } finally { - Closeables.close(stream, true) - } - case (other, _) => - throw QueryExecutionErrors.unsupportedFieldNameError(other) + if (archiveReadEnabled && SupportsArchiveFormat.isArchivePath(path)) { + val ignoredPathSegmentRegex = + new FileSourceOptions(caseInsensitiveOptions).ignoredPathSegmentRegexPattern + SupportsArchiveFormat.readArchiveEntries( + path, broadcastedHadoopConf.value.value, ignoredPathSegmentRegex) { (entry, in) => + val entryStatus = new FileStatus( + entry.getSize, false, 0, 0, status.getModificationTime, + new Path(s"${status.getPath}!/${entry.getName}")) + BinaryFileFormat.parse(() => in, entryStatus, requiredSchema, maxLength, filterFuncs) } - Iterator.single(writer.getRow) } else { - Iterator.empty + BinaryFileFormat.parse( + () => fs.open(status.getPath), status, requiredSchema, maxLength, filterFuncs) } } } @@ -144,6 +143,45 @@ object BinaryFileFormat { private[binaryfile] val LENGTH = "length" private[binaryfile] val CONTENT = "content" private[binaryfile] val BINARY_FILE = "binaryFile" + private[binaryfile] val WHOLE_FILE = "wholeFile" + + /** + * Builds one row from `status` (path/length/modificationTime) and the bytes of `getContent`, + * skipping the row when `filterFuncs` reject `status`. Used for both a standalone file and a + * single archive entry (whose `status` describes the entry). + */ + def parse( + getContent: () => InputStream, + status: FileStatus, + schema: StructType, + maxLength: Int, + filterFuncs: Seq[FileStatus => Boolean]): Iterator[InternalRow] = { + if (filterFuncs.forall(_.apply(status))) { + val writer = new UnsafeRowWriter(schema.length) + writer.resetRowWriter() + schema.fieldNames.zipWithIndex.foreach { + case (PATH, i) => writer.write(i, UTF8String.fromString(status.getPath.toString)) + case (LENGTH, i) => writer.write(i, status.getLen) + case (MODIFICATION_TIME, i) => + writer.write(i, DateTimeUtils.millisToMicros(status.getModificationTime)) + case (CONTENT, i) => + if (status.getLen > maxLength) { + throw QueryExecutionErrors.fileLengthExceedsMaxLengthError(status, maxLength) + } + val content = getContent() + try { + writer.write(i, content.readAllBytes()) + } finally { + Closeables.close(content, true) + } + case (other, _) => + throw QueryExecutionErrors.unsupportedFieldNameError(other) + } + Iterator.single(writer.getRow) + } else { + Iterator.empty + } + } /** * Schema for the binary file data source. diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/csv/CSVDataSource.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/csv/CSVDataSource.scala index 8d2e904ff27f7..2eb0ad44f4723 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/csv/CSVDataSource.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/csv/CSVDataSource.scala @@ -146,9 +146,9 @@ abstract class CSVDataSource extends Serializable with Logging with SupportsArch parseEntry: (UnivocityParser, CSVHeaderChecker, InputStream) => Iterator[InternalRow]) : Iterator[InternalRow] = { SupportsArchiveFormat.readArchiveEntries( - file.toPath, conf, ignoredPathSegmentRegex) { (entryName, in) => + file.toPath, conf, ignoredPathSegmentRegex) { (entry, in) => val headerChecker = - getHeaderChecker(true, s"CSV archive entry: ${file.urlEncodedPath}!/$entryName") + getHeaderChecker(true, s"CSV archive entry: ${file.urlEncodedPath}!/${entry.getName}") val parser = getParser() headerChecker.setHeaderForSingleVariantColumn = CSVDataSource.setHeaderForSingleVariantColumn(conf, file, parser) 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 new file mode 100644 index 0000000000000..bcbfb89d12415 --- /dev/null +++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/BinaryFileArchiveReadBase.scala @@ -0,0 +1,234 @@ +/* + * 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.charset.StandardCharsets + +import org.apache.spark.{SparkConf, SparkException} +import org.apache.spark.sql.{DataFrame, QueryTest, Row} +import org.apache.spark.sql.internal.SQLConf +import org.apache.spark.sql.test.SharedSparkSession + +/** + * Reads of binary files packed in archives via the [[SupportsArchiveFormat]] path. The `wholeFile` + * option (default true) selects the behavior: true keeps today's whole-archive-as-one-record + * contract; false emits one row per inner entry -- `content` is the entry's unpacked bytes, `path` + * is `!/`, and `length` is the entry's size (`modificationTime` stays the + * parent archive's, since entry timestamps are optional). + * + * Unlike CSV/JSON this does not extend [[ArchiveReadSuiteBase]]: binaryFile's four-column row shape + * (path/modificationTime/length/content) differs from the text formats' shape, so the shared tests + * there do not apply. + */ +trait BinaryFileArchiveReadBase extends QueryTest with SharedSparkSession { + + /** Archive extensions to exercise; the head is the default. Supplied by the container trait. */ + protected def archiveExtensions: Seq[String] + + /** Writes `entries` (name -> bytes) into the archive at `dest`. From the container trait. */ + protected def writeArchive(dest: File, entries: Seq[(String, Array[Byte])]): Unit + + /** Writes bytes that are not a readable archive at `dest`. From the container trait. */ + protected def writeCorruptArchive(dest: File): Unit + + /** 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") + + private def bytes(s: String): Array[Byte] = s.getBytes(StandardCharsets.UTF_8) + + private def withArchiveFile( + extension: String = archiveExtensions.head)(f: File => Unit): Unit = + withTempDir(dir => f(new File(dir, s"archive.$extension"))) + + private def read(path: String, options: Map[String, String]): DataFrame = + spark.read.format("binaryFile").options(options).load(path) + + test("wholeFile=true (default) reads the whole archive as a single record") { + archiveExtensions.foreach { ext => + withArchiveFile(ext) { archive => + writeArchive(archive, Seq("a.bin" -> bytes("aaa"), "b.bin" -> bytes("bb"))) + val df = read(archive.getCanonicalPath, Map.empty) + assert(df.count() == 1L) + val row = df.select("path", "length", "content").head() + assert(row.getString(0) == archive.toPath.toUri.toString || + row.getString(0).endsWith(archive.getName)) + assert(row.getLong(1) == archive.length()) + // The single record is the raw archive bytes, not any one entry's bytes. + assert(row.getAs[Array[Byte]](2).length == archive.length()) + } + } + } + + test("wholeFile=false emits one row per entry with the entry's unpacked content") { + archiveExtensions.foreach { ext => + withArchiveFile(ext) { archive => + writeArchive(archive, Seq("a.bin" -> bytes("aaa"), "b.bin" -> bytes("bbbb"))) + checkAnswer( + read(archive.getCanonicalPath, Map("wholeFile" -> "false")).select("content"), + Seq(Row(bytes("aaa")), Row(bytes("bbbb")))) + } + } + } + + 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")) + .select("path", "length", "modificationTime").collect() + assert(rows.length == 2) + // path is `!/`, distinct per entry. + val paths = rows.map(_.getString(0)) + assert(paths.forall(_.contains(s"${archive.getName}!/"))) + assert(paths.map(_.split("!/").last).sorted === Array("a.bin", "b.bin")) + // length is the entry's own unpacked size (3 for "aaa", 4 for "bbbb"), not the archive's. + val byEntry = rows.map(r => r.getString(0).split("!/").last -> r.getLong(1)).toMap + assert(byEntry === Map("a.bin" -> 3L, "b.bin" -> 4L)) + // modificationTime stays the parent archive's, identical across entries. + assert(rows.map(_.getTimestamp(2)).distinct.length == 1) + } + } + + test("wholeFile=false on an empty archive yields no rows") { + withArchiveFile() { archive => + writeArchive(archive, Seq.empty) + checkAnswer(read(archive.getCanonicalPath, Map("wholeFile" -> "false")), Seq.empty[Row]) + } + } + + test("wholeFile=false skips hidden entries") { + withArchiveFile() { archive => + writeArchive(archive, Seq( + "a.bin" -> bytes("keep"), + "_hidden.bin" -> bytes("drop"), + ".dotfile.bin" -> bytes("drop"))) + checkAnswer( + read(archive.getCanonicalPath, Map("wholeFile" -> "false")).select("content"), + Seq(Row(bytes("keep")))) + } + } + + 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") { + val e = intercept[SparkException] { + read(archive.getCanonicalPath, Map("wholeFile" -> "false")).collect() + } + assert(e.getMessage.contains("exceeds") || e.getCause != null) + } + } + } + + 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. + val df = read(archive.getCanonicalPath, Map("wholeFile" -> "false")) + checkAnswer(df.where("length = 3").select("content"), Seq(Row(bytes("aaa")))) + checkAnswer( + df.where("length <= 4").select("content"), Seq(Row(bytes("aaa")), Row(bytes("bbbb")))) + checkAnswer(df.where("length > 4").select("content"), Seq.empty[Row]) + } + } + + test("wholeFile=false reads both archive entries and loose files in the same input") { + withTempDir { dir => + val ext = archiveExtensions.head + writeArchive( + new File(dir, s"data.$ext"), + Seq("a.bin" -> bytes("in-archive-a"), "b.bin" -> bytes("in-archive-b"))) + java.nio.file.Files.write(new File(dir, "loose.bin").toPath, bytes("loose")) + checkAnswer( + read(dir.getCanonicalPath, Map("wholeFile" -> "false")).select("content"), + Seq(Row(bytes("in-archive-a")), Row(bytes("in-archive-b")), Row(bytes("loose")))) + } + } + + test("wholeFile=false with count and no content column reads the right number of rows") { + withArchiveFile() { archive => + writeArchive(archive, Seq( + "a.bin" -> bytes("a"), "b.bin" -> bytes("bb"), "c.bin" -> bytes("ccc"))) + assert(read(archive.getCanonicalPath, Map("wholeFile" -> "false")).count() == 3L) + } + } + + test("an archive always yields a single partition regardless of size") { + withArchiveFile() { archive => + val big = bytes("x" * 4096) + writeArchive(archive, (0 until 4).map(i => s"part-$i.bin" -> big)) + withSQLConf(SQLConf.FILES_MAX_PARTITION_BYTES.key -> "1024") { + val df = read(archive.getCanonicalPath, Map("wholeFile" -> "false")) + assert(df.rdd.getNumPartitions == 1, + s"archive should be a single partition; got ${df.rdd.getNumPartitions}") + assert(df.count() == 4L) + } + } + } + + Seq(true, false).foreach { ignoreCorrupt => + test(s"wholeFile=false ignoreCorruptFiles=$ignoreCorrupt controls skipping a corrupt archive") { + withArchiveFile(corruptArchiveExtension) { archive => + writeCorruptArchive(archive) + withSQLConf(SQLConf.IGNORE_CORRUPT_FILES.key -> ignoreCorrupt.toString) { + val df = read(archive.getCanonicalPath, Map("wholeFile" -> "false")) + if (ignoreCorrupt) { + checkAnswer(df, Seq.empty[Row]) + } else { + intercept[SparkException](df.collect()) + } + } + } + } + } +} + +class BinaryFileTarArchiveReadSuite extends BinaryFileArchiveReadBase with TarArchiveTestUtils { + + override protected def corruptArchiveExtension: String = "tar.gz" +} + +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 + extends BinaryFileArchiveReadBase with SevenZArchiveTestUtils { + + override protected def corruptArchiveExtension: String = "7z" +} 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 5905f0ab8d605..9ce6d0b785ace 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 @@ -158,8 +158,8 @@ class SupportsArchiveFormatSuite extends SparkFunSuite { /** Drains every entry into `(name, decodedText)` pairs through `SupportsArchiveFormat`. */ private def collect(file: File): Seq[(String, String)] = SupportsArchiveFormat.readArchiveEntries(new Path(file.toURI), new Configuration()) { - (name, in) => - Iterator.single((name, new String(readAll(in), StandardCharsets.UTF_8))) + (entry, in) => + Iterator.single((entry.getName, new String(readAll(in), StandardCharsets.UTF_8))) }.toList // ----- isArchivePath ------------------------------------------------------ @@ -256,8 +256,8 @@ class SupportsArchiveFormatSuite extends SparkFunSuite { // HadoopFSUtils.shouldFilterOutPathName still apply -- mirroring a loose-file listing with // the ignoredPathSegmentRegex option set to the same regex. val entries = SupportsArchiveFormat.readArchiveEntries( - new Path(tar.toURI), new Configuration(), Pattern.compile("(?!)")) { (name, in) => - Iterator.single((name, new String(readAll(in), StandardCharsets.UTF_8))) + new Path(tar.toURI), new Configuration(), Pattern.compile("(?!)")) { (entry, in) => + Iterator.single((entry.getName, new String(readAll(in), StandardCharsets.UTF_8))) }.toList assert(entries == Seq("_SUCCESS" -> "marker", "real.csv" -> "kept")) } @@ -272,9 +272,9 @@ class SupportsArchiveFormatSuite extends SparkFunSuite { // parseEntry yields a single element without reading the stream, so each invocation maps to // exactly one consumed output element -- letting us observe when the next entry is opened. val it = SupportsArchiveFormat.readArchiveEntries(new Path(tar.toURI), new Configuration()) { - (name, _) => - opened += name - Iterator.single(name) + (entry, _) => + opened += entry.getName + Iterator.single(entry.getName) } // Construction opens only the first entry; later entries open on demand as iteration @@ -300,11 +300,11 @@ class SupportsArchiveFormatSuite extends SparkFunSuite { val seen = ArrayBuffer[String]() val it = SupportsArchiveFormat.readArchiveEntries(new Path(tar.toURI), new Configuration()) { - (name, in) => + (entry, in) => val body = new String(readAll(in), StandardCharsets.UTF_8) in.close() // must NOT close the underlying archive seen += body - Iterator.single(name) + Iterator.single(entry.getName) } assert(it.toList == List("a.csv", "b.csv")) assert(seen.toList == List("a", "b")) @@ -317,7 +317,7 @@ class SupportsArchiveFormatSuite extends SparkFunSuite { writeTar(tar, Seq(textEntry("a.csv", "a"), textEntry("b.csv", "b"))) val it = SupportsArchiveFormat.readArchiveEntries(new Path(tar.toURI), new Configuration()) { - (name, _) => Iterator.single(name) + (entry, _) => Iterator.single(entry.getName) } assert(it.hasNext) it.asInstanceOf[Closeable].close() @@ -346,7 +346,7 @@ class SupportsArchiveFormatSuite extends SparkFunSuite { try { val it = SupportsArchiveFormat.readArchiveEntries( new Path(tar.toURI), new Configuration()) { - (name, _) => Iterator.single(name) + (entry, _) => Iterator.single(entry.getName) } assert(it.hasNext) it.next() // open the archive and register the completion listener @@ -417,9 +417,9 @@ class SupportsArchiveFormatSuite extends SparkFunSuite { val opened = ArrayBuffer[String]() val it = SupportsArchiveFormat.readArchiveEntries(new Path(zip.toURI), new Configuration()) { - (name, _) => - opened += name - Iterator.single(name) + (entry, _) => + opened += entry.getName + Iterator.single(entry.getName) } // Construction opens only the first entry; advancing past each boundary opens the next. assert(opened.toList == List("a.csv")) @@ -442,11 +442,11 @@ class SupportsArchiveFormatSuite extends SparkFunSuite { val seen = ArrayBuffer[String]() val it = SupportsArchiveFormat.readArchiveEntries(new Path(zip.toURI), new Configuration()) { - (name, in) => + (entry, in) => val body = new String(readAll(in), StandardCharsets.UTF_8) in.close() // must NOT close the underlying archive seen += body - Iterator.single(name) + Iterator.single(entry.getName) } assert(it.toList == List("a.csv", "b.csv")) assert(seen.toList == List("a", "b")) From ab9f4ae4f4f51299b7092d14934c7c43bdb7cb85 Mon Sep 17 00:00:00 2001 From: Akshat Shenoi Date: Mon, 27 Jul 2026 19:20:38 +0000 Subject: [PATCH 2/2] [SPARK-58382][SQL] Note maxLength guard is skipped for unknown-size entries --- .../execution/datasources/binaryfile/BinaryFileFormat.scala | 3 +++ 1 file changed, 3 insertions(+) diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/binaryfile/BinaryFileFormat.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/binaryfile/BinaryFileFormat.scala index dd430444bdd51..33de63072ef39 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/binaryfile/BinaryFileFormat.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/binaryfile/BinaryFileFormat.scala @@ -165,6 +165,9 @@ object BinaryFileFormat { case (MODIFICATION_TIME, i) => writer.write(i, DateTimeUtils.millisToMicros(status.getModificationTime)) case (CONTENT, i) => + // Skipped when the size is unknown (`status.getLen < 0`), which is the case for a + // streaming-zip entry sized only by a trailing data descriptor; such an entry is read + // unbounded until zip reads move to ZipFile (which exposes entry sizes up front). if (status.getLen > maxLength) { throw QueryExecutionErrors.fileLengthExceedsMaxLengthError(status, maxLength) }