cloud-fan commented on code in PR #56784:
URL: https://github.com/apache/spark/pull/56784#discussion_r3479025094
##########
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
Review Comment:
The Scaladoc cites "a **deflated** entry whose size is recorded only in a
trailing data descriptor" as not streamable, but per commons-compress 1.28.0 a
DEFLATED entry with a data descriptor *is* streamable:
`supportsDataDescriptorFor` and `supportsCompressedSizeFor` both return true
when `method == DEFLATED`, and `readDeflated` detects end-of-stream via
`inf.finished()` independent of the declared size, so `canReadEntryData` is
true and `read()` doesn't throw.
The actually non-streamable case is a **STORED** entry with a data
descriptor — `supportsDataDescriptorFor` returns false (STORED + data
descriptor, with `allowStoredEntriesWithDataDescriptor` false by default), so
`read()` throws `UnsupportedZipFeatureException`. Your own self-review says
"STORED", and `ZipArchiveTestUtils.writeArchive` (writing through a
non-seekable `FileOutputStream`, which forces data descriptors on DEFLATED
entries) produces DEFLATED+data-descriptor entries that stream fine — which
confirms the doc's example is the streamable case. Same wording is worth fixing
in the PR description too.
```suggestion
* model: a few unusual zips (e.g. a stored entry whose size is recorded
only in a trailing data
```
--
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]
