HyukjinKwon commented on code in PR #56784:
URL: https://github.com/apache/spark/pull/56784#discussion_r3478114506
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/ArchiveReader.scala:
##########
@@ -67,25 +106,109 @@ abstract class ArchiveReader(path: Path) {
def readEntries[T](
conf: Configuration,
ignoredPathSegmentRegex: Pattern =
HadoopFSUtils.defaultIgnoredPathSegmentRegexPattern)(
- parseEntry: (String, InputStream) => Iterator[T]): Iterator[T]
+ 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
Review Comment:
The advance loop calls `parseEntry` on every non-skipped entry without
consulting `ArchiveInputStream.canReadEntryData(entry)`. For
`ZipArchiveInputStream` that returns false for entries it can't stream — a
STORED entry with a trailing data descriptor, an unsupported compression
method, or encryption — i.e. exactly the "a few unusual zips … are not
streamable this way" case the new Scaladoc documents.
Question: on such an entry, does the current path fail loudly with a clear
error, or can it silently yield a truncated/garbled entry? If the latter is
possible, consider guarding here:
```scala
if (entry != null && !archive.canReadEntryData(entry)) {
throw <clear error naming the unsupported zip feature>
}
```
so the documented limitation surfaces deterministically rather than
depending on `ZipArchiveInputStream`'s default read behavior.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/ArchiveReader.scala:
##########
@@ -181,90 +286,19 @@ class TarArchiveReader(path: Path) extends
ArchiveReader(path) {
throw e
}
}
+}
- /**
- * Wraps the shared tar stream as a view over exactly the current entry's
bytes
- * (`TarArchiveInputStream.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(tar: TarArchiveInputStream): InputStream =
- CloseShieldInputStream.wrap(tar)
-
- override def readEntries[T](
- conf: Configuration,
- ignoredPathSegmentRegex: Pattern)(
- parseEntry: (String, InputStream) => Iterator[T]): Iterator[T] = {
- val tar = openTarStream(conf)
- var closed = false
-
- def cleanup(): Unit = {
- if (!closed) {
- closed = true
- try tar.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 = tar.getNextEntry
- while (entry != null && shouldSkipEntry(entry,
ignoredPathSegmentRegex)) {
- entry = tar.getNextEntry
- }
- if (entry == null) {
- done = true
- cleanup()
- } else {
- currentIter = parseEntry(entry.getName, entryStream(tar))
- }
- }
- }
-
- 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()
- }
- }
+/**
+ * [[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 deflated entry whose size is recorded
only in a trailing data
+ * descriptor) are not streamable this way.
+ */
+class ZipArchiveReader(path: Path) extends ArchiveReader(path) {
- // 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
- }
+ override protected def openArchiveStream(
+ conf: Configuration): ArchiveInputStream[_ <: ArchiveEntry] =
+ new
ZipArchiveInputStream(CodecStreams.createInputStreamWithCloseResource(conf,
path))
Review Comment:
The "not streamable" limitation this `ZipArchiveReader` documents has no
test pinning its behavior. `ZipArchiveTestUtils.writeArchive` uses
`ZipArchiveOutputStream` (good — an independent producer, distinct from the
reader) and there's a `writeCorruptArchive` negative case, but nothing
exercises a *valid-but-non-streamable* zip. Consider adding a fixture (e.g. a
STORED entry written with a data descriptor) that asserts the chosen behavior,
so a future change can't silently turn the `canReadEntryData` gap above into
corrupt rows.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
