fs: temporarily disable prefix-based search for fsspec-implementations (#6096)

* fs: temporarily disable prefix-based search for fsspec-implementations

* tests: add the other hash

dvc/fs/fsspec_wrapper.py

@@ -11,6 +11,8 @@ from .base import BaseFileSystem
 
 # pylint: disable=no-member
 class FSSpecWrapper(BaseFileSystem):
+    TRAVERSE_PREFIX_LEN = 2
+
     def __init__(self, **kwargs):
         super().__init__(**kwargs)
         self.fs_args = {"skip_instance_cache": True}

dvc/objects/db/base.py

@@ -399,8 +399,15 @@ class ObjectDB:
         # hashes_exist() (see ssh, local)
         assert self.fs.TRAVERSE_PREFIX_LEN >= 2
 
+        # During the tests, for ensuring that the traverse behavior
+        # is working we turn on this option. It will ensure the
+        # list_hashes_traverse() is called.
+        always_traverse = getattr(self.fs, "_ALWAYS_TRAVERSE", False)
+
         hashes = set(hashes)
-        if len(hashes) == 1 or not self.fs.CAN_TRAVERSE:
+        if (
+            len(hashes) == 1 or not self.fs.CAN_TRAVERSE
+        ) and not always_traverse:
             remote_hashes = self.list_hashes_exists(hashes, jobs, name)
             return remote_hashes
 
@@ -418,7 +425,7 @@ class ObjectDB:
             )
         else:
             traverse_weight = traverse_pages
-        if len(hashes) < traverse_weight:
+        if len(hashes) < traverse_weight and not always_traverse:
             logger.debug(
                 "Large remote ('{}' hashes < '{}' traverse weight), "
                 "using object_exists for remaining hashes".format(

tests/func/test_data_cloud.py

@@ -41,6 +41,14 @@ all_clouds = [
     )
 ]
 
+# Clouds that implement the general methods that can be tested
+# for functional tests that require extensive apis (e.g traversing
+# via walk_files())
+full_clouds = [
+    pytest.lazy_fixture(cloud)
+    for cloud in ["s3", "gs", "azure", "ssh", "hdfs"]
+]
+
 
 @pytest.mark.needs_internet
 @pytest.mark.parametrize("remote", all_clouds, indirect=True)
@@ -548,3 +556,23 @@ def test_pull_partial(tmp_dir, dvc, local_remote):
     stats = dvc.pull(os.path.join("foo", "bar"))
     assert stats["fetched"] == 1
     assert (tmp_dir / "foo").read_text() == {"bar": {"baz": "baz"}}
+
+
+@pytest.mark.parametrize("remote", full_clouds, indirect=True)
+def test_pull_00_prefix(tmp_dir, dvc, remote, monkeypatch):
+    # Related: https://github.com/iterative/dvc/issues/6089
+
+    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 00411460f7c92d2124a67ea0f4cb5f85
+    # bar's md5 checksum is 0000000018e6137ac2caab16074784a6
+    tmp_dir.dvc_gen({"foo": "363", "bar": "jk8ssl"})
+
+    dvc.push()
+    clean(["foo", "bar"], dvc)
+
+    stats = dvc.pull()
+    assert stats["fetched"] == 2
+    assert set(stats["added"]) == {"foo", "bar"}