dvcfilesystem.get: prefer downloading from repo.fs before dvc.fs (#10543)

Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>

.github/workflows/benchmarks.yaml

@@ -5,6 +5,10 @@ concurrency:
   group: ${{ github.workflow }}-${{ github.head_ref || github.run_id }}
   cancel-in-progress: true
 
+env:
+  FORCE_COLOR: "1"
+  PY_COLORS: "1"
+
 jobs:
   bench:
     name: run benchmarks
@@ -13,6 +17,24 @@ jobs:
       - uses: actions/setup-python@v5
         with:
             python-version: "3.12"
-      - uses: iterative/dvc-bench@main
+
+      - uses: hynek/setup-cached-uv@v2
+      - uses: actions/checkout@v4
         with:
-            pytest_options: "-k 'test_init or test_help'"
+          ref: ${{ github.event.pull_request.base.sha }}
+          fetch-depth: 0
+
+      - run: uv pip install '.[tests]' --system
+      - name: run benchmarks on base branch
+        run: pytest --benchmark-autosave dvc/testing/benchmarks/ -k 'test_init or test_help'
+
+      - uses: actions/checkout@v4
+        with:
+          fetch-depth: 0
+          clean: false
+      - run: uv pip install '.[tests]' --system
+      - name: run benchmarks for PR
+        run: >
+          pytest --benchmark-compare --benchmark-compare-fail=min:5%
+          --benchmark-group-by name
+          dvc/testing/benchmarks/ -k 'test_init or test_help'

.github/workflows/tests.yaml

@@ -99,7 +99,7 @@ jobs:
       env:
         PYTHONUTF8: 1
       run: >
-        pytest ${{ matrix.pytestargs }} -n=logical --timeout=300 --durations=0
+        pytest ${{ matrix.pytestargs }} -n=logical --dist=worksteal --timeout=300 --durations=0
         --cov --cov-report=xml --cov-report=term  --durations-path=./.github/.test_durations
     - name: upload coverage report
       uses: codecov/codecov-action@v3

.pre-commit-config.yaml

@@ -22,7 +22,7 @@ repos:
       - id: sort-simple-yaml
       - id: trailing-whitespace
   - repo: https://github.com/astral-sh/ruff-pre-commit
-    rev: 'v0.6.0'
+    rev: 'v0.6.2'
     hooks:
       - id: ruff
         args: [--fix, --exit-non-zero-on-fix]

dvc/dependency/base.py

@@ -45,7 +45,7 @@ class Dependency(Output):
             self.fs_path = self.fs.version_path(self.fs_path, self.meta.version_id)
 
     def download(self, to, jobs=None):
-        fs_download(self.fs, self.fs_path, to.fs_path, jobs=jobs)
+        return fs_download(self.fs, self.fs_path, to.fs_path, jobs=jobs)
 
     def save(self):
         super().save()

dvc/dependency/repo.py

@@ -3,7 +3,6 @@ from typing import TYPE_CHECKING, Any, ClassVar, Optional, Union
 
 import voluptuous as vol
 
-from dvc.prompt import confirm
 from dvc.utils import as_posix
 
 from .base import Dependency
@@ -12,6 +11,7 @@ if TYPE_CHECKING:
     from dvc.fs import DVCFileSystem
     from dvc.output import Output
     from dvc.stage import Stage
+    from dvc_data.hashfile.hash_info import HashInfo
 
 
 class RepoDependency(Dependency):
@@ -94,29 +94,25 @@ class RepoDependency(Dependency):
         }
 
     def download(self, to: "Output", jobs: Optional[int] = None):
-        from dvc_data.hashfile.build import build
-        from dvc_data.hashfile.checkout import CheckoutError, checkout
+        from dvc.fs import LocalFileSystem
 
-        try:
-            repo = self._make_fs(locked=True).repo
+        files = super().download(to=to, jobs=jobs)
+        if not isinstance(to.fs, LocalFileSystem):
+            return
 
-            _, _, obj = build(
-                repo.cache.local,
-                self.fs_path,
-                repo.dvcfs,
-                repo.cache.local.fs.PARAM_CHECKSUM,
-            )
-            checkout(
-                to.fs_path,
-                to.fs,
-                obj,
-                self.repo.cache.local,
-                ignore=None,
-                state=self.repo.state,
-                prompt=confirm,
-            )
-        except (CheckoutError, FileNotFoundError):
-            super().download(to=to, jobs=jobs)
+        hashes: list[tuple[str, HashInfo, dict[str, Any]]] = []
+        for src_path, dest_path, *rest in files:
+            try:
+                info = rest[0] if rest else self.fs.info(src_path)
+                hash_info = info["dvc_info"]["entry"].hash_info
+                dest_info = to.fs.info(dest_path)
+            except (KeyError, AttributeError):
+                # If no hash info found, just keep going and output will be hashed later
+                continue
+            if hash_info:
+                hashes.append((dest_path, hash_info, dest_info))
+        cache = to.cache if to.use_cache else to.local_cache
+        cache.state.save_many(hashes, to.fs)
 
     def update(self, rev: Optional[str] = None):
         if rev:

dvc/fs/__init__.py

@@ -1,5 +1,5 @@
 import glob
-from typing import Optional
+from typing import Optional, Union
 from urllib.parse import urlparse
 
 from dvc.config import ConfigError as RepoConfigError
@@ -47,28 +47,12 @@ known_implementations.update(
 
 def download(
     fs: "FileSystem", fs_path: str, to: str, jobs: Optional[int] = None
-) -> int:
+) -> list[Union[tuple[str, str], tuple[str, str, dict]]]:
     from dvc.scm import lfs_prefetch
 
     from .callbacks import TqdmCallback
 
     with TqdmCallback(desc=f"Downloading {fs.name(fs_path)}", unit="files") as cb:
-        # NOTE: We use dvc-objects generic.copy over fs.get since it makes file
-        # download atomic and avoids fsspec glob/regex path expansion.
-        if fs.isdir(fs_path):
-            from_infos = [
-                path for path in fs.find(fs_path) if not path.endswith(fs.flavour.sep)
-            ]
-            if not from_infos:
-                localfs.makedirs(to, exist_ok=True)
-                return 0
-            to_infos = [
-                localfs.join(to, *fs.relparts(info, fs_path)) for info in from_infos
-            ]
-        else:
-            from_infos = [fs_path]
-            to_infos = [to]
-
         if isinstance(fs, DVCFileSystem):
             lfs_prefetch(
                 fs,
@@ -78,10 +62,29 @@ def download(
                     else glob.escape(fs_path)
                 ],
             )
+            if not glob.has_magic(fs_path):
+                return fs._get(fs_path, to, batch_size=jobs, callback=cb)
+
+        # NOTE: We use dvc-objects generic.copy over fs.get since it makes file
+        # download atomic and avoids fsspec glob/regex path expansion.
+        if fs.isdir(fs_path):
+            from_infos = [
+                path for path in fs.find(fs_path) if not path.endswith(fs.flavour.sep)
+            ]
+            if not from_infos:
+                localfs.makedirs(to, exist_ok=True)
+                return []
+            to_infos = [
+                localfs.join(to, *fs.relparts(info, fs_path)) for info in from_infos
+            ]
+        else:
+            from_infos = [fs_path]
+            to_infos = [to]
+
         cb.set_size(len(from_infos))
         jobs = jobs or fs.jobs
         generic.copy(fs, from_infos, localfs, to_infos, callback=cb, batch_size=jobs)
-        return len(to_infos)
+        return list(zip(from_infos, to_infos))
 
 
 def parse_external_url(url, fs_config=None, config=None):

dvc/fs/dvc.py

@@ -6,13 +6,15 @@ import posixpath
 import threading
 from collections import deque
 from contextlib import ExitStack, suppress
+from glob import has_magic
 from typing import TYPE_CHECKING, Any, Callable, Optional, Union
 
-from fsspec.spec import AbstractFileSystem
+from fsspec.spec import DEFAULT_CALLBACK, AbstractFileSystem
 from funcy import wrap_with
 
 from dvc.log import logger
-from dvc_objects.fs.base import FileSystem
+from dvc.utils.threadpool import ThreadPoolExecutor
+from dvc_objects.fs.base import AnyFSPath, FileSystem
 
 from .data import DataFileSystem
 
@@ -20,6 +22,8 @@ if TYPE_CHECKING:
     from dvc.repo import Repo
     from dvc.types import DictStrAny, StrPath
 
+    from .callbacks import Callback
+
 logger = logger.getChild(__name__)
 
 RepoFactory = Union[Callable[..., "Repo"], type["Repo"]]
@@ -474,9 +478,114 @@ class _DVCFileSystem(AbstractFileSystem):
         info["name"] = path
         return info
 
+    def get(
+        self,
+        rpath,
+        lpath,
+        recursive=False,
+        callback=DEFAULT_CALLBACK,
+        maxdepth=None,
+        batch_size=None,
+        **kwargs,
+    ):
+        self._get(
+            rpath,
+            lpath,
+            recursive=recursive,
+            callback=callback,
+            maxdepth=maxdepth,
+            batch_size=batch_size,
+            **kwargs,
+        )
+
+    def _get(  # noqa: C901
+        self,
+        rpath,
+        lpath,
+        recursive=False,
+        callback=DEFAULT_CALLBACK,
+        maxdepth=None,
+        batch_size=None,
+        **kwargs,
+    ) -> list[Union[tuple[str, str], tuple[str, str, dict]]]:
+        if (
+            isinstance(rpath, list)
+            or isinstance(lpath, list)
+            or has_magic(rpath)
+            or not self.exists(rpath)
+            or not recursive
+        ):
+            super().get(
+                rpath,
+                lpath,
+                recursive=recursive,
+                callback=callback,
+                maxdepth=maxdepth,
+                **kwargs,
+            )
+            return []
+
+        if os.path.isdir(lpath) or lpath.endswith(os.path.sep):
+            lpath = self.join(lpath, os.path.basename(rpath))
+
+        if self.isfile(rpath):
+            with callback.branched(rpath, lpath) as child:
+                self.get_file(rpath, lpath, callback=child, **kwargs)
+                return [(rpath, lpath)]
+
+        _files = []
+        _dirs: list[str] = []
+        for root, dirs, files in self.walk(rpath, maxdepth=maxdepth, detail=True):
+            if files:
+                callback.set_size((callback.size or 0) + len(files))
+
+            parts = self.relparts(root, rpath)
+            if parts in ((os.curdir,), ("",)):
+                parts = ()
+            dest_root = os.path.join(lpath, *parts)
+            if not maxdepth or len(parts) < maxdepth - 1:
+                _dirs.extend(f"{dest_root}{os.path.sep}{d}" for d in dirs)
+
+            key = self._get_key_from_relative(root)
+            _, dvc_fs, _ = self._get_subrepo_info(key)
+
+            for name, info in files.items():
+                src_path = f"{root}{self.sep}{name}"
+                dest_path = f"{dest_root}{os.path.sep}{name}"
+                _files.append((dvc_fs, src_path, dest_path, info))
+
+        os.makedirs(lpath, exist_ok=True)
+        for d in _dirs:
+            os.mkdir(d)
+
+        repo_fs = self.repo.fs
+
+        def _get_file(arg):
+            dvc_fs, src, dest, info = arg
+            dvc_info = info.get("dvc_info")
+            fs_info = info.get("fs_info")
+            if dvc_fs and dvc_info and not fs_info:
+                dvc_path = dvc_info["name"]
+                dvc_fs.get_file(
+                    dvc_path, dest, callback=callback, info=dvc_info, **kwargs
+                )
+            else:
+                fs_path = fs_info["name"]
+                repo_fs.get_file(fs_path, dest, callback=callback, **kwargs)
+            return src, dest, info
+
+        with ThreadPoolExecutor(max_workers=batch_size) as executor:
+            return list(executor.imap_unordered(_get_file, _files))
+
     def get_file(self, rpath, lpath, **kwargs):
+        dvc_info = kwargs.pop("info", {}).pop("dvc_info", None)
         key = self._get_key_from_relative(rpath)
         fs_path = self._from_key(key)
+        dirpath = os.path.dirname(lpath)
+        if dirpath:
+            # makedirs raises error if the string is empty
+            os.makedirs(dirpath, exist_ok=True)
+
         try:
             return self.repo.fs.get_file(fs_path, lpath, **kwargs)
         except FileNotFoundError:
@@ -485,7 +594,7 @@ class _DVCFileSystem(AbstractFileSystem):
                 raise
 
         dvc_path = _get_dvc_path(dvc_fs, subkey)
-        return dvc_fs.get_file(dvc_path, lpath, **kwargs)
+        return dvc_fs.get_file(dvc_path, lpath, info=dvc_info, **kwargs)
 
     def du(self, path, total=True, maxdepth=None, withdirs=False, **kwargs):
         if maxdepth is not None:
@@ -553,6 +662,45 @@ class DVCFileSystem(FileSystem):
     def getcwd(self):
         return self.fs.getcwd()
 
+    def _get(
+        self,
+        from_info: Union[AnyFSPath, list[AnyFSPath]],
+        to_info: Union[AnyFSPath, list[AnyFSPath]],
+        callback: "Callback" = DEFAULT_CALLBACK,
+        recursive: bool = False,
+        batch_size: Optional[int] = None,
+        **kwargs,
+    ) -> list[Union[tuple[str, str], tuple[str, str, dict]]]:
+        # FileSystem.get is non-recursive by default if arguments are lists
+        # otherwise, it's recursive.
+        recursive = not (isinstance(from_info, list) and isinstance(to_info, list))
+        return self.fs._get(
+            from_info,
+            to_info,
+            callback=callback,
+            recursive=recursive,
+            batch_size=batch_size,
+            **kwargs,
+        )
+
+    def get(
+        self,
+        from_info: Union[AnyFSPath, list[AnyFSPath]],
+        to_info: Union[AnyFSPath, list[AnyFSPath]],
+        callback: "Callback" = DEFAULT_CALLBACK,
+        recursive: bool = False,
+        batch_size: Optional[int] = None,
+        **kwargs,
+    ) -> None:
+        self._get(
+            from_info,
+            to_info,
+            callback=callback,
+            batch_size=batch_size,
+            recursive=recursive,
+            **kwargs,
+        )
+
     @property
     def fsid(self) -> str:
         return self.fs.fsid

dvc/repo/artifacts.py

@@ -220,7 +220,7 @@ class Artifacts:
 
             out = resolve_output(path, out, force=force)
             fs = self.repo.dvcfs
-            count = fs_download(fs, path, os.path.abspath(out), jobs=jobs)
+            count = len(fs_download(fs, path, os.path.abspath(out), jobs=jobs))
         return count, out
 
     @staticmethod

dvc/testing/benchmarks/cli/commands/test_data_status.py

@@ -1,9 +1,5 @@
 from shutil import rmtree
 
-import pytest
-
-pytestmark = pytest.mark.requires(minversion=(2, 15, 0), reason="new command")
-
 
 def test_data_status(bench_dvc, tmp_dir, scm, dvc, make_dataset):
     args = ("data", "status")

dvc/testing/benchmarks/cli/commands/test_exp_show.py

@@ -1,7 +1,3 @@
-import pytest
-
-
-@pytest.mark.requires(minversion=(2, 28, 0))
 def test_exp_show(make_project, monkeypatch, bench_dvc, dvc_bin):
     url = "https://github.com/iterative/example-get-started"
     rev = "main"

dvc/testing/benchmarks/cli/commands/test_import.py

@@ -2,6 +2,12 @@ import pytest
 
 
 @pytest.mark.flaky(reruns=3)
+@pytest.mark.requires(
+    "!=3.53.*,!=3.54.0",
+    reason="Takes 10 mins to run. Regression in 3.53.0, fixed in 3.54.1",
+)
+# Introduced in https://github.com/iterative/dvc/pull/10388.
+# Fixed in https://github.com/iterative/dvc/pull/10531.
 def test_import(bench_dvc, tmp_dir, scm, dvc, make_dataset, remote):
     dataset = make_dataset(
         cache=False, files=False, dvcfile=True, commit=True, remote=True

dvc/testing/benchmarks/cli/commands/test_plots.py

@@ -1,10 +1,7 @@
-import pytest
-
 from dvc.repo import Repo
 from dvc.testing.benchmarks.fixtures import _pull
 
 
-@pytest.mark.requires(minversion=(2, 34, 0), reason="top-level plots not supported")
 def test_plots(project, bench_dvc):
     with Repo() as dvc:
         _pull(dvc)

dvc/testing/benchmarks/cli/stories/use_cases/test_sharing.py

@@ -1,9 +1,8 @@
 import shutil
 
 
-def test_sharing(bench_dvc, tmp_dir, dvc, dataset, remote):
-    bench_dvc("add", dataset)
-    bench_dvc("add", dataset, name="noop")
+def test_sharing(bench_dvc, tmp_dir, dvc, make_dataset, remote):
+    dataset = make_dataset(cache=True, dvcfile=True)
 
     bench_dvc("push")
     bench_dvc("push", name="noop")
@@ -13,6 +12,3 @@ def test_sharing(bench_dvc, tmp_dir, dvc, dataset, remote):
 
     bench_dvc("fetch")
     bench_dvc("fetch", name="noop")
-
-    bench_dvc("checkout")
-    bench_dvc("checkout", name="noop")

dvc/testing/benchmarks/conftest.py

@@ -1 +1,3 @@
 from .fixtures import *  # noqa: F403
+
+pytest_plugins = ["dvc.testing.plugin"]

dvc/testing/benchmarks/fixtures.py

@@ -60,8 +60,8 @@ def dvc_venvs():
 
 
 @pytest.fixture(scope="session")
-def dvc_git_repo(tmp_path_factory, bench_config):
-    url = bench_config.dvc_git_repo
+def dvc_repo(tmp_path_factory, bench_config):
+    url = bench_config.dvc_repo
 
     if os.path.isdir(url):
         return url
@@ -73,8 +73,10 @@ def dvc_git_repo(tmp_path_factory, bench_config):
 
 
 @pytest.fixture(scope="session")
-def dvc_bench_git_repo(tmp_path_factory, bench_config):
-    url = bench_config.dvc_bench_git_repo
+def dvc_bench_repo(tmp_path_factory, bench_config):
+    url = bench_config.dvc_bench_repo
+    if url is None:
+        pytest.skip("--dvc-bench-repo is not set")
 
     if os.path.isdir(url):
         return Path(url)
@@ -90,7 +92,7 @@ def make_dvc_bin(
     dvc_rev,
     dvc_venvs,
     make_dvc_venv,
-    dvc_git_repo,
+    dvc_repo,
     bench_config,
     request,
 ):
@@ -102,7 +104,7 @@ def make_dvc_bin(
                 pkg = f"dvc[{bench_config.dvc_install_deps}]"
             else:
                 pkg = "dvc"
-            packages = [f"{pkg} @ git+file://{dvc_git_repo}@{dvc_rev}"]
+            packages = [f"{pkg} @ git+file://{dvc_repo}@{dvc_rev}"]
             try:
                 if version.Version(dvc_rev) < version.Version("3.50.3"):
                     packages.append("pygit2==1.14.1")
@@ -116,34 +118,29 @@ def make_dvc_bin(
         dvc_bin = bench_config.dvc_bin
 
     def _dvc_bin(*args):
-        return check_output([dvc_bin, *args], text=True)  # noqa: S603
+        check_call([dvc_bin, *args])  # noqa: S603
 
-    _dvc_bin.version = parse_tuple(_dvc_bin("--version"))  # type: ignore[attr-defined]
+    _dvc_bin.version = check_output([dvc_bin, "--version"], text=True)  # type: ignore[attr-defined]  # noqa: S603
     return _dvc_bin
 
 
-def parse_tuple(version_string):
-    from packaging.version import Version
-
-    parsed = version.parse(version_string)
-    assert isinstance(parsed, Version)
-    return (parsed.major, parsed.minor, parsed.micro)
-
-
 @pytest.fixture
 def dvc_bin(request, make_dvc_bin):
     if marker := request.node.get_closest_marker("requires"):
-        minversion = marker.kwargs.get("minversion") or first(marker.args)
-        assert minversion, (
-            "'minversion' needs to be specified as"
-            " a positional or a keyword argument"
-        )
-        reason = marker.kwargs.get("reason", "")
-        if isinstance(minversion, str):
-            minversion = parse_tuple(minversion)
-        if make_dvc_bin.version < minversion:
-            version_repr = ".".join(map(str, minversion))
-            pytest.skip(f"requires dvc>={version_repr}: {reason}")
+        from packaging.specifiers import SpecifierSet
+        from packaging.version import Version, parse
+
+        spec = first(marker.args)
+        assert spec is not None
+        spec = SpecifierSet(spec) if isinstance(spec, str) else spec
+        reason = marker.kwargs["reason"]
+        dvc_version = make_dvc_bin.version
+        version = Version(parse(dvc_version).base_version)
+        if version not in spec:
+            pytest.skip(
+                f"Version {dvc_version} "
+                f"does not satisfy requirement {spec!r}: {reason}"
+            )
     return make_dvc_bin
 
 
@@ -170,11 +167,19 @@ def make_bench(request):
 
 
 @pytest.fixture
-def bench_dvc(dvc_bin, make_bench):
+def bench_dvc(request, dvc_bin, make_bench):
     def _bench_dvc(*args, **kwargs):
         name = kwargs.pop("name", None)
         name = f"-{name}" if name else ""
         bench = make_bench(args[0] + name)
+        if request.config.getoption("--benchmark-cprofile-dump") or kwargs.pop(
+            "cprofile", False
+        ):
+            cprofile_results = request.config.invocation_params.dir / "prof"
+            cprofile_results.mkdir(exist_ok=True)
+            stats_file = cprofile_results / f"{bench.name}.prof"
+            args = (*args, "--cprofile-dump", stats_file)
+
         return bench.pedantic(dvc_bin, args=args, **kwargs)
 
     return _bench_dvc
@@ -191,14 +196,14 @@ def _pull(repo, *args):
 
 
 @pytest.fixture
-def make_dataset(request, bench_config, tmp_dir, dvc_bench_git_repo):
+def make_dataset(request, bench_config, tmp_dir, dvc_bench_repo):
     def _make_dataset(
         dvcfile=False, files=True, cache=False, commit=False, remote=False
     ):
         from dvc.repo import Repo
 
         path = tmp_dir / "dataset"
-        root = dvc_bench_git_repo
+        root = dvc_bench_repo
         src = root / "data" / bench_config.dataset / "dataset"
         src_dvc = src.with_suffix(".dvc")
 
@@ -257,7 +262,7 @@ def make_project(tmp_path_factory):
 @pytest.fixture
 def project(bench_config, monkeypatch, make_project):
     rev = bench_config.project_rev
-    url = bench_config.project_git_repo
+    url = bench_config.project_repo
 
     if os.path.isdir(url):
         path = url

dvc/testing/benchmarks/plugin.py

@@ -1,57 +1,84 @@
+import os
+from dataclasses import dataclass, fields
+from pathlib import Path
+from typing import Optional
+
 DEFAULT_DVC_BIN = "dvc"
-DEFAULT_DVC_GIT_REPO = "https://github.com/iterative/dvc"
-DEFAULT_DVC_BENCH_GIT_REPO = "https://github.com/iterative/dvc-bench"
-DEFAULT_PROJECT_GIT_REPO = "https://github.com/iterative/example-get-started"
+DEFAULT_DVC_REPO = os.fspath(Path(__file__).parents[3])
+DEFAULT_PROJECT_REPO = "https://github.com/iterative/example-get-started"
 
 
 def pytest_report_header(config):
     bconf = config.bench_config
-    return f"dvc-bench: (dataset: {bconf.dataset!r}, revs: {bconf.dvc_revs!r})"
+    return f"dvc-bench: {bconf}"
 
 
 def pytest_generate_tests(metafunc):
-    str_revs = metafunc.config.getoption("--dvc-revs")
-    revs = str_revs.split(",") if str_revs else [None]
+    revs = metafunc.config.getoption("--dvc-revs")
+    if not revs:
+        revs = [None]
     if "dvc_rev" in metafunc.fixturenames:
         metafunc.parametrize("dvc_rev", revs, scope="session")
 
 
+@dataclass
 class DVCBenchConfig:
-    def __init__(self):
-        self.dataset = "small"
-        self.dvc_bin = DEFAULT_DVC_BIN
-        self.dvc_revs = None
-        self.dvc_git_repo = DEFAULT_DVC_GIT_REPO
-        self.dvc_install_deps = None
-        self.dvc_bench_git_repo = DEFAULT_DVC_BENCH_GIT_REPO
-        self.project_rev = None
-        self.project_git_repo = DEFAULT_PROJECT_GIT_REPO
+    dataset: str = "tiny"
+    dvc_repo: str = DEFAULT_DVC_REPO
+    dvc_bench_repo: Optional[str] = None
+    project_repo: str = DEFAULT_PROJECT_REPO
+    project_rev: Optional[str] = None
+    dvc_bin: str = DEFAULT_DVC_BIN
+    dvc_revs: Optional[list[str]] = None
+    dvc_install_deps: Optional[str] = None
+
+    def __repr__(self):
+        args = ", ".join(
+            f"{f.name}={val!r}"
+            for f in fields(self)
+            if (val := getattr(self, f.name)) != f.default
+        )
+        return f"{self.__class__.__name__}({args})"
 
 
 def pytest_configure(config):
     config.addinivalue_line(
-        "markers", "requires(minversion): mark a test as requiring minimum DVC version"
+        "markers",
+        "requires(spec): mark a test to run only on versions that satisfy the spec",
+    )
+    config.bench_config = DVCBenchConfig(
+        dataset=config.getoption("--dataset"),
+        dvc_repo=config.getoption("--dvc-repo"),
+        dvc_bench_repo=config.getoption("--dvc-bench-repo"),
+        project_repo=config.getoption("--project-repo"),
+        project_rev=config.getoption("--project-rev"),
+        dvc_bin=config.getoption("--dvc-bin"),
+        dvc_revs=config.getoption("--dvc-revs"),
+        dvc_install_deps=config.getoption("--dvc-install-deps"),
     )
 
-    config.bench_config = DVCBenchConfig()
-    config.bench_config.dataset = config.getoption("--dataset")
-    config.bench_config.dvc_bin = config.getoption("--dvc-bin")
-    config.bench_config.dvc_revs = config.getoption("--dvc-revs")
-    config.bench_config.dvc_git_repo = config.getoption("--dvc-git-repo")
-    config.bench_config.dvc_install_deps = config.getoption("--dvc-install-deps")
-    config.bench_config.dvc_bench_git_repo = config.getoption("--dvc-bench-git-repo")
-    config.bench_config.project_rev = config.getoption("--project-rev")
-    config.bench_config.project_git_repo = config.getoption("--project-git-repo")
+
+def resolve_path(path):
+    if os.path.isdir(path):
+        return os.path.abspath(path)
+    return path
 
 
 def pytest_addoption(parser):
     parser.addoption(
         "--dataset",
         type=str,
-        default="small",
+        default="tiny",
         help="Dataset name to use in tests (e.g. tiny/small/large/mnist/etc)",
     )
 
+    parser.addoption(
+        "--benchmark-cprofile-dump",
+        action="store_true",
+        default=False,
+        help="Save cprofile results",
+    )
+
     parser.addoption(
         "--dvc-bin",
         type=str,
@@ -61,36 +88,34 @@ def pytest_addoption(parser):
 
     parser.addoption(
         "--dvc-revs",
-        type=str,
+        type=lambda revs: revs.split(","),
         help=("Comma-separated list of DVC revisions to test (overrides `--dvc-bin`)"),
     )
 
     parser.addoption(
-        "--dvc-git-repo",
-        type=str,
-        default=DEFAULT_DVC_GIT_REPO,
+        "--dvc-repo",
+        type=resolve_path,
+        default=DEFAULT_DVC_REPO,
         help="Path or url to dvc git repo",
     )
 
     parser.addoption(
         "--dvc-install-deps",
         type=str,
-        default="",
         help="Comma-separated list of DVC installation packages",
     )
 
     parser.addoption(
-        "--dvc-bench-git-repo",
-        type=str,
-        default=DEFAULT_DVC_BENCH_GIT_REPO,
+        "--dvc-bench-repo",
+        type=resolve_path,
+        default=None,
         help="Path or url to dvc-bench git repo (for loading benchmark datasets)",
     )
 
     parser.addoption("--project-rev", type=str, help="Project revision to test")
-
     parser.addoption(
-        "--project-git-repo",
-        type=str,
-        default=DEFAULT_PROJECT_GIT_REPO,
+        "--project-repo",
+        type=resolve_path,
+        default=DEFAULT_PROJECT_REPO,
         help="Path or url to dvc project",
     )

dvc/testing/plugin.py

@@ -18,3 +18,9 @@ def pytest_configure(config):
     from .benchmarks.plugin import pytest_configure as bench_configure
 
     bench_configure(config)
+
+
+def pytest_report_header(config):
+    from .benchmarks.plugin import pytest_report_header as bench_report_header
+
+    return bench_report_header(config)

pyproject.toml

@@ -83,7 +83,7 @@ gdrive = ["dvc-gdrive>=3,<4"]
 gs = ["dvc-gs>=3,<4"]
 hdfs = ["dvc-hdfs>=3,<4"]
 lint = [
-    "mypy==1.11.1",
+    "mypy==1.11.2",
     "pandas-stubs",
     "types-colorama",
     "types-psutil",
@@ -159,7 +159,7 @@ namespaces = false
 write_to = "dvc/_dvc_version.py"
 
 [tool.pytest.ini_options]
-addopts = "-ra --cov-config pyproject.toml --dist worksteal --benchmark-skip"
+addopts = "-ra --cov-config pyproject.toml"
 filterwarnings = [
     "error::ResourceWarning",
     "error::pytest.PytestUnraisableExceptionWarning",

tests/__main__.py

@@ -3,4 +3,4 @@ if __name__ == "__main__":
 
     import pytest
 
-    sys.exit(pytest.main(["-v", "-n=logical", *sys.argv[1:]]))
+    sys.exit(pytest.main(["-v", "-n=logical", "--dist=worksteal", *sys.argv[1:]]))

tests/func/test_import.py

@@ -6,14 +6,14 @@ from funcy import first
 
 from dvc.cachemgr import CacheManager
 from dvc.config import NoRemoteError
-from dvc.dependency import base
 from dvc.dvcfile import load_file
 from dvc.fs import system
 from dvc.scm import Git
 from dvc.stage.exceptions import StagePathNotFoundError
 from dvc.testing.tmp_dir import make_subrepo
 from dvc.utils.fs import remove
-from dvc_data.index.index import DataIndexDirError
+from dvc_data.hashfile import hash
+from dvc_data.index.index import DataIndex, DataIndexDirError
 
 
 def test_import(tmp_dir, scm, dvc, erepo_dir):
@@ -725,14 +725,41 @@ def test_import_invalid_configs(tmp_dir, scm, dvc, erepo_dir):
         )
 
 
-def test_reimport(tmp_dir, scm, dvc, erepo_dir, mocker):
+@pytest.mark.parametrize(
+    "files,expected_info_calls",
+    [
+        ({"foo": "foo"}, {("foo",)}),
+        (
+            {
+                "dir": {
+                    "bar": "bar",
+                    "subdir": {"lorem": "ipsum", "nested": {"lorem": "lorem"}},
+                }
+            },
+            # info calls should be made for only directories
+            {("dir",), ("dir", "subdir"), ("dir", "subdir", "nested")},
+        ),
+    ],
+)
+def test_import_no_hash(
+    tmp_dir, scm, dvc, erepo_dir, mocker, files, expected_info_calls
+):
     with erepo_dir.chdir():
-        erepo_dir.dvc_gen("foo", "foo content", commit="create foo")
+        erepo_dir.dvc_gen(files, commit="create foo")
 
-    spy = mocker.spy(base, "fs_download")
-    dvc.imp(os.fspath(erepo_dir), "foo", "foo_imported")
-    assert spy.called
+    file_md5_spy = mocker.spy(hash, "file_md5")
+    index_info_spy = mocker.spy(DataIndex, "info")
+    name = next(iter(files))
 
-    spy.reset_mock()
-    dvc.imp(os.fspath(erepo_dir), "foo", "foo_imported", force=True)
-    assert not spy.called
+    dvc.imp(os.fspath(erepo_dir), name, "out")
+
+    local_hashes = [
+        call.args[0]
+        for call in file_md5_spy.call_args_list
+        if call.args[1].protocol == "local"
+    ]
+    # no files should be hashed, should use existing metadata
+    assert not local_hashes
+    assert {
+        call.args[1] for call in index_info_spy.call_args_list
+    } == expected_info_calls

tests/unit/fs/test_dvcfs.py

@@ -0,0 +1,805 @@
+import os
+import posixpath
+from hashlib import md5
+from itertools import product
+
+import pytest
+from fsspec.implementations.local import LocalFileSystem, make_path_posix
+from fsspec.tests.abstract.common import GLOB_EDGE_CASES_TESTS
+
+from dvc.api import DVCFileSystem
+
+
+class DVCFixtures:
+    """The fixtures imitate the fsspec.tests.abstract.AbstractFixtures.
+
+    This has been modified to use dvc's fixture, as DVCFileSystem is a read-only
+    filesystem, and cannot be used to create directories or files.
+
+    The `Output.ignore()` is mocked to avoid `.gitignore` files in the directories,
+    as we can reuse the tests from fsspec with minimal modifications.
+    `.gitignore` file is manually created with required patterns at the root of the
+    repository.
+    """
+
+    @pytest.fixture
+    def fs_bulk_operations_scenario_0(self, tmp_dir):
+        """
+        Scenario that is used for many cp/get/put tests. Creates the following
+        directory and file structure:
+
+        📁 source
+        ├── 📄 file1
+        ├── 📄 file2
+        └── 📁 subdir
+            ├── 📄 subfile1
+            ├── 📄 subfile2
+            └── 📁 nesteddir
+                └── 📄 nestedfile
+        """
+        source = tmp_dir / "source"
+        source.mkdir()
+        tmp_dir.scm_gen(
+            ".gitignore", "/source/file2/\nsource/subdir", commit="add .gitignore"
+        )
+
+        tmp_dir.scm_gen("source/file1", "file1", commit="add file1")
+        tmp_dir.dvc_gen("source/file2", "file2", commit="add file2")
+        tmp_dir.dvc_gen(
+            {
+                "source/subdir": {
+                    "subfile1": "subfile1",
+                    "subfile2": "subfile2",
+                    "nesteddir": {"nestedfile": "nestedfile"},
+                }
+            },
+            commit="add subdir",
+        )
+        return "/source"
+
+    @pytest.fixture
+    def fs_10_files_with_hashed_names(self, tmp_dir, local_fs, local_join, local_path):
+        """
+        Scenario that is used to check cp/get/put files order when source and
+        destination are lists. Creates the following directory and file structure:
+
+        📁 source
+        └── 📄 {hashed([0-9])}.txt
+        """
+        dir_contents = {
+            md5(str(i).encode("utf-8")).hexdigest() + ".txt": str(i) for i in range(10)
+        }
+        tmp_dir.dvc_gen({"source": dir_contents}, commit="add source")
+        tmp_dir.scm_gen(".gitignore", "/source", commit="add .gitignore")
+        return "/source"
+
+    @pytest.fixture
+    def src_directory(self, tmp_dir):
+        # https://github.com/fsspec/filesystem_spec/issues/1062
+        # Recursive cp/get/put of source directory into non-existent target directory.
+        tmp_dir.dvc_gen({"src": {"file": "file"}}, commit="add source")
+        return "/src"
+
+    @pytest.fixture
+    def fs_dir_and_file_with_same_name_prefix(self, tmp_dir):
+        """
+        Scenario that is used to check cp/get/put on directory and file with
+        the same name prefixes. Creates the following directory and file structure:
+
+        📁 source
+        ├── 📄 subdir.txt
+        └── 📁 subdir
+            └── 📄 subfile.txt
+        """
+        source = tmp_dir / "source"
+        source.mkdir()
+
+        tmp_dir.scm_gen(".gitignore", "/source/subdir", commit="add .gitignore")
+        tmp_dir.scm_gen("source/subdir.txt", "subdir.txt", commit="add subdir.txt")
+        tmp_dir.dvc_gen(
+            {"source/subdir": {"subfile.txt": "subfile.txt"}}, commit="add subdir"
+        )
+        return "/source"
+
+    @pytest.fixture
+    def fs_glob_edge_cases_files(self, tmp_dir):
+        """
+        Scenario that is used for glob edge cases cp/get/put tests.
+        Creates the following directory and file structure:
+
+        📁 source
+        ├── 📄 file1
+        ├── 📄 file2
+        ├── 📁 subdir0
+        │   ├── 📄 subfile1
+        │   ├── 📄 subfile2
+        │   └── 📁 nesteddir
+        │       └── 📄 nestedfile
+        └── 📁 subdir1
+            ├── 📄 subfile1
+            ├── 📄 subfile2
+            └── 📁 nesteddir
+                └── 📄 nestedfile
+        """
+        source = tmp_dir / "source"
+        source.mkdir()
+
+        tmp_dir.scm_gen(
+            ".gitignore", "/source/file1\n/source/subdir1", commit="add .gitignore"
+        )
+        tmp_dir.scm_gen("source/file1", "file1", commit="add file1")
+        tmp_dir.dvc_gen("source/file2", "file2", commit="add file2")
+
+        dir_contents = {
+            "subfile1": "subfile1",
+            "subfile2": "subfile2",
+            "nesteddir": {"nestedfile": "nestedfile"},
+        }
+        tmp_dir.scm_gen({"source/subdir0": dir_contents}, commit="add subdir0")
+        tmp_dir.dvc_gen({"source/subdir1": dir_contents}, commit="add subdir1")
+        return "/source"
+
+    @pytest.fixture(params=[{"rev": "HEAD"}, {}])
+    def fs(self, request, tmp_dir, dvc, scm):
+        return DVCFileSystem(tmp_dir, **request.param)
+
+    @pytest.fixture(autouse=True)
+    def mock_ignore(self, mocker):
+        mocker.patch("dvc.output.Output.ignore")
+
+    @pytest.fixture
+    def fs_join(self):
+        return posixpath.join
+
+    @pytest.fixture
+    def fs_path(self, fs):
+        return fs.root_marker
+
+    @pytest.fixture(scope="class")
+    def local_fs(self):
+        # Maybe need an option for auto_mkdir=False?  This is only relevant
+        # for certain implementations.
+        return LocalFileSystem(auto_mkdir=True)
+
+    @pytest.fixture
+    def local_join(self):
+        """
+        Return a function that joins its arguments together into a path, on
+        the local filesystem.
+        """
+        return os.path.join
+
+    @pytest.fixture
+    def local_path(self, tmpdir):
+        return tmpdir
+
+    @pytest.fixture
+    def local_target(self, local_fs, local_join, local_path):
+        """
+        Return name of local directory that does not yet exist to copy into.
+
+        Cleans up at the end of each test it which it is used.
+        """
+        target = local_join(local_path, "target")
+        yield target
+        if local_fs.exists(target):
+            local_fs.rm(target, recursive=True)
+
+
+class TestDVCFileSystemGet(DVCFixtures):
+    """
+    This test is adapted from `fsspec.tests.abstract.get.AbstractGetTests`
+    with minor modifications to work with DVCFixtures and DVCFileSystem.
+    """
+
+    def test_get_file_to_existing_directory(
+        self,
+        fs,
+        fs_join,
+        fs_bulk_operations_scenario_0,
+        local_fs,
+        local_join,
+        local_target,
+    ):
+        # Copy scenario 1a
+        source = fs_bulk_operations_scenario_0
+
+        target = local_target
+        local_fs.mkdir(target)
+        assert local_fs.isdir(target)
+
+        target_file2 = local_join(target, "file2")
+        target_subfile1 = local_join(target, "subfile1")
+
+        # Copy from source directory
+        fs.get(fs_join(source, "file2"), target)
+        assert local_fs.isfile(target_file2)
+
+        # Copy from sub directory
+        fs.get(fs_join(source, "subdir", "subfile1"), target)
+        assert local_fs.isfile(target_subfile1)
+
+        # Remove copied files
+        local_fs.rm([target_file2, target_subfile1])
+        assert not local_fs.exists(target_file2)
+        assert not local_fs.exists(target_subfile1)
+
+        # Repeat with trailing slash on target
+        fs.get(fs_join(source, "file2"), target + "/")
+        assert local_fs.isdir(target)
+        assert local_fs.isfile(target_file2)
+
+        fs.get(fs_join(source, "subdir", "subfile1"), target + "/")
+        assert local_fs.isfile(target_subfile1)
+
+    def test_get_file_to_new_directory(
+        self,
+        fs,
+        fs_join,
+        fs_bulk_operations_scenario_0,
+        local_fs,
+        local_join,
+        local_target,
+    ):
+        # Copy scenario 1b
+        source = fs_bulk_operations_scenario_0
+
+        target = local_target
+        local_fs.mkdir(target)
+
+        fs.get(
+            fs_join(source, "subdir", "subfile1"), local_join(target, "newdir/")
+        )  # Note trailing slash
+
+        assert local_fs.isdir(target)
+        assert local_fs.isdir(local_join(target, "newdir"))
+        assert local_fs.isfile(local_join(target, "newdir", "subfile1"))
+
+    def test_get_file_to_file_in_existing_directory(
+        self,
+        fs,
+        fs_join,
+        fs_bulk_operations_scenario_0,
+        local_fs,
+        local_join,
+        local_target,
+    ):
+        # Copy scenario 1c
+        source = fs_bulk_operations_scenario_0
+
+        target = local_target
+        local_fs.mkdir(target)
+
+        fs.get(fs_join(source, "subdir", "subfile1"), local_join(target, "newfile"))
+        assert local_fs.isfile(local_join(target, "newfile"))
+
+    def test_get_file_to_file_in_new_directory(
+        self,
+        fs,
+        fs_join,
+        fs_bulk_operations_scenario_0,
+        local_fs,
+        local_join,
+        local_target,
+    ):
+        # Copy scenario 1d
+        source = fs_bulk_operations_scenario_0
+
+        target = local_target
+        local_fs.mkdir(target)
+
+        fs.get(
+            fs_join(source, "subdir", "subfile1"),
+            local_join(target, "newdir", "newfile"),
+        )
+        assert local_fs.isdir(local_join(target, "newdir"))
+        assert local_fs.isfile(local_join(target, "newdir", "newfile"))
+
+    def test_get_directory_to_existing_directory(
+        self,
+        fs,
+        fs_join,
+        fs_bulk_operations_scenario_0,
+        local_fs,
+        local_join,
+        local_target,
+    ):
+        # Copy scenario 1e
+        source = fs_bulk_operations_scenario_0
+
+        target = local_target
+        local_fs.mkdir(target)
+        assert local_fs.isdir(target)
+
+        for source_slash, target_slash in zip([False, True], [False, True]):
+            s = fs_join(source, "subdir")
+            if source_slash:
+                s += "/"
+            t = target + "/" if target_slash else target
+
+            # Without recursive does nothing
+            fs.get(s, t)
+            assert local_fs.ls(target) == []
+
+            # With recursive
+            fs.get(s, t, recursive=True)
+            if source_slash:
+                assert local_fs.isfile(local_join(target, "subfile1"))
+                assert local_fs.isfile(local_join(target, "subfile2"))
+                assert local_fs.isdir(local_join(target, "nesteddir"))
+                assert local_fs.isfile(local_join(target, "nesteddir", "nestedfile"))
+                assert not local_fs.exists(local_join(target, "subdir"))
+
+                local_fs.rm(
+                    [
+                        local_join(target, "subfile1"),
+                        local_join(target, "subfile2"),
+                        local_join(target, "nesteddir"),
+                    ],
+                    recursive=True,
+                )
+            else:
+                assert local_fs.isdir(local_join(target, "subdir"))
+                assert local_fs.isfile(local_join(target, "subdir", "subfile1"))
+                assert local_fs.isfile(local_join(target, "subdir", "subfile2"))
+                assert local_fs.isdir(local_join(target, "subdir", "nesteddir"))
+                assert local_fs.isfile(
+                    local_join(target, "subdir", "nesteddir", "nestedfile")
+                )
+
+                local_fs.rm(local_join(target, "subdir"), recursive=True)
+            assert local_fs.ls(target) == []
+
+            # Limit recursive by maxdepth
+            fs.get(s, t, recursive=True, maxdepth=1)
+            if source_slash:
+                assert local_fs.isfile(local_join(target, "subfile1"))
+                assert local_fs.isfile(local_join(target, "subfile2"))
+                assert not local_fs.exists(local_join(target, "nesteddir"))
+                assert not local_fs.exists(local_join(target, "subdir"))
+
+                local_fs.rm(
+                    [
+                        local_join(target, "subfile1"),
+                        local_join(target, "subfile2"),
+                    ],
+                    recursive=True,
+                )
+            else:
+                assert local_fs.isdir(local_join(target, "subdir"))
+                assert local_fs.isfile(local_join(target, "subdir", "subfile1"))
+                assert local_fs.isfile(local_join(target, "subdir", "subfile2"))
+                assert not local_fs.exists(local_join(target, "subdir", "nesteddir"))
+
+                local_fs.rm(local_join(target, "subdir"), recursive=True)
+            assert local_fs.ls(target) == []
+
+    def test_get_directory_to_new_directory(
+        self,
+        fs,
+        fs_join,
+        fs_bulk_operations_scenario_0,
+        local_fs,
+        local_join,
+        local_target,
+    ):
+        # Copy scenario 1f
+        source = fs_bulk_operations_scenario_0
+
+        target = local_target
+        local_fs.mkdir(target)
+
+        for source_slash, target_slash in zip([False, True], [False, True]):
+            s = fs_join(source, "subdir")
+            if source_slash:
+                s += "/"
+            t = local_join(target, "newdir")
+            if target_slash:
+                t += "/"
+
+            # Without recursive does nothing
+            fs.get(s, t)
+            assert local_fs.ls(target) == []
+
+            # With recursive
+            fs.get(s, t, recursive=True)
+            assert local_fs.isdir(local_join(target, "newdir"))
+            assert local_fs.isfile(local_join(target, "newdir", "subfile1"))
+            assert local_fs.isfile(local_join(target, "newdir", "subfile2"))
+            assert local_fs.isdir(local_join(target, "newdir", "nesteddir"))
+            assert local_fs.isfile(
+                local_join(target, "newdir", "nesteddir", "nestedfile")
+            )
+            assert not local_fs.exists(local_join(target, "subdir"))
+
+            local_fs.rm(local_join(target, "newdir"), recursive=True)
+            assert local_fs.ls(target) == []
+
+            # Limit recursive by maxdepth
+            fs.get(s, t, recursive=True, maxdepth=1)
+            assert local_fs.isdir(local_join(target, "newdir"))
+            assert local_fs.isfile(local_join(target, "newdir", "subfile1"))
+            assert local_fs.isfile(local_join(target, "newdir", "subfile2"))
+            assert not local_fs.exists(local_join(target, "newdir", "nesteddir"))
+            assert not local_fs.exists(local_join(target, "subdir"))
+
+            local_fs.rm(local_join(target, "newdir"), recursive=True)
+            assert not local_fs.exists(local_join(target, "newdir"))
+
+    def test_get_glob_to_existing_directory(
+        self,
+        fs,
+        fs_join,
+        fs_bulk_operations_scenario_0,
+        local_fs,
+        local_join,
+        local_target,
+    ):
+        # Copy scenario 1g
+        source = fs_bulk_operations_scenario_0
+
+        target = local_target
+        local_fs.mkdir(target)
+
+        for target_slash in [False, True]:
+            t = target + "/" if target_slash else target
+
+            # Without recursive
+            fs.get(fs_join(source, "subdir", "*"), t)
+            assert local_fs.isfile(local_join(target, "subfile1"))
+            assert local_fs.isfile(local_join(target, "subfile2"))
+            assert not local_fs.isdir(local_join(target, "nesteddir"))
+            assert not local_fs.exists(local_join(target, "nesteddir", "nestedfile"))
+            assert not local_fs.exists(local_join(target, "subdir"))
+
+            local_fs.rm(
+                [
+                    local_join(target, "subfile1"),
+                    local_join(target, "subfile2"),
+                ],
+                recursive=True,
+            )
+            assert local_fs.ls(target) == []
+
+            # With recursive
+            for glob, recursive in zip(["*", "**"], [True, False]):
+                fs.get(fs_join(source, "subdir", glob), t, recursive=recursive)
+                assert local_fs.isfile(local_join(target, "subfile1"))
+                assert local_fs.isfile(local_join(target, "subfile2"))
+                assert local_fs.isdir(local_join(target, "nesteddir"))
+                assert local_fs.isfile(local_join(target, "nesteddir", "nestedfile"))
+                assert not local_fs.exists(local_join(target, "subdir"))
+
+                local_fs.rm(
+                    [
+                        local_join(target, "subfile1"),
+                        local_join(target, "subfile2"),
+                        local_join(target, "nesteddir"),
+                    ],
+                    recursive=True,
+                )
+                assert local_fs.ls(target) == []
+
+                # Limit recursive by maxdepth
+                fs.get(
+                    fs_join(source, "subdir", glob), t, recursive=recursive, maxdepth=1
+                )
+                assert local_fs.isfile(local_join(target, "subfile1"))
+                assert local_fs.isfile(local_join(target, "subfile2"))
+                assert not local_fs.exists(local_join(target, "nesteddir"))
+                assert not local_fs.exists(local_join(target, "subdir"))
+
+                local_fs.rm(
+                    [
+                        local_join(target, "subfile1"),
+                        local_join(target, "subfile2"),
+                    ],
+                    recursive=True,
+                )
+                assert local_fs.ls(target) == []
+
+    def test_get_glob_to_new_directory(
+        self,
+        fs,
+        fs_join,
+        fs_bulk_operations_scenario_0,
+        local_fs,
+        local_join,
+        local_target,
+    ):
+        # Copy scenario 1h
+        source = fs_bulk_operations_scenario_0
+
+        target = local_target
+        local_fs.mkdir(target)
+
+        for target_slash in [False, True]:
+            t = fs_join(target, "newdir")
+            if target_slash:
+                t += "/"
+
+            # Without recursive
+            fs.get(fs_join(source, "subdir", "*"), t)
+            assert local_fs.isdir(local_join(target, "newdir"))
+            assert local_fs.isfile(local_join(target, "newdir", "subfile1"))
+            assert local_fs.isfile(local_join(target, "newdir", "subfile2"))
+            assert not local_fs.exists(local_join(target, "newdir", "nesteddir"))
+            assert not local_fs.exists(
+                local_join(target, "newdir", "nesteddir", "nestedfile")
+            )
+            assert not local_fs.exists(local_join(target, "subdir"))
+            assert not local_fs.exists(local_join(target, "newdir", "subdir"))
+
+            local_fs.rm(local_join(target, "newdir"), recursive=True)
+            assert local_fs.ls(target) == []
+
+            # With recursive
+            for glob, recursive in zip(["*", "**"], [True, False]):
+                fs.get(fs_join(source, "subdir", glob), t, recursive=recursive)
+                assert local_fs.isdir(local_join(target, "newdir"))
+                assert local_fs.isfile(local_join(target, "newdir", "subfile1"))
+                assert local_fs.isfile(local_join(target, "newdir", "subfile2"))
+                assert local_fs.isdir(local_join(target, "newdir", "nesteddir"))
+                assert local_fs.isfile(
+                    local_join(target, "newdir", "nesteddir", "nestedfile")
+                )
+                assert not local_fs.exists(local_join(target, "subdir"))
+                assert not local_fs.exists(local_join(target, "newdir", "subdir"))
+
+                local_fs.rm(local_join(target, "newdir"), recursive=True)
+                assert not local_fs.exists(local_join(target, "newdir"))
+
+                # Limit recursive by maxdepth
+                fs.get(
+                    fs_join(source, "subdir", glob), t, recursive=recursive, maxdepth=1
+                )
+                assert local_fs.isdir(local_join(target, "newdir"))
+                assert local_fs.isfile(local_join(target, "newdir", "subfile1"))
+                assert local_fs.isfile(local_join(target, "newdir", "subfile2"))
+                assert not local_fs.exists(local_join(target, "newdir", "nesteddir"))
+                assert not local_fs.exists(local_join(target, "subdir"))
+                assert not local_fs.exists(local_join(target, "newdir", "subdir"))
+
+                local_fs.rm(local_fs.ls(target, detail=False), recursive=True)
+                assert not local_fs.exists(local_join(target, "newdir"))
+
+    @pytest.mark.parametrize(
+        GLOB_EDGE_CASES_TESTS["argnames"],
+        GLOB_EDGE_CASES_TESTS["argvalues"],
+    )
+    def test_get_glob_edge_cases(
+        self,
+        path,
+        recursive,
+        maxdepth,
+        expected,
+        fs,
+        fs_join,
+        fs_glob_edge_cases_files,
+        local_fs,
+        local_join,
+        local_target,
+    ):
+        # Copy scenario 1g
+        source = fs_glob_edge_cases_files
+
+        target = local_target
+
+        for new_dir, target_slash in product([True, False], [True, False]):
+            local_fs.mkdir(target)
+
+            t = local_join(target, "newdir") if new_dir else target
+            t = t + "/" if target_slash else t
+
+            fs.get(fs_join(source, path), t, recursive=recursive, maxdepth=maxdepth)
+
+            output = local_fs.find(target)
+            if new_dir:
+                prefixed_expected = [
+                    make_path_posix(local_join(target, "newdir", p)) for p in expected
+                ]
+            else:
+                prefixed_expected = [
+                    make_path_posix(local_join(target, p)) for p in expected
+                ]
+            assert sorted(output) == sorted(prefixed_expected)
+
+            try:
+                local_fs.rm(target, recursive=True)
+            except FileNotFoundError:
+                pass
+
+    def test_get_list_of_files_to_existing_directory(
+        self,
+        fs,
+        fs_join,
+        fs_bulk_operations_scenario_0,
+        local_fs,
+        local_join,
+        local_target,
+    ):
+        # Copy scenario 2a
+        source = fs_bulk_operations_scenario_0
+
+        target = local_target
+        local_fs.mkdir(target)
+
+        source_files = [
+            fs_join(source, "file1"),
+            fs_join(source, "file2"),
+            fs_join(source, "subdir", "subfile1"),
+        ]
+
+        for target_slash in [False, True]:
+            t = target + "/" if target_slash else target
+
+            fs.get(source_files, t)
+            assert local_fs.isfile(local_join(target, "file1"))
+            assert local_fs.isfile(local_join(target, "file2"))
+            assert local_fs.isfile(local_join(target, "subfile1"))
+
+            local_fs.rm(
+                [
+                    local_join(target, "file1"),
+                    local_join(target, "file2"),
+                    local_join(target, "subfile1"),
+                ],
+                recursive=True,
+            )
+            assert local_fs.ls(target) == []
+
+    def test_get_list_of_files_to_new_directory(
+        self,
+        fs,
+        fs_join,
+        fs_bulk_operations_scenario_0,
+        local_fs,
+        local_join,
+        local_target,
+    ):
+        # Copy scenario 2b
+        source = fs_bulk_operations_scenario_0
+
+        target = local_target
+        local_fs.mkdir(target)
+
+        source_files = [
+            fs_join(source, "file1"),
+            fs_join(source, "file2"),
+            fs_join(source, "subdir", "subfile1"),
+        ]
+
+        fs.get(source_files, local_join(target, "newdir") + "/")  # Note trailing slash
+        assert local_fs.isdir(local_join(target, "newdir"))
+        assert local_fs.isfile(local_join(target, "newdir", "file1"))
+        assert local_fs.isfile(local_join(target, "newdir", "file2"))
+        assert local_fs.isfile(local_join(target, "newdir", "subfile1"))
+
+    def test_get_directory_recursive(
+        self, src_directory, fs, fs_join, fs_path, local_fs, local_join, local_target
+    ):
+        target = local_target
+        src = src_directory
+
+        # get without slash
+        assert not local_fs.exists(target)
+        for loop in range(2):
+            fs.get(src, target, recursive=True)
+            assert local_fs.isdir(target)
+
+            if loop == 0:
+                assert local_fs.isfile(local_join(target, "file"))
+                assert not local_fs.exists(local_join(target, "src"))
+            else:
+                assert local_fs.isfile(local_join(target, "file"))
+                assert local_fs.isdir(local_join(target, "src"))
+                assert local_fs.isfile(local_join(target, "src", "file"))
+
+        local_fs.rm(target, recursive=True)
+
+        # get with slash
+        assert not local_fs.exists(target)
+        for _ in range(2):
+            fs.get(src + "/", target, recursive=True)
+            assert local_fs.isdir(target)
+            assert local_fs.isfile(local_join(target, "file"))
+            assert not local_fs.exists(local_join(target, "src"))
+
+    def test_get_directory_without_files_with_same_name_prefix(
+        self,
+        fs,
+        fs_join,
+        local_fs,
+        local_join,
+        local_target,
+        fs_dir_and_file_with_same_name_prefix,
+    ):
+        # Create the test dirs
+        source = fs_dir_and_file_with_same_name_prefix
+        target = local_target
+
+        # Test without glob
+        fs.get(fs_join(source, "subdir"), target, recursive=True)
+
+        assert local_fs.isfile(local_join(target, "subfile.txt"))
+        assert not local_fs.isfile(local_join(target, "subdir.txt"))
+
+        local_fs.rm([local_join(target, "subfile.txt")])
+        assert local_fs.ls(target) == []
+
+        # Test with glob
+        fs.get(fs_join(source, "subdir*"), target, recursive=True)
+
+        assert local_fs.isdir(local_join(target, "subdir"))
+        assert local_fs.isfile(local_join(target, "subdir", "subfile.txt"))
+        assert local_fs.isfile(local_join(target, "subdir.txt"))
+
+    def test_get_with_source_and_destination_as_list(
+        self,
+        fs,
+        fs_join,
+        local_fs,
+        local_join,
+        local_target,
+        fs_10_files_with_hashed_names,
+    ):
+        # Create the test dir
+        source = fs_10_files_with_hashed_names
+        target = local_target
+
+        # Create list of files for source and destination
+        source_files = []
+        destination_files = []
+        for i in range(10):
+            hashed_i = md5(str(i).encode("utf-8")).hexdigest()
+            source_files.append(fs_join(source, f"{hashed_i}.txt"))
+            destination_files.append(
+                make_path_posix(local_join(target, f"{hashed_i}.txt"))
+            )
+
+        # Copy and assert order was kept
+        fs.get(rpath=source_files, lpath=destination_files)
+
+        for i in range(10):
+            file_content = local_fs.cat(destination_files[i]).decode("utf-8")
+            assert file_content == str(i)
+
+
+def test_maxdepth(tmp_dir, dvc, scm):
+    tmp_dir.dvc_gen(
+        {
+            "dir": {
+                "file1": "file1",
+                "subdir": {
+                    "file2": "file2",
+                    "subdir2": {"file3": "file3", "subdir3": {"file4": "file4"}},
+                },
+            }
+        },
+        commit="add dir",
+    )
+
+    fs = DVCFileSystem(url=tmp_dir)
+    fs.get("dir", "dir1", recursive=True, maxdepth=1)
+    assert (tmp_dir / "dir1").read_text() == {"file1": "file1"}
+
+    fs.get("dir", "dir2", recursive=True, maxdepth=2)
+    assert (tmp_dir / "dir2").read_text() == {
+        "file1": "file1",
+        "subdir": {"file2": "file2"},
+    }
+
+    fs.get("dir", "dir3", recursive=True, maxdepth=3)
+    assert (tmp_dir / "dir3").read_text() == {
+        "file1": "file1",
+        "subdir": {"file2": "file2", "subdir2": {"file3": "file3"}},
+    }
+
+    fs.get("dir", "dir4", recursive=True, maxdepth=4)
+    assert (tmp_dir / "dir4").read_text() == {
+        "file1": "file1",
+        "subdir": {
+            "file2": "file2",
+            "subdir2": {"file3": "file3", "subdir3": {"file4": "file4"}},
+        },
+    }