[SPARK-58382][SQL] binaryFile archive support via wholeFile option

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 {

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,48 @@ 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) =>
+          // 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)
+          }
+          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.

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)