fix: decompress whole-file cache downloads made by cat

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)

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)

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"