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 c9a390c74a15 [SPARK-58110][SQL] Extract SupportsArchiveFormat trait
for archive-read data sources
c9a390c74a15 is described below
commit c9a390c74a1568fc0f508540ecb25730bf9184c4
Author: akshatshenoi-db <[email protected]>
AuthorDate: Wed Jul 15 09:48:11 2026 +0800
[SPARK-58110][SQL] Extract SupportsArchiveFormat trait for archive-read
data sources
### What changes were proposed in this pull request?
Reading tar/zip archives of data files (gated by
`spark.sql.files.archive.reader.enabled`, default off) is currently wired
through a standalone `ArchiveReader` abstract class with
`TarArchiveReader`/`ZipArchiveReader` subclasses; a data source obtains archive
support by calling into that free-standing class from its read and inference
paths.
This refactor replaces that class hierarchy with a `SupportsArchiveFormat`
trait that a file-based data source mixes in to gain archive support, treating
an archive like a directory of the entries it contains. The trait owns the
shared streaming machinery -- `isArchivePath`, `readArchiveEntries`,
`lineIterator` -- for formats whose per-file reader consumes an `InputStream`
(CSV, JSON, XML, text, Avro).
Container selection (tar/tgz/zip) moves into a single `openArchiveStream`
extension match, so the `TarArchiveReader`/`ZipArchiveReader` subclasses are
removed. Schema inference is deliberately not on the trait -- each format owns
its own -- so there is no shared `inferArchiveSchema`. The companion object
keeps two statics for callers that have no trait instance: `isArchivePath` and
a `readArchiveEntries` forwarder (for test suites and executor-side inference
whose `mapPartitions` clos [...]
Class hierarchy before:
```
abstract class ArchiveReader
<- TarArchiveReader
<- ZipArchiveReader
```
after:
```
trait SupportsArchiveFormat // mixed into
CSVDataSource/JsonDataSource/XmlDataSource/TextFileFormat
object SupportsArchiveFormat // isArchivePath, readArchiveEntries
(statics for instance-less callers)
```
The CSV/JSON/XML data sources and `TextFileFormat` now `extends ... with
SupportsArchiveFormat` and call the inherited
`readArchiveEntries`/`lineIterator` -- a mechanical swap. Avro reads archives
too, but via the companion-object statics rather than the trait: its
executor-side inference runs in a `mapPartitions` closure that cannot capture a
trait instance, so `AvroFileFormat`/`AvroUtils` call
`SupportsArchiveFormat.isArchivePath`/`readArchiveEntries` directly.
### Why are the changes needed?
The standalone-class shape made adding archive support to a new format a
matter of threading calls to a free-standing class through the format's read
and inference paths. Modeling "this data source can read archives" as a trait
the format mixes in is a better fit: the shared streaming/localization
machinery lives in one place, a format opts in by mixing in the trait and
calling inherited methods, and container selection is centralized in one
extension match instead of a subclass per c [...]
### Does this PR introduce _any_ user-facing change?
No. This is a behavior-preserving refactor; the archive read/infer
semantics are unchanged.
### How was this patch tested?
No behavior change, so this relies on the existing archive test suites: the
`ArchiveReaderSuite` engine tests (`isArchivePath` dispatch and
`readArchiveEntries` -- entry ordering, gzip handling, dir/dotfile skipping,
lazy advance, the non-closing entry stream, cleanup, and the non-streamable-zip
case) and the per-format tar/zip read suites. The call-site references were
updated from `ArchiveReader` to `SupportsArchiveFormat` and the suites' test
names/docs updated to match.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Opus 4.8)
Closes #57238 from akshatshenoi-db/archive-supports-trait.
Authored-by: akshatshenoi-db <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>
(cherry picked from commit d8b8f3993b893428b6c859b1b4ab2585377002fe)
Signed-off-by: Wenchen Fan <[email protected]>
---
.../org/apache/spark/sql/avro/AvroFileFormat.scala | 8 +-
.../org/apache/spark/sql/avro/AvroUtils.scala | 10 +-
.../sql/execution/datasources/ArchiveReader.scala | 304 ---------------------
.../datasources/SupportsArchiveFormat.scala | 233 ++++++++++++++++
.../execution/datasources/csv/CSVDataSource.scala | 20 +-
.../execution/datasources/csv/CSVFileFormat.scala | 4 +-
.../datasources/json/JsonDataSource.scala | 23 +-
.../datasources/json/JsonFileFormat.scala | 4 +-
.../datasources/text/TextFileFormat.scala | 13 +-
.../execution/datasources/xml/XmlDataSource.scala | 22 +-
.../execution/datasources/xml/XmlFileFormat.scala | 4 +-
.../datasources/ArchiveReadSuiteBase.scala | 4 +-
...uite.scala => SupportsArchiveFormatSuite.scala} | 116 ++++----
.../datasources/TextArchiveReadBase.scala | 6 +-
14 files changed, 358 insertions(+), 413 deletions(-)
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/avro/AvroFileFormat.scala
b/sql/core/src/main/scala/org/apache/spark/sql/avro/AvroFileFormat.scala
index aa9992ba6656..c2a05a8c9d8b 100755
--- a/sql/core/src/main/scala/org/apache/spark/sql/avro/AvroFileFormat.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/avro/AvroFileFormat.scala
@@ -33,7 +33,7 @@ import org.apache.spark.TaskContext
import org.apache.spark.internal.Logging
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.catalyst.{InternalRow, NoopFilters, OrderedFilters}
-import org.apache.spark.sql.execution.datasources.{ArchiveReader,
DataSourceUtils, FileFormat, OutputWriterFactory, PartitionedFile}
+import org.apache.spark.sql.execution.datasources.{DataSourceUtils,
FileFormat, OutputWriterFactory, PartitionedFile, SupportsArchiveFormat}
import org.apache.spark.sql.internal.{SessionStateHelper, SQLConf}
import org.apache.spark.sql.sources.{DataSourceRegister, Filter}
import org.apache.spark.sql.types._
@@ -71,7 +71,7 @@ private[sql] class AvroFileFormat extends FileFormat
options: Map[String, String],
path: Path): Boolean = {
val parsedOptions = new AvroOptions(options,
sparkSession.sessionState.newHadoopConf())
- if (parsedOptions.archiveFormatEnabled &&
ArchiveReader.isArchivePath(path)) {
+ if (parsedOptions.archiveFormatEnabled &&
SupportsArchiveFormat.isArchivePath(path)) {
// A tar archive is read as one sequential stream (entry by entry), so
it is never split.
return false
}
@@ -102,7 +102,7 @@ private[sql] class AvroFileFormat extends FileFormat
(file: PartitionedFile) => {
val conf = broadcastedConf.value.value
- if (parsedOptions.archiveFormatEnabled &&
ArchiveReader.isArchivePath(file.toPath)) {
+ if (parsedOptions.archiveFormatEnabled &&
SupportsArchiveFormat.isArchivePath(file.toPath)) {
// A tar archive (always a single split, see `isSplitable`) is
streamed entry by entry when
// archive reads are enabled; otherwise the file is read directly. The
V2 data source has
// no archive support, so this dispatch lives here.
@@ -199,7 +199,7 @@ private[sql] class AvroFileFormat extends FileFormat
} else {
new NoopFilters
}
- ArchiveReader(file.toPath).readEntries(conf) { (_, in) =>
+ SupportsArchiveFormat.readArchiveEntries(file.toPath, conf) { (_, in) =>
val datumReader = userProvidedSchema match {
case Some(schema) => new GenericDatumReader[GenericRecord](schema)
case None => new GenericDatumReader[GenericRecord]()
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/avro/AvroUtils.scala
b/sql/core/src/main/scala/org/apache/spark/sql/avro/AvroUtils.scala
index 9b6ed4d77a6d..a1bf478f14f1 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/avro/AvroUtils.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/avro/AvroUtils.scala
@@ -38,7 +38,7 @@ import org.apache.spark.sql.avro.AvroCompressionCodec._
import org.apache.spark.sql.avro.AvroOptions.IGNORE_EXTENSION
import org.apache.spark.sql.catalyst.{FileSourceOptions, InternalRow}
import org.apache.spark.sql.catalyst.util.CaseInsensitiveMap
-import org.apache.spark.sql.execution.datasources.{ArchiveReader,
DataSourceUtils, OutputWriterFactory}
+import org.apache.spark.sql.execution.datasources.{DataSourceUtils,
OutputWriterFactory, SupportsArchiveFormat}
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types._
import org.apache.spark.util.Utils
@@ -91,7 +91,7 @@ private[sql] object AvroUtils extends Logging {
// (streamed, never unpacked to disk), then any loose files. Avro has
no DSv2 reader and
// does not merge schemas, so this is V1-only and uses the first
readable writer schema.
val (archives, nonArchives) = if
(fileSourceOptions.archiveFormatEnabled) {
- files.partition(f => ArchiveReader.isArchivePath(f.getPath))
+ files.partition(f => SupportsArchiveFormat.isArchivePath(f.getPath))
} else {
(Seq.empty[FileStatus], files)
}
@@ -276,9 +276,9 @@ private[sql] object AvroUtils extends Logging {
ignoreCorruptFiles: Boolean,
ignoreMissingFiles: Boolean): Option[Schema] = {
try {
- // `readEntries` returns a Closeable iterator; take the first entry's
schema and close it so
- // the archive stream is released without draining the remaining entries.
- val entries = ArchiveReader(path).readEntries(conf) { (_, in) =>
+ // `readArchiveEntries` returns a Closeable iterator; take the first
entry's schema and close
+ // it so the archive stream is released without draining the remaining
entries.
+ val entries = SupportsArchiveFormat.readArchiveEntries(path, conf) { (_,
in) =>
val stream = new DataFileStream[GenericRecord](in, new
GenericDatumReader[GenericRecord]())
try {
Iterator.single(stream.getSchema)
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/ArchiveReader.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/ArchiveReader.scala
deleted file mode 100644
index 1c928fd19da8..000000000000
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/ArchiveReader.scala
+++ /dev/null
@@ -1,304 +0,0 @@
-/*
- * 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.{Closeable, InputStream}
-import java.util.Locale
-import java.util.regex.Pattern
-import java.util.zip.GZIPInputStream
-
-import scala.util.control.NonFatal
-
-import org.apache.commons.compress.archivers.{ArchiveEntry, ArchiveInputStream}
-import org.apache.commons.compress.archivers.tar.TarArchiveInputStream
-import org.apache.commons.compress.archivers.zip.ZipArchiveInputStream
-import org.apache.commons.io.ByteOrderMark
-import org.apache.commons.io.input.{BOMInputStream, CloseShieldInputStream}
-import org.apache.hadoop.conf.Configuration
-import org.apache.hadoop.fs.Path
-import org.apache.hadoop.io.Text
-import org.apache.hadoop.util.LineReader
-
-import org.apache.spark.TaskContext
-import org.apache.spark.util.HadoopFSUtils
-
-/**
- * Streaming reader for a single archive file. The archive is opened once and
decompressed/unpacked
- * as a stream -- entries are never materialized to local disk.
[[readEntries]] hands each entry's
- * bytes to a caller-supplied parse function as a bounded [[InputStream]] and
concatenates the
- * per-entry results into a single iterator, advancing to the next entry only
once the current one
- * is fully consumed. At most one entry is in flight at a time, so memory
stays bounded regardless
- * of archive size.
- *
- * This is format-agnostic: a data source whose per-file reader can consume an
`InputStream` wires
- * up archive support by calling [[readEntries]] from its read/inference paths
and supplying a
- * `parseEntry` that turns one entry stream into rows (or tokens). Formats
that need random access
- * within a file (e.g. Parquet/ORC footers) cannot use this streaming path.
- *
- * The entry-streaming engine ([[readEntries]]) is shared across archive
formats -- tar and zip
- * differ only in how the container stream is opened, and both use
commons-compress
- * `ArchiveInputStream`. A concrete subclass implements only
[[openArchiveStream]]. Obtain the
- * reader for a path via `ArchiveReader(path)`, which selects the
implementation by file
- * extension; new archive formats are added by writing another subclass rather
than modifying the
- * shared engine.
- */
-abstract class ArchiveReader(path: Path) {
-
- /**
- * Opens the archive at `path` as a commons-compress stream, transparently
handling its
- * compression. The shared [[readEntries]] engine steps through entries via
`getNextEntry`; a
- * subclass only chooses the container type (e.g. tar vs zip).
- */
- protected def openArchiveStream(conf: Configuration): ArchiveInputStream[_
<: ArchiveEntry]
-
- /**
- * Whether an entry is not a real data file and must be skipped: a
directory, or a name Spark's
- * own file listing would filter out. Applying
[[HadoopFSUtils.shouldFilterOutPathName]] (the
- * `InMemoryFileIndex` filter) with the same effective
`ignoredPathSegmentRegex` to every path
- * component keeps archive reads in parity with reading the same entries as
loose files,
- * including when the user supplies a custom `ignoredPathSegmentRegex`
option: under the default
- * filter, `.`-prefixed sidecars (macOS `._x`, `.DS_Store`), `_`-prefixed
markers (`_SUCCESS`,
- * `_committed_*`), and anything under a `.`/`_`-prefixed directory (e.g. a
leftover
- * `_temporary/` from a failed write) are skipped, while data files are
kept. The per-component
- * check matters because `InMemoryFileIndex` never recurses into such
directories, so a
- * basename-only filter would read `_temporary/part-0.csv` that a loose-file
scan drops.
- */
- private def shouldSkipEntry(entry: ArchiveEntry, ignoredPathSegmentRegex:
Pattern): Boolean = {
- if (entry.isDirectory) return true
- entry.getName.split("/").exists(c =>
- c.nonEmpty && HadoopFSUtils.shouldFilterOutPathName(c,
ignoredPathSegmentRegex))
- }
-
- /**
- * Wraps the shared archive stream as a view over exactly the current
entry's bytes
- * (`ArchiveInputStream.read` returns -1 at the entry boundary).
[[CloseShieldInputStream]]
- * ignores `close()`, so a parser closing its input does not close the
underlying archive; any
- * unread remainder of an entry is skipped by `getNextEntry()` when
advancing.
- */
- private def entryStream(archive: ArchiveInputStream[_ <: ArchiveEntry]):
InputStream =
- CloseShieldInputStream.wrap(archive)
-
- /**
- * Streams the archive entry by entry, applying `parseEntry` to each
non-skipped entry's
- * `(name, stream)` and concatenating the results into a single iterator.
The next entry is opened
- * only once the current entry's iterator is exhausted, so nothing is
buffered to disk and at most
- * one entry's bytes are read at a time. The archive stream is closed when
the returned iterator
- * is exhausted, when [[Closeable.close]] is called on it, and (defensively)
on task completion.
- * Entry skipping mirrors Spark's file listing: `ignoredPathSegmentRegex` is
the effective filter
- * (the `ignoredPathSegmentRegex` data source option), so callers reading
with a custom filter
- * must pass its compiled form here for archive entries to be filtered like
loose files.
- */
- def readEntries[T](
- conf: Configuration,
- ignoredPathSegmentRegex: Pattern =
HadoopFSUtils.defaultIgnoredPathSegmentRegexPattern)(
- parseEntry: (String, InputStream) => Iterator[T]): Iterator[T] = {
- val archive = openArchiveStream(conf)
- var closed = false
-
- def cleanup(): Unit = {
- if (!closed) {
- closed = true
- try archive.close() catch { case NonFatal(_) => }
- }
- }
-
- Option(TaskContext.get()).foreach(_.addTaskCompletionListener[Unit](_ =>
cleanup()))
-
- val entries = new Iterator[T] with Closeable {
- private var currentIter: Iterator[T] = Iterator.empty
- private var done = false
-
- // Move to the next entry whose iterator has elements (releasing each
exhausted entry's
- // reader and skipping any unread bytes), or mark the stream done once
entries run out.
- // Advancing here -- driven by `hasNext` -- rather than eagerly after
producing a row in
- // `next` is essential for parsers that reuse a single mutable row and
look ahead on
- // `hasNext`: probing the current entry right after returning a row
would overwrite that row's
- // contents before the caller has copied it.
- private def advance(): Unit = {
- while (!done && !currentIter.hasNext) {
- currentIter match {
- case c: Closeable => try c.close() catch { case NonFatal(_) => }
- case _ =>
- }
- var entry = archive.getNextEntry
- while (entry != null && shouldSkipEntry(entry,
ignoredPathSegmentRegex)) {
- entry = archive.getNextEntry
- }
- if (entry == null) {
- done = true
- cleanup()
- } else {
- currentIter = parseEntry(entry.getName, entryStream(archive))
- }
- }
- }
-
- override def hasNext: Boolean = {
- advance()
- !done && currentIter.hasNext
- }
-
- override def next(): T = {
- if (!hasNext) throw new NoSuchElementException
- currentIter.next()
- }
-
- override def close(): Unit = {
- done = true
- currentIter = Iterator.empty
- cleanup()
- }
- }
-
- // Open the first entry eagerly so the construction cost (and any failure)
surfaces here rather
- // than at the first `hasNext`. A corrupt archive throws before the caller
ever holds the
- // iterator, leaving it nothing to close: executors release the stream
through the task-
- // completion listener, but driver-side callers (e.g. Avro's header-only
schema inference) have
- // no task, so close it here before propagating.
- try {
- entries.hasNext
- } catch {
- case NonFatal(e) =>
- cleanup()
- throw e
- }
- entries
- }
-}
-
-object ArchiveReader {
-
- /**
- * Whether `path` names an archive this reader can stream. Dispatched purely
on the file
- * extension -- `.tar`, `.tar.gz`, `.tgz`, or `.zip` -- since the bytes are
not inspected here.
- */
- def isArchivePath(path: Path): Boolean = {
- val name = path.getName.toLowerCase(Locale.ROOT)
- name.endsWith(".tar") || name.endsWith(".tar.gz") || name.endsWith(".tgz")
||
- name.endsWith(".zip")
- }
-
- /**
- * Returns the [[ArchiveReader]] implementation for `path`, selected by its
file extension. Only
- * paths for which [[isArchivePath]] is true are supported; new archive
formats add a case here.
- */
- def apply(path: Path): ArchiveReader = {
- val name = path.getName.toLowerCase(Locale.ROOT)
- name match {
- case n if n.endsWith(".tar") || n.endsWith(".tar.gz") ||
n.endsWith(".tgz") =>
- new TarArchiveReader(path)
- case n if n.endsWith(".zip") =>
- new ZipArchiveReader(path)
- case _ =>
- throw new IllegalArgumentException(
- s"$path is not a supported archive (expected .tar, .tar.gz, .tgz, or
.zip)")
- }
- }
-
- /**
- * Splits one already-decompressed archive entry's bytes into lines. The
reusable, format-agnostic
- * line source for archive entries; the entry stream is not closed here (the
reader owns the
- * underlying stream).
- *
- * @param in bytes of one archive entry.
- * @param lineSeparatorInRead the explicit read line separator, or `None` to
detect line breaks.
- * @return an iterator over the entry's lines as [[Text]], without the
trailing separator.
- */
- def lineIterator(in: InputStream, lineSeparatorInRead: Option[Array[Byte]]):
Iterator[Text] = {
- // A leading byte-order mark is stripped (LineReader does not strip it on
its own) so the lines
- // match the non-archive read path.
- val bomInputStream = BOMInputStream.builder()
- .setInputStream(in)
- .setByteOrderMarks(
- ByteOrderMark.UTF_8,
- ByteOrderMark.UTF_16LE,
- ByteOrderMark.UTF_16BE,
- ByteOrderMark.UTF_32LE,
- ByteOrderMark.UTF_32BE)
- .setInclude(false)
- .get()
- val reader = lineSeparatorInRead match {
- case Some(sep) => new LineReader(bomInputStream, sep)
- case _ => new LineReader(bomInputStream)
- }
- new Iterator[Text] {
- private val text = new Text()
- private var finished = false
- private var hasValue = false
-
- override def hasNext: Boolean = {
- if (!finished && !hasValue) {
- finished = reader.readLine(text) == 0
- hasValue = !finished
- }
- !finished
- }
-
- override def next(): Text = {
- if (!hasNext) throw new NoSuchElementException
- hasValue = false
- text
- }
- }
- }
-}
-
-/**
- * [[ArchiveReader]] for tar archives: plain `.tar`, gzipped `.tar.gz`, and
`.tgz`.
- *
- * Gzip handling: Hadoop's `CompressionCodecFactory` matches the trailing
`.gz` extension and
- * auto-decompresses `.tar.gz` via `CodecStreams`, so we just wrap that stream
in
- * `TarArchiveInputStream`. `.tgz` is not a registered Hadoop codec extension,
so the gzip layer is
- * unwrapped explicitly here.
- */
-class TarArchiveReader(path: Path) extends ArchiveReader(path) {
-
- // Paths Hadoop's codec factory won't auto-decompress: we apply the gzip
layer here.
- private def needsExplicitGunzip: Boolean =
- path.getName.toLowerCase(Locale.ROOT).endsWith(".tgz")
-
- override protected def openArchiveStream(
- conf: Configuration): ArchiveInputStream[_ <: ArchiveEntry] = {
- 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 (needsExplicitGunzip) new GZIPInputStream(base) else
base
- new TarArchiveInputStream(tarBytes)
- } catch {
- case NonFatal(e) =>
- try base.close() catch { case NonFatal(_) => }
- throw e
- }
- }
-}
-
-/**
- * [[ArchiveReader]] for zip archives (`.zip`). Each entry is individually
compressed inside the
- * container (the container itself is not gzip-wrapped), so
`ZipArchiveInputStream` decompresses
- * entries as they are streamed and no Hadoop codec layer is applied. The
stream reads local file
- * headers sequentially rather than the central directory, matching the tar
reader's pure-streaming
- * model: a few unusual zips (e.g. a stored entry whose size is recorded only
in a trailing data
- * descriptor) are not streamable this way.
- */
-class ZipArchiveReader(path: Path) extends ArchiveReader(path) {
-
- override protected def openArchiveStream(
- conf: Configuration): ArchiveInputStream[_ <: ArchiveEntry] =
- new
ZipArchiveInputStream(CodecStreams.createInputStreamWithCloseResource(conf,
path))
-}
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
new file mode 100644
index 000000000000..105bad108967
--- /dev/null
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/SupportsArchiveFormat.scala
@@ -0,0 +1,233 @@
+/*
+ * 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.{Closeable, InputStream}
+import java.util.Locale
+import java.util.regex.Pattern
+import java.util.zip.GZIPInputStream
+
+import scala.util.control.NonFatal
+
+import org.apache.commons.compress.archivers.{ArchiveEntry, ArchiveInputStream}
+import org.apache.commons.compress.archivers.tar.TarArchiveInputStream
+import org.apache.commons.compress.archivers.zip.ZipArchiveInputStream
+import org.apache.commons.io.ByteOrderMark
+import org.apache.commons.io.input.{BOMInputStream, CloseShieldInputStream}
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.fs.Path
+import org.apache.hadoop.io.Text
+import org.apache.hadoop.util.LineReader
+
+import org.apache.spark.TaskContext
+import org.apache.spark.internal.Logging
+import org.apache.spark.util.HadoopFSUtils
+
+/**
+ * Mixed into a file-based data source to let it read tar/zip archives of its
files, treating an
+ * archive like a directory of the entries it contains. Schema inference is
left to each format.
+ */
+trait SupportsArchiveFormat extends Logging {
+
+ /**
+ * Splits one already-decompressed archive entry's bytes into lines (`None`
separator = detect
+ * line breaks), yielding each line as [[Text]] without the trailing
separator. Does not close the
+ * entry stream (the reader owns it).
+ */
+ protected def lineIterator(
+ in: InputStream,
+ lineSeparatorInRead: Option[Array[Byte]]): Iterator[Text] = {
+ // Strip a leading BOM (LineReader does not) so lines match the
non-archive read path.
+ val bomInputStream = BOMInputStream.builder()
+ .setInputStream(in)
+ .setByteOrderMarks(
+ ByteOrderMark.UTF_8,
+ ByteOrderMark.UTF_16LE,
+ ByteOrderMark.UTF_16BE,
+ ByteOrderMark.UTF_32LE,
+ ByteOrderMark.UTF_32BE)
+ .setInclude(false)
+ .get()
+ val reader = lineSeparatorInRead match {
+ case Some(sep) => new LineReader(bomInputStream, sep)
+ case _ => new LineReader(bomInputStream)
+ }
+ new Iterator[Text] {
+ private val text = new Text()
+ private var finished = false
+ private var hasValue = false
+
+ override def hasNext: Boolean = {
+ if (!finished && !hasValue) {
+ finished = reader.readLine(text) == 0
+ hasValue = !finished
+ }
+ !finished
+ }
+
+ override def next(): Text = {
+ if (!hasNext) throw new NoSuchElementException
+ hasValue = false
+ text
+ }
+ }
+ }
+
+}
+
+object SupportsArchiveFormat {
+
+ /**
+ * Whether the file path is a supported archive format.
+ */
+ def isArchivePath(path: Path): Boolean = {
+ val name = path.getName.toLowerCase(Locale.ROOT)
+ name.endsWith(".tar") || name.endsWith(".tar.gz") || name.endsWith(".tgz")
||
+ name.endsWith(".zip")
+ }
+
+ /**
+ * Opens the archive at `path` as a commons-compress stream, selecting the
container by extension.
+ */
+ private def openArchiveStream(
+ path: Path,
+ conf: Configuration): ArchiveInputStream[_ <: ArchiveEntry] = {
+ 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
+ }
+ case n if n.endsWith(".zip") =>
+ new
ZipArchiveInputStream(CodecStreams.createInputStreamWithCloseResource(conf,
path))
+ case _ =>
+ throw new IllegalArgumentException(
+ s"$path is not a supported archive (expected .tar, .tar.gz, .tgz, or
.zip)")
+ }
+ }
+
+ /**
+ * Whether an entry is a directory or a name Spark's file listing would
filter out, so archive
+ * reads match a loose-file scan.
+ *
+ * @param entry the archive entry to test
+ * @param ignoredPathSegmentRegex per-segment filter matched against each
`/`-separated component
+ * @return true if the entry is a directory or any path component is
filtered out
+ */
+ private def shouldSkipEntry(entry: ArchiveEntry, ignoredPathSegmentRegex:
Pattern): Boolean = {
+ if (entry.isDirectory) return true
+ entry.getName.split("/").exists(c =>
+ c.nonEmpty && HadoopFSUtils.shouldFilterOutPathName(c,
ignoredPathSegmentRegex))
+ }
+
+ /**
+ * Streams the archive at `path` entry by entry, applying `parseEntry` to
each non-skipped
+ * `(name, 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
+ * @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] = {
+ val archive = openArchiveStream(path, conf)
+ var closed = false
+
+ def cleanup(): Unit = {
+ if (!closed) {
+ closed = true
+ try archive.close() catch { case NonFatal(_) => }
+ }
+ }
+
+ Option(TaskContext.get()).foreach(_.addTaskCompletionListener[Unit](_ =>
cleanup()))
+
+ val entries = new Iterator[T] with Closeable {
+ private var currentIter: Iterator[T] = Iterator.empty
+ private var done = false
+
+ // Move to the next entry with elements (releasing the exhausted one),
or mark done. Advancing
+ // on `hasNext` rather than eagerly in `next` matters for parsers that
reuse a mutable row: an
+ // early probe would overwrite the returned row before the caller copies
it.
+ private def advance(): Unit = {
+ while (!done && !currentIter.hasNext) {
+ currentIter match {
+ case c: Closeable => try c.close() catch { case NonFatal(_) => }
+ case _ =>
+ }
+ var entry = archive.getNextEntry
+ while (entry != null && shouldSkipEntry(entry,
ignoredPathSegmentRegex)) {
+ entry = archive.getNextEntry
+ }
+ if (entry == 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.getName,
CloseShieldInputStream.wrap(archive))
+ }
+ }
+ }
+
+ override def hasNext: Boolean = {
+ advance()
+ !done && currentIter.hasNext
+ }
+
+ override def next(): T = {
+ if (!hasNext) throw new NoSuchElementException
+ currentIter.next()
+ }
+
+ override def close(): Unit = {
+ done = true
+ currentIter = Iterator.empty
+ cleanup()
+ }
+ }
+
+ // Open the first entry eagerly so a corrupt-archive failure surfaces
here, not at first
+ // `hasNext`. The caller then holds no iterator to close, and a
driver-side caller (e.g. Avro
+ // header inference) has no task-completion listener, so close the stream
here before rethrow.
+ try {
+ entries.hasNext
+ } catch {
+ case NonFatal(e) =>
+ cleanup()
+ throw e
+ }
+ entries
+ }
+
+}
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 2bdc47e8e2da..8d2e904ff27f 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
@@ -49,7 +49,7 @@ import org.apache.spark.util.Utils
/**
* Common functions for parsing CSV files
*/
-abstract class CSVDataSource extends Serializable with Logging {
+abstract class CSVDataSource extends Serializable with Logging with
SupportsArchiveFormat {
def isSplitable: Boolean
/**
@@ -74,7 +74,7 @@ abstract class CSVDataSource extends Serializable with
Logging {
case Some(columnName) => Some(StructType(Array(StructField(columnName,
VariantType))))
case None =>
val hasArchive = parsedOptions.archiveFormatEnabled &&
- inputPaths.exists(f => ArchiveReader.isArchivePath(f.getPath))
+ inputPaths.exists(f =>
SupportsArchiveFormat.isArchivePath(f.getPath))
if (hasArchive && supportsArchiveScan) {
// Archives (and any loose files alongside them) are inferred in a
single CSVInferSchema
// pass over all inputs -- archive entries are streamed, never
unpacked -- so the result
@@ -145,7 +145,8 @@ abstract class CSVDataSource extends Serializable with
Logging {
ignoredPathSegmentRegex: Pattern)(
parseEntry: (UnivocityParser, CSVHeaderChecker, InputStream) =>
Iterator[InternalRow])
: Iterator[InternalRow] = {
- ArchiveReader(file.toPath).readEntries(conf, ignoredPathSegmentRegex) {
(entryName, in) =>
+ SupportsArchiveFormat.readArchiveEntries(
+ file.toPath, conf, ignoredPathSegmentRegex) { (entryName, in) =>
val headerChecker =
getHeaderChecker(true, s"CSV archive entry:
${file.urlEncodedPath}!/$entryName")
val parser = getParser()
@@ -174,8 +175,8 @@ abstract class CSVDataSource extends Serializable with
Logging {
def tokens(dropHeader: Boolean): RDD[Array[String]] = baseRdd.flatMap {
stream =>
val path = new Path(stream.getPath())
try {
- if (ArchiveReader.isArchivePath(path)) {
- ArchiveReader(path).readEntries(stream.getConfiguration) { (_, in) =>
+ if (SupportsArchiveFormat.isArchivePath(path)) {
+ SupportsArchiveFormat.readArchiveEntries(path,
stream.getConfiguration) { (_, in) =>
tokenizeForInference(in, dropHeader, parsedOptions)
}
} else {
@@ -313,9 +314,10 @@ object TextInputCSVDataSource extends CSVDataSource {
/**
* Decodes one archive entry's bytes into the same CSV line strings the
non-archive [[readFile]]
- * path feeds to the parser: [[ArchiveReader.lineIterator]] splits the entry
into lines (honoring
- * a custom line separator) and each line is decoded with the configured
charset. Like `readFile`,
- * the decoded lines are fed to `UnivocityParser.parseIterator` without a
re-appended terminator.
+ * path feeds to the parser: [[SupportsArchiveFormat.lineIterator]] splits
the entry into lines
+ * (honoring a custom line separator) and each line is decoded with the
configured charset. Like
+ * `readFile`, the decoded lines are fed to `UnivocityParser.parseIterator`
without a re-appended
+ * terminator.
*
* @param in bytes of one already-decompressed archive entry; not closed
here (the archive owns
* the underlying stream).
@@ -323,7 +325,7 @@ object TextInputCSVDataSource extends CSVDataSource {
* @return an iterator over the entry's lines.
*/
private def entryLines(in: InputStream, options: CSVOptions):
Iterator[String] = {
- ArchiveReader.lineIterator(in, options.lineSeparatorInRead).map { line =>
+ lineIterator(in, options.lineSeparatorInRead).map { line =>
new String(line.getBytes, 0, line.getLength, options.charset)
}
}
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/csv/CSVFileFormat.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/csv/CSVFileFormat.scala
index ab7de0ffadb6..a406fbe4844d 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/csv/CSVFileFormat.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/csv/CSVFileFormat.scala
@@ -46,7 +46,7 @@ case class CSVFileFormat() extends TextBasedFileFormat with
DataSourceRegister {
val parsedOptions = getCsvOptions(sparkSession, options)
// A tar archive is decompressed/unpacked as a sequential stream, so it
must be read as a
// single split rather than carved into byte ranges.
- if (parsedOptions.archiveFormatEnabled &&
ArchiveReader.isArchivePath(path)) {
+ if (parsedOptions.archiveFormatEnabled &&
SupportsArchiveFormat.isArchivePath(path)) {
return false
}
CSVDataSource(parsedOptions).isSplitable &&
super.isSplitable(sparkSession, options, path)
@@ -142,7 +142,7 @@ case class CSVFileFormat() extends TextBasedFileFormat with
DataSourceRegister {
// A tar archive (always a single split, see `isSplitable`) is streamed
entry by entry when
// archive reads are enabled; otherwise the file is parsed directly.
- if (parsedOptions.archiveFormatEnabled &&
ArchiveReader.isArchivePath(file.toPath)) {
+ if (parsedOptions.archiveFormatEnabled &&
SupportsArchiveFormat.isArchivePath(file.toPath)) {
CSVDataSource(parsedOptions).readArchive(
conf, file, () => newParser(), getHeaderChecker, requiredSchema,
ignoredPathSegmentRegex)
} else {
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/json/JsonDataSource.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/json/JsonDataSource.scala
index 483c3c1f47f9..a5f19ae353d4 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/json/JsonDataSource.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/json/JsonDataSource.scala
@@ -51,7 +51,7 @@ import org.apache.spark.util.Utils
/**
* Common functions for parsing JSON files
*/
-abstract class JsonDataSource extends Serializable with Logging {
+abstract class JsonDataSource extends Serializable with Logging with
SupportsArchiveFormat {
def isSplitable: Boolean
/**
@@ -91,7 +91,7 @@ abstract class JsonDataSource extends Serializable with
Logging {
file: PartitionedFile,
parser: () => JacksonParser,
schema: StructType): Iterator[InternalRow] =
- ArchiveReader(file.toPath).readEntries(conf) { (_, in) =>
+ SupportsArchiveFormat.readArchiveEntries(file.toPath, conf) { (_, in) =>
readStream(in, parser(), schema)
}
@@ -104,7 +104,7 @@ abstract class JsonDataSource extends Serializable with
Logging {
case Some(columnName) => Some(StructType(Array(StructField(columnName,
VariantType))))
case None =>
val hasArchive = parsedOptions.archiveFormatEnabled &&
- inputPaths.exists(f => ArchiveReader.isArchivePath(f.getPath))
+ inputPaths.exists(f =>
SupportsArchiveFormat.isArchivePath(f.getPath))
if (hasArchive && supportsArchiveScan) {
// Archives (and any loose files alongside them) are inferred in a
single JsonInferSchema
// pass over all inputs -- archive entries are streamed, never
unpacked -- so the result
@@ -130,9 +130,10 @@ abstract class JsonDataSource extends Serializable with
Logging {
/**
* Infers a JSON schema when at least one input is a tar archive. Every
archive entry (streamed
- * via `ArchiveReader`, never unpacked to disk) and every loose file is read
as JSON records --
- * each line is a record, or the whole input is one document in multi-line
mode -- and all of them
- * feed a single [[JsonInferSchema]] pass, exactly as a directory of the
same files would infer.
+ * via `SupportsArchiveFormat`, never unpacked to disk) and every loose file
is read as JSON
+ * records -- each line is a record, or the whole input is one document in
multi-line mode -- and
+ * all of them feed a single [[JsonInferSchema]] pass, exactly as a
directory of the same files
+ * would infer.
* Because [[JsonInferSchema]] already merges every record's type by field
name across all inputs,
* one pass is itself the union: a field empty in one input but typed in
another widens to the
* real type, and a `NullType` field survives to the single final
canonicalization rather than
@@ -166,8 +167,10 @@ abstract class JsonDataSource extends Serializable with
Logging {
stream =>
val path = new Path(stream.getPath())
try {
- if (ArchiveReader.isArchivePath(path)) {
- ArchiveReader(path).readEntries(stream.getConfiguration) { (_, in)
=> perEntry(in) }
+ if (SupportsArchiveFormat.isArchivePath(path)) {
+ SupportsArchiveFormat.readArchiveEntries(path,
stream.getConfiguration) { (_, in) =>
+ perEntry(in)
+ }
} else {
perEntry(
CodecStreams.createInputStreamWithCloseResource(stream.getConfiguration, path))
@@ -201,7 +204,7 @@ abstract class JsonDataSource extends Serializable with
Logging {
} else {
// Line-delimited: each line is a record, copied off the reused line
buffer and parsed from
// its bytes (`CreateJacksonParser.bytes`, matching
TextInputJsonDataSource).
- val lines = perInput(in => ArchiveReader.lineIterator(in,
lineSeparator).map(_.copyBytes()))
+ val lines = perInput(in => lineIterator(in,
lineSeparator).map(_.copyBytes()))
val lineParser: (JsonFactory, Array[Byte]) => JsonParser = encoding
.map(enc => CreateJacksonParser.bytes(enc, _: JsonFactory, _:
Array[Byte]))
.getOrElse(CreateJacksonParser.bytes(_: JsonFactory, _: Array[Byte]))
@@ -324,7 +327,7 @@ object TextInputJsonDataSource extends JsonDataSource {
parser.options.parseMode,
schema,
parser.options.columnNameOfCorruptRecord)
- ArchiveReader.lineIterator(in,
parser.options.lineSeparatorInRead).flatMap(safeParser.parse)
+ lineIterator(in,
parser.options.lineSeparatorInRead).flatMap(safeParser.parse)
}
private def textToUTF8String(value: Text): UTF8String = {
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/json/JsonFileFormat.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/json/JsonFileFormat.scala
index ee01d67d1f7b..4ede461a4d51 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/json/JsonFileFormat.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/json/JsonFileFormat.scala
@@ -40,7 +40,7 @@ case class JsonFileFormat() extends TextBasedFileFormat with
DataSourceRegister
options: Map[String, String],
path: Path): Boolean = {
val parsedOptions = getJsonOptions(sparkSession, options)
- if (parsedOptions.archiveFormatEnabled &&
ArchiveReader.isArchivePath(path)) {
+ if (parsedOptions.archiveFormatEnabled &&
SupportsArchiveFormat.isArchivePath(path)) {
// A tar archive is read as one sequential stream (entry by entry), so
it is never split.
return false
}
@@ -107,7 +107,7 @@ case class JsonFileFormat() extends TextBasedFileFormat
with DataSourceRegister
parsedOptions,
allowArrayAsStructs = true,
filters)
- if (parsedOptions.archiveFormatEnabled &&
ArchiveReader.isArchivePath(file.toPath)) {
+ if (parsedOptions.archiveFormatEnabled &&
SupportsArchiveFormat.isArchivePath(file.toPath)) {
JsonDataSource(parsedOptions).readArchive(
broadcastedHadoopConf.value.value, file, () => parser(),
requiredSchema)
} else {
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/text/TextFileFormat.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/text/TextFileFormat.scala
index 863f4e1c162e..c220dd5a957f 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/text/TextFileFormat.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/text/TextFileFormat.scala
@@ -37,7 +37,10 @@ import org.apache.spark.util.{SerializableConfiguration,
Utils}
/**
* A data source for reading text files. The text files must be encoded as
UTF-8.
*/
-case class TextFileFormat() extends TextBasedFileFormat with
DataSourceRegister {
+case class TextFileFormat()
+ extends TextBasedFileFormat
+ with DataSourceRegister
+ with SupportsArchiveFormat {
override def shortName(): String = "text"
@@ -60,7 +63,7 @@ case class TextFileFormat() extends TextBasedFileFormat with
DataSourceRegister
options: Map[String, String],
path: Path): Boolean = {
val textOptions = new TextOptions(options)
- if (textOptions.archiveFormatEnabled && ArchiveReader.isArchivePath(path))
{
+ if (textOptions.archiveFormatEnabled &&
SupportsArchiveFormat.isArchivePath(path)) {
// A tar archive is read as one sequential stream (entry by entry), so
it is never split.
return false
}
@@ -118,7 +121,7 @@ case class TextFileFormat() extends TextBasedFileFormat
with DataSourceRegister
// archive reads are enabled; otherwise the file is read directly. Archive
scanning is wired
// into the V1 file source only, so this dispatch lives here rather than
in a shared reader.
(file: PartitionedFile) => {
- if (textOptions.archiveFormatEnabled &&
ArchiveReader.isArchivePath(file.toPath)) {
+ if (textOptions.archiveFormatEnabled &&
SupportsArchiveFormat.isArchivePath(file.toPath)) {
archiveReader(file)
} else {
perFileReader(file)
@@ -139,7 +142,7 @@ case class TextFileFormat() extends TextBasedFileFormat
with DataSourceRegister
textOptions: TextOptions): PartitionedFile => Iterator[UnsafeRow] = {
(file: PartitionedFile) => {
val confValue = conf.value.value
- ArchiveReader(file.toPath).readEntries(confValue) { (_, in) =>
+ SupportsArchiveFormat.readArchiveEntries(file.toPath, confValue) { (_,
in) =>
// Each entry is read as a standalone text file, so it gets its own
row writer, exactly as
// `readToUnsafeMem` builds one per file.
val emptyUnsafeRow = new UnsafeRow(0)
@@ -159,7 +162,7 @@ case class TextFileFormat() extends TextBasedFileFormat
with DataSourceRegister
val content = in.readAllBytes()
Iterator.single(toRow(content, content.length))
} else {
- ArchiveReader.lineIterator(in, textOptions.lineSeparatorInRead).map
{ line =>
+ lineIterator(in, textOptions.lineSeparatorInRead).map { line =>
toRow(line.getBytes, line.getLength)
}
}
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/xml/XmlDataSource.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/xml/XmlDataSource.scala
index 0618a4c51da1..ef34711e041e 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/xml/XmlDataSource.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/xml/XmlDataSource.scala
@@ -47,7 +47,7 @@ import org.apache.spark.util.Utils
/**
* Common functions for parsing XML files
*/
-abstract class XmlDataSource extends Serializable with Logging {
+abstract class XmlDataSource extends Serializable with Logging with
SupportsArchiveFormat {
def isSplitable: Boolean
/**
@@ -87,7 +87,7 @@ abstract class XmlDataSource extends Serializable with
Logging {
file: PartitionedFile,
parser: () => StaxXmlParser,
schema: StructType): Iterator[InternalRow] =
- ArchiveReader(file.toPath).readEntries(conf) { (_, in) =>
+ SupportsArchiveFormat.readArchiveEntries(file.toPath, conf) { (_, in) =>
readStream(in, parser(), schema)
}
@@ -106,7 +106,7 @@ abstract class XmlDataSource extends Serializable with
Logging {
// loose files -- so the result matches a directory read of the same
files. XML has no DSv2
// reader, so this archive scan is always V1.
val hasArchive = parsedOptions.archiveFormatEnabled &&
- inputPaths.exists(f => ArchiveReader.isArchivePath(f.getPath))
+ inputPaths.exists(f =>
SupportsArchiveFormat.isArchivePath(f.getPath))
if (hasArchive) {
Some(inferWithArchives(sparkSession, inputPaths, parsedOptions))
} else if (inputPaths.nonEmpty) {
@@ -124,9 +124,9 @@ abstract class XmlDataSource extends Serializable with
Logging {
/**
* Infers an XML schema when at least one input is a tar archive
(`.tar`/`.tar.gz`/`.tgz`). Every
- * archive entry (streamed through `ArchiveReader`, never unpacked to disk)
and every loose file
- * is tokenized into records and fed to a single [[XmlInferSchema]] pass,
exactly as a directory
- * of the same files would infer. Tokenization is per-mode so it matches
this mode's scan:
+ * archive entry (streamed through `SupportsArchiveFormat`, never unpacked
to disk) and every
+ * loose file is tokenized into records and fed to a single
[[XmlInferSchema]] pass, exactly as a
+ * directory of the same files would infer. Tokenization is per-mode so it
matches the scan:
* multi-line splits the whole stream into `rowTag`-delimited records,
single-line treats each
* line as a record (mirroring [[readFile]] and JSON's `inferWithArchives`).
*/
@@ -145,8 +145,10 @@ abstract class XmlDataSource extends Serializable with
Logging {
stream =>
val path = new Path(stream.getPath())
try {
- if (ArchiveReader.isArchivePath(path)) {
- ArchiveReader(path).readEntries(stream.getConfiguration) { (_, in)
=> perEntry(in) }
+ if (SupportsArchiveFormat.isArchivePath(path)) {
+ SupportsArchiveFormat.readArchiveEntries(path,
stream.getConfiguration) { (_, in) =>
+ perEntry(in)
+ }
} else {
perEntry(
CodecStreams.createInputStreamWithCloseResource(stream.getConfiguration, path))
@@ -173,7 +175,7 @@ abstract class XmlDataSource extends Serializable with
Logging {
perInput(in => StaxXmlParser.tokenizeStream(in, parsedOptions))
} else {
val charset = parsedOptions.charset
- perInput(in => ArchiveReader.lineIterator(in, None).map { line =>
+ perInput(in => lineIterator(in, None).map { line =>
new String(line.getBytes, 0, line.getLength, charset)
})
}
@@ -254,7 +256,7 @@ object TextInputXmlDataSource extends XmlDataSource {
in: InputStream,
parser: StaxXmlParser,
schema: StructType): Iterator[InternalRow] = {
- val lines = ArchiveReader.lineIterator(in, None).map { line =>
+ val lines = lineIterator(in, None).map { line =>
new String(line.getBytes, 0, line.getLength, parser.options.charset)
}
val safeParser = new FailureSafeParser[String](
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/xml/XmlFileFormat.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/xml/XmlFileFormat.scala
index da81ff327c74..26bf2cd3ccab 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/xml/XmlFileFormat.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/xml/XmlFileFormat.scala
@@ -51,7 +51,7 @@ case class XmlFileFormat() extends TextBasedFileFormat with
DataSourceRegister {
options: Map[String, String],
path: Path): Boolean = {
val xmlOptions = getXmlOptions(sparkSession, options)
- if (xmlOptions.archiveFormatEnabled && ArchiveReader.isArchivePath(path)) {
+ if (xmlOptions.archiveFormatEnabled &&
SupportsArchiveFormat.isArchivePath(path)) {
// A tar archive is read as one sequential stream (entry by entry), so
it is never split.
return false
}
@@ -126,7 +126,7 @@ case class XmlFileFormat() extends TextBasedFileFormat with
DataSourceRegister {
// A tar archive (always a single split, see `isSplitable`) is streamed
entry by entry when
// archive reads are enabled; otherwise the file is parsed directly. XML
has no DSv2 reader,
// so this dispatch lives here rather than inside the shared `readFile`.
- if (xmlOptions.archiveFormatEnabled &&
ArchiveReader.isArchivePath(file.toPath)) {
+ if (xmlOptions.archiveFormatEnabled &&
SupportsArchiveFormat.isArchivePath(file.toPath)) {
XmlDataSource(xmlOptions).readArchive(
broadcastedHadoopConf.value.value,
file,
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/ArchiveReadSuiteBase.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/ArchiveReadSuiteBase.scala
index 8ec73fbfefda..1d715295aeb0 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/ArchiveReadSuiteBase.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/ArchiveReadSuiteBase.scala
@@ -30,8 +30,8 @@ import org.apache.spark.util.Utils
/**
* Format- and archive-agnostic end-to-end tests for reading archives of data
files through the
- * streaming [[ArchiveReader]] path. Entries are streamed (never unpacked to
disk), and the central
- * contract verified throughout is parity with reading the same files from a
directory.
+ * streaming [[SupportsArchiveFormat]] path. Entries are streamed (never
unpacked to disk), and the
+ * central contract verified throughout is parity with reading the same files
from a directory.
*
* A concrete suite binds the abstract hooks below by mixing in a file-format
trait (e.g.
* [[org.apache.spark.sql.execution.datasources.CSVArchiveReadBase]]) and an
archive-format trait
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/ArchiveReaderSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/SupportsArchiveFormatSuite.scala
similarity index 78%
rename from
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/ArchiveReaderSuite.scala
rename to
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/SupportsArchiveFormatSuite.scala
index e49800ec0640..65547991c6b2 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/ArchiveReaderSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/SupportsArchiveFormatSuite.scala
@@ -33,11 +33,11 @@ import org.apache.hadoop.fs.Path
import org.apache.spark.{SparkFunSuite, TaskContext, TaskContextImpl}
/**
- * Unit tests for the streaming [[ArchiveReader]] core: `isArchivePath`
dispatch and `readEntries`
- * (entry ordering, gzip handling, dir/dotfile skipping, lazy advance, the
non-closing entry
- * stream, and cleanup). Nothing here touches local disk -- entries are
consumed as streams.
+ * 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 ArchiveReaderSuite extends SparkFunSuite {
+class SupportsArchiveFormatSuite extends SparkFunSuite {
private case class Entry(name: String, data: Array[Byte], isDir: Boolean =
false)
@@ -86,7 +86,7 @@ class ArchiveReaderSuite extends SparkFunSuite {
* 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 `ZipArchiveReader` documents;
`ZipArchiveOutputStream` cannot produce it
+ * non-streamable case the zip reader documents; `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.
*/
@@ -136,10 +136,11 @@ class ArchiveReaderSuite extends SparkFunSuite {
out.toByteArray
}
- /** Drains every entry into `(name, decodedText)` pairs through
`ArchiveReader.readEntries`. */
+ /** Drains every entry into `(name, decodedText)` pairs through
`SupportsArchiveFormat`. */
private def collect(file: File): Seq[(String, String)] =
- ArchiveReader(new Path(file.toURI)).readEntries(new Configuration()) {
(name, in) =>
- Iterator.single((name, new String(readAll(in), StandardCharsets.UTF_8)))
+ SupportsArchiveFormat.readArchiveEntries(new Path(file.toURI), new
Configuration()) {
+ (name, in) =>
+ Iterator.single((name, new String(readAll(in),
StandardCharsets.UTF_8)))
}.toList
// ----- isArchivePath ------------------------------------------------------
@@ -151,20 +152,20 @@ class ArchiveReaderSuite extends SparkFunSuite {
"foo.tgz", "FOO.TGZ", "/a/b/c/x.tgz",
"data.zip", "FOO.ZIP", "weird.ZiP", "/a/b/c/x.zip"
).foreach { p =>
- assert(ArchiveReader.isArchivePath(new Path(p)), s"expected archive
match for $p")
+ assert(SupportsArchiveFormat.isArchivePath(new Path(p)), s"expected
archive match for $p")
}
}
test("isArchivePath: negative cases") {
Seq("foo.csv", "foo.gz", "foo", "dir/", "foo.tarball",
"foo.tar.bz2", "foo.targz", "foo.zipx", "foo.gzip").foreach { p =>
- assert(!ArchiveReader.isArchivePath(new Path(p)), s"expected non-match
for $p")
+ assert(!SupportsArchiveFormat.isArchivePath(new Path(p)), s"expected
non-match for $p")
}
}
- // ----- readEntries --------------------------------------------------------
+ // ----- readArchiveEntries
--------------------------------------------------------
- test("readEntries: empty tar yields empty iterator") {
+ test("readArchiveEntries: empty tar yields empty iterator") {
withTempDir { dir =>
val tar = new File(dir, "empty.tar")
writeTar(tar, Seq.empty)
@@ -172,7 +173,7 @@ class ArchiveReaderSuite extends SparkFunSuite {
}
}
- test("readEntries: single entry exposes its name and bytes") {
+ test("readArchiveEntries: single entry exposes its name and bytes") {
withTempDir { dir =>
val tar = new File(dir, "single.tar")
writeTar(tar, Seq(textEntry("only.csv", "hello\n")))
@@ -180,7 +181,7 @@ class ArchiveReaderSuite extends SparkFunSuite {
}
}
- test("readEntries: multiple entries chained in tar order") {
+ test("readArchiveEntries: multiple entries chained in tar order") {
withTempDir { dir =>
val tar = new File(dir, "multi.tar")
writeTar(tar, Seq(textEntry("a.csv", "a"), textEntry("b.csv", "b"),
textEntry("c.csv", "c")))
@@ -188,7 +189,7 @@ class ArchiveReaderSuite extends SparkFunSuite {
}
}
- test("readEntries: gzipped tar (.tar.gz) via Hadoop codec factory") {
+ test("readArchiveEntries: gzipped tar (.tar.gz) via Hadoop codec factory") {
withTempDir { dir =>
val tarGz = new File(dir, "data.tar.gz")
writeTarGz(tarGz, Seq(textEntry("a.csv", "a"), textEntry("b.csv", "b")))
@@ -196,7 +197,7 @@ class ArchiveReaderSuite extends SparkFunSuite {
}
}
- test("readEntries: gzipped tar (.tgz) via explicit GZIPInputStream wrap") {
+ test("readArchiveEntries: gzipped tar (.tgz) via explicit GZIPInputStream
wrap") {
withTempDir { dir =>
val tgz = new File(dir, "data.tgz")
writeTarGz(tgz, Seq(textEntry("a.csv", "a"), textEntry("b.csv", "b")))
@@ -204,7 +205,7 @@ class ArchiveReaderSuite extends SparkFunSuite {
}
}
- test("readEntries: directory entries are skipped") {
+ test("readArchiveEntries: directory entries are skipped") {
withTempDir { dir =>
val tar = new File(dir, "dirs.tar")
writeTar(tar, Seq(
@@ -214,7 +215,7 @@ class ArchiveReaderSuite extends SparkFunSuite {
}
}
- test("readEntries: dotfile, underscore-marker, and prefixed-dir entries are
skipped") {
+ test("readArchiveEntries: dotfile, underscore-marker, and prefixed-dir
entries are skipped") {
withTempDir { dir =>
val tar = new File(dir, "skipped.tar")
writeTar(tar, Seq(
@@ -228,22 +229,22 @@ class ArchiveReaderSuite extends SparkFunSuite {
}
}
- test("readEntries: a custom ignoredPathSegmentRegex pattern surfaces hidden
entries") {
+ test("readArchiveEntries: a custom ignoredPathSegmentRegex pattern surfaces
hidden entries") {
withTempDir { dir =>
val tar = new File(dir, "custom-filter.tar")
writeTar(tar, Seq(textEntry("_SUCCESS", "marker"), textEntry("real.csv",
"kept")))
// "(?!)" matches nothing, so hidden-name filtering is disabled and only
the carve-outs in
// HadoopFSUtils.shouldFilterOutPathName still apply -- mirroring a
loose-file listing with
// the ignoredPathSegmentRegex option set to the same regex.
- val entries = ArchiveReader(new Path(tar.toURI))
- .readEntries(new Configuration(), Pattern.compile("(?!)")) { (name,
in) =>
- Iterator.single((name, new String(readAll(in),
StandardCharsets.UTF_8)))
- }.toList
+ val entries = SupportsArchiveFormat.readArchiveEntries(
+ new Path(tar.toURI), new Configuration(), Pattern.compile("(?!)")) {
(name, in) =>
+ Iterator.single((name, new String(readAll(in),
StandardCharsets.UTF_8)))
+ }.toList
assert(entries == Seq("_SUCCESS" -> "marker", "real.csv" -> "kept"))
}
}
- test("readEntries: advances lazily, one entry at a time") {
+ test("readArchiveEntries: advances lazily, one entry at a time") {
withTempDir { dir =>
val tar = new File(dir, "lazy.tar")
writeTar(tar, Seq(textEntry("a.csv", "a"), textEntry("b.csv", "b"),
textEntry("c.csv", "c")))
@@ -251,9 +252,10 @@ class ArchiveReaderSuite extends SparkFunSuite {
val opened = ArrayBuffer[String]()
// 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 = ArchiveReader(new Path(tar.toURI)).readEntries(new
Configuration()) { (name, _) =>
- opened += name
- Iterator.single(name)
+ val it = SupportsArchiveFormat.readArchiveEntries(new Path(tar.toURI),
new Configuration()) {
+ (name, _) =>
+ opened += name
+ Iterator.single(name)
}
// Construction opens only the first entry; later entries open on demand
as iteration
@@ -272,30 +274,31 @@ class ArchiveReaderSuite extends SparkFunSuite {
}
}
- test("readEntries: a parseEntry that closes its stream still advances to the
next entry") {
+ test("readArchiveEntries: a parseEntry that closes its stream still
advances") {
withTempDir { dir =>
val tar = new File(dir, "close.tar")
writeTar(tar, Seq(textEntry("a.csv", "a"), textEntry("b.csv", "b")))
val seen = ArrayBuffer[String]()
- val it = ArchiveReader(new Path(tar.toURI)).readEntries(new
Configuration()) { (name, in) =>
- val body = new String(readAll(in), StandardCharsets.UTF_8)
- in.close() // must NOT close the underlying archive
- seen += body
- Iterator.single(name)
+ val it = SupportsArchiveFormat.readArchiveEntries(new Path(tar.toURI),
new Configuration()) {
+ (name, in) =>
+ val body = new String(readAll(in), StandardCharsets.UTF_8)
+ in.close() // must NOT close the underlying archive
+ seen += body
+ Iterator.single(name)
}
assert(it.toList == List("a.csv", "b.csv"))
assert(seen.toList == List("a", "b"))
}
}
- test("readEntries: close() is safe, idempotent, and stops iteration") {
+ test("readArchiveEntries: close() is safe, idempotent, and stops iteration")
{
withTempDir { dir =>
val tar = new File(dir, "closeable.tar")
writeTar(tar, Seq(textEntry("a.csv", "a"), textEntry("b.csv", "b")))
- val it = ArchiveReader(new Path(tar.toURI)).readEntries(new
Configuration()) { (name, _) =>
- Iterator.single(name)
+ val it = SupportsArchiveFormat.readArchiveEntries(new Path(tar.toURI),
new Configuration()) {
+ (name, _) => Iterator.single(name)
}
assert(it.hasNext)
it.asInstanceOf[Closeable].close()
@@ -304,7 +307,7 @@ class ArchiveReaderSuite extends SparkFunSuite {
}
}
- test("readEntries: TaskContext completion cleans up without error") {
+ test("readArchiveEntries: TaskContext completion cleans up without error") {
withTempDir { dir =>
val tar = new File(dir, "ctx.tar")
writeTar(tar, Seq(textEntry("a.csv", "a"), textEntry("b.csv", "b")))
@@ -322,8 +325,9 @@ class ArchiveReaderSuite extends SparkFunSuite {
cpus = 1)
TaskContext.setTaskContext(ctx)
try {
- val it = ArchiveReader(new Path(tar.toURI)).readEntries(new
Configuration()) { (name, _) =>
- Iterator.single(name)
+ val it = SupportsArchiveFormat.readArchiveEntries(
+ new Path(tar.toURI), new Configuration()) {
+ (name, _) => Iterator.single(name)
}
assert(it.hasNext)
it.next() // open the archive and register the completion listener
@@ -339,7 +343,7 @@ class ArchiveReaderSuite extends SparkFunSuite {
// 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("readEntries: empty zip yields empty iterator") {
+ test("readArchiveEntries: empty zip yields empty iterator") {
withTempDir { dir =>
val zip = new File(dir, "empty.zip")
writeZip(zip, Seq.empty)
@@ -347,7 +351,7 @@ class ArchiveReaderSuite extends SparkFunSuite {
}
}
- test("readEntries: zip single entry exposes its name and bytes") {
+ test("readArchiveEntries: zip single entry exposes its name and bytes") {
withTempDir { dir =>
val zip = new File(dir, "single.zip")
writeZip(zip, Seq(textEntry("only.csv", "hello\n")))
@@ -355,7 +359,7 @@ class ArchiveReaderSuite extends SparkFunSuite {
}
}
- test("readEntries: zip multiple entries chained in archive order") {
+ test("readArchiveEntries: zip multiple entries chained in archive order") {
withTempDir { dir =>
val zip = new File(dir, "multi.zip")
writeZip(zip, Seq(textEntry("a.csv", "a"), textEntry("b.csv", "b"),
textEntry("c.csv", "c")))
@@ -363,7 +367,7 @@ class ArchiveReaderSuite extends SparkFunSuite {
}
}
- test("readEntries: zip directory entries are skipped") {
+ test("readArchiveEntries: zip directory entries are skipped") {
withTempDir { dir =>
val zip = new File(dir, "dirs.zip")
writeZip(zip, Seq(
@@ -373,7 +377,7 @@ class ArchiveReaderSuite extends SparkFunSuite {
}
}
- test("readEntries: zip dotfile, underscore-marker, and prefixed-dir entries
are skipped") {
+ test("readArchiveEntries: zip dotfile, underscore-marker, and prefixed-dir
entries are skipped") {
withTempDir { dir =>
val zip = new File(dir, "skipped.zip")
writeZip(zip, Seq(
@@ -387,15 +391,16 @@ class ArchiveReaderSuite extends SparkFunSuite {
}
}
- test("readEntries: zip advances lazily, one entry at a time") {
+ test("readArchiveEntries: zip advances lazily, one entry at a time") {
withTempDir { dir =>
val zip = new File(dir, "lazy.zip")
writeZip(zip, Seq(textEntry("a.csv", "a"), textEntry("b.csv", "b"),
textEntry("c.csv", "c")))
val opened = ArrayBuffer[String]()
- val it = ArchiveReader(new Path(zip.toURI)).readEntries(new
Configuration()) { (name, _) =>
- opened += name
- Iterator.single(name)
+ val it = SupportsArchiveFormat.readArchiveEntries(new Path(zip.toURI),
new Configuration()) {
+ (name, _) =>
+ opened += name
+ Iterator.single(name)
}
// Construction opens only the first entry; advancing past each boundary
opens the next.
assert(opened.toList == List("a.csv"))
@@ -411,24 +416,25 @@ class ArchiveReaderSuite extends SparkFunSuite {
}
}
- test("readEntries: a zip parseEntry that closes its stream still advances to
the next entry") {
+ test("readArchiveEntries: a zip parseEntry that closes its stream still
advances") {
withTempDir { dir =>
val zip = new File(dir, "close.zip")
writeZip(zip, Seq(textEntry("a.csv", "a"), textEntry("b.csv", "b")))
val seen = ArrayBuffer[String]()
- val it = ArchiveReader(new Path(zip.toURI)).readEntries(new
Configuration()) { (name, in) =>
- val body = new String(readAll(in), StandardCharsets.UTF_8)
- in.close() // must NOT close the underlying archive
- seen += body
- Iterator.single(name)
+ val it = SupportsArchiveFormat.readArchiveEntries(new Path(zip.toURI),
new Configuration()) {
+ (name, in) =>
+ val body = new String(readAll(in), StandardCharsets.UTF_8)
+ in.close() // must NOT close the underlying archive
+ seen += body
+ Iterator.single(name)
}
assert(it.toList == List("a.csv", "b.csv"))
assert(seen.toList == List("a", "b"))
}
}
- test("readEntries: a non-streamable zip entry fails loudly rather than
yielding garbled bytes") {
+ test("readArchiveEntries: a non-streamable zip entry fails loudly, not with
garbled bytes") {
withTempDir { dir =>
val zip = new File(dir, "stored-dd.zip")
writeStoredEntryWithDataDescriptor(zip, "a.csv", "hello")
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/TextArchiveReadBase.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/TextArchiveReadBase.scala
index e2506afa6b34..ba0cc64e1b35 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/TextArchiveReadBase.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/TextArchiveReadBase.scala
@@ -28,9 +28,9 @@ import org.apache.spark.sql.test.SharedSparkSession
import org.apache.spark.util.Utils
/**
- * Reads of text files packed in archives, streamed through the
[[ArchiveReader]] path. Entries are
- * streamed (never unpacked to disk), and the central contract verified
throughout is parity with
- * reading the same files from a directory.
+ * Reads of text files packed in archives, streamed through the
[[SupportsArchiveFormat]] path.
+ * Entries are streamed (never unpacked to disk), and the central contract
verified throughout is
+ * parity with reading the same files from a directory.
*
* Unlike CSV/JSON this does not extend [[ArchiveReadSuiteBase]]: the text
data source has a single
* fixed `value` column (one row per line, or per entry with `wholetext`) and
no schema inference,
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
