Repository: codalab/codalab-worksheets (language: python)

Issue:
Azure Blob Storage - find right archive structure, use .tar.gz (ratarmount) instead of .zip
Currently, uploads and downloads are tar.gz, but in Azure blob store, they will be zip files.


Discussion / hints:
Need zip for random access.  Also look into bzip.
See https://blastedbio.blogspot.com/2013/04/random-access-to-blocked-xz-format-bxzf.html for more info about xz files.
Another option, which might actually be optimal because it keeps the extension .tar.gz: BGZF (https://blastedbio.blogspot.com/2011/11/bgzf-blocked-bigger-better-gzip.html)

A library for Python that works well with BGZF: https://biopython.org/docs/1.75/api/Bio.bgzf.html
@percyliang I've added a short writeup of what we can use for the archive structure at the top of this design doc: https://docs.google.com/document/d/1hfpfXBvoooIV6_cerKhnDu7X7t9o5hzGD7rtlTlDK_s/edit#

Fix patch:
diff --git a/codalab/common.py b/codalab/common.py
index fb13e3b31..862b877b5 100644
--- a/codalab/common.py
+++ b/codalab/common.py
@@ -158,27 +158,27 @@ class LinkedBundlePath:
 
         bundle_path (str): Path to the bundle contents in that particular storage.
 
-        is_zip (bool): Whether this bundle is stored as a zip file on this storage medium stores folders. Only done currently by Azure Blob Storage.
+        is_archive (bool): Whether this bundle is stored as a .tar.gz file on this storage medium stores folders. Only done currently by Azure Blob Storage.
 
         uses_beam (bool): Whether this bundle's storage type requires using Apache Beam to interact with it.
 
-        zip_subpath (str): If is_zip is True, returns the subpath within the zip file for the file that this BundlePath points to.
+        archive_subpath (str): If is_archive is True, returns the subpath within the archive file for the file that this BundlePath points to.
 
         bundle_uuid (str): UUID of the bundle that this path refers to.
     """
 
     storage_type: StorageType
     bundle_path: str
-    is_zip: bool
+    is_archive: bool
     uses_beam: bool
-    zip_subpath: str
+    archive_subpath: str
     bundle_uuid: str
 
 
 def parse_linked_bundle_url(url):
     """Parses a linked bundle URL. This bundle URL can refer to:
         - a single file: "azfs://storageclwsdev0/bundles/uuid/contents"
-        - a single file that is stored within a zip file: "azfs://storageclwsdev0/bundles/uuid/contents.zip/file1"
+        - a single file that is stored within an archive file: "azfs://storageclwsdev0/bundles/uuid/contents.tar.gz/file1"
 
         Returns a LinkedBundlePath instance to encode this information.
     """
@@ -188,21 +188,21 @@ def parse_linked_bundle_url(url):
         url = url[len("azfs://") :]
         storage_account, container, bundle_uuid, contents_file, *remainder = url.split("/", 4)
         bundle_path = f"azfs://{storage_account}/{container}/{bundle_uuid}/{contents_file}"
-        is_zip = contents_file.endswith(".zip")
-        zip_subpath = remainder[0] if is_zip and len(remainder) else None
+        is_archive = contents_file.endswith(".tar.gz")
+        archive_subpath = remainder[0] if is_archive and len(remainder) else None
     else:
         storage_type = StorageType.FILE_STORAGE
         bundle_path = url
-        is_zip = False
+        is_archive = False
         uses_beam = False
-        zip_subpath = None
+        archive_subpath = None
         bundle_uuid = None
     return LinkedBundlePath(
         storage_type=storage_type,
         bundle_path=bundle_path,
-        is_zip=is_zip,
+        is_archive=is_archive,
         uses_beam=uses_beam,
-        zip_subpath=zip_subpath,
+        archive_subpath=archive_subpath,
         bundle_uuid=bundle_uuid,
     )
 
diff --git a/codalab/lib/beam/mockblobstoragefilesystem.py b/codalab/lib/beam/mockblobstoragefilesystem.py
index 09714a522..a22b7f774 100644
--- a/codalab/lib/beam/mockblobstoragefilesystem.py
+++ b/codalab/lib/beam/mockblobstoragefilesystem.py
@@ -9,6 +9,7 @@
 from apache_beam.io.filesystem import FileSystem
 from apache_beam.io.localfilesystem import LocalFileSystem
 import os
+from io import BytesIO
 
 __all__ = ['MockBlobStorageFileSystem']
 
@@ -63,7 +64,7 @@ def open(
       self,
       path,
       *args, **kwargs):
-    return super().open(self._azfs_to_local(path), *args, **kwargs)
+    return BytesIO(super().open(self._azfs_to_local(path), *args, **kwargs).read())
 
   def copy(self, source_file_names, destination_file_names):
     return super().copy([self._azfs_to_local(p) for p in source_file_names], [self._azfs_to_local(p) for p in destination_file_names])
diff --git a/codalab/lib/beam/ratarmount.py b/codalab/lib/beam/ratarmount.py
new file mode 100644
index 000000000..a149095f8
--- /dev/null
+++ b/codalab/lib/beam/ratarmount.py
@@ -0,0 +1,2065 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+
+import argparse
+import bisect
+import collections
+import io
+import json
+import os
+import re
+import sqlite3
+import stat
+import sys
+import tarfile
+import tempfile
+import time
+import traceback
+from timeit import default_timer as timer
+import typing
+from typing import Any, AnyStr, BinaryIO, Dict, IO, Iterable, List, Optional, Set, Tuple, Union
+
+
+# Can't do this dynamically with importlib.import_module and using supportedCompressions
+# because then the static checkers like mypy and pylint won't recognize the modules!
+try:
+    import indexed_bzip2
+except ImportError:
+    pass
+try:
+    import indexed_gzip
+except ImportError:
+    pass
+try:
+    import indexed_zstd
+except ImportError:
+    pass
+try:
+    import lzmaffi
+except ImportError:
+    pass
+
+
+__version__ = '0.7.0'
+
+
+# Defining lambdas does not yet check the names of entities used inside the lambda!
+CompressionInfo = collections.namedtuple(
+    'CompressionInfo', ['suffixes', 'doubleSuffixes', 'moduleName', 'checkHeader', 'open']
+)
+supportedCompressions = {
+    'bz2': CompressionInfo(
+        ['bz2', 'bzip2'],
+        ['tb2', 'tbz', 'tbz2', 'tz2'],
+        'indexed_bzip2',
+        lambda x: (x.read(4)[:3] == b'BZh' and x.read(6) == (0x314159265359).to_bytes(6, 'big')),
+        lambda x: indexed_bzip2.IndexedBzip2File(x.fileno()),
+    ),
+    'gz': CompressionInfo(
+        ['gz', 'gzip'],
+        ['taz', 'tgz'],
+        'indexed_gzip',
+        lambda x: x.read(2) == b'\x1F\x8B',
+        lambda x: indexed_gzip.IndexedGzipFile(fileobj=x),
+    ),
+    'xz': CompressionInfo(
+        ['xz'], ['txz'], 'lzmaffi', lambda x: x.read(6) == b"\xFD7zXZ\x00", lambda x: lzmaffi.open(x)
+    ),
+    'zst': CompressionInfo(
+        ['zst', 'zstd'],
+        ['tzst'],
+        'indexed_zstd',
+        lambda x: x.read(4) == (0xFD2FB528).to_bytes(4, 'little'),
+        lambda x: indexed_zstd.IndexedZstdFile(x.fileno()),
+    ),
+}
+
+
+def stripSuffixFromCompressedFile(path: str) -> str:
+    """Strips compression suffixes like .bz2, .gz, ..."""
+    for compression in supportedCompressions.values():
+        for suffix in compression.suffixes:
+            if path.lower().endswith('.' + suffix.lower()):
+                return path[: -(len(suffix) + 1)]
+    return path
+
+
+def stripSuffixFromTarFile(path: str) -> str:
+    """Strips extensions like .tar.gz or .gz or .tgz, ..."""
+    # 1. Try for conflated suffixes first
+    for compression in supportedCompressions.values():
+        for suffix in compression.doubleSuffixes + ['t' + s for s in compression.suffixes]:
+            if path.lower().endswith('.' + suffix.lower()):
+                return path[: -(len(suffix) + 1)]
+
+    # 2. Remove compression suffixes
+    path = stripSuffixFromCompressedFile(path)
+
+    # 3. Remove .tar if we are left with it after the compression suffix removal
+    if path.lower().endswith('.tar'):
+        path = path[:-4]
+
+    return path
+
+
+printDebug = 1
+
+
+class RatarmountError(Exception):
+    """Base exception for ratarmount module."""
+
+
+class IndexNotOpenError(RatarmountError):
+    """Exception for operations executed on a closed index database."""
+
+
+class InvalidIndexError(RatarmountError):
+    """Exception for indexes being invalid, outdated, or created with different arguments."""
+
+
+class CompressionError(RatarmountError):
+    """Exception for trying to open files with unsupported compression or unavailable decompression module."""
+
+
+def overrides(parentClass):
+    """Simple decorator that checks that a method with the same name exists in the parent class"""
+
+    def overrider(method):
+        assert method.__name__ in dir(parentClass)
+        assert callable(getattr(parentClass, method.__name__))
+        return method
+
+    return overrider
+
+
+class ProgressBar:
+    """Simple progress bar which keeps track of changes and prints the progress and a time estimate."""
+
+    def __init__(self, maxValue: float):
+        # fmt: off
+        self.maxValue        = maxValue
+        self.lastUpdateTime  = time.time()
+        self.lastUpdateValue = 0.
+        self.updateInterval  = 2.  # seconds
+        self.creationTime    = time.time()
+        # fmt: on
+
+    def update(self, value: float) -> None:
+        """Should be called whenever the monitored value changes. The progress bar is updated accordingly."""
+        if self.lastUpdateTime is not None and (time.time() - self.lastUpdateTime) < self.updateInterval:
+            return
+
+        # Use whole interval since start to estimate time
+        eta1 = int((time.time() - self.creationTime) / value * (self.maxValue - value))
+        # Use only a shorter window interval to estimate time.
+        # Accounts better for higher speeds in beginning, e.g., caused by caching effects.
+        # However, this estimate might vary a lot while the other one stabilizes after some time!
+        eta2 = int((time.time() - self.lastUpdateTime) / (value - self.lastUpdateValue) * (self.maxValue - value))
+        print(
+            "Currently at position {} of {} ({:.2f}%). "
+            "Estimated time remaining with current rate: {} min {} s, with average rate: {} min {} s.".format(
+                # fmt:off
+                value, self.maxValue, value / self.maxValue * 100.0,
+                eta2 // 60, eta2 % 60,
+                eta1 // 60, eta1 % 60
+                # fmt:on
+            ),
+            flush=True,
+        )
+
+        self.lastUpdateTime = time.time()
+        self.lastUpdateValue = value
+
+
+class StenciledFile(io.BufferedIOBase):
+    """A file abstraction layer giving a stenciled view to an underlying file."""
+
+    def __init__(self, fileobj: IO, stencils: List[Tuple[int, int]]) -> None:
+        """
+        stencils: A list tuples specifying the offset and length of the underlying file to use.
+                  The order of these tuples will be kept.
+                  The offset must be non-negative and the size must be positive.
+
+        Examples:
+            stencil = [(5,7)]
+                Makes a new 7B sized virtual file starting at offset 5 of fileobj.
+            stencil = [(0,3),(5,3)]
+                Make a new 6B sized virtual file containing bytes [0,1,2,5,6,7] of fileobj.
+            stencil = [(0,3),(0,3)]
+                Make a 6B size file containing the first 3B of fileobj twice concatenated together.
+        """
+
+        # fmt: off
+        self.fileobj = fileobj
+        self.offsets = [x[0] for x in stencils]
+        self.sizes   = [x[1] for x in stencils]
+        self.offset  = 0
+        # fmt: on
+
+        # Calculate cumulative sizes
+        self.cumsizes = [0]
+        for offset, size in stencils:
+            assert offset >= 0
+            assert size > 0
+            self.cumsizes.append(self.cumsizes[-1] + size)
+
+        # Seek to the first stencil offset in the underlying file so that "read" will work out-of-the-box
+        self.seek(0)
+
+    def _findStencil(self, offset: int) -> int:
+        """
+        Return index to stencil where offset belongs to. E.g., for stencils [(3,5),(8,2)], offsets 0 to
+        and including 4 will still be inside stencil (3,5), i.e., index 0 will be returned. For offset 6,
+        index 1 would be returned because it now is in the second contiguous region / stencil.
+        """
+        # bisect_left( value ) gives an index for a lower range: value < x for all x in list[0:i]
+        # Because value >= 0 and list starts with 0 we can therefore be sure that the returned i>0
+        # Consider the stencils [(11,2),(22,2),(33,2)] -> cumsizes [0,2,4,6]. Seek to offset 2 should seek to 22.
+        assert offset >= 0
+        i = bisect.bisect_left(self.cumsizes, offset + 1) - 1
+        assert i >= 0
+        return i
+
+    @overrides(io.BufferedIOBase)
+    def close(self) -> None:
+        # Don't close the object given to us
+        # self.fileobj.close()
+        pass
+
+    @overrides(io.BufferedIOBase)
+    def fileno(self) -> int:
+        return self.fileobj.fileno()
+
+    @overrides(io.BufferedIOBase)
+    def seekable(self) -> bool:
+        return self.fileobj.seekable()
+
+    @overrides(io.BufferedIOBase)
+    def readable(self) -> bool:
+        return self.fileobj.readable()
+
+    @overrides(io.BufferedIOBase)
+    def writable(self) -> bool:
+        return False
+
+    @overrides(io.BufferedIOBase)
+    def read(self, size: int = -1) -> bytes:
+        if size == -1:
+            size = self.cumsizes[-1] - self.offset
+
+        # This loop works in a kind of leapfrog fashion. On each even loop iteration it seeks to the next stencil
+        # and on each odd iteration it reads the data and increments the offset inside the stencil!
+        result = b''
+        i = self._findStencil(self.offset)
+        while size > 0 and i < len(self.sizes):
+            # Read as much as requested or as much as the current contiguous region / stencil still contains
+            readableSize = min(size, self.sizes[i] - (self.offset - self.cumsizes[i]))
+            if readableSize == 0:
+                # Go to next stencil
+                i += 1
+                if i >= len(self.offsets):
+                    break
+                self.fileobj.seek(self.offsets[i])
+            else:
+                # Actually read data
+                tmp = self.fileobj.read(readableSize)
+                self.offset += len(tmp)
+                result += tmp
+                size -= readableSize
+                # Now, either size is 0 or readableSize will be 0 in the next iteration
+
+        return result
+
+    @overrides(io.BufferedIOBase)
+    def seek(self, offset: int, whence: int = io.SEEK_SET) -> int:
+        if whence == io.SEEK_CUR:
+            self.offset += offset
+        elif whence == io.SEEK_END:
+            self.offset = self.cumsizes[-1] + offset
+        elif whence == io.SEEK_SET:
+            self.offset = offset
+
+        if self.offset < 0:
+            raise ValueError("Trying to seek before the start of the file!")
+        if self.offset >= self.cumsizes[-1]:
+            return self.offset
+
+        i = self._findStencil(self.offset)
+        offsetInsideStencil = self.offset - self.cumsizes[i]
+        assert offsetInsideStencil >= 0
+        assert offsetInsideStencil < self.sizes[i]
+        self.fileobj.seek(self.offsets[i] + offsetInsideStencil, io.SEEK_SET)
+
+        return self.offset
+
+    @overrides(io.BufferedIOBase)
+    def tell(self) -> int:
+        return self.offset
+
+
+# Names must be identical to the SQLite column headers!
+FileInfo = collections.namedtuple(
+    "FileInfo", "offsetheader offset size mtime mode type linkname uid gid istar issparse"
+)
+
+
+class SQLiteIndexedTar:
+    """
+    This class reads once through the whole TAR archive and stores TAR file offsets
+    for all contained files in an index to support fast seeking to a given file.
+    """
+
+    # Version 0.1.0:
+    #   - Initial version
+    # Version 0.2.0:
+    #   - Add sparse support and 'offsetheader' and 'issparse' columns to the SQLite database
+    #   - Add TAR file size metadata in order to quickly check whether the TAR changed
+    #   - Add 'offsetheader' to the primary key of the 'files' table so that files which were
+    #     updated in the TAR can still be accessed if necessary.
+    # Version 0.3.0:
+    #   - Add arguments influencing the created index to metadata (ignore-zeros, recursive, ...)
+    __version__ = '0.3.0'
+
+    def __init__(
+        # fmt: off
+        self,
+        tarFileName                : Optional[str]       = None,
+        fileObject                 : Optional[BinaryIO]  = None,
+        writeIndex                 : bool                = False,
+        clearIndexCache            : bool                = False,
+        indexFileName              : Optional[str]       = None,
+        indexFolders               : Optional[List[str]] = None,
+        recursive                  : bool                = False,
+        gzipSeekPointSpacing       : int                 = 4*1024*1024,
+        encoding                   : str                 = tarfile.ENCODING,
+        stripRecursiveTarExtension : bool                = False,
+        ignoreZeros                : bool                = False,
+        verifyModificationTime     : bool                = False,
+        # fmt: on
+    ) -> None:
+        """
+        tarFileName : Path to the TAR file to be opened. If not specified, a fileObject must be specified.
+                      If only a fileObject is given, the created index can't be cached (efficiently).
+        fileObject : A io.IOBase derived object. If not specified, tarFileName will be opened.
+                     If it is an instance of IndexedBzip2File, IndexedGzipFile, or IndexedZstdFile, then the offset
+                     loading and storing from and to the SQLite database is managed automatically by this class.
+        encoding : Will be forwarded to tarfile. Specifies how filenames inside the TAR are encoded.
+        ignoreZeros : Will be forwarded to tarfile. Specifies to not only skip zero blocks but also blocks with
+                      invalid data. Setting this to true can lead to some problems but is required to correctly
+                      read concatenated tars.
+        stripRecursiveTarExtension : If true and if recursive is also true, then a <file>.tar inside the current
+                                     tar will be mounted at <file>/ instead of <file>.tar/.
+        """
+
+        # stores which parent folders were last tried to add to database and therefore do exist
+        self.parentFolderCache: List[Tuple[str, str]] = []
+        self.sqlConnection: Optional[sqlite3.Connection] = None
+        self.indexFileName = None
+
+        # fmt: off
+        self.mountRecursively           = recursive
+        self.encoding                   = encoding
+        self.stripRecursiveTarExtension = stripRecursiveTarExtension
+        self.ignoreZeros                = ignoreZeros
+        self.verifyModificationTime     = verifyModificationTime
+        self.gzipSeekPointSpacing       = gzipSeekPointSpacing
+        # fmt: on
+
+        if not fileObject:
+            if not tarFileName:
+                raise ValueError("At least one of tarFileName and fileObject arguments should be set!")
+            self.tarFileName = os.path.abspath(tarFileName) if tarFileName else '<file object>'
+            fileObject = open(self.tarFileName, 'rb')
+
+        fileObject.seek(0, io.SEEK_END)
+        fileSize = fileObject.tell()
+        fileObject.seek(0)
+
+        # rawFileObject : Only set when opening a compressed file and only kept to keep the
+        #                 compressed file handle from being closed by the garbage collector.
+        # tarFileObject : File object to the uncompressed (or decompressed) TAR file to read actual data out of.
+        # compression   : Stores what kind of compression the originally specified TAR file uses.
+        # isTar         : Can be false for the degenerated case of only a bz2 or gz file not containing a TAR
+        self.tarFileObject, self.rawFileObject, self.compression, self.isTar = SQLiteIndexedTar._openCompressedFile(
+            fileObject, gzipSeekPointSpacing, encoding
+        )
+
+        if self.compression == 'xz':
+            try:
+                if len(self.tarFileObject.block_boundaries) <= 1 and (fileSize is None or fileSize > 1024 * 1024):
+                    print("[Warning] The specified file '{}'".format(self.tarFileName))
+                    print("[Warning] is compressed using xz but only contains one xz block. This makes it ")
+                    print("[Warning] impossible to use true seeking! Please (re)compress your TAR using pixz")
+                    print("[Warning] (see https://github.com/vasi/pixz) in order for ratarmount to do be able ")
+                    print("[Warning] to do fast seeking to requested files.")
+                    print("[Warning] As it is, each file access will decompress the whole TAR from the beginning!")
+                    print()
+            except Exception:
+                pass
+
+        if not tarFileName:
+            self.tarFileName = '<file object>'
+            self._createIndex(fileObject)
+            # return here because we can't find a save location without any identifying name
+            return
+        self.tarFileName = tarFileName
+
+        # will be used for storing indexes if current path is read-only
+        possibleIndexFilePaths = [self.tarFileName + ".index.sqlite"]
+        indexPathAsName = self.tarFileName.replace("/", "_") + ".index.sqlite"
+        if isinstance(indexFolders, str):
+            indexFolders = [indexFolders]
+        if indexFileName:
+            # A given index file name takes precedence and there should be no implicit fallback
+            possibleIndexFilePaths = [os.path.abspath(os.path.expanduser(indexFileName))]
+        elif indexFolders:
+            # An empty path is to be interpreted as the default path right besides the TAR
+            if '' not in indexFolders:
+                possibleIndexFilePaths = []
+            for folder in indexFolders:
+                if folder:
+                    indexPath = os.path.join(folder, indexPathAsName)
+                    possibleIndexFilePaths.append(os.path.abspath(os.path.expanduser(indexPath)))
+
+        if clearIndexCache:
+            for indexPath in possibleIndexFilePaths:
+                if os.path.isfile(indexPath):
+                    os.remove(indexPath)
+
+        # Try to find an already existing index
+        for indexPath in possibleIndexFilePaths:
+            if self._tryLoadIndex(indexPath):
+                self.indexFileName = indexPath
+                break
+        if self.indexIsLoaded():
+            self._loadOrStoreCompressionOffsets()
+            return
+
+        # Find a suitable (writable) location for the index database
+        if writeIndex:
+            for indexPath in possibleIndexFilePaths:
+                if self._pathIsWritable(indexPath) and self._pathCanBeUsedForSqlite(indexPath):
+                    self.indexFileName = indexPath
+                    break
+
+        if not self.indexFileName:
+            raise InvalidIndexError(
+                "Could not find any existing index or writable location for an index in " + str(possibleIndexFilePaths)
+            )
+
+        self._createIndex(self.tarFileObject)
+        self._loadOrStoreCompressionOffsets()  # store
+        if self.sqlConnection:
+            self._storeMetadata(self.sqlConnection)
+
+        if printDebug >= 1 and writeIndex:
+            # The 0-time is legacy for the automated tests
+            # fmt: off
+            print("Writing out TAR index to", self.indexFileName, "took 0s",
+                  "and is sized", os.stat( self.indexFileName ).st_size, "B")
+            # fmt: on
+
+    def _storeMetadata(self, connection: sqlite3.Connection) -> None:
+        self._storeVersionsMetadata(connection)
+
+        metadataTable = """
+            /* empty table whose sole existence specifies that we finished iterating the tar */
+            CREATE TABLE "metadata" (
+                "key"      VARCHAR(65535) NOT NULL, /* e.g. "tarsize" */
+                "value"    VARCHAR(65535) NOT NULL  /* e.g. size in bytes as integer */
+            );
+        """
+
+        connection.executescript(metadataTable)
+
+        # All of these require the generic "metadata" table.
+        self._storeTarMetadata(connection, self.tarFileName)
+        self._storeArgumentsMetadata(connection)
+        connection.commit()
+
+    @staticmethod
+    def _storeVersionsMetadata(connection: sqlite3.Connection) -> None:
+        versionsTable = """
+            /* This table's sole existence specifies that we finished iterating the tar for older ratarmount versions */
+            CREATE TABLE "versions" (
+                "name"     VARCHAR(65535) NOT NULL, /* which component the version belongs to */
+                "version"  VARCHAR(65535) NOT NULL, /* free form version string */
+                /* Semantic Versioning 2.0.0 (semver.org) parts if they can be specified:
+                 *   MAJOR version when you make incompatible API changes,
+                 *   MINOR version when you add functionality in a backwards compatible manner, and
+                 *   PATCH version when you make backwards compatible bug fixes. */
+                "major"    INTEGER,
+                "minor"    INTEGER,
+                "patch"    INTEGER
+            );
+        """
+        try:
+            connection.executescript(versionsTable)
+        except Exception as exception:
+            if printDebug >= 2:
+                print(exception)
+            print("[Warning] There was an error when adding metadata information. Index loading might not work.")
+
+        try:
+
+            def makeVersionRow(
+                versionName: str, version: str
+            ) -> Tuple[str, str, Optional[str], Optional[str], Optional[str]]:
+                versionNumbers = [re.sub('[^0-9]', '', x) for x in version.split('.')]
+                return (
+                    versionName,
+                    version,
+                    versionNumbers[0] if len(versionNumbers) > 0 else None,
+                    versionNumbers[1] if len(versionNumbers) > 1 else None,
+                    versionNumbers[2] if len(versionNumbers) > 2 else None,
+                )
+
+            versions = [
+                makeVersionRow('ratarmount', __version__),
+                makeVersionRow('index', SQLiteIndexedTar.__version__),
+            ]
+
+            for _, cinfo in supportedCompressions.items():
+                if cinfo.moduleName in globals():
+                    versions += [makeVersionRow(cinfo.moduleName, globals()[cinfo.moduleName].__version__)]
+
+            connection.executemany('INSERT OR REPLACE INTO "versions" VALUES (?,?,?,?,?)', versions)
+        except Exception as exception:
+            print("[Warning] There was an error when adding version information.")
+            if printDebug >= 3:
+                print(exception)
+
+    @staticmethod
+    def _storeTarMetadata(connection: sqlite3.Connection, tarPath: AnyStr) -> None:
+        """Adds some consistency meta information to recognize the need to update the cached TAR index"""
+        try:
+            tarStats = os.stat(tarPath)
+            serializedTarStats = json.dumps(
+                {attr: getattr(tarStats, attr) for attr in dir(tarStats) if attr.startswith('st_')}
+            )
+            connection.execute('INSERT INTO "metadata" VALUES (?,?)', ("tarstats", serializedTarStats))
+        except Exception as exception:
+            if printDebug >= 2:
+                print(exception)
+            print("[Warning] There was an error when adding file metadata.")
+            print("[Warning] Automatic detection of changed TAR files during index loading might not work.")
+
+    def _storeArgumentsMetadata(self, connection: sqlite3.Connection) -> None:
+        argumentsToSave = [
+            'mountRecursively',
+            'gzipSeekPointSpacing',
+            'encoding',
+            'stripRecursiveTarExtension',
+            'ignoreZeros',
+        ]
+
+        argumentsMetadata = json.dumps({argument: getattr(self, argument) for argument in argumentsToSave})
+
+        try:
+            connection.execute('INSERT INTO "metadata" VALUES (?,?)', ("arguments", argumentsMetadata))
+        except Exception as exception:
+            if printDebug >= 2:
+                print(exception)
+            print("[Warning] There was an error when adding argument metadata.")
+            print("[Warning] Automatic detection of changed arguments files during index loading might not work.")
+
+    @staticmethod
+    def _pathIsWritable(path: AnyStr) -> bool:
+        try:
+            folder = os.path.dirname(path)
+            if folder:
+                os.makedirs(folder, exist_ok=True)
+
+            f = open(path, 'wb')
+            f.write(b'\0' * 1024 * 1024)
+            f.close()
+            os.remove(path)
+
+            return True
+
+        except IOError:
+            if printDebug >= 2:
+                traceback.print_exc()
+                print("Could not create file:", path)
+
+        return False
+
+    @staticmethod
+    def _pathCanBeUsedForSqlite(path: AnyStr) -> bool:
+        fileExisted = os.path.isfile(path)
+        try:
+            folder = os.path.dirname(path)
+            if folder:
+                os.makedirs(folder, exist_ok=True)
+
+            connection = SQLiteIndexedTar._openSqlDb(path)
+            connection.executescript('CREATE TABLE "files" ( "path" VARCHAR(65535) NOT NULL );')
+            connection.commit()
+            connection.close()
+            return True
+        except sqlite3.OperationalError:
+            if printDebug >= 2:
+                traceback.print_exc()
+                print("Could not create SQLite database at:", path)
+        finally:
+            if not fileExisted and os.path.isfile(path):
+                SQLiteIndexedTar._uncheckedRemove(path)
+
+        return False
+
+    @staticmethod
+    def _openSqlDb(path: AnyStr) -> sqlite3.Connection:
+        sqlConnection = sqlite3.connect(path)
+        sqlConnection.row_factory = sqlite3.Row
+        sqlConnection.executescript(
+            """
+            PRAGMA LOCKING_MODE = EXCLUSIVE;
+            PRAGMA TEMP_STORE = MEMORY;
+            PRAGMA JOURNAL_MODE = OFF;
+            PRAGMA SYNCHRONOUS = OFF;
+            """
+        )
+        return sqlConnection
+
+    @staticmethod
+    def _initializeSqlDb(indexFileName: Optional[str]) -> sqlite3.Connection:
+        if printDebug >= 1:
+            print("Creating new SQLite index database at", indexFileName)
+
+        createTables = """
+            CREATE TABLE "files" (
+                "path"          VARCHAR(65535) NOT NULL,
+                "name"          VARCHAR(65535) NOT NULL,
+                "offsetheader"  INTEGER,  /* seek offset from TAR file where these file's contents resides */
+                "offset"        INTEGER,  /* seek offset from TAR file where these file's contents resides */
+                "size"          INTEGER,
+                "mtime"         INTEGER,
+                "mode"          INTEGER,
+                "type"          INTEGER,
+                "linkname"      VARCHAR(65535),
+                "uid"           INTEGER,
+                "gid"           INTEGER,
+                /* True for valid TAR files. Internally used to determine where to mount recursive TAR files. */
+                "istar"         BOOL   ,
+                "issparse"      BOOL   ,  /* for sparse files the file size refers to the expanded size! */
+                /* See SQL benchmarks for decision on the primary key.
+                 * See also https://www.sqlite.org/optoverview.html
+                 * (path,name) tuples might appear multiple times in a TAR if it got updated.
+                 * In order to also be able to show older versions, we need to add
+                 * the offsetheader column to the primary key. */
+                PRIMARY KEY (path,name,offsetheader)
+            );
+            /* "A table created using CREATE TABLE AS has no PRIMARY KEY and no constraints of any kind"
+             * Therefore, it will not be sorted and inserting will be faster! */
+            CREATE TABLE "filestmp" AS SELECT * FROM "files" WHERE 0;
+            CREATE TABLE "parentfolders" (
+                "path"     VARCHAR(65535) NOT NULL,
+                "name"     VARCHAR(65535) NOT NULL,
+                PRIMARY KEY (path,name)
+            );
+        """
+
+        sqlConnection = SQLiteIndexedTar._openSqlDb(indexFileName if indexFileName else ':memory:')
+        tables = sqlConnection.execute('SELECT name FROM sqlite_master WHERE type = "table";')
+        if {"files", "filestmp", "parentfolders"}.intersection({t[0] for t in tables}):
+            raise InvalidIndexError(
+                "The index file {} already seems to contain a table. "
+                "Please specify --recreate-index.".format(indexFileName)
+            )
+        sqlConnection.executescript(createTables)
+        return sqlConnection
+
+    @staticmethod
+    def _tarInfoFullMode(tarInfo: tarfile.TarInfo) -> int:
+        """
+        Returns the full mode for a TarInfo object. Note that TarInfo.mode only contains the permission bits
+        and not other bits like set for directory, symbolic links, and other special files.
+        """
+        return (
+            tarInfo.mode
+            # fmt: off
+            | ( stat.S_IFDIR if tarInfo.isdir () else 0 )
+            | ( stat.S_IFREG if tarInfo.isfile() else 0 )
+            | ( stat.S_IFLNK if tarInfo.issym () else 0 )
+            | ( stat.S_IFCHR if tarInfo.ischr () else 0 )
+            | ( stat.S_IFIFO if tarInfo.isfifo() else 0 )
+            # fmt: on
+        )
+
+    def _updateProgressBar(self, progressBar, fileobj: Any) -> None:
+        try:
+            if hasattr(fileobj, 'tell_compressed') and self.compression == 'bz2':
+                # Note that because bz2 works on a bitstream the tell_compressed returns the offset in bits
+                progressBar.update(fileobj.tell_compressed() // 8)
+            elif hasattr(fileobj, 'tell_compressed'):
+                progressBar.update(fileobj.tell_compressed())
+            elif hasattr(fileobj, 'fileobj'):
+                progressBar.update(fileobj.fileobj().tell())
+            elif self.rawFileObject and hasattr(self.rawFileObject, 'tell'):
+                progressBar.update(self.rawFileObject.tell())
+            else:
+                progressBar.update(fileobj.tell())
+        except Exception:
+            pass
+
+    def _createIndex(
+        self,
+        # fmt: off
+        fileObject  : Any,
+        progressBar : Any = None,
+        pathPrefix  : str = '',
+        streamOffset: int = 0
+        # fmt: on
+    ) -> None:
+        if printDebug >= 1:
+            print(
+                "Creating offset dictionary for",
+                "<file object>" if self.tarFileName is None else self.tarFileName,
+                "...",
+            )
+        t0 = timer()
+
+        # 1. If no SQL connection was given (by recursive call), open a new database file
+        openedConnection = False
+        if not self.indexIsLoaded() or not self.sqlConnection:
+            openedConnection = True
+            self.sqlConnection = self._initializeSqlDb(self.indexFileName)
+
+        # 2. Open TAR file reader
+        loadedTarFile: Any = []  # Feign an empty TAR file if anything goes wrong
+        if self.isTar:
+            try:
+                # r: uses seeks to skip to the next file inside the TAR while r| doesn't do any seeks.
+                # r| might be slower but for compressed files we have to go over all the data once anyways.
+                # Note that with ignore_zeros = True, no invalid header issues or similar will be raised even for
+                # non TAR files!?
+                loadedTarFile = tarfile.open(
+                    # fmt:off
+                    fileobj      = fileObject,
+                    mode         = 'r|' if self.compression else 'r:',
+                    ignore_zeros = self.ignoreZeros,
+                    encoding     = self.encoding,
+                    # fmt:on
+                )
+            except tarfile.ReadError:
+                pass
+        
+        try:
+            if progressBar is None:
+                progressBar = ProgressBar(os.fstat(fileObject.fileno()).st_size)
+        except:
+            pass
+
+        # 3. Iterate over files inside TAR and add them to the database
+        try:
+            filesToMountRecursively = []
+
+            for tarInfo in loadedTarFile:
+                loadedTarFile.members = []  # Clear this in order to limit memory usage by tarfile
+                self._updateProgressBar(progressBar, fileObject)
+
+                # Add a leading '/' as a convention where '/' represents the TAR root folder
+                # Partly, done because fusepy specifies paths in a mounted directory like this
+                # os.normpath does not delete duplicate '/' at beginning of string!
+                fullPath = pathPrefix + "/" + os.path.normpath(tarInfo.name).lstrip('/')
+
+                path, name = fullPath.rsplit("/", 1)
+                # fmt: off
+                fileInfo = (
+                    path                              ,  # 0
+                    name                              ,  # 1
+                    streamOffset + tarInfo.offset     ,  # 2
+                    streamOffset + tarInfo.offset_data,  # 3
+                    tarInfo.size                      ,  # 4
+                    tarInfo.mtime                     ,  # 5
+                    self._tarInfoFullMode(tarInfo)    ,  # 6
+                    tarInfo.type                      ,  # 7
+                    tarInfo.linkname                  ,  # 8
+                    tarInfo.uid                       ,  # 9
+                    tarInfo.gid                       ,  # 10
+                    False                             ,  # 11 (isTar)
+                    tarInfo.issparse()                ,  # 12
+                )
+                # fmt: on
+
+                if self.mountRecursively and tarInfo.isfile() and tarInfo.name.lower().endswith('.tar'):
+                    filesToMountRecursively.append(fileInfo)
+                else:
+                    self._setFileInfo(fileInfo)
+        except tarfile.ReadError as e:
+            if 'unexpected end of data' in str(e):
+                print(
+                    "[Warning] The TAR file is incomplete. Ratarmount will work but some files might be cut off. "
+                    "If the TAR file size changes, ratarmount will recreate the index during the next mounting."
+                )
+
+        # 4. Open contained TARs for recursive mounting
+        oldPos = fileObject.tell()
+        oldPrintName = self.tarFileName
+        for fileInfo in filesToMountRecursively:
+            tarExtension = '.tar'
+            fullPath = os.path.join(fileInfo[0], fileInfo[1])
+            if (
+                self.stripRecursiveTarExtension
+                and len(tarExtension) > 0
+                and fullPath.lower().endswith(tarExtension.lower())
+            ):
+                modifiedFullPath = fullPath[: -len(tarExtension)]
+            else:
+                modifiedFullPath = fullPath
+
+            # Temporarily change tarFileName for the info output of the recursive call
+            self.tarFileName = fullPath
+
+            # StenciledFile's tell returns the offset inside the file chunk instead of the global one,
+            # so we have to always communicate the offset of this chunk to the recursive call no matter
+            # whether tarfile has streaming access or seeking access!
+            globalOffset = fileInfo[3]
+            size = fileInfo[4]
+            tarFileObject = StenciledFile(fileObject, [(globalOffset, size)])
+
+            isTar = False
+            try:
+                self._createIndex(tarFileObject, progressBar, modifiedFullPath, globalOffset)
+                isTar = True
+            except tarfile.ReadError:
+                pass
+            finally:
+                del tarFileObject
+
+            if isTar:
+                modifiedFileInfo = list(fileInfo)
+
+                # if the TAR file contents could be read, we need to adjust the actual
+                # TAR file's metadata to be a directory instead of a file
+                mode = modifiedFileInfo[6]
+                mode = (
+                    (mode & 0o777)
+                    | stat.S_IFDIR
+                    | (stat.S_IXUSR if mode & stat.S_IRUSR != 0 else 0)
+                    | (stat.S_IXGRP if mode & stat.S_IRGRP != 0 else 0)
+                    | (stat.S_IXOTH if mode & stat.S_IROTH != 0 else 0)
+                )
+
+                path, name = modifiedFullPath.rsplit("/", 1)
+                modifiedFileInfo[0] = path
+                modifiedFileInfo[1] = name
+                modifiedFileInfo[6] = mode
+                modifiedFileInfo[11] = isTar
+
+                self._setFileInfo(tuple(modifiedFileInfo))
+            else:
+                self._setFileInfo(fileInfo)
+
+        fileObject.seek(oldPos)
+        self.tarFileName = oldPrintName
+
+        # Everything below should not be done in a recursive call of createIndex
+        if streamOffset > 0:
+            t1 = timer()
+            if printDebug >= 1:
+                print(
+                    "Creating offset dictionary for",
+                    "<file object>" if self.tarFileName is None else self.tarFileName,
+                    "took {:.2f}s".format(t1 - t0),
+                )
+            return
+
+        # If no file is in the TAR, then it most likely indicates a possibly compressed non TAR file.
+        # In that case add that itself to the file index. This won't work when called recursively,
+        # so check stream offset.
+        fileCount = self.sqlConnection.execute('SELECT COUNT(*) FROM "files";').fetchone()[0]
+        if fileCount == 0:
+            try:
+                tarInfo = os.fstat(fileObject.fileno())
+            except:
+                tarInfo = None
+            fname = os.path.basename(self.tarFileName)
+            for suffix in ['.gz', '.bz2', '.bzip2', '.gzip', '.xz', '.zst', '.zstd']:
+                if fname.lower().endswith(suffix) and len(fname) > len(suffix):
+                    fname = fname[: -len(suffix)]
+                    break
+
+            # If the file object is actually an IndexedBzip2File or such, we can't directly use the file size
+            # from os.stat and instead have to gather it from seek. Unfortunately, indexed_gzip does not support
+            # io.SEEK_END even though it could as it has the index ...
+            while fileObject.read(1024 * 1024):
+                self._updateProgressBar(progressBar, fileObject)
+            fileSize = fileObject.tell()
+
+            # fmt: off
+            fileInfo = (
+                ""                 ,  # 0 path
+                fname              ,  # 1
+                None               ,  # 2 header offset
+                0                  ,  # 3 data offset
+                fileSize           ,  # 4
+                tarInfo.st_mtime if tarInfo else 0   ,  # 5
+                tarInfo.st_mode if tarInfo else 0    ,  # 6
+                None               ,  # 7 TAR file type. Currently unused but overlaps with mode anyways
+                None               ,  # 8 linkname
+                tarInfo.st_uid if tarInfo else 0     ,  # 9
+                tarInfo.st_gid if tarInfo else 0     ,  # 10
+                False              ,  # 11 isTar
+                False              ,  # 12 isSparse, don't care if it is actually sparse or not because it is not in TAR
+            )
+            # fmt: on
+            self._setFileInfo(fileInfo)
+
+        # All the code below is for database finalizing which should not be done in a recursive call of createIndex!
+        if not openedConnection:
+            return
+
+        # 5. Resort by (path,name). This one-time resort is faster than resorting on each INSERT (cache spill)
+        if printDebug >= 2:
+            print("Resorting files by path ...")
+
+        cleanupDatabase = """
+            INSERT OR REPLACE INTO "files" SELECT * FROM "filestmp" ORDER BY "path","name",rowid;
+            DROP TABLE "filestmp";
+            INSERT OR IGNORE INTO "files"
+                /* path name offsetheader offset size mtime mode type linkname uid gid istar issparse */
+                SELECT path,name,0,0,1,0,{},{},"",0,0,0,0
+                FROM "parentfolders" ORDER BY "path","name";
+            DROP TABLE "parentfolders";
+        """.format(
+            int(0o555 | stat.S_IFDIR), int(tarfile.DIRTYPE)
+        )
+        self.sqlConnection.executescript(cleanupDatabase)
+
+        self.sqlConnection.commit()
+
+        t1 = timer()
+        if printDebug >= 1:
+            print(
+                "Creating offset dictionary for",
+                "<file object>" if self.tarFileName is None else self.tarFileName,
+                "took {:.2f}s".format(t1 - t0),
+            )
+
+    @staticmethod
+    def _rowToFileInfo(row: Dict[str, Any]) -> FileInfo:
+        return FileInfo(
+            # fmt: off
+            offset       = row['offset'],
+            offsetheader = row['offsetheader'] if 'offsetheader' in row.keys() else 0,
+            size         = row['size'],
+            mtime        = row['mtime'],
+            mode         = row['mode'],
+            type         = row['type'],
+            linkname     = row['linkname'],
+            uid          = row['uid'],
+            gid          = row['gid'],
+            istar        = row['istar'],
+            issparse     = row['issparse'] if 'issparse' in row.keys() else False
+            # fmt: on
+        )
+
+    def getFileInfo(
+        self,
+        # fmt: off
+        fullPath     : str,
+        listDir      : bool = False,
+        listVersions : bool = False,
+        fileVersion  : int  = 0
+        # fmt: on
+    ) -> Optional[Union[FileInfo, Dict[str, FileInfo]]]:
+        """
+        This is the heart of this class' public interface!
+
+        path    : full path to file where '/' denotes TAR's root, e.g., '/', or '/foo'
+        listDir : if True, return a dictionary for the given directory path: { fileName : FileInfo, ... }
+                  if False, return simple FileInfo to given path (directory or file)
+        fileVersion : If the TAR contains the same file path multiple times, by default only the last one is shown.
+                      But with this argument other versions can be queried. Version 1 is the oldest one.
+                      Version 0 translates to the most recent one for compatibility with tar --occurrence=<number>.
+                      Version -1 translates to the second most recent, and so on.
+                      For listDir=True, the file version makes no sense and is ignored!
+                      So, even if a folder was overwritten by a file, which is already not well supported by tar,
+                      then listDir for that path will still list all contents of the overwritten folder or folders,
+                      no matter the specified version. The file system layer has to take care that a directory
+                      listing is not even requeted in the first place if it is not a directory.
+                      FUSE already does this by calling getattr for all parent folders in the specified path first.
+
+        If path does not exist, always return None
+
+        If listVersions is true, then return metadata for all versions of a file possibly appearing more than once
+        in the TAR as a directory dictionary. listDir will then be ignored!
+        """
+        # TODO cache last listDir as most often a stat over all entries will soon follow
+
+        if not isinstance(fileVersion, int):
+            raise TypeError("The specified file version must be an integer!")
+        if not self.sqlConnection:
+            raise IndexNotOpenError("This method can not be called without an opened index database!")
+
+        # also strips trailing '/' except for a single '/' and leading '/'
+        fullPath = '/' + os.path.normpath(fullPath).lstrip('/')
+
+        if listVersions:
+            path, name = fullPath.rsplit('/', 1)
+            rows = self.sqlConnection.execute(
+                'SELECT * FROM "files" WHERE "path" == (?) AND "name" == (?) ORDER BY "offsetheader" ASC', (path, name)
+            )
+            result = {str(version + 1): self._rowToFileInfo(row) for version, row in enumerate(rows)}
+            return result
+
+        if listDir:
+            # For listing directory entries the file version can't be applied meaningfully at this abstraction layer.
+            # E.g., should it affect the file version of the directory to list, or should it work on the listed files
+            # instead and if so how exactly if there aren't the same versions for all files available, ...?
+            # Or, are folders assumed to be overwritten by a new folder entry in a TAR or should they be union mounted?
+            # If they should be union mounted, like is the case now, then the folder version only makes sense for
+            # its attributes.
+            rows = self.sqlConnection.execute('SELECT * FROM "files" WHERE "path" == (?)', (fullPath.rstrip('/'),))
+            directory = {}
+            gotResults = False
+            for row in rows:
+                gotResults = True
+                if row['name']:
+                    directory[row['name']] = self._rowToFileInfo(row)
+            return directory if gotResults else None
+
+        path, name = fullPath.rsplit('/', 1)
+        row = self.sqlConnection.execute(
+            """
+            SELECT * FROM "files"
+            WHERE "path" == (?) AND "name" == (?)
+            ORDER BY "offsetheader" {}
+            LIMIT 1 OFFSET (?);
+            """.format(
+                'DESC' if fileVersion is None or fileVersion <= 0 else 'ASC'
+            ),
+            (path, name, 0 if fileVersion is None else fileVersion - 1 if fileVersion > 0 else -fileVersion),
+        ).fetchone()
+        return self._rowToFileInfo(row) if row else None
+
+    def isDir(self, path: str) -> bool:
+        """Return true if path exists and is a folder."""
+        return isinstance(self.getFileInfo(path, listDir=True), dict)
+
+    def listDir(self, path: str) -> Optional[Iterable[str]]:
+        """
+        Usability wrapper for getFileInfo(listDir=True) with FileInfo stripped if you are sure you don't need it.
+        """
+        result = self.getFileInfo(path, listDir=True)
+        if isinstance(result, dict):
+            return result.keys()
+        return None
+
+    def fileVersions(self, path: str) -> int:
+        """
+        Usability wrapper for getFileInfo(listVersions=True) with FileInfo stripped if you are sure you don't need it.
+        """
+        fileVersions = self.getFileInfo(path, listVersions=True)
+        return len(fileVersions) if isinstance(fileVersions, dict) else 0
+
+    def read(self, path: str, size: int = None, offset: int = 0, fileInfo: Optional[FileInfo] = None) -> bytes:
+        """
+        fileInfo: This argument can be specified for performance reasons. It must be the FileInfo object for path!
+        """
+        if fileInfo is None:
+            queriedFileInfo = self.getFileInfo(path)
+            if isinstance(queriedFileInfo, FileInfo):
+                fileInfo = queriedFileInfo
+        if not isinstance(fileInfo, FileInfo):
+            raise ValueError("Specified path '{}' is not a file that can be read!".format(path))
+
+        if size is None:
+            size = fileInfo.size
+
+        # Dereference hard links
+        if not stat.S_ISREG(fileInfo.mode) and not stat.S_ISLNK(fileInfo.mode) and fileInfo.linkname:
+            targetLink = '/' + fileInfo.linkname.lstrip('/')
+            if targetLink != path:
+                return self.read(targetLink, size, offset)
+
+        if not fileInfo.issparse:
+            # For non-sparse files, we can simply seek to the offset and read from it.
+            self.tarFileObject.seek(fileInfo.offset + offset, os.SEEK_SET)
+            return self.tarFileObject.read(size)
+
+        # The TAR file format is very simple. It's just a concatenation of TAR blocks. There is not even a
+        # global header, only the TAR block headers. That's why we can simply cut out the TAR block for
+        # the sparse file using StenciledFile and then use tarfile on it to expand the sparse file correctly.
+        tarBlockSize = fileInfo.offset - fileInfo.offsetheader + fileInfo.size
+        tarSubFile = StenciledFile(self.tarFileObject, [(fileInfo.offsetheader, tarBlockSize)])
+        with tarfile.open(fileobj=typing.cast(BinaryIO, tarSubFile), mode='r:', encoding=self.encoding) as tmpTarFile:
+            tmpFileObject = tmpTarFile.extractfile(next(iter(tmpTarFile)))
+            if tmpFileObject:
+                tmpFileObject.seek(offset, os.SEEK_SET)
+                result = tmpFileObject.read(size)
+            else:
+                print("tarfile.extractfile returned nothing!")
+                raise Exception("tarfile.extractfile returned nothing!")
+        return result
+
+    def _tryAddParentFolders(self, path: str) -> None:
+        # Add parent folders if they do not exist.
+        # E.g.: path = '/a/b/c' -> paths = [('', 'a'), ('/a', 'b'), ('/a/b', 'c')]
+        # Without the parentFolderCache, the additional INSERT statements increase the creation time
+        # from 8.5s to 12s, so almost 50% slowdown for the 8MiB test TAR!
+        pathParts = path.split("/")
+        paths = [
+            p
+            # fmt: off
+            for p in (
+                ( "/".join( pathParts[:i] ), pathParts[i] )
+                for i in range( 1, len( pathParts ) )
+            )
+            # fmt: on
+            if p not in self.parentFolderCache
+        ]
+        if not paths:
+            return
+
+        self.parentFolderCache += paths
+        # Assuming files in the TAR are sorted by hierarchy, the maximum parent folder cache size
+        # gives the maximum cacheable file nesting depth. High numbers lead to higher memory usage and lookup times.
+        if len(self.parentFolderCache) > 16:
+            self.parentFolderCache = self.parentFolderCache[-8:]
+
+        if not self.sqlConnection:
+            raise IndexNotOpenError("This method can not be called without an opened index database!")
+        self.sqlConnection.executemany(
+            'INSERT OR IGNORE INTO "parentfolders" VALUES (?,?)', [(p[0], p[1]) for p in paths]
+        )
+
+    def _setFileInfo(self, row: tuple) -> None:
+        if not self.sqlConnection:
+            raise IndexNotOpenError("This method can not be called without an opened index database!")
+
+        try:
+            self.sqlConnection.execute('INSERT OR REPLACE INTO "files" VALUES (' + ','.join('?' * len(row)) + ');', row)
+        except UnicodeEncodeError:
+            print("[Warning] Problem caused by file name encoding when trying to insert this row:", row)
+            print("[Warning] The file name will now be stored with the bad character being escaped")
+            print("[Warning] instead of being correctly interpreted.")
+            print("[Warning] Please specify a suitable file name encoding using, e.g., --encoding iso-8859-1!")
+            print("[Warning] A list of possible encodings can be found here:")
+            print("[Warning] https://docs.python.org/3/library/codecs.html#standard-encodings")
+
+            checkedRow = []
+            for x in list(row):  # check strings
+                if isinstance(x, str):
+                    try:
+                        x.encode()
+                        checkedRow += [x]
+                    except UnicodeEncodeError:
+                        # fmt: off
+                        checkedRow += [
+                            x.encode( self.encoding, 'surrogateescape' )
+                             .decode( self.encoding, 'backslashreplace' )
+                         ]
+                        # fmt: on
+                else:
+                    checkedRow += [x]
+
+            self.sqlConnection.execute(
+                'INSERT OR REPLACE INTO "files" VALUES (' + ','.join('?' * len(row)) + ');', tuple(checkedRow)
+            )
+            print("[Warning] The escaped inserted row is now:", row)
+            print()
+
+        self._tryAddParentFolders(row[0])
+
+    def setFileInfo(self, fullPath: str, fileInfo: FileInfo) -> None:
+        """
+        fullPath : the full path to the file with leading slash (/) for which to set the file info
+        """
+        assert fullPath[0] == "/"
+
+        # os.normpath does not delete duplicate '/' at beginning of string!
+        path, name = fullPath.rsplit("/", 1)
+        row = (
+            path,
+            name,
+            fileInfo.offsetheader,
+            fileInfo.offset,
+            fileInfo.size,
+            fileInfo.mtime,
+            fileInfo.mode,
+            fileInfo.type,
+            fileInfo.linkname,
+            fileInfo.uid,
+            fileInfo.gid,
+            fileInfo.istar,
+            fileInfo.issparse,
+        )
+        self._setFileInfo(row)
+
+    def indexIsLoaded(self) -> bool:
+        """Returns true if the SQLite database has been opened for reading and a "files" table exists."""
+        if not self.sqlConnection:
+            return False
+
+        try:
+            self.sqlConnection.execute('SELECT * FROM "files" WHERE 0 == 1;')
+        except sqlite3.OperationalError:
+            self.sqlConnection = None
+            return False
+
+        return True
+
+    def loadIndex(self, indexFileName: AnyStr) -> None:
+        """Loads the given index SQLite database and checks it for validity."""
+        if self.indexIsLoaded():
+            return
+
+        t0 = time.time()
+        self.sqlConnection = self._openSqlDb(indexFileName)
+        tables = [x[0] for x in self.sqlConnection.execute('SELECT name FROM sqlite_master WHERE type="table"')]
+        versions = None
+        try:
+            rows = self.sqlConnection.execute('SELECT * FROM versions;')
+            versions = {}
+            for row in rows:
+                versions[row[0]] = (row[2], row[3], row[4])
+        except sqlite3.OperationalError:
+            pass
+
+        try:
+            # Check indexes created with bugged bz2 decoder (bug existed when I did not store versions yet)
+            if 'bzip2blocks' in tables and 'versions' not in tables:
+                raise InvalidIndexError(
+                    "The indexes created with version 0.3.0 through 0.3.3 for bzip2 compressed archives "
+                    "are very likely to be wrong because of a bzip2 decoder bug.\n"
+                    "Please delete the index or call ratarmount with the --recreate-index option!"
+                )
+
+            # Check for empty or incomplete indexes. Pretty safe to rebuild the index for these as they
+            # are so invalid, noone should miss them. So, recreate index by default for these cases.
+            if 'files' not in tables:
+                raise InvalidIndexError("SQLite index is empty")
+
+            if 'filestmp' in tables or 'parentfolders' in tables:
+                raise InvalidIndexError("SQLite index is incomplete")
+
+            # Check for pre-sparse support indexes
+            if (
+                'versions' not in tables
+                or 'index' not in versions
+                or len(versions['index']) < 2
+                or versions['index'][1] < 2
+            ):
+                print("[Warning] The found outdated index does not contain any sparse file information.")
+                print("[Warning] The index will also miss data about multiple versions of a file.")
+                print("[Warning] Please recreate the index if you have problems with those.")
+
+            if 'metadata' in tables:
+                metadata = dict(self.sqlConnection.execute('SELECT * FROM metadata;'))
+
+                if 'tarstats' in metadata:
+                    values = json.loads(metadata['tarstats'])
+                    tarStats = os.stat(self.tarFileName)
+
+                    # fmt: off
+                    if (
+                        hasattr( tarStats, "st_size" )
+                        and 'st_size' in values
+                        and tarStats.st_size != values['st_size']
+                    ):
+                        raise InvalidIndexError( "TAR file for this SQLite index has changed size from",
+                                                 values['st_size'], "to", tarStats.st_size)
+                    # fmt: on
+
+                    if (
+                        self.verifyModificationTime
+                        and hasattr(tarStats, "st_mtime")
+                        and 'st_mtime' in values
+                        and tarStats.st_mtime != values['st_mtime']
+                    ):
+                        raise InvalidIndexError(
+                            "The modification date for the TAR file",
+                            values['st_mtime'],
+                            "to this SQLite index has changed (" + str(tarStats.st_mtime) + ")",
+                        )
+
+                # Check arguments used to create the found index. These are only warnings and not forcing a rebuild
+                # by default.
+                # TODO: Add --force options?
+                if 'arguments' in metadata:
+                    indexArgs = json.loads(metadata['arguments'])
+                    argumentsToCheck = [
+                        'mountRecursively',
+                        'gzipSeekPointSpacing',
+                        'encoding',
+                        'stripRecursiveTarExtension',
+                        'ignoreZeros',
+                    ]
+                    differingArgs = []
+                    for arg in argumentsToCheck:
+                        if arg in indexArgs and hasattr(self, arg) and indexArgs[arg] != getattr(self, arg):
+                            differingArgs.append((arg, indexArgs[arg], getattr(self, arg)))
+                    if differingArgs:
+                        print("[Warning] The arguments used for creating the found index differ from the arguments ")
+                        print("[Warning] given for mounting the archive now. In order to apply these changes, ")
+                        print("[Warning] recreate the index using the --recreate-index option!")
+                        for arg, oldState, newState in differingArgs:
+                            print("[Warning] {}: index: {}, current: {}".format(arg, oldState, newState))
+
+        except Exception as e:
+            # indexIsLoaded checks self.sqlConnection, so close it before returning because it was found to be faulty
+            try:
+                self.sqlConnection.close()
+            except sqlite3.Error:
+                pass
+            self.sqlConnection = None
+
+            raise e
+
+        if printDebug >= 1:
+            # Legacy output for automated tests
+            print("Loading offset dictionary from", indexFileName, "took {:.2f}s".format(time.time() - t0))
+
+    def _tryLoadIndex(self, indexFileName: AnyStr) -> bool:
+        """calls loadIndex if index is not loaded already and provides extensive error handling"""
+
+        if self.indexIsLoaded():
+            return True
+
+        if not os.path.isfile(indexFileName):
+            return False
+
+        try:
+            self.loadIndex(indexFileName)
+        except Exception as exception:
+            if printDebug >= 3:
+                traceback.print_exc()
+
+            print("[Warning] Could not load file:", indexFileName)
+            print("[Info] Exception:", exception)
+            print("[Info] Some likely reasons for not being able to load the index file:")
+            print("[Info]   - The index file has incorrect read permissions")
+            print("[Info]   - The index file is incomplete because ratarmount was killed during index creation")
+            print("[Info]   - The index file was detected to contain errors because of known bugs of older versions")
+            print("[Info]   - The index file got corrupted because of:")
+            print("[Info]     - The program exited while it was still writing the index because of:")
+            print("[Info]       - the user sent SIGINT to force the program to quit")
+            print("[Info]       - an internal error occured while writing the index")
+            print("[Info]       - the disk filled up while writing the index")
+            print("[Info]     - Rare lowlevel corruptions caused by hardware failure")
+
+            print("[Info] This might force a time-costly index recreation, so if it happens often")
+            print("       and mounting is slow, try to find out why loading fails repeatedly,")
+            print("       e.g., by opening an issue on the public github page.")
+
+            try:
+                os.remove(indexFileName)
+            except OSError:
+                print("[Warning] Failed to remove corrupted old cached index file:", indexFileName)
+
+        if printDebug >= 3 and self.indexIsLoaded():
+            print("Loaded index", indexFileName)
+
+        return self.indexIsLoaded()
+
+    @staticmethod
+    def _detectCompression(fileobj: BinaryIO) -> Optional[str]:
+        if not isinstance(fileobj, io.IOBase) or not fileobj.seekable():
+            return None
+
+        oldOffset = fileobj.tell()
+        for compressionId, compression in supportedCompressions.items():
+            # The header check is a necessary condition not a sufficient condition.
+            # Especially for gzip, which only has 2 magic bytes, false positives might happen.
+            # Therefore, only use the magic bytes based check if the module could not be found
+            # in order to still be able to print pinpoint error messages.
+            matches = compression.checkHeader(fileobj)
+            fileobj.seek(oldOffset)
+            if not matches:
+                continue
+
+            if compression.moduleName not in globals() and matches:
+                return compressionId
+
+            try:
+                compressedFileobj = compression.open(fileobj)
+                # Reading 1B from a single-frame zst file might require decompressing it fully in order
+                # to get uncompressed file size! Avoid that. The magic bytes should suffice mostly.
+                # TODO: Make indexed_zstd not require the uncompressed size for the read call.
+                if compressionId != 'zst':
+                    compressedFileobj.read(1)
+                compressedFileobj.close()
+                fileobj.seek(oldOffset)
+                return compressionId
+            except Exception:
+                fileobj.seek(oldOffset)
+
+        return None
+
+    @staticmethod
+    def _detectTar(fileobj: BinaryIO, encoding: str) -> bool:
+        if not isinstance(fileobj, io.IOBase) or not fileobj.seekable():
+            return False
+
+        oldOffset = fileobj.tell()
+        isTar = False
+        try:
+            with tarfile.open(fileobj=fileobj, mode='r:', encoding=encoding):
+                isTar = True
+        except (tarfile.ReadError, tarfile.CompressionError):
+            pass
+
+        fileobj.seek(oldOffset)
+        return isTar
+
+    @staticmethod
+    def _openCompressedFile(fileobj: BinaryIO, gzipSeekPointSpacing: int, encoding: str) -> Any:
+        """
+        Opens a file possibly undoing the compression.
+        Returns (tar_file_obj, raw_file_obj, compression, isTar).
+        raw_file_obj will be none if compression is None.
+        """
+        compression = SQLiteIndexedTar._detectCompression(fileobj)
+        if compression not in supportedCompressions:
+            return fileobj, None, compression, SQLiteIndexedTar._detectTar(fileobj, encoding)
+
+        cinfo = supportedCompressions[compression]
+        if cinfo.moduleName not in globals():
+            raise CompressionError(
+                "Can't open a {} compressed file '{}' without {} module!".format(
+                    compression, fileobj.name, cinfo.moduleName
+                )
+            )
+
+        if compression == 'gz':
+            # drop_handles keeps a file handle opening as is required to call tell() during decoding
+            tar_file = indexed_gzip.IndexedGzipFile(fileobj=fileobj, drop_handles=False, spacing=gzipSeekPointSpacing)
+        else:
+            tar_file = cinfo.open(fileobj)
+
+        return tar_file, fileobj, compression, SQLiteIndexedTar._detectTar(tar_file, encoding)
+
+    @staticmethod
+    def _uncheckedRemove(path: Optional[AnyStr]):
+        """
+        Often cleanup is good manners but it would only be obnoxious if ratarmount crashed on unnecessary cleanup.
+        """
+        if not path or not os.path.exists(path):
+            return
+
+        try:
+            os.remove(path)
+        except Exception:
+            print("[Warning] Could not remove:", path)
+
+    def _loadOrStoreCompressionOffsets(self):
+        # This should be called after the TAR file index is complete (loaded or created).
+        # If the TAR file index was created, then tarfile has iterated over the whole file once
+        # and therefore completed the implicit compression offset creation.
+        if not self.sqlConnection:
+            raise IndexNotOpenError("This method can not be called without an opened index database!")
+        db = self.sqlConnection
+        fileObject = self.tarFileObject
+
+        if (
+            hasattr(fileObject, 'set_block_offsets')
+            and hasattr(fileObject, 'block_offsets')
+            and self.compression in ['bz2', 'zst']
+        ):
+            if self.compression == 'bz2':
+                table_name = 'bzip2blocks'
+            elif self.compression == 'zst':
+                table_name = 'zstdblocks'
+
+            try:
+                offsets = dict(db.execute('SELECT blockoffset,dataoffset FROM {};'.format(table_name)))
+                fileObject.set_block_offsets(offsets)
+            except Exception:
+                if printDebug >= 2:
+                    print(
+                        "[Info] Could not load {} block offset data. Will create it from scratch.".format(
+                            self.compression
+                        )
+                    )
+
+                tables = [x[0] for x in db.execute('SELECT name FROM sqlite_master WHERE type="table";')]
+                if table_name in tables:
+                    db.execute('DROP TABLE {}'.format(table_name))
+                db.execute('CREATE TABLE {} ( blockoffset INTEGER PRIMARY KEY, dataoffset INTEGER )'.format(table_name))
+                db.executemany('INSERT INTO {} VALUES (?,?)'.format(table_name), fileObject.block_offsets().items())
+                db.commit()
+            return
+
+        if (
+            # fmt: off
+            hasattr( fileObject, 'import_index' )
+            and hasattr( fileObject, 'export_index' )
+            and self.compression == 'gz'
+            # fmt: on
+        ):
+            # indexed_gzip index only has a file based API, so we need to write all the index data from the SQL
+            # database out into a temporary file. For that, let's first try to use the same location as the SQLite
+            # database because it should have sufficient writing rights and free disk space.
+            gzindex = None
+            for tmpDir in [os.path.dirname(self.indexFileName), None]:
+                # Try to export data from SQLite database. Note that no error checking against the existence of
+                # gzipindex table is done because the exported data itself might also be wrong and we can't check
+                # against this. Therefore, collate all error checking by catching exceptions.
+                try:
+                    gzindex = tempfile.mkstemp(dir=tmpDir)[1]
+                    with open(gzindex, 'wb') as file:
+                        file.write(db.execute('SELECT data FROM gzipindex').fetchone()[0])
+                    break
+                except Exception:
+                    self._uncheckedRemove(gzindex)
+                    gzindex = None
+
+            if gzindex:
+                try:
+                    fileObject.import_index(filename=gzindex)
+                    return
+                except Exception:
+                    pass
+                finally:
+                    self._uncheckedRemove(gzindex)
+
+            # Store the offsets into a temporary file and then into the SQLite database
+            if printDebug >= 2:
+                print("[Info] Could not load GZip Block offset data. Will create it from scratch.")
+
+            # Transparently force index to be built if not already done so. build_full_index was buggy for me.
+            # Seeking from end not supported, so we have to read the whole data in in a loop
+            while fileObject.read(1024 * 1024):
+                pass
+
+            # The created index can unfortunately be pretty large and tmp might actually run out of memory!
+            # Therefore, try different paths, starting with the location where the index resides.
+            gzindex = None
+            for tmpDir in [os.path.dirname(self.indexFileName), None]:
+                gzindex = tempfile.mkstemp(dir=tmpDir)[1]
+                try:
+                    fileObject.export_index(filename=gzindex)
+                    break
+                except indexed_gzip.ZranError:
+                    self._uncheckedRemove(gzindex)
+                    gzindex = None
+
+            if not gzindex or not os.path.isfile(gzindex):
+                print("[Warning] The GZip index required for seeking could not be stored in a temporary file!")
+                print("[Info] This might happen when you are out of space in your temporary file and at the")
+                print("[Info] the index file location. The gzipindex size takes roughly 32kiB per 4MiB of")
+                print("[Info] uncompressed(!) bytes (0.8% of the uncompressed data) by default.")
+                raise RuntimeError("Could not initialize the GZip seek cache.")
+            if printDebug >= 2:
+                print("Exported GZip index size:", os.stat(gzindex).st_size)
+
+            # Store contents of temporary file into the SQLite database
+            tables = [x[0] for x in db.execute('SELECT name FROM sqlite_master WHERE type="table"')]
+            if 'gzipindex' in tables:
+                db.execute('DROP TABLE gzipindex')
+            db.execute('CREATE TABLE gzipindex ( data BLOB )')
+            with open(gzindex, 'rb') as file:
+                db.execute('INSERT INTO gzipindex VALUES (?)', (file.read(),))
+            db.commit()
+            os.remove(gzindex)
+            return
+
+        # Note that for xz seeking, loading and storing block indexes is unnecessary because it has an index included!
+        if self.compression in [None, 'xz']:
+            return
+
+        assert (
+            False
+        ), "Could not load or store block offsets for {} probably because adding support was forgotten!".format(
+            self.compression
+        )
+
+
+def _makeMountPointFileInfoFromStats(stats: os.stat_result) -> FileInfo:
+    # make the mount point read only and executable if readable, i.e., allow directory listing
+    # clear higher bits like S_IFREG and set the directory bit instead
+    mountMode = (
+        (stats.st_mode & 0o777)
+        | stat.S_IFDIR
+        | (stat.S_IXUSR if stats.st_mode & stat.S_IRUSR != 0 else 0)
+        | (stat.S_IXGRP if stats.st_mode & stat.S_IRGRP != 0 else 0)
+        | (stat.S_IXOTH if stats.st_mode & stat.S_IROTH != 0 else 0)
+    )
+
+    return FileInfo(
+        # fmt: off
+        offset       = None           ,
+        offsetheader = None           ,
+        size         = stats.st_size  ,
+        mtime        = stats.st_mtime ,
+        mode         = mountMode      ,
+        type         = tarfile.DIRTYPE,
+        linkname     = ""             ,
+        uid          = stats.st_uid   ,
+        gid          = stats.st_gid   ,
+        istar        = True           ,
+        issparse     = False
+        # fmt: on
+    )
+
+
+class FolderMountSource:
+    """
+    This class manages one folder as mount source offering methods for listing folders, reading files, and others.
+    """
+
+    __slots__ = ('root', 'mountedTars', 'rootFileInfos')
+
+    def __init__(self, path: str, **sqliteIndexedTarOptions) -> None:
+        self.root: str = os.path.realpath(path)
+        # stores mounted TARs per mount point relative (without leading '/') to self.root.
+        self.mountedTars: Dict[str, SQLiteIndexedTar] = {}
+        self.rootFileInfos: Dict[str, FileInfo] = {}
+
+        # Find TAR files in this folder and mount them recursively if so requested
+        if sqliteIndexedTarOptions.get('recursive', False) and os.path.isdir(self.root):
+            stripSuffix = sqliteIndexedTarOptions.get('stripRecursiveTarExtension', False)
+            encoding = sqliteIndexedTarOptions.get('encoding', tarfile.ENCODING)
+
+            for folder, _, files in os.walk(self.root):
+                assert folder.startswith(self.root)
+                folder = folder[len(self.root) + 1 :]
+
+                for filePath in files:
+                    fileName = stripSuffixFromTarFile(filePath)
+                    if fileName == filePath:
+                        continue
+
+                    fullPath = os.path.realpath(os.path.join(self.root, folder, filePath))
+                    try:
+                        TarFileType(encoding=encoding)(fullPath)
+                    except argparse.ArgumentTypeError:
+                        continue
+
+                    try:
+                        indexedTar = SQLiteIndexedTar(fullPath, writeIndex=True, **sqliteIndexedTarOptions)
+                    except Exception:
+                        continue
+
+                    mountPoint = os.path.join(folder, fileName if stripSuffix else filePath)
+                    self.mountedTars[mountPoint] = indexedTar
+                    self.rootFileInfos[mountPoint] = _makeMountPointFileInfoFromStats(os.stat(fullPath))
+
+    def setFolderDescriptor(self, fd: int) -> None:
+        """
+        Make this mount source manage the special "." folder by changing to that directory.
+        Because we change to that directory it may only be used for one mount source but it also works
+        when that mount source is mounted on!
+        """
+        os.fchdir(fd)
+        self.root = '.'
+
+    def _findMountedTar(self, path: str) -> Optional[Tuple[str, str]]:
+        """
+        Returns the mount point, which can be found in self.mountedTars, and the rest of the path.
+        Basically, it splits path at the appropriate mount point boundary.
+        """
+        if not self.mountedTars:
+            return None
+
+        # TODO Not sure how performance-critical this can turn out, but maybe do something like bisection instead?
+        parts = path.lstrip(os.path.sep).split(os.path.sep)
+        subPath = ""
+        for i, part in enumerate(parts):
+            subPath = os.path.join(subPath, part)
+            if subPath in self.mountedTars:
+                pathInsideTar = os.path.join(*parts[i + 1 :]) if i + 1 < len(parts) else "/"
+                return subPath, pathInsideTar
+        return None
+
+    def _realpath(self, path: str) -> str:
+        """Path given relative to folder root. Leading '/' is acceptable"""
+        return os.path.join(self.root, path.lstrip(os.path.sep))
+
+    def _exists(self, path: str) -> bool:
+        """Check if path exists."""
+        return os.path.lexists(self._realpath(path))
+
+    @staticmethod
+    def _getFileInfoFromRealFile(filePath: str) -> FileInfo:
+        stats = os.lstat(filePath)
+        return FileInfo(
+            # fmt: off
+            offset       = None          ,
+            offsetheader = None          ,
+            size         = stats.st_size ,
+            mtime        = stats.st_mtime,
+            mode         = stats.st_mode ,
+            type         = None          ,  # I think this is completely unused and mostly contained in mode
+            linkname     = os.readlink( filePath ) if os.path.islink( filePath ) else None,
+            uid          = stats.st_uid  ,
+            gid          = stats.st_gid  ,
+            istar        = False         ,
+            issparse     = False
+            # fmt: on
+        )
+
+    def getFileInfo(self, filePath: str, fileVersion: int = 0) -> Optional[FileInfo]:
+        """Return file info for given path."""
+        # TODO: Add support for the .versions API in order to access the underlying TARs if stripRecursiveTarExtension
+        #       is false? Then again, SQLiteIndexedTar is not able to do this either, so it might be inconsistent.
+
+        pathSplitAtMountPoint = self._findMountedTar(filePath)
+        if pathSplitAtMountPoint:
+            mountPoint, pathInMountPoint = pathSplitAtMountPoint
+            if pathInMountPoint and pathInMountPoint != '/':
+                fileInfo = self.mountedTars[mountPoint].getFileInfo(pathInMountPoint, fileVersion=fileVersion)
+
+                if isinstance(fileInfo, FileInfo):
+                    # Dereference hard links
+                    if not stat.S_ISREG(fileInfo.mode) and not stat.S_ISLNK(fileInfo.mode) and fileInfo.linkname:
+                        targetLink = fileInfo.linkname.lstrip('/')
+                        if targetLink != filePath:
+                            return self.getFileInfo(os.path.join(mountPoint, targetLink), fileVersion)
+                    return fileInfo
+                return None
+            return self.rootFileInfos[mountPoint]
+
+        # This is a bit of problematic design, however, the fileVersions count from 1 for the user.
+        # And as -1 means the last version, 0 should also mean the first version ...
+        # Basically, I did accidentally mix user-visible versions 1+ versinos with API 0+ versions,
+        # leading to this problematic clash of 0 and 1.
+        if fileVersion in [0, 1] and self._exists(filePath):
+            return self._getFileInfoFromRealFile(self._realpath(filePath))
+        return None
+
+    def listDir(self, path: str) -> Optional[Iterable[str]]:
+        """
+        This method is different from SQLiteIndexedTar.getFileInfo(listDir=True) because stat'ing each file
+        does not come for free here, in contrast to SQLiteIndexedTar.
+        """
+        pathSplitAtMountPoint = self._findMountedTar(path)
+        if pathSplitAtMountPoint:
+            mountPoint, pathInMountPoint = pathSplitAtMountPoint
+            return self.mountedTars[mountPoint].listDir(pathInMountPoint)
+
+        realpath = self._realpath(path)
+        if os.path.isdir(realpath):
+            return os.listdir(realpath)
+        return None
+
+    def fileVersions(self, path: str) -> int:
+        """Returns available versions for a file."""
+        pathSplitAtMountPoint = self._findMountedTar(path)
+        if pathSplitAtMountPoint:
+            mountPoint, pathInMountPoint = pathSplitAtMountPoint
+            return self.mountedTars[mountPoint].fileVersions(pathInMountPoint)
+        return 1 if self._exists(path) else 0
+
+    def read(self, path: str, size: int, offset: int, fileInfo: Optional[FileInfo] = None) -> bytes:
+        """
+        fileInfo: This argument can be specified for performance reasons. It must be the FileInfo object for path!
+        """
+        pathSplitAtMountPoint = self._findMountedTar(path)
+        if pathSplitAtMountPoint:
+            mountPoint, pathInMountPoint = pathSplitAtMountPoint
+            return self.mountedTars[mountPoint].read(pathInMountPoint, size, offset, fileInfo)
+
+        realpath = self._realpath(path)
+        if not self._exists(path):
+            raise ValueError("Specified path '{}' is not a file that can be read!".format(realpath))
+
+        # TODO: Avoid opening the file on each read? I guess that's what the fh argument in fusepy is for!
+        #       Note that it does not matter for TAR file read because the TAR itself is kept open and only the
+        #       StenciledFile is opened on each read and then only for sparse files.
+        with open(realpath, 'rb') as file:
+            file.seek(offset)
+            return file.read(size)
+
+
+class TarFileType:
+    """
+    Similar to argparse.FileType but raises an exception if it is not a valid TAR file.
+    """
+
+    def __init__(self, encoding: str = tarfile.ENCODING) -> None:
+        self.encoding = encoding
+
+    def __call__(self, tarFile: str) -> Tuple[str, Optional[str]]:
+        if not os.path.exists(tarFile):
+            raise argparse.ArgumentTypeError("File '{}' does not exist!".format(tarFile))
+
+        with open(tarFile, 'rb') as fileobj:
+            fileSize = os.stat(tarFile).st_size
+            compression = SQLiteIndexedTar._detectCompression(fileobj)
+
+            try:
+                # Determining if there are many frames in zstd is O(1) with is_multiframe
+                if compression != 'zst' or supportedCompressions[compression].moduleName not in globals():
+                    raise Exception()  # early exit because we catch it ourself anyways
+
+                zstdFile = supportedCompressions[compression].open(fileobj)
+
+                if not zstdFile.is_multiframe() and fileSize > 1024 * 1024:
+                    print("[Warning] The specified file '{}'".format(tarFile))
+                    print("[Warning] is compressed using zstd but only contains one zstd frame. This makes it ")
+                    print("[Warning] impossible to use true seeking! Please (re)compress your TAR using multiple ")
+                    print("[Warning] frames in order for ratarmount to do be able to do fast seeking to requested ")
+                    print("[Warning] files. Else, each file access will decompress the whole TAR from the beginning!")
+                    print("[Warning] You can try out t2sz for creating such archives:")
+                    print("[Warning] https://github.com/martinellimarco/t2sz")
+                    print("[Warning] Here you can find a simple bash script demonstrating how to do this:")
+                    print("[Warning] https://github.com/mxmlnkn/ratarmount#xz-and-zst-files")
+                    print()
+            except Exception:
+                pass
+
+            if compression not in supportedCompressions:
+                if SQLiteIndexedTar._detectTar(fileobj, self.encoding):
+                    return tarFile, compression
+                raise argparse.ArgumentTypeError("Archive '{}' can't be opened!\n".format(tarFile))
+
+        cinfo = supportedCompressions[compression]
+        if cinfo.moduleName not in globals():
+            raise argparse.ArgumentTypeError(
+                "Can't open a {} compressed TAR file '{}' without {} module!".format(
+                    compression, fileobj.name, cinfo.moduleName
+                )
+            )
+
+        return tarFile, compression
+
+
+class _CustomFormatter(argparse.ArgumentDefaultsHelpFormatter, argparse.RawDescriptionHelpFormatter):
+    pass
+
+
+def _parseArgs(rawArgs: Optional[List[str]] = None):
+    parser = argparse.ArgumentParser(
+        formatter_class=_CustomFormatter,
+        description='''\
+With ratarmount, you can:
+  - Mount a (compressed) TAR file to a folder for read-only access
+  - Mount a compressed file to `<mountpoint>/<filename>`
+  - Bind mount a folder to another folder for read-only access
+  - Union mount a list of TARs, compressed files, and folders to a mount point
+    for read-only access
+''',
+        epilog='''\
+# Metadata Index Cache
+
+In order to reduce the mounting time, the created index for random access
+to files inside the tar will be saved to one of these locations. These
+locations are checked in order and the first, which works sufficiently, will
+be used. This is the default location order:
+
+  1. <path to tar>.index.sqlite
+  2. ~/.ratarmount/<path to tar: '/' -> '_'>.index.sqlite
+     E.g., ~/.ratarmount/_media_cdrom_programm.tar.index.sqlite
+
+This list of fallback folders can be overwritten using the `--index-folders`
+option. Furthermore, an explicitly named index file may be specified using
+the `--index-file` option. If `--index-file` is used, then the fallback
+folders, including the default ones, will be ignored!
+
+# Bind Mounting
+
+The mount sources can be TARs and/or folders.  Because of that, ratarmount
+can also be used to bind mount folders read-only to another path similar to
+`bindfs` and `mount --bind`. So, for:
+
+    ratarmount folder mountpoint
+
+all files in `folder` will now be visible in mountpoint.
+
+# Union Mounting
+
+If multiple mount sources are specified, the sources on the right side will be
+added to or update existing files from a mount source left of it. For example:
+
+    ratarmount folder1 folder2 mountpoint
+
+will make both, the files from folder1 and folder2, visible in mountpoint.
+If a file exists in both multiple source, then the file from the rightmost
+mount source will be used, which in the above example would be `folder2`.
+
+If you want to update / overwrite a folder with the contents of a given TAR,
+you can specify the folder both as a mount source and as the mount point:
+
+    ratarmount folder file.tar folder
+
+The FUSE option -o nonempty will be automatically added if such a usage is
+detected. If you instead want to update a TAR with a folder, you only have to
+swap the two mount sources:
+
+    ratarmount file.tar folder folder
+
+# File versions
+
+If a file exists multiple times in a TAR or in multiple mount sources, then
+the hidden versions can be accessed through special <file>.versions folders.
+For example, consider:
+
+    ratarmount folder updated.tar mountpoint
+
+and the file `foo` exists both in the folder and as two different versions
+in `updated.tar`. Then, you can list all three versions using:
+
+    ls -la mountpoint/foo.versions/
+        dr-xr-xr-x 2 user group     0 Apr 25 21:41 .
+        dr-x------ 2 user group 10240 Apr 26 15:59 ..
+        -r-x------ 2 user group   123 Apr 25 21:41 1
+        -r-x------ 2 user group   256 Apr 25 21:53 2
+        -r-x------ 2 user group  1024 Apr 25 22:13 3
+
+In this example, the oldest version has only 123 bytes while the newest and
+by default shown version has 1024 bytes. So, in order to look at the oldest
+version, you can simply do:
+    cat mountpoint/foo.versions/1
+
+# Compressed non-TAR files
+
+If you want a compressed file not containing a TAR, e.g., `foo.bz2`, then
+you can also use ratarmount for that. The uncompressed view will then be
+mounted to `<mountpoint>/foo` and you will be able to leverage ratarmount's
+seeking capabilities when opening that file.
+''',
+    )
+
+    # fmt: off
+    parser.add_argument(
+        '-f', '--foreground', action='store_true', default = False,
+        help = 'Keeps the python program in foreground so it can print debug '
+               'output when the mounted path is accessed.' )
+
+    parser.add_argument(
+        '-d', '--debug', type = int, default = 1,
+        help = 'Sets the debugging level. Higher means more output. Currently, 3 is the highest.' )
+
+    parser.add_argument(
+        '-c', '--recreate-index', action='store_true', default = False,
+        help = 'If specified, pre-existing .index files will be deleted and newly created.' )
+
+    parser.add_argument(
+        '-r', '--recursive', action='store_true', default = False,
+        help = 'Mount TAR archives inside the mounted TAR recursively. '
+               'Note that this only has an effect when creating an index. '
+               'If an index already exists, then this option will be effectively ignored. '
+               'Recreate the index if you want change the recursive mounting policy anyways.' )
+
+    # Considerations for the default value:
+    #   - seek times for the bz2 backend are between 0.01s and 0.1s
+    #   - seek times for the gzip backend are roughly 1/10th compared to bz2 at a default spacing of 4MiB
+    #     -> we could do a spacing of 40MiB (however the comparison are for another test archive, so it might not apply)
+    #   - ungziping firefox 66 inflates the compressed size of 66MiB to 184MiB (~3 times more) and takes 1.4s on my PC
+    #     -> to have a response time of 0.1s, it would require a spacing < 13MiB
+    #   - the gzip index takes roughly 32kiB per seek point
+    #   - the bzip2 index takes roughly 16B per 100-900kiB of compressed data
+    #     -> for the gzip index to have the same space efficiency assuming a compression ratio of only 1,
+    #        the spacing would have to be 1800MiB at which point it would become almost useless
+    parser.add_argument(
+        '-gs', '--gzip-seek-point-spacing', type = float, default = 16,
+        help =
+        'This only is applied when the index is first created or recreated with the -c option. '
+        'The spacing given in MiB specifies the seek point distance in the uncompressed data. '
+        'A distance of 16MiB means that archives smaller than 16MiB in uncompressed size will '
+        'not benefit from faster seek times. A seek point takes roughly 32kiB. '
+        'So, smaller distances lead to more responsive seeking but may explode the index size!' )
+
+    parser.add_argument(
+        '-p', '--prefix', type = str, default = '',
+        help = '[deprecated] Use "-o modules=subdir,subdir=<prefix>" instead. '
+               'This standard way utilizes FUSE itself and will also work for other FUSE '
+               'applications. So, it is preferable even if a bit more verbose.'
+               'The specified path to the folder inside the TAR will be mounted to root. '
+               'This can be useful when the archive as created with absolute paths. '
+               'E.g., for an archive created with `tar -P cf /var/log/apt/history.log`, '
+               '-p /var/log/apt/ can be specified so that the mount target directory '
+               '>directly< contains history.log.' )
+
+    parser.add_argument(
+        '-e', '--encoding', type = str, default = tarfile.ENCODING,
+        help = 'Specify an input encoding used for file names among others in the TAR. '
+               'This must be used when, e.g., trying to open a latin1 encoded TAR on an UTF-8 system. '
+               'Possible encodings: https://docs.python.org/3/library/codecs.html#standard-encodings' )
+
+    parser.add_argument(
+        '-i', '--ignore-zeros', action = 'store_true',
+        help = 'Ignore zeroed blocks in archive. Normally, two consecutive 512-blocks filled with zeroes mean EOF '
+               'and ratarmount stops reading after encountering them. This option instructs it to read further and '
+               'is useful when reading archives created with the -A option.' )
+
+    parser.add_argument(
+        '--verify-mtime', action = 'store_true',
+        help = 'By default, only the TAR file size is checked to match the one in the found existing ratarmount index. '
+               'If this option is specified, then also check the modification timestamp. But beware that the mtime '
+               'might change during copying or downloading without the contents changing. So, this check might cause '
+               'false positives.' )
+
+    parser.add_argument(
+        '-s', '--strip-recursive-tar-extension', action = 'store_true',
+        help = 'If true, then recursively mounted TARs named <file>.tar will be mounted at <file>/. '
+               'This might lead to folders of the same name being overwritten, so use with care. '
+               'The index needs to be (re)created to apply this option!' )
+
+    parser.add_argument(
+        '--index-file', type = str,
+        help = 'Specify a path to the .index.sqlite file. Setting this will disable fallback index folders.' )
+
+    parser.add_argument(
+        '--index-folders', default = "," + os.path.join( "~", ".ratarmount" ),
+        help = 'Specify one or multiple paths for storing .index.sqlite files. Paths will be tested for suitability '
+               'in the given order. An empty path will be interpreted as the location in which the TAR resides. '
+               'If the argument begins with a bracket "[", then it will be interpreted as a JSON-formatted list. '
+               'If the argument contains a comma ",", it will be interpreted as a comma-separated list of folders. '
+               'Else, the whole string will be interpreted as one folder path. Examples: '
+               '--index-folders ",~/.foo" will try to save besides the TAR and if that does not work, in ~/.foo. '
+               '--index-folders \'["~/.ratarmount", "foo,9000"]\' will never try to save besides the TAR. '
+               '--index-folder ~/.ratarmount will only test ~/.ratarmount as a storage location and nothing else. '
+               'Instead, it will first try ~/.ratarmount and the folder "foo,9000". ' )
+
+    parser.add_argument(
+        '-o', '--fuse', type = str, default = '',
+        help = 'Comma separated FUSE options. See "man mount.fuse" for help. '
+               'Example: --fuse "allow_other,entry_timeout=2.8,gid=0". ' )
+
+    parser.add_argument(
+        '-v', '--version', action='store_true', help = 'Print version string.' )
+
+    parser.add_argument(
+        'mount_source', nargs = '+',
+        help = 'The path to the TAR archive to be mounted. '
+               'If multiple archives and/or folders are specified, then they will be mounted as if the arguments '
+               'coming first were updated with the contents of the archives or folders specified thereafter, '
+               'i.e., the list of TARs and folders will be union mounted.' )
+    parser.add_argument(
+        'mount_point', nargs = '?',
+        help = 'The path to a folder to mount the TAR contents into. '
+               'If no mount path is specified, the TAR will be mounted to a folder of the same name '
+               'but without a file extension.' )
+    # fmt: on
+
+    args = parser.parse_args(rawArgs)
+
+    args.gzipSeekPointSpacing = args.gzip_seek_point_spacing * 1024 * 1024
+
+    # This is a hack but because we have two positional arguments (and want that reflected in the auto-generated help),
+    # all positional arguments, including the mountpath will be parsed into the tarfilepaths namespace and we have to
+    # manually separate them depending on the type.
+    if os.path.isdir(args.mount_source[-1]) or not os.path.exists(args.mount_source[-1]):
+        args.mount_point = args.mount_source[-1]
+        args.mount_source = args.mount_source[:-1]
+    if not args.mount_source:
+        print("[Error] You must at least specify one path to a valid TAR file or union mount source directory!")
+        sys.exit(1)
+
+    # Manually check that all specified TARs and folders exist
+    args.mount_source = [
+        TarFileType(encoding=args.encoding)(tarFile)[0] if not os.path.isdir(tarFile) else os.path.realpath(tarFile)
+        for tarFile in args.mount_source
+    ]
+
+    # Automatically generate a default mount path
+    if not args.mount_point:
+        autoMountPoint = stripSuffixFromTarFile(args.mount_source[0])
+        if args.mount_point == autoMountPoint:
+            args.mount_point = os.path.splitext(args.mount_source[0])[0]
+        else:
+            args.mount_point = autoMountPoint
+    args.mount_point = os.path.abspath(args.mount_point)
+
+    # Preprocess the --index-folders list as a string argument
+    if args.index_folders:
+        if args.index_folders[0] == '[':
+            args.index_folders = json.loads(args.index_folders)
+        elif ',' in args.index_folders:
+            args.index_folders = args.index_folders.split(',')
+
+    return args
diff --git a/codalab/lib/beam/seekable_zipfile.py b/codalab/lib/beam/seekable_zipfile.py
deleted file mode 100644
index 73afc5027..000000000
--- a/codalab/lib/beam/seekable_zipfile.py
+++ /dev/null
@@ -1,2143 +0,0 @@
-# type: ignore
-# This file comes from https://raw.githubusercontent.com/python/cpython/ad4f64d58c020016bd438de0e863a0d31d0f0dac/Lib/zipfile.py.
-# It incorporates https://github.com/python/cpython/pull/4966 (allowing seeking for zip files), which was released in Python 3.7.
-# Since we do not require Python 3.7, we are just using this file instead for the seekable zip file functionality.
-# TODO (Ashwin): Once we can upgrade the minimum Python version of CodaLab to be 3.7,
-# remove this file and replace instances of SeekableZipFile with just ZipFile.
-# 
-# To use it:
-# from codalab.lib.beam.seekable_zipfile import ZipFile as SeekableZipFile
-
-"""
-Read and write ZIP files.
-
-XXX references to utf-8 need further investigation.
-"""
-import io
-import os
-import importlib.util
-import sys
-import time
-import stat
-import shutil
-import struct
-import binascii
-import threading
-
-try:
-    import zlib # We may need its compression method
-    crc32 = zlib.crc32
-except ImportError:
-    zlib = None
-    crc32 = binascii.crc32
-
-try:
-    import bz2 # We may need its compression method
-except ImportError:
-    bz2 = None
-
-try:
-    import lzma # We may need its compression method
-except ImportError:
-    lzma = None
-
-__all__ = ["BadZipFile", "BadZipfile", "error",
-           "ZIP_STORED", "ZIP_DEFLATED", "ZIP_BZIP2", "ZIP_LZMA",
-           "is_zipfile", "ZipInfo", "ZipFile", "PyZipFile", "LargeZipFile"]
-
-class BadZipFile(Exception):
-    pass
-
-
-class LargeZipFile(Exception):
-    """
-    Raised when writing a zipfile, the zipfile requires ZIP64 extensions
-    and those extensions are disabled.
-    """
-
-error = BadZipfile = BadZipFile      # Pre-3.2 compatibility names
-
-
-ZIP64_LIMIT = (1 << 31) - 1
-ZIP_FILECOUNT_LIMIT = (1 << 16) - 1
-ZIP_MAX_COMMENT = (1 << 16) - 1
-
-# constants for Zip file compression methods
-ZIP_STORED = 0
-ZIP_DEFLATED = 8
-ZIP_BZIP2 = 12
-ZIP_LZMA = 14
-# Other ZIP compression methods not supported
-
-DEFAULT_VERSION = 20
-ZIP64_VERSION = 45
-BZIP2_VERSION = 46
-LZMA_VERSION = 63
-# we recognize (but not necessarily support) all features up to that version
-MAX_EXTRACT_VERSION = 63
-
-# Below are some formats and associated data for reading/writing headers using
-# the struct module.  The names and structures of headers/records are those used
-# in the PKWARE description of the ZIP file format:
-#     http://www.pkware.com/documents/casestudies/APPNOTE.TXT
-# (URL valid as of January 2008)
-
-# The "end of central directory" structure, magic number, size, and indices
-# (section V.I in the format document)
-structEndArchive = b"<4s4H2LH"
-stringEndArchive = b"PK\005\006"
-sizeEndCentDir = struct.calcsize(structEndArchive)
-
-_ECD_SIGNATURE = 0
-_ECD_DISK_NUMBER = 1
-_ECD_DISK_START = 2
-_ECD_ENTRIES_THIS_DISK = 3
-_ECD_ENTRIES_TOTAL = 4
-_ECD_SIZE = 5
-_ECD_OFFSET = 6
-_ECD_COMMENT_SIZE = 7
-# These last two indices are not part of the structure as defined in the
-# spec, but they are used internally by this module as a convenience
-_ECD_COMMENT = 8
-_ECD_LOCATION = 9
-
-# The "central directory" structure, magic number, size, and indices
-# of entries in the structure (section V.F in the format document)
-structCentralDir = "<4s4B4HL2L5H2L"
-stringCentralDir = b"PK\001\002"
-sizeCentralDir = struct.calcsize(structCentralDir)
-
-# indexes of entries in the central directory structure
-_CD_SIGNATURE = 0
-_CD_CREATE_VERSION = 1
-_CD_CREATE_SYSTEM = 2
-_CD_EXTRACT_VERSION = 3
-_CD_EXTRACT_SYSTEM = 4
-_CD_FLAG_BITS = 5
-_CD_COMPRESS_TYPE = 6
-_CD_TIME = 7
-_CD_DATE = 8
-_CD_CRC = 9
-_CD_COMPRESSED_SIZE = 10
-_CD_UNCOMPRESSED_SIZE = 11
-_CD_FILENAME_LENGTH = 12
-_CD_EXTRA_FIELD_LENGTH = 13
-_CD_COMMENT_LENGTH = 14
-_CD_DISK_NUMBER_START = 15
-_CD_INTERNAL_FILE_ATTRIBUTES = 16
-_CD_EXTERNAL_FILE_ATTRIBUTES = 17
-_CD_LOCAL_HEADER_OFFSET = 18
-
-# The "local file header" structure, magic number, size, and indices
-# (section V.A in the format document)
-structFileHeader = "<4s2B4HL2L2H"
-stringFileHeader = b"PK\003\004"
-sizeFileHeader = struct.calcsize(structFileHeader)
-
-_FH_SIGNATURE = 0
-_FH_EXTRACT_VERSION = 1
-_FH_EXTRACT_SYSTEM = 2
-_FH_GENERAL_PURPOSE_FLAG_BITS = 3
-_FH_COMPRESSION_METHOD = 4
-_FH_LAST_MOD_TIME = 5
-_FH_LAST_MOD_DATE = 6
-_FH_CRC = 7
-_FH_COMPRESSED_SIZE = 8
-_FH_UNCOMPRESSED_SIZE = 9
-_FH_FILENAME_LENGTH = 10
-_FH_EXTRA_FIELD_LENGTH = 11
-
-# The "Zip64 end of central directory locator" structure, magic number, and size
-structEndArchive64Locator = "<4sLQL"
-stringEndArchive64Locator = b"PK\x06\x07"
-sizeEndCentDir64Locator = struct.calcsize(structEndArchive64Locator)
-
-# The "Zip64 end of central directory" record, magic number, size, and indices
-# (section V.G in the format document)
-structEndArchive64 = "<4sQ2H2L4Q"
-stringEndArchive64 = b"PK\x06\x06"
-sizeEndCentDir64 = struct.calcsize(structEndArchive64)
-
-_CD64_SIGNATURE = 0
-_CD64_DIRECTORY_RECSIZE = 1
-_CD64_CREATE_VERSION = 2
-_CD64_EXTRACT_VERSION = 3
-_CD64_DISK_NUMBER = 4
-_CD64_DISK_NUMBER_START = 5
-_CD64_NUMBER_ENTRIES_THIS_DISK = 6
-_CD64_NUMBER_ENTRIES_TOTAL = 7
-_CD64_DIRECTORY_SIZE = 8
-_CD64_OFFSET_START_CENTDIR = 9
-
-def _check_zipfile(fp):
-    try:
-        if _EndRecData(fp):
-            return True         # file has correct magic number
-    except OSError:
-        pass
-    return False
-
-def is_zipfile(filename):
-    """Quickly see if a file is a ZIP file by checking the magic number.
-
-    The filename argument may be a file or file-like object too.
-    """
-    result = False
-    try:
-        if hasattr(filename, "read"):
-            result = _check_zipfile(fp=filename)
-        else:
-            with open(filename, "rb") as fp:
-                result = _check_zipfile(fp)
-    except OSError:
-        pass
-    return result
-
-def _EndRecData64(fpin, offset, endrec):
-    """
-    Read the ZIP64 end-of-archive records and use that to update endrec
-    """
-    try:
-        fpin.seek(offset - sizeEndCentDir64Locator, 2)
-    except OSError:
-        # If the seek fails, the file is not large enough to contain a ZIP64
-        # end-of-archive record, so just return the end record we were given.
-        return endrec
-
-    data = fpin.read(sizeEndCentDir64Locator)
-    if len(data) != sizeEndCentDir64Locator:
-        return endrec
-    sig, diskno, reloff, disks = struct.unpack(structEndArchive64Locator, data)
-    if sig != stringEndArchive64Locator:
-        return endrec
-
-    if diskno != 0 or disks != 1:
-        raise BadZipFile("zipfiles that span multiple disks are not supported")
-
-    # Assume no 'zip64 extensible data'
-    fpin.seek(offset - sizeEndCentDir64Locator - sizeEndCentDir64, 2)
-    data = fpin.read(sizeEndCentDir64)
-    if len(data) != sizeEndCentDir64:
-        return endrec
-    sig, sz, create_version, read_version, disk_num, disk_dir, \
-        dircount, dircount2, dirsize, diroffset = \
-        struct.unpack(structEndArchive64, data)
-    if sig != stringEndArchive64:
-        return endrec
-
-    # Update the original endrec using data from the ZIP64 record
-    endrec[_ECD_SIGNATURE] = sig
-    endrec[_ECD_DISK_NUMBER] = disk_num
-    endrec[_ECD_DISK_START] = disk_dir
-    endrec[_ECD_ENTRIES_THIS_DISK] = dircount
-    endrec[_ECD_ENTRIES_TOTAL] = dircount2
-    endrec[_ECD_SIZE] = dirsize
-    endrec[_ECD_OFFSET] = diroffset
-    return endrec
-
-
-def _EndRecData(fpin):
-    """Return data from the "End of Central Directory" record, or None.
-
-    The data is a list of the nine items in the ZIP "End of central dir"
-    record followed by a tenth item, the file seek offset of this record."""
-
-    # Determine file size
-    fpin.seek(0, 2)
-    filesize = fpin.tell()
-
-    # Check to see if this is ZIP file with no archive comment (the
-    # "end of central directory" structure should be the last item in the
-    # file if this is the case).
-    try:
-        fpin.seek(-sizeEndCentDir, 2)
-    except OSError:
-        return None
-    data = fpin.read()
-    if (len(data) == sizeEndCentDir and
-        data[0:4] == stringEndArchive and
-        data[-2:] == b"\000\000"):
-        # the signature is correct and there's no comment, unpack structure
-        endrec = struct.unpack(structEndArchive, data)
-        endrec=list(endrec)
-
-        # Append a blank comment and record start offset
-        endrec.append(b"")
-        endrec.append(filesize - sizeEndCentDir)
-
-        # Try to read the "Zip64 end of central directory" structure
-        return _EndRecData64(fpin, -sizeEndCentDir, endrec)
-
-    # Either this is not a ZIP file, or it is a ZIP file with an archive
-    # comment.  Search the end of the file for the "end of central directory"
-    # record signature. The comment is the last item in the ZIP file and may be
-    # up to 64K long.  It is assumed that the "end of central directory" magic
-    # number does not appear in the comment.
-    maxCommentStart = max(filesize - (1 << 16) - sizeEndCentDir, 0)
-    fpin.seek(maxCommentStart, 0)
-    data = fpin.read()
-    start = data.rfind(stringEndArchive)
-    if start >= 0:
-        # found the magic number; attempt to unpack and interpret
-        recData = data[start:start+sizeEndCentDir]
-        if len(recData) != sizeEndCentDir:
-            # Zip file is corrupted.
-            return None
-        endrec = list(struct.unpack(structEndArchive, recData))
-        commentSize = endrec[_ECD_COMMENT_SIZE] #as claimed by the zip file
-        comment = data[start+sizeEndCentDir:start+sizeEndCentDir+commentSize]
-        endrec.append(comment)
-        endrec.append(maxCommentStart + start)
-
-        # Try to read the "Zip64 end of central directory" structure
-        return _EndRecData64(fpin, maxCommentStart + start - filesize,
-                             endrec)
-
-    # Unable to find a valid end of central directory structure
-    return None
-
-
-class ZipInfo (object):
-    """Class with attributes describing each file in the ZIP archive."""
-
-    __slots__ = (
-        'orig_filename',
-        'filename',
-        'date_time',
-        'compress_type',
-        '_compresslevel',
-        'comment',
-        'extra',
-        'create_system',
-        'create_version',
-        'extract_version',
-        'reserved',
-        'flag_bits',
-        'volume',
-        'internal_attr',
-        'external_attr',
-        'header_offset',
-        'CRC',
-        'compress_size',
-        'file_size',
-        '_raw_time',
-    )
-
-    def __init__(self, filename="NoName", date_time=(1980,1,1,0,0,0)):
-        self.orig_filename = filename   # Original file name in archive
-
-        # Terminate the file name at the first null byte.  Null bytes in file
-        # names are used as tricks by viruses in archives.
-        null_byte = filename.find(chr(0))
-        if null_byte >= 0:
-            filename = filename[0:null_byte]
-        # This is used to ensure paths in generated ZIP files always use
-        # forward slashes as the directory separator, as required by the
-        # ZIP format specification.
-        if os.sep != "/" and os.sep in filename:
-            filename = filename.replace(os.sep, "/")
-
-        self.filename = filename        # Normalized file name
-        self.date_time = date_time      # year, month, day, hour, min, sec
-
-        if date_time[0] < 1980:
-            raise ValueError('ZIP does not support timestamps before 1980')
-
-        # Standard values:
-        self.compress_type = ZIP_STORED # Type of compression for the file
-        self._compresslevel = None      # Level for the compressor
-        self.comment = b""              # Comment for each file
-        self.extra = b""                # ZIP extra data
-        if sys.platform == 'win32':
-            self.create_system = 0          # System which created ZIP archive
-        else:
-            # Assume everything else is unix-y
-            self.create_system = 3          # System which created ZIP archive
-        self.create_version = DEFAULT_VERSION  # Version which created ZIP archive
-        self.extract_version = DEFAULT_VERSION # Version needed to extract archive
-        self.reserved = 0               # Must be zero
-        self.flag_bits = 0              # ZIP flag bits
-        self.volume = 0                 # Volume number of file header
-        self.internal_attr = 0          # Internal attributes
-        self.external_attr = 0          # External file attributes
-        # Other attributes are set by class ZipFile:
-        # header_offset         Byte offset to the file header
-        # CRC                   CRC-32 of the uncompressed file
-        # compress_size         Size of the compressed file
-        # file_size             Size of the uncompressed file
-
-    def __repr__(self):
-        result = ['<%s filename=%r' % (self.__class__.__name__, self.filename)]
-        if self.compress_type != ZIP_STORED:
-            result.append(' compress_type=%s' %
-                          compressor_names.get(self.compress_type,
-                                               self.compress_type))
-        hi = self.external_attr >> 16
-        lo = self.external_attr & 0xFFFF
-        if hi:
-            result.append(' filemode=%r' % stat.filemode(hi))
-        if lo:
-            result.append(' external_attr=%#x' % lo)
-        isdir = self.is_dir()
-        if not isdir or self.file_size:
-            result.append(' file_size=%r' % self.file_size)
-        if ((not isdir or self.compress_size) and
-            (self.compress_type != ZIP_STORED or
-             self.file_size != self.compress_size)):
-            result.append(' compress_size=%r' % self.compress_size)
-        result.append('>')
-        return ''.join(result)
-
-    def FileHeader(self, zip64=None):
-        """Return the per-file header as a string."""
-        dt = self.date_time
-        dosdate = (dt[0] - 1980) << 9 | dt[1] << 5 | dt[2]
-        dostime = dt[3] << 11 | dt[4] << 5 | (dt[5] // 2)
-        if self.flag_bits & 0x08:
-            # Set these to zero because we write them after the file data
-            CRC = compress_size = file_size = 0
-        else:
-            CRC = self.CRC
-            compress_size = self.compress_size
-            file_size = self.file_size
-
-        extra = self.extra
-
-        min_version = 0
-        if zip64 is None:
-            zip64 = file_size > ZIP64_LIMIT or compress_size > ZIP64_LIMIT
-        if zip64:
-            fmt = '<HHQQ'
-            extra = extra + struct.pack(fmt,
-                                        1, struct.calcsize(fmt)-4, file_size, compress_size)
-        if file_size > ZIP64_LIMIT or compress_size > ZIP64_LIMIT:
-            if not zip64:
-                raise LargeZipFile("Filesize would require ZIP64 extensions")
-            # File is larger than what fits into a 4 byte integer,
-            # fall back to the ZIP64 extension
-            file_size = 0xffffffff
-            compress_size = 0xffffffff
-            min_version = ZIP64_VERSION
-
-        if self.compress_type == ZIP_BZIP2:
-            min_version = max(BZIP2_VERSION, min_version)
-        elif self.compress_type == ZIP_LZMA:
-            min_version = max(LZMA_VERSION, min_version)
-
-        self.extract_version = max(min_version, self.extract_version)
-        self.create_version = max(min_version, self.create_version)
-        filename, flag_bits = self._encodeFilenameFlags()
-        header = struct.pack(structFileHeader, stringFileHeader,
-                             self.extract_version, self.reserved, flag_bits,
-                             self.compress_type, dostime, dosdate, CRC,
-                             compress_size, file_size,
-                             len(filename), len(extra))
-        return header + filename + extra
-
-    def _encodeFilenameFlags(self):
-        try:
-            return self.filename.encode('ascii'), self.flag_bits
-        except UnicodeEncodeError:
-            return self.filename.encode('utf-8'), self.flag_bits | 0x800
-
-    def _decodeExtra(self):
-        # Try to decode the extra field.
-        extra = self.extra
-        unpack = struct.unpack
-        while len(extra) >= 4:
-            tp, ln = unpack('<HH', extra[:4])
-            if ln+4 > len(extra):
-                raise BadZipFile("Corrupt extra field %04x (size=%d)" % (tp, ln))
-            if tp == 0x0001:
-                if ln >= 24:
-                    counts = unpack('<QQQ', extra[4:28])
-                elif ln == 16:
-                    counts = unpack('<QQ', extra[4:20])
-                elif ln == 8:
-                    counts = unpack('<Q', extra[4:12])
-                elif ln == 0:
-                    counts = ()
-                else:
-                    raise BadZipFile("Corrupt extra field %04x (size=%d)" % (tp, ln))
-
-                idx = 0
-
-                # ZIP64 extension (large files and/or large archives)
-                if self.file_size in (0xffffffffffffffff, 0xffffffff):
-                    self.file_size = counts[idx]
-                    idx += 1
-
-                if self.compress_size == 0xFFFFFFFF:
-                    self.compress_size = counts[idx]
-                    idx += 1
-
-                if self.header_offset == 0xffffffff:
-                    old = self.header_offset
-                    self.header_offset = counts[idx]
-                    idx+=1
-
-            extra = extra[ln+4:]
-
-    @classmethod
-    def from_file(cls, filename, arcname=None):
-        """Construct an appropriate ZipInfo for a file on the filesystem.
-
-        filename should be the path to a file or directory on the filesystem.
-
-        arcname is the name which it will have within the archive (by default,
-        this will be the same as filename, but without a drive letter and with
-        leading path separators removed).
-        """
-        if isinstance(filename, os.PathLike):
-            filename = os.fspath(filename)
-        st = os.stat(filename)
-        isdir = stat.S_ISDIR(st.st_mode)
-        mtime = time.localtime(st.st_mtime)
-        date_time = mtime[0:6]
-        # Create ZipInfo instance to store file information
-        if arcname is None:
-            arcname = filename
-        arcname = os.path.normpath(os.path.splitdrive(arcname)[1])
-        while arcname[0] in (os.sep, os.altsep):
-            arcname = arcname[1:]
-        if isdir:
-            arcname += '/'
-        zinfo = cls(arcname, date_time)
-        zinfo.external_attr = (st.st_mode & 0xFFFF) << 16  # Unix attributes
-        if isdir:
-            zinfo.file_size = 0
-            zinfo.external_attr |= 0x10  # MS-DOS directory flag
-        else:
-            zinfo.file_size = st.st_size
-
-        return zinfo
-
-    def is_dir(self):
-        """Return True if this archive member is a directory."""
-        return self.filename[-1] == '/'
-
-
-# ZIP encryption uses the CRC32 one-byte primitive for scrambling some
-# internal keys. We noticed that a direct implementation is faster than
-# relying on binascii.crc32().
-
-_crctable = None
-def _gen_crc(crc):
-    for j in range(8):
-        if crc & 1:
-            crc = (crc >> 1) ^ 0xEDB88320
-        else:
-            crc >>= 1
-    return crc
-
-# ZIP supports a password-based form of encryption. Even though known
-# plaintext attacks have been found against it, it is still useful
-# to be able to get data out of such a file.
-#
-# Usage:
-#     zd = _ZipDecrypter(mypwd)
-#     plain_bytes = zd(cypher_bytes)
-
-def _ZipDecrypter(pwd):
-    key0 = 305419896
-    key1 = 591751049
-    key2 = 878082192
-
-    global _crctable
-    if _crctable is None:
-        _crctable = list(map(_gen_crc, range(256)))
-    crctable = _crctable
-
-    def crc32(ch, crc):
-        """Compute the CRC32 primitive on one byte."""
-        return (crc >> 8) ^ crctable[(crc ^ ch) & 0xFF]
-
-    def update_keys(c):
-        nonlocal key0, key1, key2
-        key0 = crc32(c, key0)
-        key1 = (key1 + (key0 & 0xFF)) & 0xFFFFFFFF
-        key1 = (key1 * 134775813 + 1) & 0xFFFFFFFF
-        key2 = crc32(key1 >> 24, key2)
-
-    for p in pwd:
-        update_keys(p)
-
-    def decrypter(data):
-        """Decrypt a bytes object."""
-        result = bytearray()
-        append = result.append
-        for c in data:
-            k = key2 | 2
-            c ^= ((k * (k^1)) >> 8) & 0xFF
-            update_keys(c)
-            append(c)
-        return bytes(result)
-
-    return decrypter
-
-
-class LZMACompressor:
-
-    def __init__(self):
-        self._comp = None
-
-    def _init(self):
-        props = lzma._encode_filter_properties({'id': lzma.FILTER_LZMA1})
-        self._comp = lzma.LZMACompressor(lzma.FORMAT_RAW, filters=[
-            lzma._decode_filter_properties(lzma.FILTER_LZMA1, props)
-        ])
-        return struct.pack('<BBH', 9, 4, len(props)) + props
-
-    def compress(self, data):
-        if self._comp is None:
-            return self._init() + self._comp.compress(data)
-        return self._comp.compress(data)
-
-    def flush(self):
-        if self._comp is None:
-            return self._init() + self._comp.flush()
-        return self._comp.flush()
-
-
-class LZMADecompressor:
-
-    def __init__(self):
-        self._decomp = None
-        self._unconsumed = b''
-        self.eof = False
-
-    def decompress(self, data):
-        if self._decomp is None:
-            self._unconsumed += data
-            if len(self._unconsumed) <= 4:
-                return b''
-            psize, = struct.unpack('<H', self._unconsumed[2:4])
-            if len(self._unconsumed) <= 4 + psize:
-                return b''
-
-            self._decomp = lzma.LZMADecompressor(lzma.FORMAT_RAW, filters=[
-                lzma._decode_filter_properties(lzma.FILTER_LZMA1,
-                                               self._unconsumed[4:4 + psize])
-            ])
-            data = self._unconsumed[4 + psize:]
-            del self._unconsumed
-
-        result = self._decomp.decompress(data)
-        self.eof = self._decomp.eof
-        return result
-
-
-compressor_names = {
-    0: 'store',
-    1: 'shrink',
-    2: 'reduce',
-    3: 'reduce',
-    4: 'reduce',
-    5: 'reduce',
-    6: 'implode',
-    7: 'tokenize',
-    8: 'deflate',
-    9: 'deflate64',
-    10: 'implode',
-    12: 'bzip2',
-    14: 'lzma',
-    18: 'terse',
-    19: 'lz77',
-    97: 'wavpack',
-    98: 'ppmd',
-}
-
-def _check_compression(compression):
-    if compression == ZIP_STORED:
-        pass
-    elif compression == ZIP_DEFLATED:
-        if not zlib:
-            raise RuntimeError(
-                "Compression requires the (missing) zlib module")
-    elif compression == ZIP_BZIP2:
-        if not bz2:
-            raise RuntimeError(
-                "Compression requires the (missing) bz2 module")
-    elif compression == ZIP_LZMA:
-        if not lzma:
-            raise RuntimeError(
-                "Compression requires the (missing) lzma module")
-    else:
-        raise NotImplementedError("That compression method is not supported")
-
-
-def _get_compressor(compress_type, compresslevel=None):
-    if compress_type == ZIP_DEFLATED:
-        if compresslevel is not None:
-            return zlib.compressobj(compresslevel, zlib.DEFLATED, -15)
-        return zlib.compressobj(zlib.Z_DEFAULT_COMPRESSION, zlib.DEFLATED, -15)
-    elif compress_type == ZIP_BZIP2:
-        if compresslevel is not None:
-            return bz2.BZ2Compressor(compresslevel)
-        return bz2.BZ2Compressor()
-    # compresslevel is ignored for ZIP_LZMA
-    elif compress_type == ZIP_LZMA:
-        return LZMACompressor()
-    else:
-        return None
-
-
-def _get_decompressor(compress_type):
-    if compress_type == ZIP_STORED:
-        return None
-    elif compress_type == ZIP_DEFLATED:
-        return zlib.decompressobj(-15)
-    elif compress_type == ZIP_BZIP2:
-        return bz2.BZ2Decompressor()
-    elif compress_type == ZIP_LZMA:
-        return LZMADecompressor()
-    else:
-        descr = compressor_names.get(compress_type)
-        if descr:
-            raise NotImplementedError("compression type %d (%s)" % (compress_type, descr))
-        else:
-            raise NotImplementedError("compression type %d" % (compress_type,))
-
-
-class _SharedFile:
-    def __init__(self, file, pos, close, lock, writing):
-        self._file = file
-        self._pos = pos
-        self._close = close
-        self._lock = lock
-        self._writing = writing
-        self.seekable = file.seekable
-        self.tell = file.tell
-
-    def seek(self, offset, whence=0):
-        with self._lock:
-            if self._writing():
-                raise ValueError("Can't reposition in the ZIP file while "
-                        "there is an open writing handle on it. "
-                        "Close the writing handle before trying to read.")
-            self._file.seek(offset, whence)
-            self._pos = self._file.tell()
-            return self._pos
-
-    def read(self, n=-1):
-        with self._lock:
-            if self._writing():
-                raise ValueError("Can't read from the ZIP file while there "
-                        "is an open writing handle on it. "
-                        "Close the writing handle before trying to read.")
-            self._file.seek(self._pos)
-            data = self._file.read(n)
-            self._pos = self._file.tell()
-            return data
-
-    def close(self):
-        if self._file is not None:
-            fileobj = self._file
-            self._file = None
-            self._close(fileobj)
-
-# Provide the tell method for unseekable stream
-class _Tellable:
-    def __init__(self, fp):
-        self.fp = fp
-        self.offset = 0
-
-    def write(self, data):
-        n = self.fp.write(data)
-        self.offset += n
-        return n
-
-    def tell(self):
-        return self.offset
-
-    def flush(self):
-        self.fp.flush()
-
-    def close(self):
-        self.fp.close()
-
-
-class ZipExtFile(io.BufferedIOBase):
-    """File-like object for reading an archive member.
-       Is returned by ZipFile.open().
-    """
-
-    # Max size supported by decompressor.
-    MAX_N = 1 << 31 - 1
-
-    # Read from compressed files in 4k blocks.
-    MIN_READ_SIZE = 4096
-
-    # Chunk size to read during seek
-    MAX_SEEK_READ = 1 << 24
-
-    def __init__(self, fileobj, mode, zipinfo, decrypter=None,
-                 close_fileobj=False):
-        self._fileobj = fileobj
-        self._decrypter = decrypter
-        self._close_fileobj = close_fileobj
-
-        self._compress_type = zipinfo.compress_type
-        self._compress_left = zipinfo.compress_size
-        self._left = zipinfo.file_size
-
-        self._decompressor = _get_decompressor(self._compress_type)
-
-        self._eof = False
-        self._readbuffer = b''
-        self._offset = 0
-
-        self.newlines = None
-
-        # Adjust read size for encrypted files since the first 12 bytes
-        # are for the encryption/password information.
-        if self._decrypter is not None:
-            self._compress_left -= 12
-
-        self.mode = mode
-        self.name = zipinfo.filename
-
-        if hasattr(zipinfo, 'CRC'):
-            self._expected_crc = zipinfo.CRC
-            self._running_crc = crc32(b'')
-        else:
-            self._expected_crc = None
-
-        self._seekable = False
-        try:
-            if fileobj.seekable():
-                self._orig_compress_start = fileobj.tell()
-                self._orig_compress_size = zipinfo.compress_size
-                self._orig_file_size = zipinfo.file_size
-                self._orig_start_crc = self._running_crc
-                self._seekable = True
-        except AttributeError:
-            pass
-
-    def __repr__(self):
-        result = ['<%s.%s' % (self.__class__.__module__,
-                              self.__class__.__qualname__)]
-        if not self.closed:
-            result.append(' name=%r mode=%r' % (self.name, self.mode))
-            if self._compress_type != ZIP_STORED:
-                result.append(' compress_type=%s' %
-                              compressor_names.get(self._compress_type,
-                                                   self._compress_type))
-        else:
-            result.append(' [closed]')
-        result.append('>')
-        return ''.join(result)
-
-    def readline(self, limit=-1):
-        """Read and return a line from the stream.
-
-        If limit is specified, at most limit bytes will be read.
-        """
-
-        if limit < 0:
-            # Shortcut common case - newline found in buffer.
-            i = self._readbuffer.find(b'\n', self._offset) + 1
-            if i > 0:
-                line = self._readbuffer[self._offset: i]
-                self._offset = i
-                return line
-
-        return io.BufferedIOBase.readline(self, limit)
-
-    def peek(self, n=1):
-        """Returns buffered bytes without advancing the position."""
-        if n > len(self._readbuffer) - self._offset:
-            chunk = self.read(n)
-            if len(chunk) > self._offset:
-                self._readbuffer = chunk + self._readbuffer[self._offset:]
-                self._offset = 0
-            else:
-                self._offset -= len(chunk)
-
-        # Return up to 512 bytes to reduce allocation overhead for tight loops.
-        return self._readbuffer[self._offset: self._offset + 512]
-
-    def readable(self):
-        return True
-
-    def read(self, n=-1):
-        """Read and return up to n bytes.
-        If the argument is omitted, None, or negative, data is read and returned until EOF is reached..
-        """
-        if n is None or n < 0:
-            buf = self._readbuffer[self._offset:]
-            self._readbuffer = b''
-            self._offset = 0
-            while not self._eof:
-                buf += self._read1(self.MAX_N)
-            return buf
-
-        end = n + self._offset
-        if end < len(self._readbuffer):
-            buf = self._readbuffer[self._offset:end]
-            self._offset = end
-            return buf
-
-        n = end - len(self._readbuffer)
-        buf = self._readbuffer[self._offset:]
-        self._readbuffer = b''
-        self._offset = 0
-        while n > 0 and not self._eof:
-            data = self._read1(n)
-            if n < len(data):
-                self._readbuffer = data
-                self._offset = n
-                buf += data[:n]
-                break
-            buf += data
-            n -= len(data)
-        return buf
-
-    def _update_crc(self, newdata):
-        # Update the CRC using the given data.
-        if self._expected_crc is None:
-            # No need to compute the CRC if we don't have a reference value
-            return
-        self._running_crc = crc32(newdata, self._running_crc)
-        # Check the CRC if we're at the end of the file
-        if self._eof and self._running_crc != self._expected_crc:
-            raise BadZipFile("Bad CRC-32 for file %r" % self.name)
-
-    def read1(self, n):
-        """Read up to n bytes with at most one read() system call."""
-
-        if n is None or n < 0:
-            buf = self._readbuffer[self._offset:]
-            self._readbuffer = b''
-            self._offset = 0
-            while not self._eof:
-                data = self._read1(self.MAX_N)
-                if data:
-                    buf += data
-                    break
-            return buf
-
-        end = n + self._offset
-        if end < len(self._readbuffer):
-            buf = self._readbuffer[self._offset:end]
-            self._offset = end
-            return buf
-
-        n = end - len(self._readbuffer)
-        buf = self._readbuffer[self._offset:]
-        self._readbuffer = b''
-        self._offset = 0
-        if n > 0:
-            while not self._eof:
-                data = self._read1(n)
-                if n < len(data):
-                    self._readbuffer = data
-                    self._offset = n
-                    buf += data[:n]
-                    break
-                if data:
-                    buf += data
-                    break
-        return buf
-
-    def _read1(self, n):
-        # Read up to n compressed bytes with at most one read() system call,
-        # decrypt and decompress them.
-        if self._eof or n <= 0:
-            return b''
-
-        # Read from file.
-        if self._compress_type == ZIP_DEFLATED:
-            ## Handle unconsumed data.
-            data = self._decompressor.unconsumed_tail
-            if n > len(data):
-                data += self._read2(n - len(data))
-        else:
-            data = self._read2(n)
-
-        if self._compress_type == ZIP_STORED:
-            self._eof = self._compress_left <= 0
-        elif self._compress_type == ZIP_DEFLATED:
-            n = max(n, self.MIN_READ_SIZE)
-            data = self._decompressor.decompress(data, n)
-            self._eof = (self._decompressor.eof or
-                         self._compress_left <= 0 and
-                         not self._decompressor.unconsumed_tail)
-            if self._eof:
-                data += self._decompressor.flush()
-        else:
-            data = self._decompressor.decompress(data)
-            self._eof = self._decompressor.eof or self._compress_left <= 0
-
-        data = data[:self._left]
-        self._left -= len(data)
-        if self._left <= 0:
-            self._eof = True
-        self._update_crc(data)
-        return data
-
-    def _read2(self, n):
-        if self._compress_left <= 0:
-            return b''
-
-        n = max(n, self.MIN_READ_SIZE)
-        n = min(n, self._compress_left)
-
-        data = self._fileobj.read(n)
-        self._compress_left -= len(data)
-        if not data:
-            raise EOFError
-
-        if self._decrypter is not None:
-            data = self._decrypter(data)
-        return data
-
-    def close(self):
-        try:
-            if self._close_fileobj:
-                self._fileobj.close()
-        finally:
-            super().close()
-
-    def seekable(self):
-        return self._seekable
-
-    def seek(self, offset, whence=0):
-        if not self._seekable:
-            raise io.UnsupportedOperation("underlying stream is not seekable")
-        curr_pos = self.tell()
-        if whence == 0: # Seek from start of file
-            new_pos = offset
-        elif whence == 1: # Seek from current position
-            new_pos = curr_pos + offset
-        elif whence == 2: # Seek from EOF
-            new_pos = self._orig_file_size + offset
-        else:
-            raise ValueError("whence must be os.SEEK_SET (0), "
-                             "os.SEEK_CUR (1), or os.SEEK_END (2)")
-
-        if new_pos > self._orig_file_size:
-            new_pos = self._orig_file_size
-
-        if new_pos < 0:
-            new_pos = 0
-
-        read_offset = new_pos - curr_pos
-        buff_offset = read_offset + self._offset
-
-        if buff_offset >= 0 and buff_offset < len(self._readbuffer):
-            # Just move the _offset index if the new position is in the _readbuffer
-            self._offset = buff_offset
-            read_offset = 0
-        elif read_offset < 0:
-            # Position is before the current position. Reset the ZipExtFile
-            self._fileobj.seek(self._orig_compress_start)
-            self._running_crc = self._orig_start_crc
-            self._compress_left = self._orig_compress_size
-            self._left = self._orig_file_size
-            self._readbuffer = b''
-            self._offset = 0
-            self._decompressor = _get_decompressor(self._compress_type)
-            self._eof = False
-            read_offset = new_pos
-
-        while read_offset > 0:
-            read_len = min(self.MAX_SEEK_READ, read_offset)
-            self.read(read_len)
-            read_offset -= read_len
-
-        return self.tell()
-
-    def tell(self):
-        if not self._seekable:
-            raise io.UnsupportedOperation("underlying stream is not seekable")
-        filepos = self._orig_file_size - self._left - len(self._readbuffer) + self._offset
-        return filepos
-
-
-class _ZipWriteFile(io.BufferedIOBase):
-    def __init__(self, zf, zinfo, zip64):
-        self._zinfo = zinfo
-        self._zip64 = zip64
-        self._zipfile = zf
-        self._compressor = _get_compressor(zinfo.compress_type,
-                                           zinfo._compresslevel)
-        self._file_size = 0
-        self._compress_size = 0
-        self._crc = 0
-
-    @property
-    def _fileobj(self):
-        return self._zipfile.fp
-
-    def writable(self):
-        return True
-
-    def write(self, data):
-        if self.closed:
-            raise ValueError('I/O operation on closed file.')
-        nbytes = len(data)
-        self._file_size += nbytes
-        self._crc = crc32(data, self._crc)
-        if self._compressor:
-            data = self._compressor.compress(data)
-            self._compress_size += len(data)
-        self._fileobj.write(data)
-        return nbytes
-
-    def close(self):
-        if self.closed:
-            return
-        super().close()
-        # Flush any data from the compressor, and update header info
-        if self._compressor:
-            buf = self._compressor.flush()
-            self._compress_size += len(buf)
-            self._fileobj.write(buf)
-            self._zinfo.compress_size = self._compress_size
-        else:
-            self._zinfo.compress_size = self._file_size
-        self._zinfo.CRC = self._crc
-        self._zinfo.file_size = self._file_size
-
-        # Write updated header info
-        if self._zinfo.flag_bits & 0x08:
-            # Write CRC and file sizes after the file data
-            fmt = '<LQQ' if self._zip64 else '<LLL'
-            self._fileobj.write(struct.pack(fmt, self._zinfo.CRC,
-                self._zinfo.compress_size, self._zinfo.file_size))
-            self._zipfile.start_dir = self._fileobj.tell()
-        else:
-            if not self._zip64:
-                if self._file_size > ZIP64_LIMIT:
-                    raise RuntimeError('File size unexpectedly exceeded ZIP64 '
-                                       'limit')
-                if self._compress_size > ZIP64_LIMIT:
-                    raise RuntimeError('Compressed size unexpectedly exceeded '
-                                       'ZIP64 limit')
-            # Seek backwards and write file header (which will now include
-            # correct CRC and file sizes)
-
-            # Preserve current position in file
-            self._zipfile.start_dir = self._fileobj.tell()
-            self._fileobj.seek(self._zinfo.header_offset)
-            self._fileobj.write(self._zinfo.FileHeader(self._zip64))
-            self._fileobj.seek(self._zipfile.start_dir)
-
-        self._zipfile._writing = False
-
-        # Successfully written: Add file to our caches
-        self._zipfile.filelist.append(self._zinfo)
-        self._zipfile.NameToInfo[self._zinfo.filename] = self._zinfo
-
-class ZipFile:
-    """ Class with methods to open, read, write, close, list zip files.
-
-    z = ZipFile(file, mode="r", compression=ZIP_STORED, allowZip64=True,
-                compresslevel=None)
-
-    file: Either the path to the file, or a file-like object.
-          If it is a path, the file will be opened and closed by ZipFile.
-    mode: The mode can be either read 'r', write 'w', exclusive create 'x',
-          or append 'a'.
-    compression: ZIP_STORED (no compression), ZIP_DEFLATED (requires zlib),
-                 ZIP_BZIP2 (requires bz2) or ZIP_LZMA (requires lzma).
-    allowZip64: if True ZipFile will create files with ZIP64 extensions when
-                needed, otherwise it will raise an exception when this would
-                be necessary.
-    compresslevel: None (default for the given compression type) or an integer
-                   specifying the level to pass to the compressor.
-                   When using ZIP_STORED or ZIP_LZMA this keyword has no effect.
-                   When using ZIP_DEFLATED integers 0 through 9 are accepted.
-                   When using ZIP_BZIP2 integers 1 through 9 are accepted.
-
-    """
-
-    fp = None                   # Set here since __del__ checks it
-    _windows_illegal_name_trans_table = None
-
-    def __init__(self, file, mode="r", compression=ZIP_STORED, allowZip64=True,
-                 compresslevel=None):
-        """Open the ZIP file with mode read 'r', write 'w', exclusive create 'x',
-        or append 'a'."""
-        if mode not in ('r', 'w', 'x', 'a'):
-            raise ValueError("ZipFile requires mode 'r', 'w', 'x', or 'a'")
-
-        _check_compression(compression)
-
-        self._allowZip64 = allowZip64
-        self._didModify = False
-        self.debug = 0  # Level of printing: 0 through 3
-        self.NameToInfo = {}    # Find file info given name
-        self.filelist = []      # List of ZipInfo instances for archive
-        self.compression = compression  # Method of compression
-        self.compresslevel = compresslevel
-        self.mode = mode
-        self.pwd = None
-        self._comment = b''
-
-        # Check if we were passed a file-like object
-        if isinstance(file, os.PathLike):
-            file = os.fspath(file)
-        if isinstance(file, str):
-            # No, it's a filename
-            self._filePassed = 0
-            self.filename = file
-            modeDict = {'r' : 'rb', 'w': 'w+b', 'x': 'x+b', 'a' : 'r+b',
-                        'r+b': 'w+b', 'w+b': 'wb', 'x+b': 'xb'}
-            filemode = modeDict[mode]
-            while True:
-                try:
-                    self.fp = io.open(file, filemode)
-                except OSError:
-                    if filemode in modeDict:
-                        filemode = modeDict[filemode]
-                        continue
-                    raise
-                break
-        else:
-            self._filePassed = 1
-            self.fp = file
-            self.filename = getattr(file, 'name', None)
-        self._fileRefCnt = 1
-        self._lock = threading.RLock()
-        self._seekable = True
-        self._writing = False
-
-        try:
-            if mode == 'r':
-                self._RealGetContents()
-            elif mode in ('w', 'x'):
-                # set the modified flag so central directory gets written
-                # even if no files are added to the archive
-                self._didModify = True
-                try:
-                    self.start_dir = self.fp.tell()
-                except (AttributeError, OSError):
-                    self.fp = _Tellable(self.fp)
-                    self.start_dir = 0
-                    self._seekable = False
-                else:
-                    # Some file-like objects can provide tell() but not seek()
-                    try:
-                        self.fp.seek(self.start_dir)
-                    except (AttributeError, OSError):
-                        self._seekable = False
-            elif mode == 'a':
-                try:
-                    # See if file is a zip file
-                    self._RealGetContents()
-                    # seek to start of directory and overwrite
-                    self.fp.seek(self.start_dir)
-                except BadZipFile:
-                    # file is not a zip file, just append
-                    self.fp.seek(0, 2)
-
-                    # set the modified flag so central directory gets written
-                    # even if no files are added to the archive
-                    self._didModify = True
-                    self.start_dir = self.fp.tell()
-            else:
-                raise ValueError("Mode must be 'r', 'w', 'x', or 'a'")
-        except:
-            fp = self.fp
-            self.fp = None
-            self._fpclose(fp)
-            raise
-
-    def __enter__(self):
-        return self
-
-    def __exit__(self, type, value, traceback):
-        self.close()
-
-    def __repr__(self):
-        result = ['<%s.%s' % (self.__class__.__module__,
-                              self.__class__.__qualname__)]
-        if self.fp is not None:
-            if self._filePassed:
-                result.append(' file=%r' % self.fp)
-            elif self.filename is not None:
-                result.append(' filename=%r' % self.filename)
-            result.append(' mode=%r' % self.mode)
-        else:
-            result.append(' [closed]')
-        result.append('>')
-        return ''.join(result)
-
-    def _RealGetContents(self):
-        """Read in the table of contents for the ZIP file."""
-        fp = self.fp
-        try:
-            endrec = _EndRecData(fp)
-        except OSError:
-            raise BadZipFile("File is not a zip file")
-        if not endrec:
-            raise BadZipFile("File is not a zip file")
-        if self.debug > 1:
-            print(endrec)
-        size_cd = endrec[_ECD_SIZE]             # bytes in central directory
-        offset_cd = endrec[_ECD_OFFSET]         # offset of central directory
-        self._comment = endrec[_ECD_COMMENT]    # archive comment
-
-        # "concat" is zero, unless zip was concatenated to another file
-        concat = endrec[_ECD_LOCATION] - size_cd - offset_cd
-        if endrec[_ECD_SIGNATURE] == stringEndArchive64:
-            # If Zip64 extension structures are present, account for them
-            concat -= (sizeEndCentDir64 + sizeEndCentDir64Locator)
-
-        if self.debug > 2:
-            inferred = concat + offset_cd
-            print("given, inferred, offset", offset_cd, inferred, concat)
-        # self.start_dir:  Position of start of central directory
-        self.start_dir = offset_cd + concat
-        fp.seek(self.start_dir, 0)
-        data = fp.read(size_cd)
-        fp = io.BytesIO(data)
-        total = 0
-        while total < size_cd:
-            centdir = fp.read(sizeCentralDir)
-            if len(centdir) != sizeCentralDir:
-                raise BadZipFile("Truncated central directory")
-            centdir = struct.unpack(structCentralDir, centdir)
-            if centdir[_CD_SIGNATURE] != stringCentralDir:
-                raise BadZipFile("Bad magic number for central directory")
-            if self.debug > 2:
-                print(centdir)
-            filename = fp.read(centdir[_CD_FILENAME_LENGTH])
-            flags = centdir[5]
-            if flags & 0x800:
-                # UTF-8 file names extension
-                filename = filename.decode('utf-8')
-            else:
-                # Historical ZIP filename encoding
-                filename = filename.decode('cp437')
-            # Create ZipInfo instance to store file information
-            x = ZipInfo(filename)
-            x.extra = fp.read(centdir[_CD_EXTRA_FIELD_LENGTH])
-            x.comment = fp.read(centdir[_CD_COMMENT_LENGTH])
-            x.header_offset = centdir[_CD_LOCAL_HEADER_OFFSET]
-            (x.create_version, x.create_system, x.extract_version, x.reserved,
-             x.flag_bits, x.compress_type, t, d,
-             x.CRC, x.compress_size, x.file_size) = centdir[1:12]
-            if x.extract_version > MAX_EXTRACT_VERSION:
-                raise NotImplementedError("zip file version %.1f" %
-                                          (x.extract_version / 10))
-            x.volume, x.internal_attr, x.external_attr = centdir[15:18]
-            # Convert date/time code to (year, month, day, hour, min, sec)
-            x._raw_time = t
-            x.date_time = ( (d>>9)+1980, (d>>5)&0xF, d&0x1F,
-                            t>>11, (t>>5)&0x3F, (t&0x1F) * 2 )
-
-            x._decodeExtra()
-            x.header_offset = x.header_offset + concat
-            self.filelist.append(x)
-            self.NameToInfo[x.filename] = x
-
-            # update total bytes read from central directory
-            total = (total + sizeCentralDir + centdir[_CD_FILENAME_LENGTH]
-                     + centdir[_CD_EXTRA_FIELD_LENGTH]
-                     + centdir[_CD_COMMENT_LENGTH])
-
-            if self.debug > 2:
-                print("total", total)
-
-
-    def namelist(self):
-        """Return a list of file names in the archive."""
-        return [data.filename for data in self.filelist]
-
-    def infolist(self):
-        """Return a list of class ZipInfo instances for files in the
-        archive."""
-        return self.filelist
-
-    def printdir(self, file=None):
-        """Print a table of contents for the zip file."""
-        print("%-46s %19s %12s" % ("File Name", "Modified    ", "Size"),
-              file=file)
-        for zinfo in self.filelist:
-            date = "%d-%02d-%02d %02d:%02d:%02d" % zinfo.date_time[:6]
-            print("%-46s %s %12d" % (zinfo.filename, date, zinfo.file_size),
-                  file=file)
-
-    def testzip(self):
-        """Read all the files and check the CRC."""
-        chunk_size = 2 ** 20
-        for zinfo in self.filelist:
-            try:
-                # Read by chunks, to avoid an OverflowError or a
-                # MemoryError with very large embedded files.
-                with self.open(zinfo.filename, "r") as f:
-                    while f.read(chunk_size):     # Check CRC-32
-                        pass
-            except BadZipFile:
-                return zinfo.filename
-
-    def getinfo(self, name):
-        """Return the instance of ZipInfo given 'name'."""
-        info = self.NameToInfo.get(name)
-        if info is None:
-            raise KeyError(
-                'There is no item named %r in the archive' % name)
-
-        return info
-
-    def setpassword(self, pwd):
-        """Set default password for encrypted files."""
-        if pwd and not isinstance(pwd, bytes):
-            raise TypeError("pwd: expected bytes, got %s" % type(pwd).__name__)
-        if pwd:
-            self.pwd = pwd
-        else:
-            self.pwd = None
-
-    @property
-    def comment(self):
-        """The comment text associated with the ZIP file."""
-        return self._comment
-
-    @comment.setter
-    def comment(self, comment):
-        if not isinstance(comment, bytes):
-            raise TypeError("comment: expected bytes, got %s" % type(comment).__name__)
-        # check for valid comment length
-        if len(comment) > ZIP_MAX_COMMENT:
-            import warnings
-            warnings.warn('Archive comment is too long; truncating to %d bytes'
-                          % ZIP_MAX_COMMENT, stacklevel=2)
-            comment = comment[:ZIP_MAX_COMMENT]
-        self._comment = comment
-        self._didModify = True
-
-    def read(self, name, pwd=None):
-        """Return file bytes (as a string) for name."""
-        with self.open(name, "r", pwd) as fp:
-            return fp.read()
-
-    def open(self, name, mode="r", pwd=None, *, force_zip64=False):
-        """Return file-like object for 'name'.
-
-        name is a string for the file name within the ZIP file, or a ZipInfo
-        object.
-
-        mode should be 'r' to read a file already in the ZIP file, or 'w' to
-        write to a file newly added to the archive.
-
-        pwd is the password to decrypt files (only used for reading).
-
-        When writing, if the file size is not known in advance but may exceed
-        2 GiB, pass force_zip64 to use the ZIP64 format, which can handle large
-        files.  If the size is known in advance, it is best to pass a ZipInfo
-        instance for name, with zinfo.file_size set.
-        """
-        if mode not in {"r", "w"}:
-            raise ValueError('open() requires mode "r" or "w"')
-        if pwd and not isinstance(pwd, bytes):
-            raise TypeError("pwd: expected bytes, got %s" % type(pwd).__name__)
-        if pwd and (mode == "w"):
-            raise ValueError("pwd is only supported for reading files")
-        if not self.fp:
-            raise ValueError(
-                "Attempt to use ZIP archive that was already closed")
-
-        # Make sure we have an info object
-        if isinstance(name, ZipInfo):
-            # 'name' is already an info object
-            zinfo = name
-        elif mode == 'w':
-            zinfo = ZipInfo(name)
-            zinfo.compress_type = self.compression
-            zinfo._compresslevel = self.compresslevel
-        else:
-            # Get info object for name
-            zinfo = self.getinfo(name)
-
-        if mode == 'w':
-            return self._open_to_write(zinfo, force_zip64=force_zip64)
-
-        if self._writing:
-            raise ValueError("Can't read from the ZIP file while there "
-                    "is an open writing handle on it. "
-                    "Close the writing handle before trying to read.")
-
-        # Open for reading:
-        self._fileRefCnt += 1
-        zef_file = _SharedFile(self.fp, zinfo.header_offset,
-                               self._fpclose, self._lock, lambda: self._writing)
-        try:
-            # Skip the file header:
-            fheader = zef_file.read(sizeFileHeader)
-            if len(fheader) != sizeFileHeader:
-                raise BadZipFile("Truncated file header")
-            fheader = struct.unpack(structFileHeader, fheader)
-            if fheader[_FH_SIGNATURE] != stringFileHeader:
-                raise BadZipFile("Bad magic number for file header")
-
-            fname = zef_file.read(fheader[_FH_FILENAME_LENGTH])
-            if fheader[_FH_EXTRA_FIELD_LENGTH]:
-                zef_file.read(fheader[_FH_EXTRA_FIELD_LENGTH])
-
-            if zinfo.flag_bits & 0x20:
-                # Zip 2.7: compressed patched data
-                raise NotImplementedError("compressed patched data (flag bit 5)")
-
-            if zinfo.flag_bits & 0x40:
-                # strong encryption
-                raise NotImplementedError("strong encryption (flag bit 6)")
-
-            if zinfo.flag_bits & 0x800:
-                # UTF-8 filename
-                fname_str = fname.decode("utf-8")
-            else:
-                fname_str = fname.decode("cp437")
-
-            if fname_str != zinfo.orig_filename:
-                raise BadZipFile(
-                    'File name in directory %r and header %r differ.'
-                    % (zinfo.orig_filename, fname))
-
-            # check for encrypted flag & handle password
-            is_encrypted = zinfo.flag_bits & 0x1
-            zd = None
-            if is_encrypted:
-                if not pwd:
-                    pwd = self.pwd
-                if not pwd:
-                    raise RuntimeError("File %r is encrypted, password "
-                                       "required for extraction" % name)
-
-                zd = _ZipDecrypter(pwd)
-                # The first 12 bytes in the cypher stream is an encryption header
-                #  used to strengthen the algorithm. The first 11 bytes are
-                #  completely random, while the 12th contains the MSB of the CRC,
-                #  or the MSB of the file time depending on the header type
-                #  and is used to check the correctness of the password.
-                header = zef_file.read(12)
-                h = zd(header[0:12])
-                if zinfo.flag_bits & 0x8:
-                    # compare against the file type from extended local headers
-                    check_byte = (zinfo._raw_time >> 8) & 0xff
-                else:
-                    # compare against the CRC otherwise
-                    check_byte = (zinfo.CRC >> 24) & 0xff
-                if h[11] != check_byte:
-                    raise RuntimeError("Bad password for file %r" % name)
-
-            return ZipExtFile(zef_file, mode, zinfo, zd, True)
-        except:
-            zef_file.close()
-            raise
-
-    def _open_to_write(self, zinfo, force_zip64=False):
-        if force_zip64 and not self._allowZip64:
-            raise ValueError(
-                "force_zip64 is True, but allowZip64 was False when opening "
-                "the ZIP file."
-            )
-        if self._writing:
-            raise ValueError("Can't write to the ZIP file while there is "
-                             "another write handle open on it. "
-                             "Close the first handle before opening another.")
-
-        # Sizes and CRC are overwritten with correct data after processing the file
-        if not hasattr(zinfo, 'file_size'):
-            zinfo.file_size = 0
-        zinfo.compress_size = 0
-        zinfo.CRC = 0
-
-        zinfo.flag_bits = 0x00
-        if zinfo.compress_type == ZIP_LZMA:
-            # Compressed data includes an end-of-stream (EOS) marker
-            zinfo.flag_bits |= 0x02
-        if not self._seekable:
-            zinfo.flag_bits |= 0x08
-
-        if not zinfo.external_attr:
-            zinfo.external_attr = 0o600 << 16  # permissions: ?rw-------
-
-        # Compressed size can be larger than uncompressed size
-        zip64 = self._allowZip64 and \
-                (force_zip64 or zinfo.file_size * 1.05 > ZIP64_LIMIT)
-
-        if self._seekable:
-            self.fp.seek(self.start_dir)
-        zinfo.header_offset = self.fp.tell()
-
-        self._writecheck(zinfo)
-        self._didModify = True
-
-        self.fp.write(zinfo.FileHeader(zip64))
-
-        self._writing = True
-        return _ZipWriteFile(self, zinfo, zip64)
-
-    def extract(self, member, path=None, pwd=None):
-        """Extract a member from the archive to the current working directory,
-           using its full name. Its file information is extracted as accurately
-           as possible. `member' may be a filename or a ZipInfo object. You can
-           specify a different directory using `path'.
-        """
-        if path is None:
-            path = os.getcwd()
-        else:
-            path = os.fspath(path)
-
-        return self._extract_member(member, path, pwd)
-
-    def extractall(self, path=None, members=None, pwd=None):
-        """Extract all members from the archive to the current working
-           directory. `path' specifies a different directory to extract to.
-           `members' is optional and must be a subset of the list returned
-           by namelist().
-        """
-        if members is None:
-            members = self.namelist()
-
-        if path is None:
-            path = os.getcwd()
-        else:
-            path = os.fspath(path)
-
-        for zipinfo in members:
-            self._extract_member(zipinfo, path, pwd)
-
-    @classmethod
-    def _sanitize_windows_name(cls, arcname, pathsep):
-        """Replace bad characters and remove trailing dots from parts."""
-        table = cls._windows_illegal_name_trans_table
-        if not table:
-            illegal = ':<>|"?*'
-            table = str.maketrans(illegal, '_' * len(illegal))
-            cls._windows_illegal_name_trans_table = table
-        arcname = arcname.translate(table)
-        # remove trailing dots
-        arcname = (x.rstrip('.') for x in arcname.split(pathsep))
-        # rejoin, removing empty parts.
-        arcname = pathsep.join(x for x in arcname if x)
-        return arcname
-
-    def _extract_member(self, member, targetpath, pwd):
-        """Extract the ZipInfo object 'member' to a physical
-           file on the path targetpath.
-        """
-        if not isinstance(member, ZipInfo):
-            member = self.getinfo(member)
-
-        # build the destination pathname, replacing
-        # forward slashes to platform specific separators.
-        arcname = member.filename.replace('/', os.path.sep)
-
-        if os.path.altsep:
-            arcname = arcname.replace(os.path.altsep, os.path.sep)
-        # interpret absolute pathname as relative, remove drive letter or
-        # UNC path, redundant separators, "." and ".." components.
-        arcname = os.path.splitdrive(arcname)[1]
-        invalid_path_parts = ('', os.path.curdir, os.path.pardir)
-        arcname = os.path.sep.join(x for x in arcname.split(os.path.sep)
-                                   if x not in invalid_path_parts)
-        if os.path.sep == '\\':
-            # filter illegal characters on Windows
-            arcname = self._sanitize_windows_name(arcname, os.path.sep)
-
-        targetpath = os.path.join(targetpath, arcname)
-        targetpath = os.path.normpath(targetpath)
-
-        # Create all upper directories if necessary.
-        upperdirs = os.path.dirname(targetpath)
-        if upperdirs and not os.path.exists(upperdirs):
-            os.makedirs(upperdirs)
-
-        if member.is_dir():
-            if not os.path.isdir(targetpath):
-                os.mkdir(targetpath)
-            return targetpath
-
-        with self.open(member, pwd=pwd) as source, \
-             open(targetpath, "wb") as target:
-            shutil.copyfileobj(source, target)
-
-        return targetpath
-
-    def _writecheck(self, zinfo):
-        """Check for errors before writing a file to the archive."""
-        if zinfo.filename in self.NameToInfo:
-            import warnings
-            warnings.warn('Duplicate name: %r' % zinfo.filename, stacklevel=3)
-        if self.mode not in ('w', 'x', 'a'):
-            raise ValueError("write() requires mode 'w', 'x', or 'a'")
-        if not self.fp:
-            raise ValueError(
-                "Attempt to write ZIP archive that was already closed")
-        _check_compression(zinfo.compress_type)
-        if not self._allowZip64:
-            requires_zip64 = None
-            if len(self.filelist) >= ZIP_FILECOUNT_LIMIT:
-                requires_zip64 = "Files count"
-            elif zinfo.file_size > ZIP64_LIMIT:
-                requires_zip64 = "Filesize"
-            elif zinfo.header_offset > ZIP64_LIMIT:
-                requires_zip64 = "Zipfile size"
-            if requires_zip64:
-                raise LargeZipFile(requires_zip64 +
-                                   " would require ZIP64 extensions")
-
-    def write(self, filename, arcname=None,
-              compress_type=None, compresslevel=None):
-        """Put the bytes from filename into the archive under the name
-        arcname."""
-        if not self.fp:
-            raise ValueError(
-                "Attempt to write to ZIP archive that was already closed")
-        if self._writing:
-            raise ValueError(
-                "Can't write to ZIP archive while an open writing handle exists"
-            )
-
-        zinfo = ZipInfo.from_file(filename, arcname)
-
-        if zinfo.is_dir():
-            zinfo.compress_size = 0
-            zinfo.CRC = 0
-        else:
-            if compress_type is not None:
-                zinfo.compress_type = compress_type
-            else:
-                zinfo.compress_type = self.compression
-
-            if compresslevel is not None:
-                zinfo._compresslevel = compresslevel
-            else:
-                zinfo._compresslevel = self.compresslevel
-
-        if zinfo.is_dir():
-            with self._lock:
-                if self._seekable:
-                    self.fp.seek(self.start_dir)
-                zinfo.header_offset = self.fp.tell()  # Start of header bytes
-                if zinfo.compress_type == ZIP_LZMA:
-                # Compressed data includes an end-of-stream (EOS) marker
-                    zinfo.flag_bits |= 0x02
-
-                self._writecheck(zinfo)
-                self._didModify = True
-
-                self.filelist.append(zinfo)
-                self.NameToInfo[zinfo.filename] = zinfo
-                self.fp.write(zinfo.FileHeader(False))
-                self.start_dir = self.fp.tell()
-        else:
-            with open(filename, "rb") as src, self.open(zinfo, 'w') as dest:
-                shutil.copyfileobj(src, dest, 1024*8)
-
-    def writestr(self, zinfo_or_arcname, data,
-                 compress_type=None, compresslevel=None):
-        """Write a file into the archive.  The contents is 'data', which
-        may be either a 'str' or a 'bytes' instance; if it is a 'str',
-        it is encoded as UTF-8 first.
-        'zinfo_or_arcname' is either a ZipInfo instance or
-        the name of the file in the archive."""
-        if isinstance(data, str):
-            data = data.encode("utf-8")
-        if not isinstance(zinfo_or_arcname, ZipInfo):
-            zinfo = ZipInfo(filename=zinfo_or_arcname,
-                            date_time=time.localtime(time.time())[:6])
-            zinfo.compress_type = self.compression
-            zinfo._compresslevel = self.compresslevel
-            if zinfo.filename[-1] == '/':
-                zinfo.external_attr = 0o40775 << 16   # drwxrwxr-x
-                zinfo.external_attr |= 0x10           # MS-DOS directory flag
-            else:
-                zinfo.external_attr = 0o600 << 16     # ?rw-------
-        else:
-            zinfo = zinfo_or_arcname
-
-        if not self.fp:
-            raise ValueError(
-                "Attempt to write to ZIP archive that was already closed")
-        if self._writing:
-            raise ValueError(
-                "Can't write to ZIP archive while an open writing handle exists."
-            )
-
-        if compress_type is not None:
-            zinfo.compress_type = compress_type
-
-        if compresslevel is not None:
-            zinfo._compresslevel = compresslevel
-
-        zinfo.file_size = len(data)            # Uncompressed size
-        with self._lock:
-            with self.open(zinfo, mode='w') as dest:
-                dest.write(data)
-
-    def __del__(self):
-        """Call the "close()" method in case the user forgot."""
-        self.close()
-
-    def close(self):
-        """Close the file, and for mode 'w', 'x' and 'a' write the ending
-        records."""
-        if self.fp is None:
-            return
-
-        if self._writing:
-            raise ValueError("Can't close the ZIP file while there is "
-                             "an open writing handle on it. "
-                             "Close the writing handle before closing the zip.")
-
-        try:
-            if self.mode in ('w', 'x', 'a') and self._didModify: # write ending records
-                with self._lock:
-                    if self._seekable:
-                        self.fp.seek(self.start_dir)
-                    self._write_end_record()
-        finally:
-            fp = self.fp
-            self.fp = None
-            self._fpclose(fp)
-
-    def _write_end_record(self):
-        for zinfo in self.filelist:         # write central directory
-            dt = zinfo.date_time
-            dosdate = (dt[0] - 1980) << 9 | dt[1] << 5 | dt[2]
-            dostime = dt[3] << 11 | dt[4] << 5 | (dt[5] // 2)
-            extra = []
-            if zinfo.file_size > ZIP64_LIMIT \
-               or zinfo.compress_size > ZIP64_LIMIT:
-                extra.append(zinfo.file_size)
-                extra.append(zinfo.compress_size)
-                file_size = 0xffffffff
-                compress_size = 0xffffffff
-            else:
-                file_size = zinfo.file_size
-                compress_size = zinfo.compress_size
-
-            if zinfo.header_offset > ZIP64_LIMIT:
-                extra.append(zinfo.header_offset)
-                header_offset = 0xffffffff
-            else:
-                header_offset = zinfo.header_offset
-
-            extra_data = zinfo.extra
-            min_version = 0
-            if extra:
-                # Append a ZIP64 field to the extra's
-                extra_data = struct.pack(
-                    '<HH' + 'Q'*len(extra),
-                    1, 8*len(extra), *extra) + extra_data
-
-                min_version = ZIP64_VERSION
-
-            if zinfo.compress_type == ZIP_BZIP2:
-                min_version = max(BZIP2_VERSION, min_version)
-            elif zinfo.compress_type == ZIP_LZMA:
-                min_version = max(LZMA_VERSION, min_version)
-
-            extract_version = max(min_version, zinfo.extract_version)
-            create_version = max(min_version, zinfo.create_version)
-            try:
-                filename, flag_bits = zinfo._encodeFilenameFlags()
-                centdir = struct.pack(structCentralDir,
-                                      stringCentralDir, create_version,
-                                      zinfo.create_system, extract_version, zinfo.reserved,
-                                      flag_bits, zinfo.compress_type, dostime, dosdate,
-                                      zinfo.CRC, compress_size, file_size,
-                                      len(filename), len(extra_data), len(zinfo.comment),
-                                      0, zinfo.internal_attr, zinfo.external_attr,
-                                      header_offset)
-            except DeprecationWarning:
-                print((structCentralDir, stringCentralDir, create_version,
-                       zinfo.create_system, extract_version, zinfo.reserved,
-                       zinfo.flag_bits, zinfo.compress_type, dostime, dosdate,
-                       zinfo.CRC, compress_size, file_size,
-                       len(zinfo.filename), len(extra_data), len(zinfo.comment),
-                       0, zinfo.internal_attr, zinfo.external_attr,
-                       header_offset), file=sys.stderr)
-                raise
-            self.fp.write(centdir)
-            self.fp.write(filename)
-            self.fp.write(extra_data)
-            self.fp.write(zinfo.comment)
-
-        pos2 = self.fp.tell()
-        # Write end-of-zip-archive record
-        centDirCount = len(self.filelist)
-        centDirSize = pos2 - self.start_dir
-        centDirOffset = self.start_dir
-        requires_zip64 = None
-        if centDirCount > ZIP_FILECOUNT_LIMIT:
-            requires_zip64 = "Files count"
-        elif centDirOffset > ZIP64_LIMIT:
-            requires_zip64 = "Central directory offset"
-        elif centDirSize > ZIP64_LIMIT:
-            requires_zip64 = "Central directory size"
-        if requires_zip64:
-            # Need to write the ZIP64 end-of-archive records
-            if not self._allowZip64:
-                raise LargeZipFile(requires_zip64 +
-                                   " would require ZIP64 extensions")
-            zip64endrec = struct.pack(
-                structEndArchive64, stringEndArchive64,
-                44, 45, 45, 0, 0, centDirCount, centDirCount,
-                centDirSize, centDirOffset)
-            self.fp.write(zip64endrec)
-
-            zip64locrec = struct.pack(
-                structEndArchive64Locator,
-                stringEndArchive64Locator, 0, pos2, 1)
-            self.fp.write(zip64locrec)
-            centDirCount = min(centDirCount, 0xFFFF)
-            centDirSize = min(centDirSize, 0xFFFFFFFF)
-            centDirOffset = min(centDirOffset, 0xFFFFFFFF)
-
-        endrec = struct.pack(structEndArchive, stringEndArchive,
-                             0, 0, centDirCount, centDirCount,
-                             centDirSize, centDirOffset, len(self._comment))
-        self.fp.write(endrec)
-        self.fp.write(self._comment)
-        self.fp.flush()
-
-    def _fpclose(self, fp):
-        assert self._fileRefCnt > 0
-        self._fileRefCnt -= 1
-        if not self._fileRefCnt and not self._filePassed:
-            fp.close()
-
-
-class PyZipFile(ZipFile):
-    """Class to create ZIP archives with Python library files and packages."""
-
-    def __init__(self, file, mode="r", compression=ZIP_STORED,
-                 allowZip64=True, optimize=-1):
-        ZipFile.__init__(self, file, mode=mode, compression=compression,
-                         allowZip64=allowZip64)
-        self._optimize = optimize
-
-    def writepy(self, pathname, basename="", filterfunc=None):
-        """Add all files from "pathname" to the ZIP archive.
-
-        If pathname is a package directory, search the directory and
-        all package subdirectories recursively for all *.py and enter
-        the modules into the archive.  If pathname is a plain
-        directory, listdir *.py and enter all modules.  Else, pathname
-        must be a Python *.py file and the module will be put into the
-        archive.  Added modules are always module.pyc.
-        This method will compile the module.py into module.pyc if
-        necessary.
-        If filterfunc(pathname) is given, it is called with every argument.
-        When it is False, the file or directory is skipped.
-        """
-        pathname = os.fspath(pathname)
-        if filterfunc and not filterfunc(pathname):
-            if self.debug:
-                label = 'path' if os.path.isdir(pathname) else 'file'
-                print('%s %r skipped by filterfunc' % (label, pathname))
-            return
-        dir, name = os.path.split(pathname)
-        if os.path.isdir(pathname):
-            initname = os.path.join(pathname, "__init__.py")
-            if os.path.isfile(initname):
-                # This is a package directory, add it
-                if basename:
-                    basename = "%s/%s" % (basename, name)
-                else:
-                    basename = name
-                if self.debug:
-                    print("Adding package in", pathname, "as", basename)
-                fname, arcname = self._get_codename(initname[0:-3], basename)
-                if self.debug:
-                    print("Adding", arcname)
-                self.write(fname, arcname)
-                dirlist = sorted(os.listdir(pathname))
-                dirlist.remove("__init__.py")
-                # Add all *.py files and package subdirectories
-                for filename in dirlist:
-                    path = os.path.join(pathname, filename)
-                    root, ext = os.path.splitext(filename)
-                    if os.path.isdir(path):
-                        if os.path.isfile(os.path.join(path, "__init__.py")):
-                            # This is a package directory, add it
-                            self.writepy(path, basename,
-                                         filterfunc=filterfunc)  # Recursive call
-                    elif ext == ".py":
-                        if filterfunc and not filterfunc(path):
-                            if self.debug:
-                                print('file %r skipped by filterfunc' % path)
-                            continue
-                        fname, arcname = self._get_codename(path[0:-3],
-                                                            basename)
-                        if self.debug:
-                            print("Adding", arcname)
-                        self.write(fname, arcname)
-            else:
-                # This is NOT a package directory, add its files at top level
-                if self.debug:
-                    print("Adding files from directory", pathname)
-                for filename in sorted(os.listdir(pathname)):
-                    path = os.path.join(pathname, filename)
-                    root, ext = os.path.splitext(filename)
-                    if ext == ".py":
-                        if filterfunc and not filterfunc(path):
-                            if self.debug:
-                                print('file %r skipped by filterfunc' % path)
-                            continue
-                        fname, arcname = self._get_codename(path[0:-3],
-                                                            basename)
-                        if self.debug:
-                            print("Adding", arcname)
-                        self.write(fname, arcname)
-        else:
-            if pathname[-3:] != ".py":
-                raise RuntimeError(
-                    'Files added with writepy() must end with ".py"')
-            fname, arcname = self._get_codename(pathname[0:-3], basename)
-            if self.debug:
-                print("Adding file", arcname)
-            self.write(fname, arcname)
-
-    def _get_codename(self, pathname, basename):
-        """Return (filename, archivename) for the path.
-
-        Given a module name path, return the correct file path and
-        archive name, compiling if necessary.  For example, given
-        /python/lib/string, return (/python/lib/string.pyc, string).
-        """
-        def _compile(file, optimize=-1):
-            import py_compile
-            if self.debug:
-                print("Compiling", file)
-            try:
-                py_compile.compile(file, doraise=True, optimize=optimize)
-            except py_compile.PyCompileError as err:
-                print(err.msg)
-                return False
-            return True
-
-        file_py  = pathname + ".py"
-        file_pyc = pathname + ".pyc"
-        pycache_opt0 = importlib.util.cache_from_source(file_py, optimization='')
-        pycache_opt1 = importlib.util.cache_from_source(file_py, optimization=1)
-        pycache_opt2 = importlib.util.cache_from_source(file_py, optimization=2)
-        if self._optimize == -1:
-            # legacy mode: use whatever file is present
-            if (os.path.isfile(file_pyc) and
-                  os.stat(file_pyc).st_mtime >= os.stat(file_py).st_mtime):
-                # Use .pyc file.
-                arcname = fname = file_pyc
-            elif (os.path.isfile(pycache_opt0) and
-                  os.stat(pycache_opt0).st_mtime >= os.stat(file_py).st_mtime):
-                # Use the __pycache__/*.pyc file, but write it to the legacy pyc
-                # file name in the archive.
-                fname = pycache_opt0
-                arcname = file_pyc
-            elif (os.path.isfile(pycache_opt1) and
-                  os.stat(pycache_opt1).st_mtime >= os.stat(file_py).st_mtime):
-                # Use the __pycache__/*.pyc file, but write it to the legacy pyc
-                # file name in the archive.
-                fname = pycache_opt1
-                arcname = file_pyc
-            elif (os.path.isfile(pycache_opt2) and
-                  os.stat(pycache_opt2).st_mtime >= os.stat(file_py).st_mtime):
-                # Use the __pycache__/*.pyc file, but write it to the legacy pyc
-                # file name in the archive.
-                fname = pycache_opt2
-                arcname = file_pyc
-            else:
-                # Compile py into PEP 3147 pyc file.
-                if _compile(file_py):
-                    if sys.flags.optimize == 0:
-                        fname = pycache_opt0
-                    elif sys.flags.optimize == 1:
-                        fname = pycache_opt1
-                    else:
-                        fname = pycache_opt2
-                    arcname = file_pyc
-                else:
-                    fname = arcname = file_py
-        else:
-            # new mode: use given optimization level
-            if self._optimize == 0:
-                fname = pycache_opt0
-                arcname = file_pyc
-            else:
-                arcname = file_pyc
-                if self._optimize == 1:
-                    fname = pycache_opt1
-                elif self._optimize == 2:
-                    fname = pycache_opt2
-                else:
-                    msg = "invalid value for 'optimize': {!r}".format(self._optimize)
-                    raise ValueError(msg)
-            if not (os.path.isfile(fname) and
-                    os.stat(fname).st_mtime >= os.stat(file_py).st_mtime):
-                if not _compile(file_py, optimize=self._optimize):
-                    fname = arcname = file_py
-        archivename = os.path.split(arcname)[1]
-        if basename:
-            archivename = "%s/%s" % (basename, archivename)
-        return (fname, archivename)
-
-
-def main(args=None):
-    import argparse
-
-    description = 'A simple command-line interface for zipfile module.'
-    parser = argparse.ArgumentParser(description=description)
-    group = parser.add_mutually_exclusive_group(required=True)
-    group.add_argument('-l', '--list', metavar='<zipfile>',
-                       help='Show listing of a zipfile')
-    group.add_argument('-e', '--extract', nargs=2,
-                       metavar=('<zipfile>', '<output_dir>'),
-                       help='Extract zipfile into target dir')
-    group.add_argument('-c', '--create', nargs='+',
-                       metavar=('<name>', '<file>'),
-                       help='Create zipfile from sources')
-    group.add_argument('-t', '--test', metavar='<zipfile>',
-                       help='Test if a zipfile is valid')
-    args = parser.parse_args(args)
-
-    if args.test is not None:
-        src = args.test
-        with ZipFile(src, 'r') as zf:
-            badfile = zf.testzip()
-        if badfile:
-            print("The following enclosed file is corrupted: {!r}".format(badfile))
-        print("Done testing")
-
-    elif args.list is not None:
-        src = args.list
-        with ZipFile(src, 'r') as zf:
-            zf.printdir()
-
-    elif args.extract is not None:
-        src, curdir = args.extract
-        with ZipFile(src, 'r') as zf:
-            zf.extractall(curdir)
-
-    elif args.create is not None:
-        zip_name = args.create.pop(0)
-        files = args.create
-
-        def addToZip(zf, path, zippath):
-            if os.path.isfile(path):
-                zf.write(path, zippath, ZIP_DEFLATED)
-            elif os.path.isdir(path):
-                if zippath:
-                    zf.write(path, zippath)
-                for nm in sorted(os.listdir(path)):
-                    addToZip(zf,
-                             os.path.join(path, nm), os.path.join(zippath, nm))
-            # else: ignore
-
-        with ZipFile(zip_name, 'w') as zf:
-            for path in files:
-                zippath = os.path.basename(path)
-                if not zippath:
-                    zippath = os.path.basename(os.path.dirname(path))
-                if zippath in ('', os.curdir, os.pardir):
-                    zippath = ''
-                addToZip(zf, path, zippath)
-
-if __name__ == "__main__":
-    main()
\ No newline at end of file
diff --git a/codalab/lib/bundle_cli.py b/codalab/lib/bundle_cli.py
index d71c9d159..57f85a8f3 100644
--- a/codalab/lib/bundle_cli.py
+++ b/codalab/lib/bundle_cli.py
@@ -49,7 +49,6 @@
     precondition,
     UsageError,
     ensure_str,
-    parse_linked_bundle_url,
 )
 from codalab.lib import (
     bundle_util,
@@ -98,7 +97,7 @@
 from codalab.lib.bundle_store import MultiDiskBundleStore
 from codalab.lib.interactive_session import InteractiveSession
 from codalab.lib.print_util import FileTransferProgress
-from codalab.worker.file_util import un_tar_directory, unzip_directory
+from codalab.worker.file_util import un_tar_directory
 from codalab.worker.download_util import BundleTarget
 from codalab.worker.bundle_state import State, LinkFormat
 from codalab.rest.worksheet_block_schemas import BlockModes
@@ -1486,21 +1485,7 @@ def do_download_command(self, args):
         )
         with progress, closing(contents):
             if target_info['type'] == 'directory':
-                # The dependency can be a .tar.gz (if from local disk)
-                # or a .zip file (if on Azure Blob Storage).
-                # TODO (Ashwin): Unify some of the logic here with the code in DependencyManager._store_dependency()
-                # into common utility functions.
-                content_type = (
-                    "application/zip"
-                    if parse_linked_bundle_url(
-                        nested_dict_get(info, 'metadata', 'link_url')
-                    ).uses_beam
-                    else "application/gzip"
-                )
-                if content_type == "application/gzip":
-                    un_tar_directory(contents, final_path, 'gz', force=args.force)
-                elif content_type == "application/zip":
-                    unzip_directory(contents, final_path, force=args.force)
+                un_tar_directory(contents, final_path, 'gz', force=args.force)
             elif target_info['type'] == 'file':
                 with open(final_path, 'wb') as out:
                     shutil.copyfileobj(contents, out)
diff --git a/codalab/lib/download_manager.py b/codalab/lib/download_manager.py
index 2d1c53643..6f1de81ee 100644
--- a/codalab/lib/download_manager.py
+++ b/codalab/lib/download_manager.py
@@ -149,7 +149,7 @@ def _get_target_info_within_bundle(self, target, depth):
                 self._worker_model.deallocate_socket(response_socket_id)
 
     @retry_if_no_longer_running
-    def stream_archived_directory(self, target):
+    def stream_tarred_gzipped_directory(self, target):
         """
         Returns a file-like object containing an archive of the given directory.
         If the directory is on the local bundle store or from a worker hosting a
@@ -157,7 +157,7 @@ def stream_archived_directory(self, target):
         If it is already an archive (such as .zip) on Blob Storage, that archive
         will be returned directly.
 
-        Returns a tuple: (fileobj, mimetype, file extension)
+        Returns fileobj.
         """
         bundle_state = self._bundle_model.get_bundle_state(target.bundle_uuid)
         # Raises NotFoundException if uuid is invalid
@@ -172,13 +172,9 @@ def stream_archived_directory(self, target):
             directory_path = self._get_target_path(target)
             if parse_linked_bundle_url(directory_path).uses_beam:
                 # If streaming a folder within an Azure bundle, we need to download its contents,
-                # re-zip the folder, and return the zip file.
-                return (
-                    file_util.open_file(directory_path, sub_zip_directory=True),
-                    'application/zip',
-                    '.zip',
-                )
-            return (file_util.tar_gzip_directory(directory_path), 'application/gzip', '.tar.gz')
+                # re-archive the folder, and return the .tar.gz file.
+                return file_util.open_file(directory_path)
+            return file_util.tar_gzip_directory(directory_path)
         else:
             # stream_tarred_gzipped_directory calls are sent to the worker even
             # on a shared filesystem since
@@ -194,11 +190,7 @@ def stream_archived_directory(self, target):
                 read_args = {'type': 'stream_directory'}
                 self._send_read_message(worker, response_socket_id, target, read_args)
                 fileobj = self._get_read_response_stream(response_socket_id)
-                return (
-                    Deallocating(fileobj, self._worker_model, response_socket_id),
-                    'application/gzip',
-                    '.tar.gz',
-                )
+                return Deallocating(fileobj, self._worker_model, response_socket_id)
             except Exception:
                 self._worker_model.deallocate_socket(response_socket_id)
                 raise
diff --git a/codalab/lib/upload_manager.py b/codalab/lib/upload_manager.py
index 4545df5c8..34f3ad3bc 100644
--- a/codalab/lib/upload_manager.py
+++ b/codalab/lib/upload_manager.py
@@ -4,13 +4,15 @@
 
 from codalab.common import UsageError
 from codalab.lib import crypt_util, file_util, path_util, zip_util
-from codalab.worker.file_util import zip_directory
+from codalab.worker.file_util import tar_gzip_directory
 from codalab.lib.beam.filesystems import AZURE_BLOB_ACCOUNT_NAME
 from apache_beam.io.filesystem import CompressionTypes
 from apache_beam.io.filesystems import FileSystems
 from codalab.common import parse_linked_bundle_url
 from codalab.worker.bundle_state import LinkFormat
 import tempfile
+import logging
+from codalab.lib.beam.ratarmount import SQLiteIndexedTar
 
 
 class UploadManager(object):
@@ -124,7 +126,7 @@ def upload_to_bundle_store(
                             simplify_archive=simplify_archives,
                         )
                     else:
-                        # We reach this code path if we are uploading zip files to Azure
+                        # We reach this code path if we are uploading archive files to Azure
                         # (in which case unpack is set to False), or uploading a single file regularly.
                         with open(source_output_path, 'wb') as out:
                             shutil.copyfileobj(source[1], out)
@@ -137,12 +139,30 @@ def upload_to_bundle_store(
 
                 # is_directory is True if the bundle is a directory and False if it is a single file.
                 is_directory = os.path.isdir(bundle_path)
-                link_url = f"azfs://{AZURE_BLOB_ACCOUNT_NAME}/bundles/{bundle.uuid}{'/contents.zip' if is_directory else '/contents'}"
+                link_url = f"azfs://{AZURE_BLOB_ACCOUNT_NAME}/bundles/{bundle.uuid}{'/contents.tar.gz' if is_directory else '/contents'}"
                 with FileSystems.create(
                     link_url, compression_type=CompressionTypes.UNCOMPRESSED
-                ) as out:
+                ) as out, tempfile.NamedTemporaryFile(
+                    suffix=".tar.gz"
+                ) as tmp_tar_file, tempfile.NamedTemporaryFile(
+                    suffix=".sqlite"
+                ) as tmp_index_file:
                     if is_directory:
-                        shutil.copyfileobj(zip_directory(bundle_path), out)
+                        tar_gzip_directory(bundle_path, output_path=tmp_tar_file.name)
+                        shutil.copyfileobj(tmp_tar_file, out)
+                        with open(tmp_tar_file.name, "rb") as ttf:
+                            SQLiteIndexedTar(
+                                fileObject=ttf,
+                                tarFileName=bundle.uuid,
+                                writeIndex=True,
+                                clearIndexCache=True,
+                                indexFileName=tmp_index_file.name,
+                            )
+                        with FileSystems.create(
+                            link_url.replace("/contents.tar.gz", "/index.sqlite"),
+                            compression_type=CompressionTypes.UNCOMPRESSED,
+                        ) as out_index_file, open(tmp_index_file.name, "rb") as tif:
+                            shutil.copyfileobj(tif, out_index_file)
                     else:
                         shutil.copyfileobj(open(bundle_path, 'rb'), out)
 
diff --git a/codalab/rest/bundles.py b/codalab/rest/bundles.py
index 2c386b0fa..1f723b9fd 100644
--- a/codalab/rest/bundles.py
+++ b/codalab/rest/bundles.py
@@ -638,8 +638,9 @@ def _fetch_bundle_contents_blob(uuid, path=''):
             abort(http.client.BAD_REQUEST, 'Head and tail not supported for directory blobs.')
         # Always return archived directories
         gzipped_stream = False  # but don't set the encoding to 'gzip'
-        fileobj, mimetype, extension = local.download_manager.stream_archived_directory(target)
-        filename += extension
+        fileobj = local.download_manager.stream_tarred_gzipped_directory(target)
+        mimetype = "application/gzip"
+        filename += ".tar.gz"
     elif target_info['type'] == 'file':
         # Let's gzip to save bandwidth.
         # For simplicity, we do this even if the file is already a packed
diff --git a/codalab/server/bundle_manager.py b/codalab/server/bundle_manager.py
index b8ddcb192..b61984026 100644
--- a/codalab/server/bundle_manager.py
+++ b/codalab/server/bundle_manager.py
@@ -18,7 +18,7 @@
 from codalab.common import NotFoundError, PermissionError, parse_linked_bundle_url, normpath
 from codalab.lib import bundle_util, formatting, path_util
 from codalab.server.worker_info_accessor import WorkerInfoAccessor
-from codalab.worker.file_util import remove_path, un_tar_directory, unzip_directory
+from codalab.worker.file_util import remove_path, un_tar_directory
 from codalab.worker.bundle_state import State, RunResources
 from codalab.worker.download_util import get_target_info, BundleTarget
 import tempfile
@@ -263,19 +263,8 @@ def _make_bundle(self, bundle):
                         # TODO (Ashwin): Unify some of the logic here with the code in DependencyManager._store_dependency()
                         # into common utility functions.
                         if target_info['type'] == 'directory':
-                            (
-                                fileobj,
-                                content_type,
-                                extension,
-                            ) = self._download_manager.stream_archived_directory(target)
-                            # The dependency can be a .tar.gz (if from local disk)
-                            # or a .zip file (if on Azure Blob Storage).
-                            if content_type == "application/gzip":
-                                un_tar_directory(fileobj, dependency_path, 'gz')
-                            elif content_type == "application/zip":
-                                unzip_directory(fileobj, dependency_path)
-                            else:
-                                raise Exception(f"Invalid content type: {content_type}")
+                            fileobj = self._download_manager.stream_tarred_gzipped_directory(target)
+                            un_tar_directory(fileobj, dependency_path, 'gz')
                         else:
                             fileobj = self._download_manager.stream_file(target, gzipped=False)
                             with open(dependency_path, 'wb') as f:
diff --git a/codalab/worker/dependency_manager.py b/codalab/worker/dependency_manager.py
index 99a7761c8..b991669f6 100644
--- a/codalab/worker/dependency_manager.py
+++ b/codalab/worker/dependency_manager.py
@@ -9,7 +9,7 @@
 from typing import Dict
 
 from codalab.lib.formatting import size_str
-from codalab.worker.file_util import remove_path, un_tar_directory, unzip_directory
+from codalab.worker.file_util import remove_path, un_tar_directory
 from codalab.worker.fsm import BaseDependencyManager, DependencyStage, StateTransitioner
 import codalab.worker.pyjson
 from codalab.worker.worker_thread import ThreadDict
@@ -396,14 +396,7 @@ def _store_dependency(self, dependency_path, fileobj, target_type, content_type)
                 else:
                     os.remove(dependency_path)
             if target_type == 'directory':
-                # The dependency can be a .tar.gz (if from local disk)
-                # or a .zip file (if on Azure Blob Storage).
-                if content_type == "application/gzip":
-                    un_tar_directory(fileobj, dependency_path, 'gz')
-                elif content_type == "application/zip":
-                    unzip_directory(fileobj, dependency_path)
-                else:
-                    raise Exception(f"Invalid content type: {content_type}")
+                un_tar_directory(fileobj, dependency_path, 'gz')
             else:
                 with open(dependency_path, 'wb') as f:
                     logger.debug('copying file to %s', dependency_path)
diff --git a/codalab/worker/download_util.py b/codalab/worker/download_util.py
index b72771ebf..498f44093 100644
--- a/codalab/worker/download_util.py
+++ b/codalab/worker/download_util.py
@@ -1,9 +1,16 @@
 import os
 from apache_beam.io.filesystem import CompressionTypes
 from apache_beam.io.filesystems import FileSystems
-from zipfile import ZipFile
-import stat
+import tarfile
 from codalab.common import parse_linked_bundle_url
+import logging
+from codalab.worker.file_util import open_indexed_tar_gz_file
+from codalab.lib.beam.ratarmount import SQLiteIndexedTar
+
+import stat
+from io import BytesIO
+import tempfile
+import shutil
 
 
 class PathException(Exception):
@@ -68,8 +75,13 @@ def get_target_info(bundle_path, target, depth):
         # If the target is on Azure, use a special method using Apache Beam
         # to get the target info.
         try:
+            logging.info("going to _compute_target_info_beam")
             info = _compute_target_info_beam(final_path, depth)
-        except PathException:
+        except Exception:
+            import traceback
+
+            logging.info("error!")
+            logging.info("error: %s", traceback.format_exc())
             raise PathException(
                 "Path '{}' in bundle {} not found".format(target.subpath, target.bundle_uuid)
             )
@@ -159,11 +171,10 @@ def _compute_target_info_beam(path, depth):
     """Computes target info for a file that is externalized on a location
     such as Azure, by using the Apache Beam FileSystem APIs."""
 
-    # TODO (Ashwin): properly return permissions.
     linked_bundle_path = parse_linked_bundle_url(path)
     if not FileSystems.exists(linked_bundle_path.bundle_path):
         raise PathException
-    if not linked_bundle_path.is_zip:
+    if not linked_bundle_path.is_archive:
         # Single file
         file = FileSystems.match([path])[0].metadata_list[0]
         return {
@@ -173,68 +184,54 @@ def _compute_target_info_beam(path, depth):
             'perm': 0o777,
         }
 
-    with ZipFile(
-        FileSystems.open(
-            linked_bundle_path.bundle_path, compression_type=CompressionTypes.UNCOMPRESSED
-        )
-    ) as f:
-        zipinfos = [zipinfo for zipinfo in f.infolist()]
-
-        # These methods are used to detect and read symlinks created from "zip -y".
-        # The link path is stored in the contents of the file pointed to by zipinfo.
-        # See https://discuss.python.org/t/how-info-zip-represents-symlinks/4104
-        islink = lambda zipinfo: stat.S_ISLNK(zipinfo.external_attr >> 16)
-        readlink = lambda zipinfo: f.read(zipinfo).decode()
-
-        isfile = lambda zipinfo: not zipinfo.is_dir()
-        isdir = lambda zipinfo: zipinfo.is_dir()
-        listdir = lambda path: [
-            p
-            for p in [z.filename[len(path) :] for z in zipinfos if z.filename.startswith(path)]
-            if "/" not in p.strip("/") and p.strip("/")
-        ]
-
-        def _get_info(path, depth):
-            try:
-                zipinfo = next(
-                    z for z in zipinfos if z.filename == path or z.filename == path + '/'
-                )
-            except StopIteration:
-                # Not found
-                raise PathException
-            result = {}
-            result['name'] = zipinfo.filename.strip("/").split("/")[-1]  # get last part of path
-            result['size'] = zipinfo.file_size
-            result['perm'] = 0o777
-            if islink(zipinfo):
-                # See https://discuss.python.org/t/how-info-zip-represents-symlinks/4104
-                result['type'] = 'link'
-                result['link'] = readlink(zipinfo)
-            elif isfile(zipinfo):
-                result['type'] = 'file'
-            elif isdir(zipinfo):
-                result['type'] = 'directory'
-                if depth > 0:
-                    result['contents'] = [
-                        _get_info(
-                            zipinfo.filename.rstrip("/") + "/" + file_name.lstrip("/"), depth - 1
-                        )
-                        for file_name in listdir(path)
-                    ]
-            return result
-
-        if linked_bundle_path.zip_subpath:
-            # Return the contents of a subpath within a directory.
-            return _get_info(linked_bundle_path.zip_subpath, depth)
-        else:
-            # No subpath, return the entire directory.
-            file = FileSystems.match([path])[0].metadata_list[0]
-            result = {
-                'name': linked_bundle_path.bundle_uuid,
-                'type': 'directory',
-                'size': file.size_in_bytes,
-                'perm': 0o777,
-            }
+    tf, _ = open_indexed_tar_gz_file(linked_bundle_path.bundle_path)
+    islink = lambda finfo: stat.S_ISLNK(finfo.mode)
+    readlink = lambda finfo: finfo.linkname
+
+    isfile = lambda finfo: finfo.type in tarfile.REGULAR_TYPES
+    isdir = lambda finfo: finfo.type == tarfile.DIRTYPE
+    listdir = lambda path: tf.getFileInfo(path, listDir=True)
+
+    def _get_info(path, depth):
+        if not path.startswith("/"):
+            path = "/" + path
+        finfo = tf.getFileInfo(path)
+        if finfo is None:
+            # Not found
+            raise PathException
+        result = {}
+        result['name'] = path.split("/")[-1]  # get last part of path
+        result['size'] = finfo.size
+        result['perm'] = finfo.mode & 0o777
+        if islink(finfo):
+            result['type'] = 'link'
+            result['link'] = readlink(finfo)
+        elif isfile(finfo):
+            result['type'] = 'file'
+        elif isdir(finfo):
+            result['type'] = 'directory'
             if depth > 0:
-                result['contents'] = [_get_info(file_name, depth - 1) for file_name in listdir("")]
-            return result
+                result['contents'] = [
+                    _get_info(path + "/" + file_name, depth - 1)
+                    for file_name in listdir(path)
+                    if file_name != "."
+                ]
+        return result
+
+    if linked_bundle_path.archive_subpath:
+        # Return the contents of a subpath within a directory.
+        return _get_info(linked_bundle_path.archive_subpath, depth)
+    else:
+        # No subpath, return the entire directory.
+        file = FileSystems.match([path])[0].metadata_list[0]
+        result = {
+            'name': linked_bundle_path.bundle_uuid,
+            'type': 'directory',
+            'size': file.size_in_bytes,
+            'perm': 0o777,
+        }
+        if depth > 0:
+            result['contents'] = [
+                _get_info(file_name, depth - 1) for file_name in listdir("/") if file_name != "."
+            ]
+        return result
diff --git a/codalab/worker/file_util.py b/codalab/worker/file_util.py
index 43bca9f0e..0c5dc7998 100644
--- a/codalab/worker/file_util.py
+++ b/codalab/worker/file_util.py
@@ -8,14 +8,15 @@
 import tarfile
 import zlib
 import bz2
-from zipfile import ZipFile
-from codalab.lib.beam.seekable_zipfile import ZipFile as SeekableZipFile  # type: ignore
+from io import IOBase
 
 from codalab.common import BINARY_PLACEHOLDER, UsageError
 from codalab.common import parse_linked_bundle_url
 from apache_beam.io.filesystem import CompressionTypes
 from apache_beam.io.filesystems import FileSystems
 import tempfile
+import logging
+from codalab.lib.beam.ratarmount import SQLiteIndexedTar
 
 NONE_PLACEHOLDER = '<none>'
 
@@ -47,10 +48,10 @@ def tar_gzip_directory(
     exclude_patterns=None,
     exclude_names=None,
     ignore_file=None,
+    output_path=None,
 ):
     """
-    Returns a file-like object containing a tarred and gzipped archive of the
-    given directory.
+    Tars and gzips the given directory.
 
     follow_symlinks: Whether symbolic links should be followed.
     exclude_names: Any top-level directory entries with names in exclude_names
@@ -58,8 +59,10 @@ def tar_gzip_directory(
     exclude_patterns: Any directory entries with the given names at any depth in
                       the directory structure are excluded.
     ignore_file: Name of the file where exclusion patterns are read from.
+    output_path: If specified, outputs the .tar.gz file to the given path, and blocks
+    until this is complete. Otherwise, returns a file-like object with the .tar.gz stream.
     """
-    args = ['tar', 'czf', '-', '-C', directory_path]
+    args = ['tar', 'czf', output_path or '-', '-C', directory_path]
 
     # If the BSD tar library is being used, append --disable-copy to prevent creating ._* files
     if 'bsdtar' in get_tar_version_output():
@@ -85,7 +88,11 @@ def tar_gzip_directory(
     args.append('.')
     try:
         proc = subprocess.Popen(args, stdout=subprocess.PIPE)
-        return proc.stdout
+        if output_path:
+            # If output_path is specified, block until finished.
+            proc.stdout.read()
+        else:
+            return proc.stdout
     except subprocess.CalledProcessError as e:
         raise IOError(e.output)
 
@@ -205,40 +212,143 @@ def do_unzip(filename):
         do_unzip(fileobj_or_name)
 
 
-def open_file(
-    file_path, mode='r', compression_type=CompressionTypes.UNCOMPRESSED, sub_zip_directory=False
-):
+def open_indexed_tar_gz_file(path):
+    """Open a .tar.gz file specified by the provided path on Azure Blob Storage.
+    Also reads this file's associated index.sqlite file, then opens the file as an
+    SQLiteIndexedTar object.
+
+    This way, the .tar.gz file can be read and specific files can be extracted without
+    needing to download the entire .tar.gz file.
+    """
+    f = FileSystems.open(
+        path, compression_type=CompressionTypes.UNCOMPRESSED
+    )
+    with tempfile.NamedTemporaryFile(suffix=".sqlite") as index_file:
+        shutil.copyfileobj(
+            FileSystems.open(
+                path.replace("/contents.tar.gz", "/index.sqlite"),
+                compression_type=CompressionTypes.UNCOMPRESSED,
+            ),
+            index_file,
+        )
+        return (SQLiteIndexedTar(
+            fileObject=f,
+            tarFileName=parse_linked_bundle_url(path).bundle_uuid,
+            writeIndex=False,
+            clearIndexCache=False,
+            indexFileName=index_file.name,
+        ), f)
+
+
+class ClosingStreamWrapper(IOBase):
+    def __init__(self, buffer, cleanup_fn, cleanup_args):
+        self._buffer = buffer
+        self._cleanup_fn = cleanup_fn
+        self._cleanup_args = cleanup_args
+
+    def close(self, *args, **kwargs):
+        return self._buffer.close(*args, **kwargs)
+
+    @property
+    def closed(self):
+        return self._buffer.closed
+
+    def fileno(self, *args, **kwargs):
+        return self._buffer.fileno(*args, **kwargs)
+
+    def flush(self, *args, **kwargs):
+        return self._buffer.flush(*args, **kwargs)
+
+    def isatty(self, *args, **kwargs):
+        return self._buffer.isatty(*args, **kwargs)
+
+    def readable(self, *args, **kwargs):
+        return self._buffer.readable(*args, **kwargs)
+
+    def readline(self, *args, **kwargs):
+        return self._buffer.readline(*args, **kwargs)
+
+    def readlines(self, *args, **kwargs):
+        return self._buffer.readlines(*args, **kwargs)
+
+    def seek(self, *args, **kwargs):
+        return self._buffer.seek(*args, **kwargs)
+
+    def seekable(self, *args, **kwargs):
+        return self._buffer.seekable(*args, **kwargs)
+
+    def tell(self, *args, **kwargs):
+        return self._buffer.tell(*args, **kwargs)
+
+    def truncate(self, *args, **kwargs):
+        return self._buffer.truncate(*args, **kwargs)
+
+    def writable(self, *args, **kwargs):
+        return self._buffer.writable(*args, **kwargs)
+
+    def writelines(self, *args, **kwargs):
+        return self._buffer.writelines(*args, **kwargs)
+
+    def read(self, *args, **kwargs):
+        return self._buffer.read(*args, **kwargs)
+
+    def readall(self, *args, **kwargs):
+        return self._buffer.readall(*args, **kwargs)
+
+    def readinto(self, *args, **kwargs):
+        return self._buffer.readinto(*args, **kwargs)
+
+    def write(self, *args, **kwargs):
+        return self._buffer.write(*args, **kwargs)
+
+    def __del__(self, *args, **kwargs):
+        self._cleanup_fn(*self._cleanup_args)
+        return self._buffer.__del__(*args, **kwargs)
+
+
+def open_file(file_path, mode='r'):
     """
     Opens the file indicated by the given file path. Can be in a directory.
-    sub_zip_directory: If True, indicates that the given file path points to a directory that exists
-        within a zip file on Azure Blob Storage, in which case that directory will be downloaded,
-        re-zipped, and returned by this function. Default is False.
     """
     linked_bundle_path = parse_linked_bundle_url(file_path)
     if (
         linked_bundle_path.uses_beam
-        and linked_bundle_path.is_zip
-        and linked_bundle_path.zip_subpath
+        and linked_bundle_path.is_archive
+        and linked_bundle_path.archive_subpath
     ):
-        # If file path is a zip file on Azure, open the specified path within the
-        # zip file. zipfile.open only supports 'r', not 'rb', so we always pass in 'r'.
-        with SeekableZipFile(
-            FileSystems.open(linked_bundle_path.bundle_path, compression_type)
-        ) as f:
-            zinfo = f.getinfo(linked_bundle_path.zip_subpath + ("/" if sub_zip_directory else ""))
-            if zinfo.is_dir():
-                # If streaming a folder within an Azure bundle, we need to download its contents,
-                # re-archive the folder, and return the .tar.gz version of that folder.
-                with tempfile.TemporaryDirectory() as tmp_dirname:
-                    extracted_path = f.extract(zinfo, tmp_dirname)
-                    for z in f.infolist():
-                        # Extract other members of the directory.
-                        # TODO (Ashwin): Make sure this works with symlinks, too. .extract() may not work.
-                        if z.filename.startswith(linked_bundle_path.zip_subpath + "/"):
-                            f.extract(z, tmp_dirname)
-                    return zip_directory(extracted_path)
-            else:
-                return f.open(zinfo, 'r')
+        # If file path is a .tar.gz file on Azure, open the specified path within the archive.
+        tf, _ = open_indexed_tar_gz_file(linked_bundle_path.bundle_path)
+
+        isdir = lambda finfo: finfo.type == tarfile.DIRTYPE
+        listdir = lambda path: tf.getFileInfo(path, listDir=True)
+
+        fpath = "/" + linked_bundle_path.archive_subpath
+        finfo = tf.getFileInfo(fpath)
+        if finfo is None:
+            raise FileNotFoundError(fpath)
+        if isdir(finfo):
+            # If streaming a folder within an Azure bundle, we need to download its contents,
+            # re-archive the folder, and return the .tar.gz version of that folder.
+            # with tempfile.TemporaryDirectory() as tmp_dirname:
+            tmp_dir = tempfile.TemporaryDirectory()
+            extracted_path = os.path.join(tmp_dir.name, linked_bundle_path.archive_subpath)
+            os.mkdir(extracted_path)
+            for member_name, member_fileinfo in listdir(fpath).items():
+                # Extract other members of the directory.
+                # TODO (Ashwin): Make sure this works with symlinks, too.
+                with open(os.path.join(extracted_path, member_name), "wb+") as f:
+                    f.write(
+                        tf.read("", fileInfo=member_fileinfo, size=member_fileinfo.size, offset=0)
+                    )
+            return ClosingStreamWrapper(
+                tar_gzip_directory(extracted_path), lambda t: t.cleanup(), (tmp_dir,),
+            )
+        else:
+            # TODO: Implement a tf.open() function so that we don't have to read the entire file.
+            return BytesIO(tf.read(path="", fileInfo=finfo, size=finfo.size, offset=0))
+    # elif linked_bundle_path.uses_beam and linked_bundle_path.is_archive:
+        
+    #     return gzip.GzipFile(fileobj=FileSystems.open(file_path, mode, compression_type=CompressionTypes.UNCOMPRESSED))
     return FileSystems.open(file_path, mode, compression_type=CompressionTypes.UNCOMPRESSED)
 
 
@@ -384,21 +494,18 @@ def get_file_size(file_path):
     linked_bundle_path = parse_linked_bundle_url(file_path)
     if (
         linked_bundle_path.uses_beam
-        and linked_bundle_path.is_zip
-        and linked_bundle_path.zip_subpath
+        and linked_bundle_path.is_archive
+        and linked_bundle_path.archive_subpath
     ):
-        # If file path is a zip file on Azure, open the specified path within the
-        # zip file. zipfile.open only supports 'r', not 'rb', so we always pass in 'r'.
-        with ZipFile(
-            FileSystems.open(
-                linked_bundle_path.bundle_path, compression_type=CompressionTypes.UNCOMPRESSED
-            )
-        ) as f:
-            try:
-                zinfo = f.getinfo(linked_bundle_path.zip_subpath)
-            except KeyError:
-                raise FileNotFoundError(file_path)
-            return zinfo.file_size
+        # If file path is a .tar.gz file on Azure, open the specified path within the
+        # .tar.gz file.
+        tf, _ = open_indexed_tar_gz_file(linked_bundle_path.bundle_path)
+
+        fpath = "/" + linked_bundle_path.archive_subpath
+        finfo = tf.getFileInfo(fpath)
+        if finfo is None:
+            raise FileNotFoundError(fpath)
+        return finfo.size
     if not get_path_exists(file_path):
         raise FileNotFoundError(file_path)
     # TODO: add a FileSystems.size() method to Apache Beam to make this less verbose.
diff --git a/requirements.txt b/requirements.txt
index f81af1819..fd3a55c52 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -8,6 +8,7 @@ setuptools>=40.0.0
 argcomplete==1.12.1
 PyYAML==5.3.1
 psutil==5.7.2
+ratarmount==0.7.0
 six==1.15.0
 SQLAlchemy==1.3.19
 watchdog==0.10.3


Test patch:
diff --git a/tests/unit/lib/path_util_test.py b/tests/unit/lib/path_util_test.py
index af60ecbaa..0b9d3e281 100644
--- a/tests/unit/lib/path_util_test.py
+++ b/tests/unit/lib/path_util_test.py
@@ -127,34 +127,34 @@ def test_single_file(self):
         self.assertEqual(
             linked_bundle_path.bundle_path, "azfs://storageclwsdev0/bundles/uuid/contents.txt"
         )
-        self.assertEqual(linked_bundle_path.is_zip, False)
-        self.assertEqual(linked_bundle_path.zip_subpath, None)
+        self.assertEqual(linked_bundle_path.is_archive, False)
+        self.assertEqual(linked_bundle_path.archive_subpath, None)
         self.assertEqual(linked_bundle_path.bundle_uuid, "uuid")
 
     def test_directory(self):
-        """Parse a URL referring to a zipped directory."""
+        """Parse a URL referring to an archived directory."""
         linked_bundle_path = path_util.parse_linked_bundle_url(
-            "azfs://storageclwsdev0/bundles/uuid/contents.zip"
+            "azfs://storageclwsdev0/bundles/uuid/contents.tar.gz"
         )
         self.assertEqual(linked_bundle_path.storage_type, StorageType.AZURE_BLOB_STORAGE)
         self.assertEqual(
-            linked_bundle_path.bundle_path, "azfs://storageclwsdev0/bundles/uuid/contents.zip"
+            linked_bundle_path.bundle_path, "azfs://storageclwsdev0/bundles/uuid/contents.tar.gz"
         )
-        self.assertEqual(linked_bundle_path.is_zip, True)
-        self.assertEqual(linked_bundle_path.zip_subpath, None)
+        self.assertEqual(linked_bundle_path.is_archive, True)
+        self.assertEqual(linked_bundle_path.archive_subpath, None)
         self.assertEqual(linked_bundle_path.bundle_uuid, "uuid")
 
     def test_directory_with_subpath(self):
-        """Parse a URL referring to a subpath within a zipped directory."""
+        """Parse a URL referring to a subpath within an archived directory."""
         linked_bundle_path = path_util.parse_linked_bundle_url(
-            "azfs://storageclwsdev0/bundles/uuid/contents.zip/a/b.txt"
+            "azfs://storageclwsdev0/bundles/uuid/contents.tar.gz/a/b.txt"
         )
         self.assertEqual(linked_bundle_path.storage_type, StorageType.AZURE_BLOB_STORAGE)
         self.assertEqual(
-            linked_bundle_path.bundle_path, "azfs://storageclwsdev0/bundles/uuid/contents.zip"
+            linked_bundle_path.bundle_path, "azfs://storageclwsdev0/bundles/uuid/contents.tar.gz"
         )
-        self.assertEqual(linked_bundle_path.is_zip, True)
-        self.assertEqual(linked_bundle_path.zip_subpath, "a/b.txt")
+        self.assertEqual(linked_bundle_path.is_archive, True)
+        self.assertEqual(linked_bundle_path.archive_subpath, "a/b.txt")
         self.assertEqual(linked_bundle_path.bundle_uuid, "uuid")
 
     def test_non_azure_file(self):
@@ -167,4 +167,4 @@ def test_non_azure_file(self):
         self.assertEqual(
             linked_bundle_path.bundle_path, "/tmp/storageclwsdev0/bundles/uuid/contents.txt"
         )
-        self.assertEqual(linked_bundle_path.is_zip, False)
+        self.assertEqual(linked_bundle_path.is_archive, False)
diff --git a/tests/unit/server/upload_download_test.py b/tests/unit/server/upload_download_test.py
index 8be869153..d1b90337d 100644
--- a/tests/unit/server/upload_download_test.py
+++ b/tests/unit/server/upload_download_test.py
@@ -7,7 +7,6 @@
 import gzip
 import tarfile
 import unittest
-import zipfile
 
 
 class BaseUploadDownloadBundleTest(TestBase):
@@ -33,9 +32,9 @@ def check_file_target_contents(self, target):
         with gzip.GzipFile(fileobj=self.download_manager.stream_file(target, gzipped=True)) as f:
             self.assertEqual(f.read(), b"hello world")
 
-        with self.assertRaises(Exception):
+        with self.assertRaises(tarfile.ReadError):
             with tarfile.open(
-                fileobj=self.download_manager.stream_archived_directory(target)[0], mode='r:gz'
+                fileobj=self.download_manager.stream_tarred_gzipped_directory(target), mode='r:gz'
             ) as f:
                 pass
 
@@ -65,7 +64,10 @@ def check_file_target_contents(self, target):
 
     def check_folder_target_contents(self, target, expected_members=[]):
         """Checks to make sure that the specified folder has the expected contents and can be streamed, etc."""
-        raise NotImplementedError
+        with tarfile.open(
+            fileobj=self.download_manager.stream_tarred_gzipped_directory(target), mode='r:gz'
+        ) as f:
+            self.assertEqual(sorted(f.getnames()), sorted(expected_members))
 
     def test_bundle_single_file(self):
         """Running get_target_info for a bundle with a single file."""
@@ -77,8 +79,8 @@ def test_bundle_single_file(self):
         info = self.download_manager.get_target_info(target, 0)
         self.assertEqual(info["name"], bundle.uuid)
         self.assertEqual(info["size"], 11)
-        # TODO (Ashwin): reenable once permissions work.
-        # self.assertEqual(info["perm"], self.DEFAULT_PERM)
+
+        self.assertEqual(info["perm"], 511)
         self.assertEqual(info["type"], "file")
         self.assertEqual(str(info["resolved_target"]), f"{bundle.uuid}:")
         self.check_file_target_contents(target)
@@ -87,9 +89,25 @@ def test_bundle_folder(self):
         """Running get_target_info for a bundle with a folder, and with subpaths."""
         bundle = self.create_run_bundle()
         self.save_bundle(bundle)
-        self.upload_folder(
-            bundle, [("item.txt", b"hello world"), ("src/item2.txt", b"hello world")]
-        )
+
+        f = BytesIO()
+
+        def writestr(tf, name, contents):
+            tinfo = tarfile.TarInfo(name)
+            tinfo.size = len(contents)
+            tf.addfile(tinfo, BytesIO(contents.encode()))
+
+        def writedir(tf, name):
+            tinfo = tarfile.TarInfo(name)
+            tinfo.type = tarfile.DIRTYPE
+            tf.addfile(tinfo, BytesIO())
+
+        f.seek(0)
+        with tarfile.open(fileobj=f, mode="w:gz") as tf:
+            writestr(tf, "./item.txt", "hello world")
+            writestr(tf, "./src/item2.txt", "hello world")
+        f.seek(0)
+        self.upload_folder(bundle, f)
 
         target = BundleTarget(bundle.uuid, "")
         info = self.download_manager.get_target_info(target, 2)
@@ -107,7 +125,7 @@ def test_bundle_folder(self):
                     {'name': 'item.txt', 'perm': self.DEFAULT_PERM, 'type': 'file'},
                     {
                         'name': 'src',
-                        'perm': 511,
+                        'perm': 493,
                         'type': 'directory',
                         'contents': [
                             {
@@ -155,14 +173,7 @@ def test_bundle_folder(self):
 class RegularBundleStoreTest(BaseUploadDownloadBundleTest):
     """Test uploading and downloading from / to a regular, file-based bundle store."""
 
-    def upload_folder(self, bundle, contents):
-        f = BytesIO()
-        with tarfile.open(fileobj=f, mode='w:gz') as tf:
-            for item in contents:
-                tinfo = tarfile.TarInfo(name=item[0])
-                tinfo.size = len(item[1])
-                tf.addfile(tinfo, BytesIO(item[1]))
-        f.seek(0)
+    def upload_folder(self, bundle, f):
         sources = [["contents.tar.gz", f]]
         self.upload_manager.upload_to_bundle_store(
             bundle,
@@ -190,29 +201,11 @@ def upload_file(self, bundle, contents):
             use_azure_blob_beta=False,
         )
 
-    def check_folder_target_contents(self, target, expected_members=[]):
-        """Checks to make sure that the specified folder has the expected contents and can be streamed, etc."""
-        with tarfile.open(
-            fileobj=self.download_manager.stream_archived_directory(target)[0], mode='r:gz'
-        ) as f:
-            self.assertEqual(sorted(f.getnames()), sorted(expected_members))
-
 
 class AzureBlobBundleStoreTest(BaseUploadDownloadBundleTest, unittest.TestCase):
     """Test uploading and downloading from / to Azure Blob storage."""
 
-    # TODO: permissions are not yet preserved. Remove this DEFAULT_PERM setting when
-    # permissions are properly preserved.
-    DEFAULT_PERM = 511
-
-    def upload_folder(self, bundle, contents):
-        f = BytesIO()
-        with tarfile.open(fileobj=f, mode='w:gz') as tf:
-            for item in contents:
-                tinfo = tarfile.TarInfo(name=item[0])
-                tinfo.size = len(item[1])
-                tf.addfile(tinfo, BytesIO(item[1]))
-        f.seek(0)
+    def upload_folder(self, bundle, f):
         sources = [["contents.tar.gz", f]]
         self.upload_manager.upload_to_bundle_store(
             bundle,
@@ -239,13 +232,3 @@ def upload_file(self, bundle, contents):
             simplify_archives=True,
             use_azure_blob_beta=True,
         )
-
-    def check_folder_target_contents(self, target, expected_members=[]):
-        """Checks to make sure that the specified folder has the expected contents and can be streamed, etc."""
-        with zipfile.ZipFile(
-            self.download_manager.stream_archived_directory(target)[0], mode='r'
-        ) as f:
-            # Remove leading and trailing .'s and /'s from the member list, to account for the slight
-            # differences in .zip and .tar's member list formatting.
-            normalize_list = lambda lst: sorted([m.strip("/.") for m in lst if m.strip("/.")])
-            self.assertEqual(normalize_list(f.namelist()), normalize_list(expected_members))
diff --git a/tests/unit/worker/download_util_test.py b/tests/unit/worker/download_util_test.py
index ba9978d4a..2854dcf36 100644
--- a/tests/unit/worker/download_util_test.py
+++ b/tests/unit/worker/download_util_test.py
@@ -2,9 +2,13 @@
 from codalab.worker.download_util import get_target_info, BundleTarget
 import unittest
 import random
-from zipfile import ZipFile
+import tarfile
 from apache_beam.io.filesystem import CompressionTypes
 from apache_beam.io.filesystems import FileSystems
+from io import BytesIO
+import tempfile
+from codalab.lib.beam.ratarmount import SQLiteIndexedTar
+import shutil
 
 
 class AzureBlobGetTargetInfoTest(unittest.TestCase):
@@ -21,23 +25,55 @@ def test_single_file(self):
     def test_nested_directories(self):
         """Test getting target info of different files within a bundle that consists of nested directories, on Azure Blob Storage."""
         bundle_uuid = str(random.random())
-        bundle_path = f"azfs://storageclwsdev0/bundles/{bundle_uuid}/contents.zip"
-        with FileSystems.create(bundle_path, compression_type=CompressionTypes.UNCOMPRESSED) as f:
-            with ZipFile(f, "w") as zf:
-                # We need to create separate entries for each directories, as a regular
-                # zip file would have.
-                zf.writestr("README.md", "hello world")
-                zf.writestr("src/", "")
-                zf.writestr("src/test.sh", "echo hi")
-                zf.writestr("dist/", "")
-                zf.writestr("dist/a/", "")
-                zf.writestr("dist/a/b/", "")
-                zf.writestr("dist/a/b/test2.sh", "echo two")
+        bundle_path = f"azfs://storageclwsdev0/bundles/{bundle_uuid}/contents.tar.gz"
+
+        def writestr(tf, name, contents):
+            tinfo = tarfile.TarInfo(name)
+            tinfo.size = len(contents)
+            tf.addfile(tinfo, BytesIO(contents.encode()))
+
+        def writedir(tf, name):
+            tinfo = tarfile.TarInfo(name)
+            tinfo.type = tarfile.DIRTYPE
+            tf.addfile(tinfo, BytesIO())
+
+        # TODO: Unify this code with code in UploadManager.upload_to_bundle_store().
+        with FileSystems.create(
+            bundle_path, compression_type=CompressionTypes.UNCOMPRESSED
+        ) as out, tempfile.NamedTemporaryFile(
+            suffix=".tar.gz"
+        ) as tmp_tar_file, tempfile.NamedTemporaryFile(
+            suffix=".sqlite"
+        ) as tmp_index_file:
+            with tarfile.open(name=tmp_tar_file.name, mode="w:gz") as tf:
+                # We need to create separate entries for each directory, as a regular
+                # .tar.gz file would have.
+                writestr(tf, "./README.md", "hello world")
+                writedir(tf, "./src")
+                writestr(tf, "./src/test.sh", "echo hi")
+                writedir(tf, "./dist")
+                writedir(tf, "./dist/a")
+                writedir(tf, "./dist/a/b")
+                writestr(tf, "./dist/a/b/test2.sh", "echo two")
+            shutil.copyfileobj(tmp_tar_file, out)
+            with open(tmp_tar_file.name, "rb") as ttf:
+                SQLiteIndexedTar(
+                    fileObject=ttf,
+                    tarFileName=bundle_uuid,
+                    writeIndex=True,
+                    clearIndexCache=True,
+                    indexFileName=tmp_index_file.name,
+                )
+            with FileSystems.create(
+                bundle_path.replace("/contents.tar.gz", "/index.sqlite"),
+                compression_type=CompressionTypes.UNCOMPRESSED,
+            ) as out_index_file, open(tmp_index_file.name, "rb") as tif:
+                shutil.copyfileobj(tif, out_index_file)
 
         target_info = get_target_info(bundle_path, BundleTarget(bundle_uuid, None), 0)
         target_info.pop("resolved_target")
         self.assertEqual(
-            target_info, {'name': bundle_uuid, 'type': 'directory', 'size': 704, 'perm': 511}
+            target_info, {'name': bundle_uuid, 'type': 'directory', 'size': 249, 'perm': 511}
         )
 
         target_info = get_target_info(bundle_path, BundleTarget(bundle_uuid, None), 1)
@@ -47,12 +83,12 @@ def test_nested_directories(self):
             {
                 'name': bundle_uuid,
                 'type': 'directory',
-                'size': 704,
+                'size': 249,
                 'perm': 511,
                 'contents': [
-                    {'name': 'README.md', 'type': 'file', 'size': 11, 'perm': 511},
-                    {'name': 'src', 'type': 'directory', 'size': 0, 'perm': 511},
-                    {'name': 'dist', 'type': 'directory', 'size': 0, 'perm': 511},
+                    {'name': 'README.md', 'type': 'file', 'size': 11, 'perm': 420},
+                    {'name': 'dist', 'type': 'directory', 'size': 0, 'perm': 420},
+                    {'name': 'src', 'type': 'directory', 'size': 0, 'perm': 420},
                 ],
             },
         )
@@ -60,18 +96,18 @@ def test_nested_directories(self):
         target_info = get_target_info(bundle_path, BundleTarget(bundle_uuid, "README.md"), 1)
         target_info.pop("resolved_target")
         self.assertEqual(
-            target_info, {'name': 'README.md', 'type': 'file', 'size': 11, 'perm': 511}
+            target_info, {'name': 'README.md', 'type': 'file', 'size': 11, 'perm': 420}
         )
 
         target_info = get_target_info(bundle_path, BundleTarget(bundle_uuid, "src/test.sh"), 1)
         target_info.pop("resolved_target")
-        self.assertEqual(target_info, {'name': 'test.sh', 'type': 'file', 'size': 7, 'perm': 511})
+        self.assertEqual(target_info, {'name': 'test.sh', 'type': 'file', 'size': 7, 'perm': 420})
 
         target_info = get_target_info(
             bundle_path, BundleTarget(bundle_uuid, "dist/a/b/test2.sh"), 1
         )
         target_info.pop("resolved_target")
-        self.assertEqual(target_info, {'name': 'test2.sh', 'type': 'file', 'size': 8, 'perm': 511})
+        self.assertEqual(target_info, {'name': 'test2.sh', 'type': 'file', 'size': 8, 'perm': 420})
 
         target_info = get_target_info(bundle_path, BundleTarget(bundle_uuid, "src"), 1)
         target_info.pop("resolved_target")
@@ -81,29 +117,29 @@ def test_nested_directories(self):
                 'name': 'src',
                 'type': 'directory',
                 'size': 0,
-                'perm': 511,
-                'contents': [{'name': 'test.sh', 'type': 'file', 'size': 7, 'perm': 511}],
+                'perm': 420,
+                'contents': [{'name': 'test.sh', 'type': 'file', 'size': 7, 'perm': 420}],
             },
         )
 
         # Return all depths
         target_info = get_target_info(bundle_path, BundleTarget(bundle_uuid, "dist/a"), 999)
         target_info.pop("resolved_target")
-        print(target_info)
+
         self.assertEqual(
             target_info,
             {
                 'name': 'a',
                 'size': 0,
-                'perm': 511,
+                'perm': 420,
                 'type': 'directory',
                 'contents': [
                     {
                         'name': 'b',
                         'size': 0,
-                        'perm': 511,
+                        'perm': 420,
                         'type': 'directory',
-                        'contents': [{'name': 'test2.sh', 'size': 8, 'perm': 511, 'type': 'file'}],
+                        'contents': [{'name': 'test2.sh', 'size': 8, 'perm': 420, 'type': 'file'}],
                     }
                 ],
             },