Repository: fsspec/filesystem_spec
Pull request #1692 (2024-09-26 18:47:01)

Adds engine option to `LazyReferenceMapper`
Defaults to 'fastparquet', but allows for 'pyarrow'

martindurant: Are you aware of arro3, a new minimalist arrow implementation that does include a parquet reader? Maybe would be perfect for this, but perhaps still too immature to use.

norlandrhagen: > Are you aware of arro3, a new minimalist arrow implementation that does include a parquet reader? Maybe would be perfect for this, but perhaps still too immature to use.

I just stumbled on [that](https://github.com/kylebarron/arro3) yesterday! Seems like a great idea, but yeah, it might be a bit too early to use 🤷 



Patch:
diff --git a/fsspec/asyn.py b/fsspec/asyn.py
index f203fa0..de41839 100644
--- a/fsspec/asyn.py
+++ b/fsspec/asyn.py
@@ -344,6 +344,10 @@ class AsyncFileSystem(AbstractFileSystem):
     async def _cp_file(self, path1, path2, **kwargs):
         raise NotImplementedError
 
+    async def _mv_file(self, path1, path2):
+        await self._cp_file(path1, path2)
+        await self._rm_file(path1)
+
     async def _copy(
         self,
         path1,
diff --git a/fsspec/core.py b/fsspec/core.py
index de66be2..e79f2a8 100644
--- a/fsspec/core.py
+++ b/fsspec/core.py
@@ -578,7 +578,7 @@ def expand_paths_if_needed(paths, mode, num, fs, name_function):
     paths = list(paths)
 
     if "w" in mode:  # read mode
-        if sum([1 for p in paths if "*" in p]) > 1:
+        if sum(1 for p in paths if "*" in p) > 1:
             raise ValueError(
                 "When writing data, only one filename mask can be specified."
             )
diff --git a/fsspec/implementations/dirfs.py b/fsspec/implementations/dirfs.py
index 6a28909..0bee864 100644
--- a/fsspec/implementations/dirfs.py
+++ b/fsspec/implementations/dirfs.py
@@ -370,3 +370,15 @@ class DirFileSystem(AsyncFileSystem):
             *args,
             **kwargs,
         )
+
+    async def open_async(
+        self,
+        path,
+        *args,
+        **kwargs,
+    ):
+        return await self.fs.open_async(
+            self._join(path),
+            *args,
+            **kwargs,
+        )
diff --git a/fsspec/implementations/git.py b/fsspec/implementations/git.py
index 7c34d93..fde671b 100644
--- a/fsspec/implementations/git.py
+++ b/fsspec/implementations/git.py
@@ -80,7 +80,7 @@ class GitFileSystem(AbstractFileSystem):
                         {
                             "type": "directory",
                             "name": "/".join([path, obj.name]).lstrip("/"),
-                            "hex": obj.hex,
+                            "hex": str(obj.id),
                             "mode": f"{obj.filemode:o}",
                             "size": 0,
                         }
@@ -90,7 +90,7 @@ class GitFileSystem(AbstractFileSystem):
                         {
                             "type": "file",
                             "name": "/".join([path, obj.name]).lstrip("/"),
-                            "hex": obj.hex,
+                            "hex": str(obj.id),
                             "mode": f"{obj.filemode:o}",
                             "size": obj.size,
                         }
@@ -101,7 +101,7 @@ class GitFileSystem(AbstractFileSystem):
                 {
                     "type": "file",
                     "name": obj.name,
-                    "hex": obj.hex,
+                    "hex": str(obj.id),
                     "mode": f"{obj.filemode:o}",
                     "size": obj.size,
                 }
diff --git a/fsspec/implementations/reference.py b/fsspec/implementations/reference.py
index 6904d7b..d5d3f29 100644
--- a/fsspec/implementations/reference.py
+++ b/fsspec/implementations/reference.py
@@ -7,7 +7,7 @@ import math
 import os
 from itertools import chain
 from functools import lru_cache
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Literal
 
 import fsspec.core
 
@@ -104,7 +104,13 @@ class LazyReferenceMapper(collections.abc.MutableMapping):
         return pd
 
     def __init__(
-        self, root, fs=None, out_root=None, cache_size=128, categorical_threshold=10
+        self,
+        root,
+        fs=None,
+        out_root=None,
+        cache_size=128,
+        categorical_threshold=10,
+        engine: Literal["fastparquet", "pyarrow"] = "fastparquet",
     ):
         """
 
@@ -126,16 +132,25 @@ class LazyReferenceMapper(collections.abc.MutableMapping):
             Encode urls as pandas.Categorical to reduce memory footprint if the ratio
             of the number of unique urls to total number of refs for each variable
             is greater than or equal to this number. (default 10)
+        engine: Literal["fastparquet","pyarrow"]
+            Engine choice for reading parquet files. (default is "fastparquet")
         """
+
         self.root = root
         self.chunk_sizes = {}
         self.out_root = out_root or self.root
         self.cat_thresh = categorical_threshold
+        self.engine = engine
         self.cache_size = cache_size
         self.url = self.root + "/{field}/refs.{record}.parq"
         # TODO: derive fs from `root`
         self.fs = fsspec.filesystem("file") if fs is None else fs
 
+        from importlib.util import find_spec
+
+        if self.engine == "pyarrow" and find_spec("pyarrow") is None:
+            raise ImportError("engine choice `pyarrow` is not installed.")
+
     def __getattr__(self, item):
         if item in ("_items", "record_size", "zmetadata"):
             self.setup()
@@ -158,7 +173,7 @@ class LazyReferenceMapper(collections.abc.MutableMapping):
             """cached parquet file loader"""
             path = self.url.format(field=field, record=record)
             data = io.BytesIO(self.fs.cat_file(path))
-            df = self.pd.read_parquet(data, engine="fastparquet")
+            df = self.pd.read_parquet(data, engine=self.engine)
             refs = {c: df[c].to_numpy() for c in df.columns}
             return refs
 
@@ -463,18 +478,28 @@ class LazyReferenceMapper(collections.abc.MutableMapping):
 
         fn = f"{base_url or self.out_root}/{field}/refs.{record}.parq"
         self.fs.mkdirs(f"{base_url or self.out_root}/{field}", exist_ok=True)
+
+        if self.engine == "pyarrow":
+            df_backend_kwargs = {"write_statistics": False}
+        elif self.engine == "fastparquet":
+            df_backend_kwargs = {
+                "stats": False,
+                "object_encoding": object_encoding,
+                "has_nulls": has_nulls,
+            }
+        else:
+            raise NotImplementedError(f"{self.engine} not supported")
+
         df.to_parquet(
             fn,
-            engine="fastparquet",
+            engine=self.engine,
             storage_options=storage_options
             or getattr(self.fs, "storage_options", None),
             compression="zstd",
             index=False,
-            stats=False,
-            object_encoding=object_encoding,
-            has_nulls=has_nulls,
-            # **kwargs,
+            **df_backend_kwargs,
         )
+
         partition.clear()
         self._items.pop((field, record))
 
@@ -486,6 +511,7 @@ class LazyReferenceMapper(collections.abc.MutableMapping):
         base_url: str
             Location of the output
         """
+
         # write what we have so far and clear sub chunks
         for thing in list(self._items):
             if isinstance(thing, tuple):
diff --git a/fsspec/spec.py b/fsspec/spec.py
index 8229170..e67d280 100644
--- a/fsspec/spec.py
+++ b/fsspec/spec.py
@@ -428,11 +428,9 @@ class AbstractFileSystem(metaclass=_Cached):
         except (FileNotFoundError, OSError) as e:
             if on_error == "raise":
                 raise
-            elif callable(on_error):
+            if callable(on_error):
                 on_error(e)
-            if detail:
-                return path, {}, {}
-            return path, [], []
+            return
 
         for info in listing:
             # each info name must be at least [path]/part , but here


Test patch:
diff --git a/fsspec/implementations/tests/test_dirfs.py b/fsspec/implementations/tests/test_dirfs.py
index 990ab11..46ddd0c 100644
--- a/fsspec/implementations/tests/test_dirfs.py
+++ b/fsspec/implementations/tests/test_dirfs.py
@@ -589,6 +589,16 @@ def test_sign(mocker, dirfs):
     dirfs.fs.sign.assert_called_once_with(f"{PATH}/file", *ARGS, **KWARGS)
 
 
+@pytest.mark.asyncio
+async def test_open_async(mocker, adirfs):
+    adirfs.fs.open_async = mocker.AsyncMock()
+    assert (
+        await adirfs.open_async("file", *ARGS, **KWARGS)
+        == adirfs.fs.open_async.return_value
+    )
+    adirfs.fs.open_async.assert_called_once_with(f"{PATH}/file", *ARGS, **KWARGS)
+
+
 def test_open(mocker, dirfs):
     dirfs.fs.open = mocker.Mock()
     assert dirfs.open("file", *ARGS, **KWARGS) == dirfs.fs.open.return_value
diff --git a/fsspec/implementations/tests/test_http.py b/fsspec/implementations/tests/test_http.py
index 568284e..91b16ce 100644
--- a/fsspec/implementations/tests/test_http.py
+++ b/fsspec/implementations/tests/test_http.py
@@ -16,20 +16,20 @@ from fsspec.tests.conftest import data, reset_files, server, win  # noqa: F401
 
 def test_list(server):
     h = fsspec.filesystem("http")
-    out = h.glob(server + "/index/*")
-    assert out == [server + "/index/realfile"]
+    out = h.glob(server.address + "/index/*")
+    assert out == [server.realfile]
 
 
 def test_list_invalid_args(server):
     with pytest.raises(TypeError):
         h = fsspec.filesystem("http", use_foobar=True)
-        h.glob(server + "/index/*")
+        h.glob(server.address + "/index/*")
 
 
 def test_list_cache(server):
     h = fsspec.filesystem("http", use_listings_cache=True)
-    out = h.glob(server + "/index/*")
-    assert out == [server + "/index/realfile"]
+    out = h.glob(server.address + "/index/*")
+    assert out == [server.realfile]
 
 
 def test_list_cache_with_expiry_time_cached(server):
@@ -40,14 +40,14 @@ def test_list_cache_with_expiry_time_cached(server):
 
     # By querying the filesystem with "use_listings_cache=True",
     # the cache will automatically get populated.
-    out = h.glob(server + "/index/*")
-    assert out == [server + "/index/realfile"]
+    out = h.glob(server.address + "/index/*")
+    assert out == [server.realfile]
 
     # Verify cache content.
     assert len(h.dircache) == 1
 
-    out = h.glob(server + "/index/*")
-    assert out == [server + "/index/realfile"]
+    out = h.glob(server.address + "/index/*")
+    assert out == [server.realfile]
 
 
 def test_list_cache_with_expiry_time_purged(server):
@@ -58,26 +58,26 @@ def test_list_cache_with_expiry_time_purged(server):
 
     # By querying the filesystem with "use_listings_cache=True",
     # the cache will automatically get populated.
-    out = h.glob(server + "/index/*")
-    assert out == [server + "/index/realfile"]
+    out = h.glob(server.address + "/index/*")
+    assert out == [server.realfile]
     assert len(h.dircache) == 1
 
     # Verify cache content.
-    assert server + "/index/" in h.dircache
-    assert len(h.dircache.get(server + "/index/")) == 1
+    assert server.address + "/index/" in h.dircache
+    assert len(h.dircache.get(server.address + "/index/")) == 1
 
     # Wait beyond the TTL / cache expiry time.
     time.sleep(0.31)
 
     # Verify that the cache item should have been purged.
-    cached_items = h.dircache.get(server + "/index/")
+    cached_items = h.dircache.get(server.address + "/index/")
     assert cached_items is None
 
     # Verify that after clearing the item from the cache,
     # it can get populated again.
-    out = h.glob(server + "/index/*")
-    assert out == [server + "/index/realfile"]
-    cached_items = h.dircache.get(server + "/index/")
+    out = h.glob(server.address + "/index/*")
+    assert out == [server.realfile]
+    cached_items = h.dircache.get(server.address + "/index/")
     assert len(cached_items) == 1
 
 
@@ -89,8 +89,8 @@ def test_list_cache_reuse(server):
 
     # By querying the filesystem with "use_listings_cache=True",
     # the cache will automatically get populated.
-    out = h.glob(server + "/index/*")
-    assert out == [server + "/index/realfile"]
+    out = h.glob(server.address + "/index/*")
+    assert out == [server.realfile]
 
     # Verify cache content.
     assert len(h.dircache) == 1
@@ -114,53 +114,53 @@ def test_ls_raises_filenotfound(server):
     h = fsspec.filesystem("http")
 
     with pytest.raises(FileNotFoundError):
-        h.ls(server + "/not-a-key")
+        h.ls(server.address + "/not-a-key")
 
 
 def test_list_cache_with_max_paths(server):
     h = fsspec.filesystem("http", use_listings_cache=True, max_paths=5)
-    out = h.glob(server + "/index/*")
-    assert out == [server + "/index/realfile"]
+    out = h.glob(server.address + "/index/*")
+    assert out == [server.realfile]
 
 
 def test_list_cache_with_skip_instance_cache(server):
     h = fsspec.filesystem("http", use_listings_cache=True, skip_instance_cache=True)
-    out = h.glob(server + "/index/*")
-    assert out == [server + "/index/realfile"]
+    out = h.glob(server.address + "/index/*")
+    assert out == [server.realfile]
 
 
 def test_glob_return_subfolders(server):
     h = fsspec.filesystem("http")
-    out = h.glob(server + "/simple/*")
+    out = h.glob(server.address + "/simple/*")
     assert set(out) == {
-        server + "/simple/dir/",
-        server + "/simple/file",
+        server.address + "/simple/dir/",
+        server.address + "/simple/file",
     }
 
 
 def test_isdir(server):
     h = fsspec.filesystem("http")
-    assert h.isdir(server + "/index/")
-    assert not h.isdir(server + "/index/realfile")
-    assert not h.isdir(server + "doesnotevenexist")
+    assert h.isdir(server.address + "/index/")
+    assert not h.isdir(server.realfile)
+    assert not h.isdir(server.address + "doesnotevenexist")
 
 
 def test_policy_arg(server):
     h = fsspec.filesystem("http", size_policy="get")
-    out = h.glob(server + "/index/*")
-    assert out == [server + "/index/realfile"]
+    out = h.glob(server.address + "/index/*")
+    assert out == [server.realfile]
 
 
 def test_exists(server):
     h = fsspec.filesystem("http")
-    assert not h.exists(server + "/notafile")
+    assert not h.exists(server.address + "/notafile")
     with pytest.raises(FileNotFoundError):
-        h.cat(server + "/notafile")
+        h.cat(server.address + "/notafile")
 
 
 def test_read(server):
     h = fsspec.filesystem("http")
-    out = server + "/index/realfile"
+    out = server.realfile
     with h.open(out, "rb") as f:
         assert f.read() == data
     with h.open(out, "rb", block_size=0) as f:
@@ -174,7 +174,7 @@ def test_file_pickle(server):
 
     # via HTTPFile
     h = fsspec.filesystem("http", headers={"give_length": "true", "head_ok": "true"})
-    out = server + "/index/realfile"
+    out = server.realfile
 
     with fsspec.open(out, headers={"give_length": "true", "head_ok": "true"}) as f:
         pic = pickle.loads(pickle.dumps(f))
@@ -188,7 +188,7 @@ def test_file_pickle(server):
 
     # via HTTPStreamFile
     h = fsspec.filesystem("http")
-    out = server + "/index/realfile"
+    out = server.realfile
     with h.open(out, "rb") as f:
         out = pickle.dumps(f)
         assert f.read() == data
@@ -198,7 +198,7 @@ def test_file_pickle(server):
 
 def test_methods(server):
     h = fsspec.filesystem("http")
-    url = server + "/index/realfile"
+    url = server.realfile
     assert h.exists(url)
     assert h.cat(url) == data
 
@@ -219,7 +219,7 @@ def test_methods(server):
 )
 def test_random_access(server, headers):
     h = fsspec.filesystem("http", headers=headers)
-    url = server + "/index/realfile"
+    url = server.realfile
     with h.open(url, "rb") as f:
         if headers:
             assert f.size == len(data)
@@ -248,7 +248,7 @@ def test_random_access(server, headers):
 )
 def test_no_range_support(server, headers):
     h = fsspec.filesystem("http", headers=headers)
-    url = server + "/index/realfile"
+    url = server.realfile
     with h.open(url, "rb") as f:
         # Random access is not possible if the server doesn't respect Range
         with pytest.raises(ValueError):
@@ -262,7 +262,7 @@ def test_no_range_support(server, headers):
 
 def test_stream_seek(server):
     h = fsspec.filesystem("http")
-    url = server + "/index/realfile"
+    url = server.realfile
     with h.open(url, "rb") as f:
         f.seek(0)  # is OK
         data1 = f.read(5)
@@ -275,11 +275,11 @@ def test_stream_seek(server):
 
 def test_mapper_url(server):
     h = fsspec.filesystem("http")
-    mapper = h.get_mapper(server + "/index/")
+    mapper = h.get_mapper(server.address + "/index/")
     assert mapper.root.startswith("http:")
     assert list(mapper)
 
-    mapper2 = fsspec.get_mapper(server + "/index/")
+    mapper2 = fsspec.get_mapper(server.address + "/index/")
     assert mapper2.root.startswith("http:")
     assert list(mapper) == list(mapper2)
 
@@ -288,7 +288,7 @@ def test_content_length_zero(server):
     h = fsspec.filesystem(
         "http", headers={"give_length": "true", "zero_length": "true"}
     )
-    url = server + "/index/realfile"
+    url = server.realfile
 
     with h.open(url, "rb") as f:
         assert f.read() == data
@@ -298,7 +298,7 @@ def test_content_encoding_gzip(server):
     h = fsspec.filesystem(
         "http", headers={"give_length": "true", "gzip_encoding": "true"}
     )
-    url = server + "/index/realfile"
+    url = server.realfile
 
     with h.open(url, "rb") as f:
         assert isinstance(f, HTTPStreamFile)
@@ -308,7 +308,7 @@ def test_content_encoding_gzip(server):
 
 def test_download(server, tmpdir):
     h = fsspec.filesystem("http", headers={"give_length": "true", "head_ok": "true "})
-    url = server + "/index/realfile"
+    url = server.realfile
     fn = os.path.join(tmpdir, "afile")
     h.get(url, fn)
     assert open(fn, "rb").read() == data
@@ -316,8 +316,8 @@ def test_download(server, tmpdir):
 
 def test_multi_download(server, tmpdir):
     h = fsspec.filesystem("http", headers={"give_length": "true", "head_ok": "true "})
-    urla = server + "/index/realfile"
-    urlb = server + "/index/otherfile"
+    urla = server.realfile
+    urlb = server.address + "/index/otherfile"
     fna = os.path.join(tmpdir, "afile")
     fnb = os.path.join(tmpdir, "bfile")
     h.get([urla, urlb], [fna, fnb])
@@ -327,25 +327,25 @@ def test_multi_download(server, tmpdir):
 
 def test_ls(server):
     h = fsspec.filesystem("http")
-    l = h.ls(server + "/data/20020401/", detail=False)
-    nc = server + "/data/20020401/GRACEDADM_CLSM0125US_7D.A20020401.030.nc4"
+    l = h.ls(server.address + "/data/20020401/", detail=False)
+    nc = server.address + "/data/20020401/GRACEDADM_CLSM0125US_7D.A20020401.030.nc4"
     assert nc in l
     assert len(l) == 11
-    assert all(u["type"] == "file" for u in h.ls(server + "/data/20020401/"))
-    assert h.glob(server + "/data/20020401/*.nc4") == [nc]
+    assert all(u["type"] == "file" for u in h.ls(server.address + "/data/20020401/"))
+    assert h.glob(server.address + "/data/20020401/*.nc4") == [nc]
 
 
 def test_mcat(server):
     h = fsspec.filesystem("http", headers={"give_length": "true", "head_ok": "true "})
-    urla = server + "/index/realfile"
-    urlb = server + "/index/otherfile"
+    urla = server.realfile
+    urlb = server.address + "/index/otherfile"
     out = h.cat([urla, urlb])
     assert out == {urla: data, urlb: data}
 
 
 def test_cat_file_range(server):
     h = fsspec.filesystem("http", headers={"give_length": "true", "head_ok": "true "})
-    urla = server + "/index/realfile"
+    urla = server.realfile
     assert h.cat(urla, start=1, end=10) == data[1:10]
     assert h.cat(urla, start=1) == data[1:]
 
@@ -358,37 +358,37 @@ def test_cat_file_range(server):
 def test_cat_file_range_numpy(server):
     np = pytest.importorskip("numpy")
     h = fsspec.filesystem("http", headers={"give_length": "true", "head_ok": "true "})
-    urla = server + "/index/realfile"
+    urla = server.realfile
     assert h.cat(urla, start=np.int8(1), end=np.int8(10)) == data[1:10]
     out = h.cat_ranges([urla, urla], starts=np.array([1, 5]), ends=np.array([10, 15]))
     assert out == [data[1:10], data[5:15]]
 
 
 def test_mcat_cache(server):
-    urla = server + "/index/realfile"
-    urlb = server + "/index/otherfile"
+    urla = server.realfile
+    urlb = server.address + "/index/otherfile"
     fs = fsspec.filesystem("simplecache", target_protocol="http")
     assert fs.cat([urla, urlb]) == {urla: data, urlb: data}
 
 
 def test_mcat_expand(server):
     h = fsspec.filesystem("http", headers={"give_length": "true", "head_ok": "true "})
-    out = h.cat(server + "/index/*")
-    assert out == {server + "/index/realfile": data}
+    out = h.cat(server.address + "/index/*")
+    assert out == {server.realfile: data}
 
 
 def test_info(server):
     fs = fsspec.filesystem("http", headers={"give_etag": "true", "head_ok": "true"})
-    info = fs.info(server + "/index/realfile")
+    info = fs.info(server.realfile)
     assert info["ETag"] == "xxx"
 
     fs = fsspec.filesystem("http", headers={"give_mimetype": "true"})
-    info = fs.info(server + "/index/realfile")
+    info = fs.info(server.realfile)
     assert info["mimetype"] == "text/html"
 
     fs = fsspec.filesystem("http", headers={"redirect": "true"})
-    info = fs.info(server + "/redirectme")
-    assert info["url"] == server + "/index/realfile"
+    info = fs.info(server.address + "/redirectme")
+    assert info["url"] == server.realfile
 
 
 @pytest.mark.parametrize("method", ["POST", "PUT"])
@@ -400,21 +400,21 @@ def test_put_file(server, tmp_path, method, reset_files):
 
     fs = fsspec.filesystem("http", headers={"head_ok": "true", "give_length": "true"})
     with pytest.raises(FileNotFoundError):
-        fs.info(server + "/hey")
+        fs.info(server.address + "/hey")
 
-    fs.put_file(src_file, server + "/hey", method=method)
-    assert fs.info(server + "/hey")["size"] == len(data)
+    fs.put_file(src_file, server.address + "/hey", method=method)
+    assert fs.info(server.address + "/hey")["size"] == len(data)
 
-    fs.get_file(server + "/hey", dwl_file)
+    fs.get_file(server.address + "/hey", dwl_file)
     assert dwl_file.read_bytes() == data
 
     src_file.write_bytes(b"xxx")
     with open(src_file, "rb") as stream:
-        fs.put_file(stream, server + "/hey_2", method=method)
-    assert fs.cat(server + "/hey_2") == b"xxx"
+        fs.put_file(stream, server.address + "/hey_2", method=method)
+    assert fs.cat(server.address + "/hey_2") == b"xxx"
 
-    fs.put_file(io.BytesIO(b"yyy"), server + "/hey_3", method=method)
-    assert fs.cat(server + "/hey_3") == b"yyy"
+    fs.put_file(io.BytesIO(b"yyy"), server.address + "/hey_3", method=method)
+    assert fs.cat(server.address + "/hey_3") == b"yyy"
 
 
 async def get_aiohttp():
@@ -450,7 +450,7 @@ def test_async_other_thread(server):
     th.start()
     fs = fsspec.filesystem("http", asynchronous=True, loop=loop)
     asyncio.run_coroutine_threadsafe(fs.set_session(), loop=loop).result()
-    url = server + "/index/realfile"
+    url = server.realfile
     cor = fs._cat([url])
     fut = asyncio.run_coroutine_threadsafe(cor, loop=loop)
     assert fut.result() == {url: data}
@@ -463,7 +463,7 @@ def test_async_this_thread(server):
 
         session = await fs.set_session()  # creates client
 
-        url = server + "/index/realfile"
+        url = server.realfile
         with pytest.raises((NotImplementedError, RuntimeError)):
             fs.cat([url])
         out = await fs._cat([url])
@@ -493,7 +493,7 @@ def test_processes(server, method):
     if win and method != "spawn":
         pytest.skip("Windows can only spawn")
     ctx = mp.get_context(method)
-    fn = server + "/index/realfile"
+    fn = server.realfile
     fs = fsspec.filesystem("http")
 
     q = ctx.Queue()
@@ -513,7 +513,7 @@ def test_close(get_client):
 @pytest.mark.asyncio
 async def test_async_file(server):
     fs = fsspec.filesystem("http", asynchronous=True, skip_instance_cache=True)
-    fn = server + "/index/realfile"
+    fn = server.realfile
     of = await fs.open_async(fn)
     async with of as f:
         out1 = await f.read(10)
@@ -525,19 +525,21 @@ async def test_async_file(server):
 
 def test_encoded(server):
     fs = fsspec.filesystem("http", encoded=True)
-    out = fs.cat(server + "/Hello%3A%20G%C3%BCnter", headers={"give_path": "true"})
+    out = fs.cat(
+        server.address + "/Hello%3A%20G%C3%BCnter", headers={"give_path": "true"}
+    )
     assert json.loads(out)["path"] == "/Hello%3A%20G%C3%BCnter"
     with pytest.raises(aiohttp.client_exceptions.ClientError):
-        fs.cat(server + "/Hello: Günter", headers={"give_path": "true"})
+        fs.cat(server.address + "/Hello: Günter", headers={"give_path": "true"})
 
     fs = fsspec.filesystem("http", encoded=False)
-    out = fs.cat(server + "/Hello: Günter", headers={"give_path": "true"})
+    out = fs.cat(server.address + "/Hello: Günter", headers={"give_path": "true"})
     assert json.loads(out)["path"] == "/Hello:%20G%C3%BCnter"
 
 
 def test_with_cache(server):
     fs = fsspec.filesystem("http", headers={"head_ok": "true", "give_length": "true"})
-    fn = server + "/index/realfile"
+    fn = server.realfile
     fs1 = fsspec.filesystem("blockcache", fs=fs)
     with fs1.open(fn, "rb") as f:
         out = f.read()
@@ -549,16 +551,18 @@ async def test_async_expand_path(server):
     fs = fsspec.filesystem("http", asynchronous=True, skip_instance_cache=True)
 
     # maxdepth=1
-    assert await fs._expand_path(server + "/index", recursive=True, maxdepth=1) == [
-        server + "/index",
-        server + "/index/realfile",
+    assert await fs._expand_path(
+        server.address + "/index", recursive=True, maxdepth=1
+    ) == [
+        server.address + "/index",
+        server.address + "/index/realfile",
     ]
 
     # maxdepth=0
     with pytest.raises(ValueError):
-        await fs._expand_path(server + "/index", maxdepth=0)
+        await fs._expand_path(server.address + "/index", maxdepth=0)
     with pytest.raises(ValueError):
-        await fs._expand_path(server + "/index", recursive=True, maxdepth=0)
+        await fs._expand_path(server.address + "/index", recursive=True, maxdepth=0)
 
     await fs._session.close()
 
@@ -568,12 +572,12 @@ async def test_async_walk(server):
     fs = fsspec.filesystem("http", asynchronous=True, skip_instance_cache=True)
 
     # No maxdepth
-    res = [a async for a in fs._walk(server + "/index")]
-    assert res == [(server + "/index", [], ["realfile"])]
+    res = [a async for a in fs._walk(server.address + "/index")]
+    assert res == [(server.address + "/index", [], ["realfile"])]
 
     # maxdepth=0
     with pytest.raises(ValueError):
-        async for a in fs._walk(server + "/index", maxdepth=0):
+        async for a in fs._walk(server.address + "/index", maxdepth=0):
             pass
 
     await fs._session.close()
diff --git a/fsspec/implementations/tests/test_reference.py b/fsspec/implementations/tests/test_reference.py
index 7bfd7d0..d82dc17 100644
--- a/fsspec/implementations/tests/test_reference.py
+++ b/fsspec/implementations/tests/test_reference.py
@@ -10,7 +10,7 @@ from fsspec.implementations.reference import (
     ReferenceFileSystem,
     ReferenceNotReachable,
 )
-from fsspec.tests.conftest import data, realfile, reset_files, server, win  # noqa: F401
+from fsspec.tests.conftest import data, reset_files, server, win  # noqa: F401
 
 
 def test_simple(server):
@@ -20,8 +20,8 @@ def test_simple(server):
 
     refs = {
         "a": b"data",
-        "b": (realfile, 0, 5),
-        "c": (realfile, 1, 5),
+        "b": (server.realfile, 0, 5),
+        "c": (server.realfile, 1, 5),
         "d": b"base64:aGVsbG8=",
         "e": {"key": "value"},
     }
@@ -46,8 +46,8 @@ def test_simple_ver1(server):
         "version": 1,
         "refs": {
             "a": b"data",
-            "b": (realfile, 0, 5),
-            "c": (realfile, 1, 5),
+            "b": (server.realfile, 0, 5),
+            "c": (server.realfile, 1, 5),
             "d": b"base64:aGVsbG8=",
             "e": {"key": "value"},
         },
@@ -76,7 +76,7 @@ def test_target_options(m):
 
 
 def test_ls(server):
-    refs = {"a": b"data", "b": (realfile, 0, 5), "c/d": (realfile, 1, 6)}
+    refs = {"a": b"data", "b": (server.realfile, 0, 5), "c/d": (server.realfile, 1, 6)}
     h = fsspec.filesystem("http")
     fs = fsspec.filesystem("reference", fo=refs, fs=h)
 
@@ -102,9 +102,9 @@ def test_nested_dirs_ls():
 def test_info(server):
     refs = {
         "a": b"data",
-        "b": (realfile, 0, 5),
-        "c/d": (realfile, 1, 6),
-        "e": (realfile,),
+        "b": (server.realfile, 0, 5),
+        "c/d": (server.realfile, 1, 6),
+        "e": (server.realfile,),
     }
     h = fsspec.filesystem("http", headers={"give_length": "true", "head_ok": "true"})
     fs = fsspec.filesystem("reference", fo=refs, fs=h)
@@ -117,9 +117,9 @@ def test_info(server):
 def test_mutable(server, m):
     refs = {
         "a": b"data",
-        "b": (realfile, 0, 5),
-        "c/d": (realfile, 1, 6),
-        "e": (realfile,),
+        "b": (server.realfile, 0, 5),
+        "c/d": (server.realfile, 1, 6),
+        "e": (server.realfile,),
     }
     h = fsspec.filesystem("http", headers={"give_length": "true", "head_ok": "true"})
     fs = fsspec.filesystem("reference", fo=refs, fs=h)
@@ -179,7 +179,7 @@ def test_defaults(server):
         "reference",
         fo=refs,
         target_protocol="http",
-        target=realfile,
+        target=server.realfile,
         remote_protocol="http",
     )
 
@@ -761,18 +761,25 @@ def test_append_parquet(lazy_refs, m):
     assert lazy2["data/1"] == b"Adata"
 
 
-def test_deep_parq(m):
+@pytest.mark.parametrize("engine", ["fastparquet", "pyarrow"])
+def test_deep_parq(m, engine):
     pytest.importorskip("kerchunk")
     zarr = pytest.importorskip("zarr")
+
     lz = fsspec.implementations.reference.LazyReferenceMapper.create(
-        "memory://out.parq", fs=m
+        "memory://out.parq",
+        fs=m,
+        engine=engine,
     )
     g = zarr.open_group(lz, mode="w")
+
     g2 = g.create_group("instant")
     g2.create_dataset(name="one", data=[1, 2, 3])
     lz.flush()
 
-    lz = fsspec.implementations.reference.LazyReferenceMapper("memory://out.parq", fs=m)
+    lz = fsspec.implementations.reference.LazyReferenceMapper(
+        "memory://out.parq", fs=m, engine=engine
+    )
     g = zarr.open_group(lz)
     assert g.instant.one[:].tolist() == [1, 2, 3]
     assert sorted(_["name"] for _ in lz.ls("")) == [".zgroup", ".zmetadata", "instant"]
@@ -780,6 +787,7 @@ def test_deep_parq(m):
         "instant/.zgroup",
         "instant/one",
     ]
+
     assert sorted(_["name"] for _ in lz.ls("instant/one")) == [
         "instant/one/.zarray",
         "instant/one/0",
diff --git a/fsspec/implementations/tests/test_zip.py b/fsspec/implementations/tests/test_zip.py
index 8bf8155..14d0008 100644
--- a/fsspec/implementations/tests/test_zip.py
+++ b/fsspec/implementations/tests/test_zip.py
@@ -164,21 +164,25 @@ def zip_file_fixture(tmp_path):
 def _assert_all_except_context_dependent_variables(result, expected_result):
     for path in expected_result.keys():
         assert result[path]
-        result_without_date_time = result[path].copy()
-        result_without_date_time.pop("date_time")
-        result_without_date_time.pop("_raw_time")
-        result_without_date_time.pop("external_attr")
-        result_without_date_time.pop("create_system")
-        result_without_date_time.pop("_end_offset", None)
-        result_without_date_time.pop("header_offset", None)
-
-        expected_result_without_date_time = expected_result[path].copy()
-        expected_result_without_date_time.pop("date_time")
-        expected_result_without_date_time.pop("_raw_time")
-        expected_result_without_date_time.pop("external_attr")
-        expected_result_without_date_time.pop("create_system")
-        expected_result_without_date_time.pop("_end_offset", None)
-        expected_result_without_date_time.pop("header_offset", None)
+        fields = [
+            "orig_filename",
+            "filename",
+            "compress_type",
+            "comment",
+            "extra",
+            "CRC",
+            "compress_size",
+            "file_size",
+            "name",
+            "size",
+            "type",
+        ]
+
+        result_without_date_time = {k: result[path][k] for k in fields}
+
+        expected_result_without_date_time = {
+            k: expected_result[path][k] for k in fields
+        }
         assert result_without_date_time == expected_result_without_date_time
 
 
diff --git a/fsspec/tests/conftest.py b/fsspec/tests/conftest.py
index 413ed71..31d99fa 100644
--- a/fsspec/tests/conftest.py
+++ b/fsspec/tests/conftest.py
@@ -5,25 +5,34 @@ import os
 import threading
 from collections import ChainMap
 from http.server import BaseHTTPRequestHandler, HTTPServer
+from types import SimpleNamespace
 
 import pytest
 
 requests = pytest.importorskip("requests")
-port = 9898
 data = b"\n".join([b"some test data"] * 1000)
-realfile = f"http://127.0.0.1:{port}/index/realfile"
-index = b'<a href="%s">Link</a>' % realfile.encode()
 listing = open(
     os.path.join(os.path.dirname(__file__), "data", "listing.html"), "rb"
 ).read()
 win = os.name == "nt"
 
 
+def _make_realfile(baseurl):
+    return f"{baseurl}/index/realfile"
+
+
+def _make_index_listing(baseurl):
+    realfile = _make_realfile(baseurl)
+    return b'<a href="%s">Link</a>' % realfile.encode()
+
+
 def _make_listing(*paths):
-    return "\n".join(
-        f'<a href="http://127.0.0.1:{port}{f}">Link_{i}</a>'
-        for i, f in enumerate(paths)
-    ).encode()
+    def _make_listing_port(baseurl):
+        return "\n".join(
+            f'<a href="{baseurl}{f}">Link_{i}</a>' for i, f in enumerate(paths)
+        ).encode()
+
+    return _make_listing_port
 
 
 @pytest.fixture
@@ -39,7 +48,7 @@ class HTTPTestHandler(BaseHTTPRequestHandler):
     static_files = {
         "/index/realfile": data,
         "/index/otherfile": data,
-        "/index": index,
+        "/index": _make_index_listing,
         "/data/20020401": listing,
         "/simple/": _make_listing("/simple/file", "/simple/dir/"),
         "/simple/file": data,
@@ -64,14 +73,17 @@ class HTTPTestHandler(BaseHTTPRequestHandler):
             self.wfile.write(data)
 
     def do_GET(self):
+        baseurl = f"http://{self.server.server_name}:{self.server.server_port}"
         file_path = self.path
         if file_path.endswith("/") and file_path.rstrip("/") in self.files:
             file_path = file_path.rstrip("/")
         file_data = self.files.get(file_path)
+        if callable(file_data):
+            file_data = file_data(baseurl)
         if "give_path" in self.headers:
             return self._respond(200, data=json.dumps({"path": self.path}).encode())
         if "redirect" in self.headers and file_path != "/index/realfile":
-            new_url = f"http://127.0.0.1:{port}/index/realfile"
+            new_url = _make_realfile(baseurl)
             return self._respond(301, {"Location": new_url})
         if file_data is None:
             return self._respond(404)
@@ -169,13 +181,13 @@ class HTTPTestHandler(BaseHTTPRequestHandler):
 
 @contextlib.contextmanager
 def serve():
-    server_address = ("", port)
+    server_address = ("", 0)
     httpd = HTTPServer(server_address, HTTPTestHandler)
     th = threading.Thread(target=httpd.serve_forever)
     th.daemon = True
     th.start()
     try:
-        yield f"http://127.0.0.1:{port}"
+        yield f"http://{httpd.server_name}:{httpd.server_port}"
     finally:
         httpd.socket.close()
         httpd.shutdown()
@@ -185,4 +197,5 @@ def serve():
 @pytest.fixture(scope="module")
 def server():
     with serve() as s:
-        yield s
+        server = SimpleNamespace(address=s, realfile=_make_realfile(s))
+        yield server
diff --git a/fsspec/tests/test_caches.py b/fsspec/tests/test_caches.py
index 5bde713..6176d80 100644
--- a/fsspec/tests/test_caches.py
+++ b/fsspec/tests/test_caches.py
@@ -215,7 +215,7 @@ def test_background(server, monkeypatch):
     import fsspec
 
     head = {"head_ok": "true", "head_give_length": "true"}
-    urla = server + "/index/realfile"
+    urla = server.realfile
     h = fsspec.filesystem("http", headers=head)
     thread_ids = {threading.current_thread().ident}
     f = h.open(urla, block_size=5, cache_type="background")
diff --git a/fsspec/tests/test_generic.py b/fsspec/tests/test_generic.py
index fc4c8bf..bd9745f 100644
--- a/fsspec/tests/test_generic.py
+++ b/fsspec/tests/test_generic.py
@@ -7,10 +7,10 @@ from fsspec.tests.conftest import data, server  # noqa: F401
 def test_remote_async_ops(server):
     fsspec.filesystem("http", headers={"give_length": "true", "head_ok": "true"})
     fs = fsspec.filesystem("generic", default_method="current")
-    out = fs.info(server + "/index/realfile")
+    out = fs.info(server.realfile)
     assert out["size"] == len(data)
     assert out["type"] == "file"
-    assert fs.isfile(server + "/index/realfile")  # this method from superclass
+    assert fs.isfile(server.realfile)  # this method from superclass
 
 
 def test_touch_rm(m):
@@ -29,7 +29,7 @@ def test_touch_rm(m):
 def test_cp_async_to_sync(server, m):
     fsspec.filesystem("http", headers={"give_length": "true", "head_ok": "true"})
     fs = fsspec.filesystem("generic", default_method="current")
-    fs.cp([server + "/index/realfile"], ["memory://realfile"])
+    fs.cp([server.realfile], ["memory://realfile"])
     assert m.cat("realfile") == data
 
     fs.rm("memory://realfile")
@@ -45,7 +45,7 @@ def test_pipe_cat_sync(m):
 def test_cat_async(server):
     fsspec.filesystem("http", headers={"give_length": "true", "head_ok": "true"})
     fs = fsspec.filesystem("generic", default_method="current")
-    assert fs.cat(server + "/index/realfile") == data
+    assert fs.cat(server.realfile) == data
 
 
 def test_rsync(tmpdir, m):