Repository: fsspec/filesystem_spec

Issue:
Missing sub-directories when listing root in `ReferenceFileSystem`
## Minimal reproduction

```python
import fsspec
from fsspec.implementations.reference import ReferenceFileSystem

print(f"fsspec: {fsspec.__version__}")
fs = ReferenceFileSystem({"a": "A", "B/C/b": "B", "B/C/d": "d", "B/_": "_"})
print(fs.ls(""))
```

Produces following output

```
fsspec: 2023.9.2
[{'name': 'a', 'type': 'file', 'size': 1}]
```

notice missing directory `B/` directory at top level. If we move `B/_` ahead of `B/C/b`, error does not occur. Problem is here

https://github.com/fsspec/filesystem_spec/blob/1637a97aa53f04ff0e71cfb44ef54c24d7586eca/fsspec/implementations/reference.py#L989-L999








Discussion:

Patch:
diff --git a/fsspec/implementations/reference.py b/fsspec/implementations/reference.py
index b69e548..cfbd39f 100644
--- a/fsspec/implementations/reference.py
+++ b/fsspec/implementations/reference.py
@@ -985,16 +985,24 @@ class ReferenceFileSystem(AsyncFileSystem):
             elif len(part) == 1:
                 size = None
             else:
-                _, start, size = part
+                _, _, size = part
             par = path.rsplit("/", 1)[0] if "/" in path else ""
             par0 = par
+            subdirs = [par0]
             while par0 and par0 not in self.dircache:
-                # build parent directories
-                self.dircache[par0] = []
-                self.dircache.setdefault(
-                    par0.rsplit("/", 1)[0] if "/" in par0 else "", []
-                ).append({"name": par0, "type": "directory", "size": 0})
+                # collect parent directories
                 par0 = self._parent(par0)
+                subdirs.append(par0)
+
+            subdirs = subdirs[::-1]
+            for parent, child in zip(subdirs, subdirs[1:]):
+                # register newly discovered directories
+                assert child not in self.dircache
+                assert parent in self.dircache
+                self.dircache[parent].append(
+                    {"name": child, "type": "directory", "size": 0}
+                )
+                self.dircache[child] = []
 
             self.dircache[par].append({"name": path, "type": "file", "size": size})
 
diff --git a/fsspec/implementations/webhdfs.py b/fsspec/implementations/webhdfs.py
index 53e0df5..a7426b2 100644
--- a/fsspec/implementations/webhdfs.py
+++ b/fsspec/implementations/webhdfs.py
@@ -149,7 +149,7 @@ class WebHDFS(AbstractFileSystem):
             self.session.auth = HTTPBasicAuth(self.user, self.password)
 
     def _call(self, op, method="get", path=None, data=None, redirect=True, **kwargs):
-        url = self._apply_proxy(self.url + quote(path or ""))
+        url = self._apply_proxy(self.url + quote(path or "", safe="/="))
         args = kwargs.copy()
         args.update(self.pars)
         args["op"] = op.upper()


Test patch:
diff --git a/fsspec/implementations/tests/test_reference.py b/fsspec/implementations/tests/test_reference.py
index c553b35..efdca65 100644
--- a/fsspec/implementations/tests/test_reference.py
+++ b/fsspec/implementations/tests/test_reference.py
@@ -10,7 +10,6 @@ from fsspec.tests.conftest import data, realfile, reset_files, server, win  # no
 
 
 def test_simple(server):  # noqa: F811
-
     refs = {
         "a": b"data",
         "b": (realfile, 0, 5),
@@ -53,6 +52,16 @@ def test_ls(server):  # noqa: F811
     }
 
 
+def test_nested_dirs_ls():
+    # issue #1430
+    refs = {"a": "A", "B/C/b": "B", "B/C/d": "d", "B/_": "_"}
+    fs = fsspec.filesystem("reference", fo=refs)
+    assert len(fs.ls("")) == 2
+    assert set(e["name"] for e in fs.ls("")) == set(["a", "B"])
+    assert len(fs.ls("B")) == 2
+    assert set(e["name"] for e in fs.ls("B")) == set(["B/C", "B/_"])
+
+
 def test_info(server):  # noqa: F811
     refs = {
         "a": b"data",
diff --git a/fsspec/implementations/tests/test_webhdfs.py b/fsspec/implementations/tests/test_webhdfs.py
index f55a6d6..fac34c7 100644
--- a/fsspec/implementations/tests/test_webhdfs.py
+++ b/fsspec/implementations/tests/test_webhdfs.py
@@ -135,3 +135,63 @@ def test_webhdfs_cp_file(hdfs_cluster):
     assert fs.exists(src)
     assert fs.exists(dst)
     assert fs.cat(src) == fs.cat(dst)
+
+
+def test_path_with_equals(hdfs_cluster):
+    fs = WebHDFS(
+        hdfs_cluster, user="testuser", data_proxy={"worker.example.com": "localhost"}
+    )
+    path_with_equals = "/user/testuser/some_table/datestamp=2023-11-11"
+
+    fs.mkdir(path_with_equals)
+
+    result = fs.ls(path_with_equals)
+    assert result is not None
+    assert fs.exists(path_with_equals)
+
+
+def test_error_handling_with_equals_in_path(hdfs_cluster):
+    fs = WebHDFS(hdfs_cluster, user="testuser")
+    invalid_path_with_equals = (
+        "/user/testuser/some_table/invalid_path=datestamp=2023-11-11"
+    )
+
+    with pytest.raises(FileNotFoundError):
+        fs.ls(invalid_path_with_equals)
+
+
+def test_create_and_touch_file_with_equals(hdfs_cluster):
+    fs = WebHDFS(
+        hdfs_cluster,
+        user="testuser",
+        data_proxy={"worker.example.com": "localhost"},
+    )
+    base_path = "/user/testuser/some_table/datestamp=2023-11-11"
+    file_path = f"{base_path}/testfile.txt"
+
+    fs.mkdir(base_path)
+    fs.touch(file_path, "wb")
+    assert fs.exists(file_path)
+
+
+def test_write_read_verify_file_with_equals(hdfs_cluster):
+    fs = WebHDFS(
+        hdfs_cluster,
+        user="testuser",
+        data_proxy={"worker.example.com": "localhost"},
+    )
+    base_path = "/user/testuser/some_table/datestamp=2023-11-11"
+    file_path = f"{base_path}/testfile.txt"
+    content = b"This is some content!"
+
+    fs.mkdir(base_path)
+    with fs.open(file_path, "wb") as f:
+        f.write(content)
+
+    with fs.open(file_path, "rb") as f:
+        assert f.read() == content
+
+    file_info = fs.ls(base_path, detail=True)
+    assert len(file_info) == 1
+    assert file_info[0]["name"] == file_path
+    assert file_info[0]["size"] == len(content)