fsspec: loosen the prefix check to cover both None and False (#6246)

* fsspec: loosen the prefix check to cover both None and False

* gs: disable prefix-based search

dvc/fs/fsspec_wrapper.py

@@ -173,7 +173,7 @@ class ObjectFSWrapper(FSSpecWrapper):
         )
 
     def find(self, path_info, detail=False, prefix=None):
-        if prefix is not None:
+        if prefix:
             path = self._with_bucket(path_info.parent)
             files = self.fs.find(
                 path, detail=detail, prefix=path_info.parts[-1]

dvc/fs/gs.py

@@ -16,6 +16,7 @@ class GSFileSystem(ObjectFSWrapper):
     REQUIRES = {"gcsfs": "gcsfs"}
     PARAM_CHECKSUM = "etag"
     DETAIL_FIELDS = frozenset(("etag", "size"))
+    TRAVERSE_PREFIX_LEN = 2
 
     def _prepare_credentials(self, **config):
         login_info = {"consistency": None}

tests/func/test_data_cloud.py

@@ -588,3 +588,23 @@ def test_pull_00_prefix(tmp_dir, dvc, remote, monkeypatch):
     stats = dvc.pull()
     assert stats["fetched"] == 2
     assert set(stats["added"]) == {"foo", "bar"}
+
+
+@pytest.mark.parametrize("remote", full_clouds, indirect=True)
+def test_pull_no_00_prefix(tmp_dir, dvc, remote, monkeypatch):
+    # Related: https://github.com/iterative/dvc/issues/6244
+
+    fs_type = type(dvc.cloud.get_remote("upstream").fs)
+    monkeypatch.setattr(fs_type, "_ALWAYS_TRAVERSE", True, raising=False)
+    monkeypatch.setattr(fs_type, "LIST_OBJECT_PAGE_SIZE", 256, raising=False)
+
+    # foo's md5 checksum is 14ffd92a6cbf5f2f657067df0d5881a6
+    # bar's md5 checksum is 64020400f00960c0ef04052547b134b3
+    tmp_dir.dvc_gen({"foo": "dvc", "bar": "cml"})
+
+    dvc.push()
+    clean(["foo", "bar"], dvc)
+
+    stats = dvc.pull()
+    assert stats["fetched"] == 2
+    assert set(stats["added"]) == {"foo", "bar"}