diff --git a/docs/source/changelog.rst b/docs/source/changelog.rst index 5cda044eb..4983608a4 100644 --- a/docs/source/changelog.rst +++ b/docs/source/changelog.rst @@ -12,6 +12,9 @@ Enhancements Fixes +- Decompress ``filecache`` and ``simplecache`` downloads made by ``cat`` so + subsequent cached reads use the same contents as ``open``. + - Reject mixed source or destination backends in ``GenericFileSystem.copy``(#2302). - Return the current position for accepted no-op seeks on HTTP streams (#2306) diff --git a/fsspec/implementations/cached.py b/fsspec/implementations/cached.py index 5b0cd5792..ee0e26296 100644 --- a/fsspec/implementations/cached.py +++ b/fsspec/implementations/cached.py @@ -758,9 +758,13 @@ def cat( try: detail = self._check_file(p) if not detail: - fn = self._make_local_details(p) - getpaths.append(p) - storepaths.append(fn) + if self.compression: + self._get_cached_file_before_open(p) + fn = os.path.join(self.storage[-1], self._mapper(p)) + else: + fn = self._make_local_details(p) + getpaths.append(p) + storepaths.append(fn) else: detail, fn = detail if isinstance(detail, tuple) else (None, detail) fns.append(fn) diff --git a/fsspec/implementations/tests/test_cached.py b/fsspec/implementations/tests/test_cached.py index a368d06bb..97fe1e856 100644 --- a/fsspec/implementations/tests/test_cached.py +++ b/fsspec/implementations/tests/test_cached.py @@ -992,6 +992,76 @@ def test_with_compression(impl, compression): assert open(os.path.join(cachedir, "data"), "rb").read() == data +@pytest.mark.parametrize("protocol", ["simplecache", "filecache"]) +@pytest.mark.parametrize("compression", [None, "gzip", "bz2", "infer"]) +@pytest.mark.parametrize("selection", ["single", "list", "glob"]) +@pytest.mark.parametrize("warm", [False, True]) +def test_cat_with_compression(tmp_path, m, protocol, compression, selection, warm): + codec = "bz2" if compression == "bz2" else "gzip" + suffix = "bz2" if codec == "bz2" else "gz" + paths = [f"/cat-compression/{name}.{suffix}" for name in ("one", "two")] + expected = {} + for path in paths: + data = f"uncompressed contents of {path}".encode() + with m.open(path, "wb") as raw: + with compr[codec](raw, mode="wb") as compressed: + compressed.write(data) + expected[path] = data if compression else m.cat_file(path) + + options = { + "fs": m, + "cache_storage": str(tmp_path), + "compression": compression, + "skip_instance_cache": True, + } + fs = fsspec.filesystem(protocol, **options) + if warm: + for path in paths: + with fs.open(path, "rb") as f: + assert f.read() == expected[path] + + if selection == "single": + assert fs.cat(paths[0]) == expected[paths[0]] + paths = paths[:1] + else: + argument = paths if selection == "list" else f"/cat-compression/*.{suffix}" + assert fs.cat(argument) == expected + + # cat must populate the same cache representation as open, including after + # the cache is loaded by another instance. + reloaded = fsspec.filesystem(protocol, **options) + for path in paths: + for cached_fs in (fs, reloaded): + with cached_fs.open(path, "rb") as f: + assert f.read() == expected[path] + cached = fs._check_file(path) + fn = cached[1] if isinstance(cached, tuple) else cached + assert pathlib.Path(fn).read_bytes() == expected[path] + + +@pytest.mark.parametrize("protocol", ["simplecache", "filecache"]) +@pytest.mark.parametrize("compression", ["gzip", "bz2", "infer"]) +def test_cat_does_not_poison_compressed_open(tmp_path, m, protocol, compression): + codec = "bz2" if compression == "bz2" else "gzip" + path = "/cat-poison.bz2" if codec == "bz2" else "/cat-poison.gz" + data = b"decompressed cache contents" + with m.open(path, "wb") as raw: + with compr[codec](raw, mode="wb") as compressed: + compressed.write(data) + + options = {"fs": m, "compression": compression, "skip_instance_cache": True} + control = fsspec.filesystem( + protocol, cache_storage=str(tmp_path / "open"), **options + ) + with control.open(path, "rb") as f: + assert f.read() == data + + fs = fsspec.filesystem(protocol, cache_storage=str(tmp_path / "cat"), **options) + fs.cat(path) + with fs.open(path, "rb") as f: + assert f.read() == data + + @pytest.mark.parametrize("protocol", ["simplecache", "filecache"]) def test_again(protocol): fn = "memory://afile"