uros-b commented on code in PR #57588:
URL: https://github.com/apache/spark/pull/57588#discussion_r3667491709
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/SupportsArchiveFormat.scala:
##########
@@ -233,25 +227,96 @@ object SupportsArchiveFormat {
}
}
- /**
- * Opens a `.7z` archive by seeking.
- *
- * @param path the archive path
- * @param conf Hadoop configuration used to open the archive
- * @return the archive's entries as an [[ArchiveInputStream]] cursor
- */
- private def openSevenZStream(
- path: Path,
- conf: Configuration): ArchiveInputStream[_ <: ArchiveEntry] = {
+ /** Opens a `.tar`/`.tar.gz`/`.tgz` archive, streaming its entries through
one forward cursor. */
+ private def openTarStream(path: Path, conf: Configuration): ArchiveEntries =
{
+ val gzipped = path.getName.toLowerCase(Locale.ROOT).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 tar = new TarArchiveInputStream(if (gzipped) new
GZIPInputStream(base) else base)
+ val entries = Iterator.continually(tar.getNextEntry).takeWhile(_ != null)
+ .map((_, tar: InputStream))
+ closeable(entries, () => tar.close())
+ } catch {
+ case NonFatal(e) =>
+ try base.close() catch { case NonFatal(_) => }
+ throw e
+ }
+ }
+
+ /** Pairs an entry iterator with the resource it reads from, closed when the
caller is done. */
+ private def closeable(
+ entries: Iterator[(ArchiveEntry, InputStream)],
+ closeFn: () => Unit): ArchiveEntries =
+ new Iterator[(ArchiveEntry, InputStream)] with Closeable {
+ override def hasNext: Boolean = entries.hasNext
+ override def next(): (ArchiveEntry, InputStream) = entries.next()
+ override def close(): Unit = closeFn()
+ }
+
+ /** Opens a `.7z` archive by seeking. */
+ private def openSevenZStream(path: Path, conf: Configuration):
ArchiveEntries = {
val fs = path.getFileSystem(conf)
- val in = fs.open(path)
+ val length = fs.getFileStatus(path).getLen
+ var channel: SeekableByteChannel = null
+ var sevenZ: SevenZFile = null
try {
- val channel = new HadoopSeekableByteChannel(in,
fs.getFileStatus(path).getLen)
- new SevenZArchiveInputStream(
- SevenZFile.builder().setSeekableByteChannel(channel).get(), channel)
+ channel = new HadoopSeekableByteChannel(fs.open(path), length)
+ sevenZ = SevenZFile.builder().setSeekableByteChannel(channel).get()
+ // SevenZFile is a forward cursor: one stream serves whichever entry
getNextEntry selects.
+ val entryStream = new SevenZEntryInputStream(sevenZ)
+ val entries = Iterator.continually(sevenZ.getNextEntry).takeWhile(_ !=
null)
+ .map((_, entryStream))
+ closeable(entries, () => {
+ try {
+ sevenZ.close()
+ } finally {
+ channel.close()
+ }
+ })
} catch {
case NonFatal(e) =>
- try in.close() catch { case NonFatal(_) => }
+ Utils.closeQuietly(sevenZ)
+ Utils.closeQuietly(channel)
+ throw e
+ }
+ }
+
+ /** Opens a `.zip` archive by seeking, reading the central directory first.
*/
+ private def openZipStream(path: Path, conf: Configuration): ArchiveEntries =
{
+ val fs = path.getFileSystem(conf)
+ val length = fs.getFileStatus(path).getLen
+ var channel: SeekableByteChannel = null
+ var zipFile: ZipFile = null
+ try {
+ channel = new HadoopSeekableByteChannel(fs.open(path), length)
+ zipFile = ZipFile.builder().setSeekableByteChannel(channel).get()
+ var current: InputStream = null
+ def closeCurrentStream(): Unit = if (current != null) {
+ try current.close() catch { case NonFatal(_) => }
+ current = null
+ }
+ val entries = zipFile.getEntries.asScala.map { entry =>
+ closeCurrentStream()
+ if (!zipFile.canReadEntryData(entry)) {
Review Comment:
zipFile.canReadEntryData(entry) is evaluated eagerly inside the .map closure
before shouldSkipEntry is consulted in advance(). This means an entry that the
engine would otherwise skip (e.g. a directory entry or dotfile carrying an
encrypted GP flag) throws CANNOT_READ_ZIP_ENTRY instead of being silently
skipped; a semantic regression from the old ZipArchiveInputStream path. The fix
is a one-liner: guard the check with if (!entry.isDirectory &&
!zipFile.canReadEntryData(entry)), or restructure to run shouldSkipEntry before
canReadEntryData. The corner case is currently untested.
--
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]
