Repository: fsspec/filesystem_spec

Issue:
bug in `ReferenceFileSystem.cat()` scrambles and omits reference mappings
This is an edge case, happens when `ReferenceFileSystem()`'s references do not have a protocol specified and we pass `.cat()` some missing reference(s) and set  `oe_error='return` or `oe_error='omit`. See the reproducer below

### Reproducer
```python
from fsspec.implementations.reference import ReferenceFileSystem

my_fs = ReferenceFileSystem(
    fo={
        'ref1': 'first reference', 
        'ref2': 'some other reference'}
)

result = my_fs.cat(['nonexistent', 'ref1', 'ref2'], on_error='return')
```

This produces:
```bash
{'nonexistent': b'first reference', 'ref1': b'some other reference'}
```
However it should produce:
```bash
{'nonexistent': FileNotFoundError('nonexistent'), 'ref1': b'first reference', 'ref2': b'some other reference'}
```

### Downstream implications 

This currently breaks Zarr's spec's uninitialized chunk behavior in xarray, resulting in `KeyError` wherever `.cat()` is used to find chunks under the hood. 
For example when evaluating a variable of  `xarray.Dataset()` by calling `ds.my_var.values`
I can confirm this is the case on kerchunked zarrs from NetCDFs. As well as 'native' zarrs that I created via the zarr lib and open
as a key-value store without writing to disk

### Proposed solution
Checking for valid path references in `.cat()` and extending the unit tests to cover this edge case.

PR is inbound with updated test and bugfix.

Discussion:
martindurant: That was fast! :)
arongergely: @martindurant I spent 2 days troubleshooting my kerchunk+xarray code. So this reaction time is not a true indication of effort :laughing: 

Patch:
diff --git a/fsspec/implementations/reference.py b/fsspec/implementations/reference.py
index cfbd39f..ebd54bc 100644
--- a/fsspec/implementations/reference.py
+++ b/fsspec/implementations/reference.py
@@ -798,20 +798,22 @@ class ReferenceFileSystem(AsyncFileSystem):
         out = {}
         for proto, paths in proto_dict.items():
             fs = self.fss[proto]
-            urls, starts, ends = [], [], []
+            urls, starts, ends, valid_paths = [], [], [], []
             for p in paths:
                 # find references or label not-found. Early exit if any not
                 # found and on_error is "raise"
                 try:
                     u, s, e = self._cat_common(p)
-                    urls.append(u)
-                    starts.append(s)
-                    ends.append(e)
                 except FileNotFoundError as err:
                     if on_error == "raise":
                         raise
                     if on_error != "omit":
                         out[p] = err
+                else:
+                    urls.append(u)
+                    starts.append(s)
+                    ends.append(e)
+                    valid_paths.append(p)
 
             # process references into form for merging
             urls2 = []
@@ -819,7 +821,7 @@ class ReferenceFileSystem(AsyncFileSystem):
             ends2 = []
             paths2 = []
             whole_files = set()
-            for u, s, e, p in zip(urls, starts, ends, paths):
+            for u, s, e, p in zip(urls, starts, ends, valid_paths):
                 if isinstance(u, bytes):
                     # data
                     out[p] = u
@@ -831,7 +833,7 @@ class ReferenceFileSystem(AsyncFileSystem):
                     starts2.append(s)
                     ends2.append(e)
                     paths2.append(p)
-            for u, s, e, p in zip(urls, starts, ends, paths):
+            for u, s, e, p in zip(urls, starts, ends, valid_paths):
                 # second run to account for files that are to be loaded whole
                 if s is not None and u not in whole_files:
                     urls2.append(u)
@@ -851,7 +853,7 @@ class ReferenceFileSystem(AsyncFileSystem):
             bytes_out = fs.cat_ranges(new_paths, new_starts, new_ends)
 
             # unbundle from merged bytes - simple approach
-            for u, s, e, p in zip(urls, starts, ends, paths):
+            for u, s, e, p in zip(urls, starts, ends, valid_paths):
                 if p in out:
                     continue  # was bytes, already handled
                 for np, ns, ne, b in zip(new_paths, new_starts, new_ends, bytes_out):


Test patch:
diff --git a/fsspec/implementations/tests/test_reference.py b/fsspec/implementations/tests/test_reference.py
index efdca65..1b92558 100644
--- a/fsspec/implementations/tests/test_reference.py
+++ b/fsspec/implementations/tests/test_reference.py
@@ -449,6 +449,7 @@ def test_merging(m):
 def test_cat_file_ranges(m):
     other = b"other test data"
     m.pipe("/b", other)
+
     fs = fsspec.filesystem(
         "reference",
         fo={
@@ -467,15 +468,26 @@ def test_cat_file_ranges(m):
     assert fs.cat_file("d", 1, -3) == other[4:10][1:-3]
 
 
-def test_cat_missing(m):
+@pytest.mark.parametrize(
+    "fo",
+    [
+        {
+            "c": ["memory://b"],
+            "d": ["memory://unknown", 4, 6],
+        },
+        {
+            "c": ["memory://b"],
+            "d": ["//unknown", 4, 6],
+        },
+    ],
+    ids=["memory protocol", "mixed protocols: memory and unspecified"],
+)
+def test_cat_missing(m, fo):
     other = b"other test data"
     m.pipe("/b", other)
     fs = fsspec.filesystem(
         "reference",
-        fo={
-            "c": ["memory://b"],
-            "d": ["memory://unknown", 4, 6],
-        },
+        fo=fo,
     )
     with pytest.raises(FileNotFoundError):
         fs.cat("notafile")
@@ -508,6 +520,11 @@ def test_cat_missing(m):
     out = mapper.getitems(["c", "d"], on_error="return")
     assert isinstance(out["d"], ReferenceNotReachable)
 
+    out = fs.cat(["notone", "c", "d"], on_error="return")
+    assert isinstance(out["notone"], FileNotFoundError)
+    assert out["c"] == other
+    assert isinstance(out["d"], ReferenceNotReachable)
+
     out = mapper.getitems(["c", "d"], on_error="omit")
     assert list(out) == ["c"]