ls: implement --tree to show as a tree and --level to limit depth for recursion (#10598)

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

.github/ISSUE_TEMPLATE/epic_story.md

@@ -1,60 +0,0 @@
----
-name: Epic/Story Issue Template
-about: Template for "top" level issues - Epics (>2 weeks) / Stories (<2 weeks)
-title: 'Epic: New Feature'
-labels: epic
-assignees:
-
----
-
-## Summary / Background
-What do you want to achieve, why? business context
-...
-
-## Scope
-
-What will be impacted and what won't be?
-What needs implementation and what is invariant?
-e.g.
-- user should be able to run workflow X from UI
-- enable workflow Y, Z from CLI
-
-## Assumptions
-Product / UX assumptions as well as technical assumptions / limitations
-e.g.
-* Support only Python Runtime
-* Focus on DVC experiments only
-* Deployment environments don't change often and can be picked up from shared configuration
-
-## Open Questions
-e.g.
-- How should access control work for shared artifacts (workflow X)
-- Python runtime assumption - is it really valid? in light of <...>
-
-## Blockers / Dependencies
-List issues or other conditions / blockers
-
-## General Approach
-Invocation example:
-```shell
-$ mapper-run task.tar.gz --ray-cluster <ip>:<port>
-```
-
-## Steps
-
-### Must have (p1)
-- [ ] subissue2
-- [ ] step 2
- - info
- - info
-
-### Optional / followup (p2)
-- [ ] ⌛ step 3 wip
-- [ ] step 4
-
-## Timelines
-
-Put your estimations here. Update once certainty changes
-- end of week (Feb 3) for prototype with workflows X, Y
-- Feb 15 - MVP in production
-- Low priority followups can be done later

.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
@@ -12,7 +16,29 @@ jobs:
     steps:
       - uses: actions/setup-python@v5
         with:
-            python-version: "3.11"
-      - uses: iterative/dvc-bench@main
+            python-version: "3.12"
+
+      - uses: actions/checkout@v4
         with:
-            pytest_options: "-k 'test_init or test_help'"
+          ref: ${{ github.event.pull_request.base.sha }}
+          fetch-depth: 0
+
+      - uses: astral-sh/setup-uv@v3
+        with:
+          enable-cache: true
+          cache-dependency-glob: pyproject.toml
+
+      - 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/build.yaml

@@ -27,14 +27,15 @@ jobs:
       with:
         python-version: 3.12
 
-    - run: python -m pip install --upgrade pip setuptools_scm build twine
+    - uses: astral-sh/setup-uv@v3
+    - run: uv pip install --upgrade setuptools_scm twine --system
 
     - name: Force version for Test PyPI upload
       if: ${{ !startsWith(github.ref, 'refs/tags') }}
       run: echo version=$(python -m setuptools_scm | awk -F+ '{print $1}' | tail -1) >> $GITHUB_ENV
 
     - run: echo 'PKG = "pip"'>dvc/_build.py
-    - run: python -m build
+    - run: uv build
       env:
         SETUPTOOLS_SCM_PRETEND_VERSION_FOR_DVC: ${{ env.version }}
     - run: twine check --strict dist/*
@@ -66,6 +67,7 @@ jobs:
     - name: Upload package to Test PyPI
       uses: pypa/gh-action-pypi-publish@release/v1
       with:
+        attestations: true
         repository-url: https://test.pypi.org/legacy/
         skip-existing: true
 
@@ -90,3 +92,5 @@ jobs:
 
     - name: Upload package to PyPI
       uses: pypa/gh-action-pypi-publish@release/v1
+      with:
+        attestations: true

.github/workflows/plugin_tests.yaml

@@ -18,8 +18,8 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        os: [ubuntu-20.04]
-        pyv: ["3.10"]
+        os: [ubuntu-latest]
+        pyv: ["3.12"]
         plugin: ["dvc-s3"]
 
     steps:
@@ -37,16 +37,17 @@ jobs:
       uses: actions/setup-python@v5
       with:
         python-version: ${{ matrix.pyv }}
-        cache: pip
-        cache-dependency-path: |
-          dvc/pyproject.toml
-          ${{ matrix.plugin }}/setup.cfg
+
+    - uses: astral-sh/setup-uv@v3
+      with:
+        enable-cache: true
+        cache-dependency-glob: "**/pyproject.toml"
+        cache-suffix: ${{ matrix.pyv }}
 
     - name: Install plugin + DVC@PR
       run: |
-        pip install --upgrade pip wheel
-        pip install "./dvc[testing]"
-        pip install -e "./${{ matrix.plugin }}[tests]"
+        uv pip install "./dvc[testing]" --system
+        uv pip install -e "./${{ matrix.plugin }}[tests]" --system
 
     - name: Run plugin tests
       timeout-minutes: 15

.github/workflows/tests.yaml

@@ -16,15 +16,69 @@ permissions:
   contents: read
 
 jobs:
-  tests:
-    timeout-minutes: 50
+  lint:
+    timeout-minutes: 10
     runs-on: ${{ matrix.os }}
     strategy:
       fail-fast: false
       matrix:
-        os: [ubuntu-20.04, windows-latest, macos-latest]
-        pyv: ["3.9", "3.10", "3.11", "3.12"]
+        os: [ubuntu-latest, macos-latest, windows-latest]
+    steps:
+    - uses: actions/checkout@v4
+      with:
+        fetch-depth: 0
+    - name: Set up Python
+      uses: actions/setup-python@v5
+      with:
+        python-version: '3.12'
+    - uses: astral-sh/setup-uv@v3
+      with:
+        enable-cache: true
+        cache-dependency-glob: pyproject.toml
+    - run: uv pip install -e ".[dev]" --system
+      env:
+        UV_LINK_MODE: ${{ runner.os == 'Windows' && 'symlink' || 'hardlink' }}
 
+    - name: Cache mypy
+      uses: actions/cache@v4
+      with:
+        path: .mypy_cache
+        key: mypy-${{ runner.os }}-${{ env.pythonLocation }}-${{ hashFiles('pyproject.toml') }}
+    - uses: pre-commit/[email protected]
+
+  tests:
+    timeout-minutes: 20
+    runs-on: ${{ matrix.os }}
+    strategy:
+      fail-fast: false
+      matrix:
+        os: [ubuntu-latest, macos-latest]
+        pyv: ["3.9", "3.10", "3.11", "3.12"]
+        include:
+        - os: windows-latest
+          pyv: "3.9"
+          pytestargs: "--splitting-algorithm=least_duration --splits 4 --group 1"
+        - os: windows-latest
+          pyv: "3.9"
+          pytestargs: "--splitting-algorithm=least_duration --splits 4 --group 2"
+        - os: windows-latest
+          pyv: "3.9"
+          pytestargs: "--splitting-algorithm=least_duration --splits 4 --group 3"
+        - os: windows-latest
+          pyv: "3.9"
+          pytestargs: "--splitting-algorithm=least_duration --splits 4 --group 4"
+        - os: windows-latest
+          pyv: "3.12"
+          pytestargs: "--splitting-algorithm=least_duration --splits 4 --group 1"
+        - os: windows-latest
+          pyv: "3.12"
+          pytestargs: "--splitting-algorithm=least_duration --splits 4 --group 2"
+        - os: windows-latest
+          pyv: "3.12"
+          pytestargs: "--splitting-algorithm=least_duration --splits 4 --group 3"
+        - os: windows-latest
+          pyv: "3.12"
+          pytestargs: "--splitting-algorithm=least_duration --splits 4 --group 4"
     steps:
     - uses: actions/checkout@v4
       with:
@@ -33,20 +87,22 @@ jobs:
       uses: actions/setup-python@v5
       with:
         python-version: ${{ matrix.pyv }}
-        cache: pip
-        cache-dependency-path: pyproject.toml
-    - name: install
-      run: |
-        pip install --upgrade pip wheel
-        pip install -e ".[dev]"
-    - uses: pre-commit/[email protected]
+    - uses: astral-sh/setup-uv@v3
+      with:
+        enable-cache: true
+        cache-dependency-glob: pyproject.toml
+        cache-suffix: ${{ matrix.pyv }}
+    - run: uv pip install -e ".[dev]" pytest-split --system
+      env:
+        UV_LINK_MODE: ${{ runner.os == 'Windows' && 'symlink' || 'hardlink' }}
+
     - name: run tests
       timeout-minutes: 40
       env:
         PYTHONUTF8: 1
       run: >
-        pytest -n=logical --timeout=300 --durations=100
-        --cov --cov-report=xml --cov-report=term
+        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
       with:
@@ -66,7 +122,7 @@ jobs:
     runs-on: ubuntu-latest
     steps:
     - name: Slack Notification
-      uses: rtCamp/[email protected].0
+      uses: rtCamp/[email protected].1
       env:
         SLACK_WEBHOOK: ${{ secrets.SLACK_WEBHOOK }}
         SLACK_COLOR: ${{ job.status }}

.pre-commit-config.yaml

@@ -6,7 +6,7 @@ ci:
 
 repos:
   - repo: https://github.com/pre-commit/pre-commit-hooks
-    rev: v4.6.0
+    rev: v5.0.0
     hooks:
       - id: check-added-large-files
       - id: check-case-conflict
@@ -22,7 +22,7 @@ repos:
       - id: sort-simple-yaml
       - id: trailing-whitespace
   - repo: https://github.com/astral-sh/ruff-pre-commit
-    rev: 'v0.5.4'
+    rev: 'v0.7.0'
     hooks:
       - id: ruff
         args: [--fix, --exit-non-zero-on-fix]
@@ -50,7 +50,7 @@ repos:
         language: system
         name: DVC pre-commit
         stages:
-          - commit
+          - pre-commit
         verbose: true
         require_serial: true
       - args:
@@ -61,7 +61,7 @@ repos:
         language: system
         name: DVC pre-push
         stages:
-          - push
+          - pre-push
         require_serial: true
       - always_run: true
         args:

README.rst

@@ -1,4 +1,6 @@
-|Banner|
+🚀 Check out our new product `DataChain <https://github.com/iterative/datachain>`_ (and give it a ⭐!) if you need to version and process a large number of files. Contact us at [email protected] to discuss commercial solutions and support for AI reproducibility and data management scenarios.
+
+--------------------------
 
 `Website <https://dvc.org>`_
 • `Docs <https://dvc.org/doc>`_

dvc/commands/dag.py

@@ -17,7 +17,7 @@ def _show_ascii(graph: "DiGraph"):
 
     ret = []
     for pipeline in pipelines:
-        ret.append(draw(pipeline.nodes, pipeline.edges))
+        ret.append(draw(pipeline.nodes, pipeline.edges))  # noqa: PERF401
 
     return "\n".join(ret)
 

dvc/commands/gc.py

@@ -54,7 +54,7 @@ class CmdGC(CmdBase):
             msg += " and all experiments"
 
         if self.args.not_in_remote:
-            msg += " that are present in the DVC remote"
+            msg += " that are not present in the DVC remote"
 
         if self.args.repos:
             msg += " of the current and the following repos:"

dvc/commands/ls/__init__.py

@@ -1,3 +1,5 @@
+from typing import Callable
+
 from dvc.cli import completion, formatter
 from dvc.cli.command import CmdBaseNoRepo
 from dvc.cli.utils import DictAction, append_doc_link
@@ -9,54 +11,181 @@ from dvc.ui import ui
 logger = logger.getChild(__name__)
 
 
-def _format_entry(entry, fmt):
-    from dvc.utils.humanize import naturalsize
+def _get_formatter(with_color: bool = False) -> Callable[[dict], str]:
+    def fmt(entry: dict) -> str:
+        return entry["path"]
 
-    size = entry.get("size")
-    if size is None:
-        size = ""
-    else:
-        size = naturalsize(size)
-    return size, fmt(entry)
-
-
-def show_entries(entries, with_color=False, with_size=False):
     if with_color:
         ls_colors = LsColors()
-        fmt = ls_colors.format
-    else:
+        return ls_colors.format
 
-        def fmt(entry):
-            return entry["path"]
+    return fmt
 
+
+def _format_entry(entry, name, with_size=True, with_hash=False):
+    from dvc.utils.humanize import naturalsize
+
+    ret = []
     if with_size:
-        ui.table([_format_entry(entry, fmt) for entry in entries])
+        size = entry.get("size")
+        if size is None:
+            size = ""
+        else:
+            size = naturalsize(size)
+        ret.append(size)
+    if with_hash:
+        md5 = entry.get("md5", "")
+        ret.append(md5)
+    ret.append(name)
+    return ret
+
+
+def show_entries(entries, with_color=False, with_size=False, with_hash=False):
+    fmt = _get_formatter(with_color)
+    if with_size or with_hash:
+        colalign = ("right",) if with_size else None
+        ui.table(
+            [
+                _format_entry(
+                    entry,
+                    fmt(entry),
+                    with_size=with_size,
+                    with_hash=with_hash,
+                )
+                for entry in entries
+            ],
+            colalign=colalign,
+        )
         return
 
     # NOTE: this is faster than ui.table for very large number of entries
     ui.write("\n".join(fmt(entry) for entry in entries))
 
 
+class TreePart:
+    Edge = "├── "
+    Line = "│   "
+    Corner = "└── "
+    Blank = "    "
+
+
+def _build_tree_structure(
+    entries, with_color=False, with_size=False, with_hash=False, _depth=0, _prefix=""
+):
+    rows = []
+    fmt = _get_formatter(with_color)
+
+    num_entries = len(entries)
+    for i, (name, entry) in enumerate(entries.items()):
+        # show full path for root, otherwise only show the name
+        if _depth > 0:
+            entry["path"] = name
+
+        is_last = i >= num_entries - 1
+        tree_part = ""
+        if _depth > 0:
+            tree_part = TreePart.Corner if is_last else TreePart.Edge
+
+        row = _format_entry(
+            entry,
+            _prefix + tree_part + fmt(entry),
+            with_size=with_size,
+            with_hash=with_hash,
+        )
+        rows.append(row)
+
+        if contents := entry.get("contents"):
+            new_prefix = _prefix
+            if _depth > 0:
+                new_prefix += TreePart.Blank if is_last else TreePart.Line
+            new_rows = _build_tree_structure(
+                contents,
+                with_color=with_color,
+                with_size=with_size,
+                with_hash=with_hash,
+                _depth=_depth + 1,
+                _prefix=new_prefix,
+            )
+            rows.extend(new_rows)
+
+    return rows
+
+
+def show_tree(entries, with_color=False, with_size=False, with_hash=False):
+    import tabulate
+
+    rows = _build_tree_structure(
+        entries,
+        with_color=with_color,
+        with_size=with_size,
+        with_hash=with_hash,
+    )
+
+    colalign = ("right",) if with_size else None
+
+    _orig = tabulate.PRESERVE_WHITESPACE
+    tabulate.PRESERVE_WHITESPACE = True
+    try:
+        ui.table(rows, colalign=colalign)
+    finally:
+        tabulate.PRESERVE_WHITESPACE = _orig
+
+
 class CmdList(CmdBaseNoRepo):
-    def run(self):
+    def _show_tree(self):
+        from dvc.repo.ls import ls_tree
+
+        entries = ls_tree(
+            self.args.url,
+            self.args.path,
+            rev=self.args.rev,
+            dvc_only=self.args.dvc_only,
+            config=self.args.config,
+            remote=self.args.remote,
+            remote_config=self.args.remote_config,
+            maxdepth=self.args.level,
+        )
+        show_tree(
+            entries,
+            with_color=True,
+            with_size=self.args.size,
+            with_hash=self.args.show_hash,
+        )
+        return 0
+
+    def _show_list(self):
         from dvc.repo import Repo
 
-        try:
-            entries = Repo.ls(
-                self.args.url,
-                self.args.path,
-                rev=self.args.rev,
-                recursive=self.args.recursive,
-                dvc_only=self.args.dvc_only,
-                config=self.args.config,
-                remote=self.args.remote,
-                remote_config=self.args.remote_config,
+        entries = Repo.ls(
+            self.args.url,
+            self.args.path,
+            rev=self.args.rev,
+            recursive=self.args.recursive,
+            dvc_only=self.args.dvc_only,
+            config=self.args.config,
+            remote=self.args.remote,
+            remote_config=self.args.remote_config,
+            maxdepth=self.args.level,
+        )
+        if self.args.json:
+            ui.write_json(entries)
+        elif entries:
+            show_entries(
+                entries,
+                with_color=True,
+                with_size=self.args.size,
+                with_hash=self.args.show_hash,
             )
-            if self.args.json:
-                ui.write_json(entries)
-            elif entries:
-                show_entries(entries, with_color=True, with_size=self.args.size)
-            return 0
+        return 0
+
+    def run(self):
+        if self.args.tree and self.args.json:
+            raise DvcException("Cannot use --tree and --json options together.")
+
+        try:
+            if self.args.tree:
+                return self._show_tree()
+            return self._show_list()
         except FileNotFoundError:
             logger.exception("")
             return 1
@@ -85,6 +214,19 @@ def add_parser(subparsers, parent_parser):
         action="store_true",
         help="Recursively list files.",
     )
+    list_parser.add_argument(
+        "-T",
+        "--tree",
+        action="store_true",
+        help="Recurse into directories as a tree.",
+    )
+    list_parser.add_argument(
+        "-L",
+        "--level",
+        metavar="depth",
+        type=int,
+        help="Limit the depth of recursion.",
+    )
     list_parser.add_argument(
         "--dvc-only", action="store_true", help="Show only DVC outputs."
     )
@@ -123,6 +265,12 @@ def add_parser(subparsers, parent_parser):
         ),
     )
     list_parser.add_argument("--size", action="store_true", help="Show sizes.")
+    list_parser.add_argument(
+        "--show-hash",
+        help="Display hash value for each item.",
+        action="store_true",
+        default=False,
+    )
     list_parser.add_argument(
         "path",
         nargs="?",

dvc/commands/ls/ls_colors.py

@@ -32,7 +32,9 @@ class LsColors:
         if entry.get("isexec", False):
             return self._format(text, code="ex")
 
-        _, ext = os.path.splitext(text)
+        stem, ext = os.path.splitext(text)
+        if not ext and stem.startswith("."):
+            ext = stem
         return self._format(text, ext=ext)
 
     def _format(self, text, code=None, ext=None):

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/db.py

@@ -44,6 +44,10 @@ class AbstractDependency(Dependency):
         self.def_path = None  # type: ignore[assignment]
         self.info = info or {}
 
+    @property
+    def is_in_repo(self):
+        return False
+
 
 class DbDependency(AbstractDependency):
     PARAM_CONNECTION = "connection"
@@ -70,8 +74,8 @@ class DbDependency(AbstractDependency):
         return self.db_info.get(self.PARAM_QUERY) or self.db_info.get(self.PARAM_TABLE)
 
     def __repr__(self):
-        return "{}:{}".format(
-            self.__class__.__name__, "".join(f"{k}=={v}" for k, v in self.info.items())
+        return "{}: {}".format(
+            self.__class__.__name__, "".join(f"{k}={v}" for k, v in self.info.items())
         )
 
     def __str__(self):

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:
-            super().download(to=to, jobs=jobs)
+        hashes: list[tuple[str, HashInfo, dict[str, Any]]] = []
+        for src_path, dest_path, maybe_info in files:
+            try:
+                info = maybe_info or 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,4 +1,5 @@
 import glob
+from itertools import repeat
 from typing import Optional
 from urllib.parse import urlparse
 
@@ -47,28 +48,12 @@ known_implementations.update(
 
 def download(
     fs: "FileSystem", fs_path: str, to: str, jobs: Optional[int] = None
-) -> int:
+) -> list[tuple[str, str, Optional[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 +63,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, repeat(None)))
 
 
 def parse_external_url(url, fs_config=None, config=None):

dvc/fs/dvc.py

@@ -4,22 +4,28 @@ import ntpath
 import os
 import posixpath
 import threading
-from collections import deque
-from contextlib import ExitStack, suppress
+from collections import defaultdict, deque
+from contextlib import ExitStack, nullcontext, 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
 
 if TYPE_CHECKING:
+    from contextlib import AbstractContextManager
+
     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 +480,127 @@ 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, PLR0912
+        self,
+        rpath,
+        lpath,
+        recursive=False,
+        callback=DEFAULT_CALLBACK,
+        maxdepth=None,
+        batch_size=None,
+        **kwargs,
+    ) -> list[tuple[str, str, Optional[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, None)]
+
+        result: list[tuple[str, str, Optional[dict]]] = []
+        _dirs: list[str] = []
+        _files: dict[FileSystem, list[tuple[str, str, Optional[dict]]]]
+        _files = defaultdict(list)
+
+        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():
+                dvc_info = info.get("dvc_info")
+                fs_info = info.get("fs_info")
+                if dvc_fs and dvc_info and not fs_info:
+                    fs = dvc_fs
+                    fs_path = dvc_info["name"]
+                else:
+                    fs = self.repo.fs
+                    fs_path = fs_info["name"]
+
+                src_path = f"{root}{self.sep}{name}"
+                dest_path = f"{dest_root}{os.path.sep}{name}"
+                _files[fs].append((fs_path, dest_path, dvc_info))
+                result.append((src_path, dest_path, info))
+
+        os.makedirs(lpath, exist_ok=True)
+        for d in _dirs:
+            os.makedirs(d, exist_ok=True)
+
+        def get_file(arg: tuple[FileSystem, tuple[str, str, Optional[dict]]]):
+            fs, (src, dest, info) = arg
+            kw = kwargs
+            if isinstance(fs, DataFileSystem):
+                kw = kw | {"info": info}
+            return fs.get_file(src, dest, callback=callback, **kw)
+
+        if batch_size == 1:
+            ctx: AbstractContextManager = nullcontext()
+            map_fn: Callable = map
+        else:
+            ctx = ThreadPoolExecutor(max_workers=batch_size)
+            map_fn = ctx.imap_unordered
+
+        with ctx:
+            it = ((fs, f) for fs, files in _files.items() for f in files)
+            deque(map_fn(get_file, it), maxlen=0)
+        return result
+
     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 +609,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 +677,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[tuple[str, str, Optional[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/info.py

@@ -123,7 +123,7 @@ def _get_supported_remotes():
         if not fs_cls.get_missing_deps():
             dependencies = []
             for requirement in fs_cls.REQUIRES:
-                dependencies.append(
+                dependencies.append(  # noqa: PERF401
                     f"{requirement} = {importlib_metadata.version(requirement)}"
                 )
 

dvc/lock.py

@@ -4,6 +4,7 @@ import hashlib
 import os
 from abc import ABC, abstractmethod
 from datetime import timedelta
+from typing import Optional, Union
 
 import flufl.lock
 import zc.lockfile
@@ -174,9 +175,9 @@ class HardlinkLock(flufl.lock.Lock, LockBase):
         self._owned = True
         self._retry_errnos = []
 
-    def lock(self):
+    def lock(self, timeout: Optional[Union[timedelta, int]] = None):
         try:
-            super().lock(timedelta(seconds=DEFAULT_TIMEOUT))
+            super().lock(timeout or timedelta(seconds=DEFAULT_TIMEOUT))
         except flufl.lock.TimeOutError:
             raise LockError(FAILED_TO_LOCK_MESSAGE)  # noqa: B904
 

dvc/output.py

@@ -542,7 +542,9 @@ class Output:
             desc=f"Collecting files and computing hashes in {self}",
             disable=no_progress_bar,
         ) as pb:
-            return build(*args, callback=pb.as_callback(), **kwargs)
+            kwargs["callback"] = pb.as_callback()
+            kwargs.setdefault("checksum_jobs", self.fs.hash_jobs)
+            return build(*args, **kwargs)
 
     def _get_hash_meta(self):
         if self.use_cache:
@@ -730,6 +732,7 @@ class Output:
         from dvc_data.hashfile.checkout import LinkError, PromptError
 
         kwargs.setdefault("ignore", self.dvcignore)
+        kwargs.setdefault("checksum_jobs", self.fs.hash_jobs)
         try:
             return checkout(*args, **kwargs)
         except PromptError as exc:
@@ -749,11 +752,9 @@ class Output:
             granular = (
                 self.is_dir_checksum and filter_info and filter_info != self.fs_path
             )
-            # NOTE: trying to use hardlink during transfer only if we will be
-            # relinking later
-            hardlink = relink
+            hardlink = relink and next(iter(self.cache.cache_types), None) == "hardlink"
             if granular:
-                obj = self._commit_granular_dir(filter_info, hardlink)
+                obj = self._commit_granular_dir(filter_info, hardlink=hardlink)
             else:
                 staging, _, obj = self._build(
                     self.cache,
@@ -786,6 +787,7 @@ class Output:
                         state=self.repo.state,
                         prompt=prompt.confirm,
                         progress_callback=cb,
+                        old=obj,
                     )
                 self.set_exec()
 
@@ -1398,12 +1400,14 @@ class Output:
 
         assert staging
         assert obj.hash_info
+
+        hardlink = relink and next(iter(self.cache.cache_types), None) == "hardlink"
         with TqdmCallback(desc=f"Adding {self} to cache", unit="file") as cb:
             otransfer(
                 staging,
                 self.cache,
                 {obj.hash_info},
-                hardlink=relink,
+                hardlink=hardlink,
                 shallow=False,
                 callback=cb,
             )
@@ -1421,6 +1425,7 @@ class Output:
                     state=self.repo.state,
                     prompt=prompt.confirm,
                     progress_callback=callback,
+                    old=obj,
                 )
             self.set_exec()
         return obj

dvc/parsing/context.py

@@ -176,8 +176,9 @@ class Container(Node, ABC):
             return value
         if isinstance(value, (list, dict)):
             assert meta
-            container = CtxDict if isinstance(value, dict) else CtxList
-            return container(value, meta=meta)
+            if isinstance(value, dict):
+                return CtxDict(value, meta=meta)
+            return CtxList(value, meta=meta)
         msg = f"Unsupported value of type '{type(value).__name__}' in '{meta}'"
         raise TypeError(msg)
 

dvc/render/converter/vega.py

@@ -23,7 +23,7 @@ def _lists(blob: Union[dict, list]) -> Iterable[list]:
     if isinstance(blob, list):
         yield blob
     else:
-        for _, value in blob.items():
+        for value in blob.values():
             if isinstance(value, dict):
                 yield from _lists(value)
             elif isinstance(value, list):

dvc/repo/__init__.py

@@ -147,7 +147,7 @@ class Repo:
     ):
         from dvc.cachemgr import CacheManager
         from dvc.data_cloud import DataCloud
-        from dvc.fs import GitFileSystem, LocalFileSystem, localfs
+        from dvc.fs import GitFileSystem, LocalFileSystem
         from dvc.lock import LockNoop, make_lock
         from dvc.repo.artifacts import Artifacts
         from dvc.repo.datasets import Datasets
@@ -161,7 +161,7 @@ class Repo:
 
         self.url = url
         self._fs_conf = {"repo_factory": repo_factory}
-        self._fs = fs or localfs
+        self._fs = fs or LocalFileSystem()
         self._scm = scm
         self._config = config
         self._remote = remote
@@ -206,6 +206,11 @@ class Repo:
                     friendly=True,
                 )
                 os.makedirs(self.site_cache_dir, exist_ok=True)
+                if not fs and (
+                    checksum_jobs := self.config["core"].get("checksum_jobs")
+                ):
+                    self.fs.hash_jobs = checksum_jobs
+
                 self.state = State(self.root_dir, self.site_cache_dir, self.dvcignore)
             else:
                 self.lock = LockNoop()

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/repo/datasets.py

@@ -76,7 +76,7 @@ class SerDe:
         return asdict(self, filter=filter_defaults, value_serializer=value_serializer)
 
     @classmethod
-    def from_dict(cls: type["Self"], d: dict[str, Any]) -> "Self":
+    def from_dict(cls, d: dict[str, Any]) -> "Self":
         _fields = fields(cast("type[AttrsInstance]", cls))
         kwargs = {f.name: d[f.name] for f in _fields if f.name in d}
         return cls(**kwargs)

dvc/repo/experiments/__init__.py

@@ -205,7 +205,7 @@ class Experiments:
         Experiment will be derived from the current workspace.
         """
 
-        name = kwargs.get("name", None)
+        name = kwargs.get("name")
         baseline_sha = kwargs.get("baseline_rev") or self.repo.scm.get_rev()
 
         if name:
@@ -249,7 +249,7 @@ class Experiments:
         exec_results = queue.reproduce(copy_paths=copy_paths, message=message)
 
         results: dict[str, str] = {}
-        for _, exp_result in exec_results.items():
+        for exp_result in exec_results.values():
             results.update(exp_result)
         return results
 

dvc/repo/experiments/executor/base.py

@@ -288,7 +288,7 @@ class BaseExecutor(ABC):
             stages = []
             if targets:
                 for target in targets:
-                    stages.append(
+                    stages.append(  # noqa: PERF401
                         dvc.commit(
                             target, recursive=recursive, force=True, relink=False
                         )

dvc/repo/experiments/pull.py

@@ -55,7 +55,7 @@ def pull(  # noqa: C901
         rev_dict = iter_revs(repo.scm, rev, num)
         rev_set = set(rev_dict.keys())
         ref_info_dict = exp_refs_by_baseline(repo.scm, rev_set, git_remote)
-        for _, ref_info_list in ref_info_dict.items():
+        for ref_info_list in ref_info_dict.values():
             exp_ref_set.update(ref_info_list)
 
     pull_result = _pull(repo, git_remote, exp_ref_set, force)

dvc/repo/experiments/push.py

@@ -78,7 +78,7 @@ def exp_refs_from_rev(scm: "Git", rev: list[str], num: int = 1) -> set["ExpRefIn
     rev_dict = iter_revs(scm, rev, num)
     rev_set = set(rev_dict.keys())
     ref_info_dict = exp_refs_by_baseline(scm, rev_set)
-    for _, ref_info_list in ref_info_dict.items():
+    for ref_info_list in ref_info_dict.values():
         exp_ref_set.update(ref_info_list)
     return exp_ref_set
 

dvc/repo/experiments/queue/celery.py

@@ -323,7 +323,7 @@ class LocalCeleryQueue(BaseStashQueue):
     def _get_running_task_ids(self) -> set[str]:
         running_task_ids: set[str] = set()
         active_workers = self.worker_status()
-        for _, tasks in active_workers.items():
+        for tasks in active_workers.values():
             task = first(tasks)
             if task:
                 running_task_ids.add(task["id"])

dvc/repo/experiments/queue/utils.py

@@ -25,7 +25,7 @@ def get_remote_executor_refs(scm: "Git", remote_url: str) -> list[str]:
     refs = []
     for ref in iter_remote_refs(scm, remote_url, base=EXPS_NAMESPACE):
         if not ref.startswith(EXEC_NAMESPACE) and ref != EXPS_STASH:
-            refs.append(ref)
+            refs.append(ref)  # noqa: PERF401
     return refs
 
 

dvc/repo/experiments/remove.py

@@ -98,7 +98,7 @@ def _resolve_exp_by_baseline(
     rev_dict = iter_revs(repo.scm, rev, num)
     rev_set = set(rev_dict.keys())
     ref_info_dict = exp_refs_by_baseline(repo.scm, rev_set, git_remote)
-    for _, ref_info_list in ref_info_dict.items():
+    for ref_info_list in ref_info_dict.values():
         for ref_info in ref_info_list:
             commit_ref_dict[ref_info.name] = ref_info
     return commit_ref_dict

dvc/repo/experiments/serialize.py

@@ -50,7 +50,12 @@ class SerializableExp:
         Params, metrics, deps, outs are filled via repo fs/index, all other fields
         should be passed via kwargs.
         """
-        from dvc.dependency import ParamsDependency, RepoDependency
+        from dvc.dependency import (
+            DatasetDependency,
+            DbDependency,
+            ParamsDependency,
+            RepoDependency,
+        )
 
         rev = rev or repo.get_rev()
         assert rev
@@ -68,7 +73,10 @@ class SerializableExp:
                     nfiles=dep.meta.nfiles if dep.meta else None,
                 )
                 for dep in repo.index.deps
-                if not isinstance(dep, (ParamsDependency, RepoDependency))
+                if not isinstance(
+                    dep,
+                    (ParamsDependency, RepoDependency, DatasetDependency, DbDependency),
+                )
             },
             outs={
                 relpath(out.fs_path, repo.root_dir): ExpOut(

dvc/repo/imp_db.py

@@ -4,7 +4,6 @@ from funcy import compact
 
 from dvc.exceptions import OutputDuplicationError
 from dvc.repo.scm_context import scm_context
-from dvc.ui import ui
 from dvc.utils import resolve_output, resolve_paths
 
 if TYPE_CHECKING:
@@ -25,11 +24,6 @@ def imp_db(
     force: bool = False,
     connection: Optional[str] = None,
 ):
-    ui.warn("WARNING: import-db is an experimental feature.")
-    ui.warn(
-        "The functionality may change or break without notice, "
-        "which could lead to unexpected behavior."
-    )
     assert sql or table
     assert output_format in ("csv", "json")
 

dvc/repo/ls.py

@@ -4,8 +4,44 @@ from typing import TYPE_CHECKING, Any, Optional, Union
 if TYPE_CHECKING:
     from dvc.fs.dvc import DVCFileSystem
 
+
+def _open_repo(
+    url: str,
+    rev: Optional[str] = None,
+    config: Union[None, dict[str, Any], str] = None,
+    remote: Optional[str] = None,
+    remote_config: Optional[dict] = None,
+):
+    from dvc.config import Config
+
     from . import Repo
 
+    if config and not isinstance(config, dict):
+        config_dict = Config.load_file(config)
+    else:
+        config_dict = None
+
+    return Repo.open(
+        url,
+        rev=rev,
+        subrepos=True,
+        uninitialized=True,
+        config=config_dict,
+        remote=remote,
+        remote_config=remote_config,
+    )
+
+
+def _adapt_info(info: dict[str, Any]) -> dict[str, Any]:
+    dvc_info = info.get("dvc_info", {})
+    return {
+        "isout": dvc_info.get("isout", False),
+        "isdir": info["type"] == "directory",
+        "isexec": info.get("isexec", False),
+        "size": info.get("size"),
+        "md5": dvc_info.get("md5") or dvc_info.get("md5-dos2unix"),
+    }
+
 
 def ls(
     url: str,
@@ -16,6 +52,7 @@ def ls(
     config: Union[None, dict[str, Any], str] = None,
     remote: Optional[str] = None,
     remote_config: Optional[dict] = None,
+    maxdepth: Optional[int] = None,
 ):
     """Methods for getting files and outputs for the repo.
 
@@ -41,60 +78,58 @@ def ls(
             "isexec": bool,
         }
     """
-    from dvc.config import Config
-
-    from . import Repo
-
-    if config and not isinstance(config, dict):
-        config_dict = Config.load_file(config)
-    else:
-        config_dict = None
-
-    with Repo.open(
-        url,
-        rev=rev,
-        subrepos=True,
-        uninitialized=True,
-        config=config_dict,
-        remote=remote,
-        remote_config=remote_config,
-    ) as repo:
+    with _open_repo(url, rev, config, remote, remote_config) as repo:
         path = path or ""
+        fs: DVCFileSystem = repo.dvcfs
+        fs_path = fs.from_os_path(path)
+        return _ls(fs, fs_path, recursive, dvc_only, maxdepth)
 
-        ret = _ls(repo, path, recursive, dvc_only)
 
-        ret_list = []
-        for path, info in ret.items():
-            info["path"] = path
-            ret_list.append(info)
-        ret_list.sort(key=lambda f: f["path"])
-        return ret_list
+def ls_tree(
+    url: str,
+    path: Optional[str] = None,
+    rev: Optional[str] = None,
+    dvc_only: bool = False,
+    config: Union[None, dict[str, Any], str] = None,
+    remote: Optional[str] = None,
+    remote_config: Optional[dict] = None,
+    maxdepth: Optional[int] = None,
+):
+    with _open_repo(url, rev, config, remote, remote_config) as repo:
+        path = path or ""
+        fs: DVCFileSystem = repo.dvcfs
+        fs_path = fs.from_os_path(path)
+        return _ls_tree(fs, fs_path, dvc_only, maxdepth)
 
 
 def _ls(
-    repo: "Repo",
+    fs: "DVCFileSystem",
     path: str,
     recursive: Optional[bool] = None,
     dvc_only: bool = False,
+    maxdepth: Optional[int] = None,
 ):
-    fs: DVCFileSystem = repo.dvcfs
-    fs_path = fs.from_os_path(path)
-
-    fs_path = fs.info(fs_path)["name"]
+    fs_path = fs.info(path)["name"]
 
     infos = {}
-    if fs.isfile(fs_path):
-        infos[os.path.basename(path)] = fs.info(fs_path)
+
+    # ignore maxdepth only if recursive is not set
+    maxdepth = maxdepth if recursive else None
+    if maxdepth == 0 or fs.isfile(fs_path):
+        infos[os.path.basename(path) or os.curdir] = fs.info(fs_path)
     else:
         for root, dirs, files in fs.walk(
-            fs_path, dvcfiles=True, dvc_only=dvc_only, detail=True
+            fs_path,
+            dvcfiles=True,
+            dvc_only=dvc_only,
+            detail=True,
+            maxdepth=maxdepth,
         ):
-            if not recursive:
-                files.update(dirs)
-
             parts = fs.relparts(root, fs_path)
             if parts == (".",):
                 parts = ()
+            if not recursive or (maxdepth and len(parts) >= maxdepth - 1):
+                files.update(dirs)
 
             for name, entry in files.items():
                 infos[os.path.join(*parts, name)] = entry
@@ -102,14 +137,37 @@ def _ls(
             if not recursive:
                 break
 
-    ret = {}
-    for name, info in infos.items():
-        dvc_info = info.get("dvc_info", {})
-        ret[name] = {
-            "isout": dvc_info.get("isout", False),
-            "isdir": info["type"] == "directory",
-            "isexec": info.get("isexec", False),
-            "size": info.get("size"),
-        }
+    ret_list = []
+    for p, info in sorted(infos.items(), key=lambda x: x[0]):
+        _info = _adapt_info(info)
+        _info["path"] = p
+        ret_list.append(_info)
+    return ret_list
 
+
+def _ls_tree(
+    fs, path, dvc_only: bool = False, maxdepth: Optional[int] = None, _info=None
+):
+    ret = {}
+    info = _info or fs.info(path)
+
+    path = info["name"].rstrip(fs.sep) or os.curdir
+    name = path.rsplit("/", 1)[-1]
+    ls_info = _adapt_info(info)
+    ls_info["path"] = path
+
+    recurse = maxdepth is None or maxdepth > 0
+    if recurse and info["type"] == "directory":
+        infos = fs.ls(path, dvcfiles=True, dvc_only=dvc_only, detail=True)
+        infos.sort(key=lambda f: f["name"])
+        maxdepth = maxdepth - 1 if maxdepth is not None else None
+        contents = {}
+        for info in infos:
+            d = _ls_tree(
+                fs, info["name"], dvc_only=dvc_only, maxdepth=maxdepth, _info=info
+            )
+            contents.update(d)
+        ls_info["contents"] = contents
+
+    ret[name] = ls_info
     return ret

dvc/stage/__init__.py

@@ -83,7 +83,7 @@ class RawData:
 def create_stage(cls: type[_T], repo, path, **kwargs) -> _T:
     from dvc.dvcfile import check_dvcfile_path
 
-    wdir = os.path.abspath(kwargs.get("wdir", None) or os.curdir)
+    wdir = os.path.abspath(kwargs.get("wdir") or os.curdir)
     path = os.path.abspath(path)
 
     check_dvcfile_path(repo, path)
@@ -435,7 +435,7 @@ class Stage(params.StageParams):
         elif not self.changed(True, upstream) and pull:
             try:
                 logger.info("Pulling data for %s", self)
-                self.repo.pull(self.addressing, jobs=kwargs.get("jobs", None))
+                self.repo.pull(self.addressing, jobs=kwargs.get("jobs"))
                 self.checkout()
                 return None
             except CheckoutError:
@@ -609,7 +609,7 @@ class Stage(params.StageParams):
             self.remove_outs(ignore_remove=False, force=False)
 
         if (self.is_import and not self.frozen) or self.is_partial_import:
-            self._sync_import(dry, force, kwargs.get("jobs", None), no_download)
+            self._sync_import(dry, force, kwargs.get("jobs"), no_download)
         elif not self.frozen and self.cmd:
             self._run_stage(dry, force, **kwargs)
         elif not dry:

dvc/stage/run.py

@@ -154,7 +154,11 @@ def cmd_run(stage, dry=False, run_env=None):
 
 
 def _pull_missing_deps(stage):
+    from dvc.dependency import DatasetDependency, DbDependency
+
     for dep in stage.deps:
+        if isinstance(dep, (DatasetDependency, DbDependency)):
+            continue
         if not dep.exists:
             stage.repo.pull(dep.def_path)
 

dvc/stage/utils.py

@@ -302,9 +302,9 @@ def _get_stage_files(stage: "Stage") -> list[str]:
             and dep.is_in_repo
             and not stage.repo.dvcfs.isdvc(stage.repo.dvcfs.from_os_path(str(dep)))
         ):
-            ret.append(relpath(dep.fs_path))
+            ret.append(relpath(dep.fs_path))  # noqa: PERF401
 
     for out in stage.outs:
         if not out.use_scm_ignore and out.is_in_repo:
-            ret.append(relpath(out.fs_path))
+            ret.append(relpath(out.fs_path))  # noqa: PERF401
     return ret

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

@@ -1,2 +1,19 @@
-def test_add(bench_dvc, tmp_dir, dvc, dataset):
-    bench_dvc("add", dataset)
+from .test_checkout import _skip_unsupported_link
+
+
+def generate_test(*, link_type="copy"):
+    def _test_add(bench_dvc, tmp_dir, dvc, dataset):
+        _skip_unsupported_link((tmp_dir / ".dvc" / "cache"), tmp_dir, link_type)
+
+        with dvc.config.edit() as conf:
+            conf["cache"]["type"] = link_type
+
+        bench_dvc("add", dataset)
+        bench_dvc("add", dataset, name="noop")
+
+    return _test_add
+
+
+test_add_copy = generate_test(link_type="copy")
+test_add_symlink = generate_test(link_type="symlink")
+test_add_hardlink = generate_test(link_type="hardlink")

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

@@ -1,7 +1,8 @@
 import os
 import shutil
+import sys
 from pathlib import Path
-from subprocess import check_output
+from subprocess import check_call, check_output
 from typing import Optional
 
 import pytest
@@ -23,13 +24,14 @@ class VirtualEnv:
         self.bin = self.path / ("Scripts" if os.name == "nt" else "bin")
 
     def create(self) -> None:
-        import virtualenv
+        check_call([sys.executable, "-m", "uv", "venv", self.path])  # noqa:S603
 
-        virtualenv.cli_run([os.fspath(self.path)])
+    def install(self, *packages: str) -> None:
+        check_call([sys.executable, "-m", "uv", "pip", "install", *packages])  # noqa: S603
 
     def run(self, cmd: str, *args: str, env: Optional[dict[str, str]] = None) -> None:
         exe = self.which(cmd)
-        check_output([exe, *args], env=env)  # noqa: S603
+        check_call([exe, *args], env=env)  # noqa: S603
 
     def which(self, cmd: str) -> str:
         assert self.bin.exists()
@@ -58,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
@@ -71,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)
@@ -88,61 +92,55 @@ def make_dvc_bin(
     dvc_rev,
     dvc_venvs,
     make_dvc_venv,
-    dvc_git_repo,
+    dvc_repo,
     bench_config,
     request,
 ):
     if dvc_rev:
-        venv = dvc_venvs.get(dvc_rev)
+        venv: VirtualEnv = dvc_venvs.get(dvc_rev)
         if not venv:
             venv = make_dvc_venv(dvc_rev)
-            venv.run("pip", "install", "-U", "pip")
             if bench_config.dvc_install_deps:
-                egg = f"dvc[{bench_config.dvc_install_deps}]"
+                pkg = f"dvc[{bench_config.dvc_install_deps}]"
             else:
-                egg = "dvc"
-            venv.run("pip", "install", f"git+file://{dvc_git_repo}@{dvc_rev}#egg={egg}")
-            if dvc_rev in ["2.18.1", "2.11.0", "2.6.3"]:
-                venv.run("pip", "install", "fsspec==2022.11.0")
+                pkg = "dvc"
+            packages = [f"{pkg} @ git+file://{dvc_repo}@{dvc_rev}"]
             try:
                 if version.Version(dvc_rev) < version.Version("3.50.3"):
-                    venv.run("pip", "install", "pygit2==1.14.1")
+                    packages.append("pygit2==1.14.1")
             except version.InvalidVersion:
                 pass
+            venv.install(*packages)
+
             dvc_venvs[dvc_rev] = venv
         dvc_bin = venv.which("dvc")
     else:
         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
 
 
@@ -169,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
@@ -190,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")
 
@@ -256,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/path_info.py

@@ -32,7 +32,7 @@ class PathInfo(pathlib.PurePath, _BasePath):
 
         def __new__(cls, *args):
             if cls is PathInfo:
-                cls = WindowsPathInfo if os.name == "nt" else PosixPathInfo
+                cls = WindowsPathInfo if os.name == "nt" else PosixPathInfo  # noqa: PLW0642
 
             return cls._from_parts(args)  # type: ignore[attr-defined]
 
@@ -79,13 +79,13 @@ class PathInfo(pathlib.PurePath, _BasePath):
             and self._cparts[:n] == other._cparts  # type: ignore[attr-defined]
         )
 
-    def relative_to(self, other):
+    def relative_to(self, other, *args, **kwargs):
         # pathlib relative_to raises exception when one path is not a direct
         # descendant of the other when os.path.relpath would return abspath.
         # For DVC PathInfo we only need the relpath behavior.
         # See: https://bugs.python.org/issue40358
         try:
-            path = super().relative_to(other)
+            path = super().relative_to(other, *args, **kwargs)
         except ValueError:
             path = relpath(self, other)
         return self.__class__(path)

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)

dvc/testing/tmp_dir.py

@@ -71,7 +71,7 @@ class TmpDir(pathlib.Path):
 
         def __new__(cls, *args, **kwargs):
             if cls is TmpDir:
-                cls = WindowsTmpDir if os.name == "nt" else PosixTmpDir
+                cls = WindowsTmpDir if os.name == "nt" else PosixTmpDir  # noqa: PLW0642
 
             # init parameter and `_init` method has been removed in Python 3.10.
             kw = {"init": False} if sys.version_info < (3, 10) else {}

dvc/ui/__init__.py

@@ -300,6 +300,7 @@ class Console:
         header_styles: Optional[Union[dict[str, "Styles"], Sequence["Styles"]]] = None,
         row_styles: Optional[Sequence["Styles"]] = None,
         borders: Union[bool, str] = False,
+        colalign: Optional[tuple[str, ...]] = None,
     ) -> None:
         from dvc.ui import table as t
 
@@ -321,7 +322,13 @@ class Console:
             return
 
         return t.plain_table(
-            self, data, headers, markdown=markdown, pager=pager, force=force
+            self,
+            data,
+            headers,
+            markdown=markdown,
+            pager=pager,
+            force=force,
+            colalign=colalign,
         )
 
     def status(self, status: str, **kwargs: Any) -> "Status":

dvc/ui/table.py

@@ -29,6 +29,7 @@ def plain_table(
     markdown: bool = False,
     pager: bool = False,
     force: bool = True,
+    colalign: Optional[tuple[str, ...]] = None,
 ) -> None:
     from funcy import nullcontext
     from tabulate import tabulate
@@ -40,6 +41,7 @@ def plain_table(
         disable_numparse=True,
         # None will be shown as "" by default, overriding
         missingval="-",
+        colalign=colalign,
     )
     if markdown:
         # NOTE: md table is incomplete without the trailing newline

dvc/utils/__init__.py

@@ -243,7 +243,7 @@ def env2bool(var, undefined=False):
     var = os.getenv(var, None)
     if var is None:
         return undefined
-    return bool(re.search("1|y|yes|true", var, flags=re.I))
+    return bool(re.search("1|y|yes|true", var, flags=re.IGNORECASE))
 
 
 def resolve_output(inp: str, out: Optional[str], force=False) -> str:
@@ -382,7 +382,7 @@ def glob_targets(targets, glob=True, recursive=True):
 
 def error_handler(func):
     def wrapper(*args, **kwargs):
-        onerror = kwargs.get("onerror", None)
+        onerror = kwargs.get("onerror")
         result = {}
 
         try:

dvc/utils/serialize/_common.py

@@ -61,7 +61,7 @@ def _load_data(
 ):
     open_fn = fs.open if fs else open
     encoding = "utf-8"
-    with open_fn(path, encoding=encoding, **kwargs) as fd:  # type: ignore[operator]
+    with open_fn(path, encoding=encoding, **kwargs) as fd:  # type: ignore[arg-type]
         with reraise(UnicodeDecodeError, EncodingError(path, encoding)):
             return parser(fd.read(), path)
 
@@ -74,7 +74,7 @@ def _dump_data(
     **dumper_args,
 ):
     open_fn = fs.open if fs else open
-    with open_fn(path, "w+", encoding="utf-8") as fd:  # type: ignore[operator]
+    with open_fn(path, "w+", encoding="utf-8") as fd:  # type: ignore[call-overload]
         dumper(data, fd, **dumper_args)
 
 

pyproject.toml

@@ -38,14 +38,14 @@ dependencies = [
     "distro>=1.3",
     "dpath<3,>=2.1.0",
     "dulwich",
-    "dvc-data>=3.15,<3.16",
+    "dvc-data>=3.16.2,<3.17",
     "dvc-http>=2.29.0",
     "dvc-objects",
     "dvc-render>=1.0.1,<2",
     "dvc-studio-client>=0.21,<1",
     "dvc-task>=0.3.0,<1",
     "flatten_dict<1,>=0.4.1",
-    "flufl.lock>=8.1.0,<9", # https://github.com/iterative/dvc/issues/9654
+    "flufl.lock>=8.1.0,<9",
     "fsspec>=2024.2.0",
     "funcy>=1.14",
     "grandalf<1,>=0.7",
@@ -65,7 +65,7 @@ dependencies = [
     "requests>=2.22",
     "rich>=12",
     "ruamel.yaml>=0.17.11",
-    "scmrepo>=3.3.4,<4",
+    "scmrepo>=3.3.8,<4",
     "shortuuid>=0.5",
     "shtab<2,>=1.3.4",
     "tabulate>=0.8.7",
@@ -83,7 +83,7 @@ gdrive = ["dvc-gdrive>=3,<4"]
 gs = ["dvc-gs>=3,<4"]
 hdfs = ["dvc-hdfs>=3,<4"]
 lint = [
-    "mypy==1.10.1",
+    "mypy==1.12.1",
     "pandas-stubs",
     "types-colorama",
     "types-psutil",
@@ -101,7 +101,7 @@ ssh_gssapi = ["dvc-ssh[gssapi]>=4,<5"]
 testing = [
   "pytest-benchmark[histogram]",
   "pytest-test-utils",
-  "virtualenv",
+  "uv",
 ]
 tests = [
     "dvc[testing]",
@@ -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",
@@ -264,21 +264,19 @@ show-fixes = true
 
 [tool.ruff.lint]
 ignore = [
-    "N818", "S101", "ISC001", "PT004", "PT007", "RET502", "RET503", "SIM105", "SIM108", "SIM117",
-    "TRY003", "TRY300", "PLR2004", "PLW2901", "LOG007",
+    "N818", "S101", "ISC001", "PT007", "RET502", "RET503", "SIM105", "SIM108", "SIM117",
+    "TRY003", "TRY300", "PERF203", "PLR2004", "PLW2901", "LOG007",
 ]
 select = [
     "F", "E", "W", "C90", "I", "N", "UP", "YTT", "ASYNC", "S", "BLE", "B", "A", "C4", "DTZ", "T10",
-    "EXE", "ISC", "ICN", "G", "INP", "PIE", "T20", "PYI", "PT", "Q", "RSE", "RET",
+    "EXE", "ISC", "ICN", "LOG", "G", "INP", "PIE", "T20", "PYI", "PT", "Q", "RSE", "RET",
     "SLOT", "SIM", "TID", "TCH", "ARG", "PGH", "PLC", "PLE", "PLR", "PLW", "TRY",
-    "FLY", "PERF101", "LOG", "RUF", "RUF022", "RUF023", "RUF024", "RUF025", "RUF026", "RUF027"
+    "FLY", "PERF", "FURB", "RUF", "RUF022", "RUF023", "RUF024", "RUF026", "RUF027", "RUF029", "RUF030",
 ]
 preview = true
 explicit-preview-rules = true
 
 [tool.ruff.lint.flake8-pytest-style]
-fixture-parentheses = false
-mark-parentheses = false
 parametrize-names-type = "csv"
 raises-extend-require-match-for = ["dvc.exceptions.DvcException", "dvc.scm.SCMError", "scmrepo.exceptions.SCMError"]
 
@@ -305,4 +303,4 @@ max-args = 10
 "dvc/commands/**" = ["N806"]
 "dvc/testing/**" = ["ARG002"]
 "dvc/testing/benchmarks/**" = ["ARG001"]
-"tests/**" = ["S", "ARG001", "ARG002", "TRY002", "TRY301"]
+"tests/**" = ["S", "ARG001", "ARG002", "TRY002", "TRY301", "PERF"]

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_add.py

@@ -305,20 +305,20 @@ def test_should_update_state_entry_for_directory_after_add(mocker, dvc, tmp_dir)
 
     ret = main(["add", "data"])
     assert ret == 0
-    assert file_md5_counter.mock.call_count == 4
+    assert file_md5_counter.mock.call_count == 3
 
     ret = main(["status"])
     assert ret == 0
-    assert file_md5_counter.mock.call_count == 5
+    assert file_md5_counter.mock.call_count == 4
 
     os.rename("data", "data.back")
     ret = main(["checkout"])
     assert ret == 0
-    assert file_md5_counter.mock.call_count == 6
+    assert file_md5_counter.mock.call_count == 5
 
     ret = main(["status"])
     assert ret == 0
-    assert file_md5_counter.mock.call_count == 7
+    assert file_md5_counter.mock.call_count == 6
 
 
 def test_add_commit(tmp_dir, dvc):
@@ -339,16 +339,16 @@ def test_should_collect_dir_cache_only_once(mocker, tmp_dir, dvc):
     counter = mocker.spy(dvc_data.hashfile.build, "_build_tree")
     ret = main(["add", "data"])
     assert ret == 0
+    assert counter.mock.call_count == 1
+
+    ret = main(["status"])
+    assert ret == 0
     assert counter.mock.call_count == 2
 
     ret = main(["status"])
     assert ret == 0
     assert counter.mock.call_count == 3
 
-    ret = main(["status"])
-    assert ret == 0
-    assert counter.mock.call_count == 4
-
 
 def test_should_place_stage_in_data_dir_if_repository_below_symlink(
     mocker, tmp_dir, dvc
@@ -754,6 +754,7 @@ def test_add_file_in_symlink_dir(make_tmp_dir, tmp_dir, dvc):
 def test_add_with_cache_link_error(tmp_dir, dvc, mocker, capsys):
     tmp_dir.gen("foo", "foo")
 
+    dvc.cache.local.cache_types = ["symlink", "hardlink"]
     mocker.patch("dvc_data.hashfile.checkout.test_links", return_value=[])
     dvc.add("foo")
     err = capsys.readouterr()[1]

tests/func/test_daemon.py

@@ -6,7 +6,7 @@ import sys
 from collections import defaultdict
 from collections.abc import Iterator
 from contextlib import contextmanager, suppress
-from http.server import HTTPServer, SimpleHTTPRequestHandler
+from http.server import BaseHTTPRequestHandler, HTTPServer
 from pathlib import Path
 from threading import Thread
 from typing import ClassVar
@@ -41,7 +41,7 @@ UPDATER_INFO_STR = json.dumps(UPDATER_INFO).encode("utf8")
 
 
 def make_request_handler():
-    class RequestHandler(SimpleHTTPRequestHandler):
+    class RequestHandler(BaseHTTPRequestHandler):
         # save requests count for each method
         hits: ClassVar[dict[str, int]] = defaultdict(int)
 
@@ -80,7 +80,10 @@ def server():
         thread = Thread(target=httpd.serve_forever)
         thread.daemon = True
         thread.start()
-        yield httpd
+        try:
+            yield httpd
+        finally:
+            httpd.shutdown()
 
 
 def test_analytics(tmp_path, server):
@@ -94,6 +97,11 @@ def test_analytics(tmp_path, server):
     }
     env.pop("DVC_TEST", None)
     env.pop("DVC_NO_ANALYTICS", None)
+    # The `iterative-telemetry` package calls `gh api` to generate a CI id.
+    # This might hang especially on Windows,
+    # possibly due to system load from the running tests.
+    # Removing the GITHUB_ACTIONS env var avoids calling `gh api`.
+    env.pop("GITHUB_ACTIONS", None)
 
     output = subprocess.check_output(
         [*_get_dvc_args(), "config", "-l", "-vv"],
@@ -101,18 +109,24 @@ def test_analytics(tmp_path, server):
         text=True,
     )
 
-    match = re.search(r".*Saving analytics report to (.*)", output, flags=re.M)
+    match = re.search(r".*Saving analytics report to (.*)", output, flags=re.MULTILINE)
     assert match, "no match for the report file"
     report_file = match.group(1).strip()
 
-    match = re.search(r".*Spawned .*analytics.* with pid (.*)", output, flags=re.M)
+    match = re.search(
+        r".*Spawned .*analytics.* with pid (.*)", output, flags=re.MULTILINE
+    )
     assert match, "no match for the pid"
     pid = int(match.group(1).strip())
 
     with suppress(psutil.NoSuchProcess):
         psutil.Process(pid).wait(timeout=10)
+
+    log_contents = logfile.read_text(encoding="utf8")
+    expected_line = (f"Process {pid} " if os.name != "nt" else "") + "exiting with 0"
+    assert expected_line in log_contents
+
     assert not os.path.exists(report_file)
-    assert f"Process {pid} exiting with 0" in logfile.read_text(encoding="utf8")
     assert server.RequestHandlerClass.hits == {"POST": 1}
 
 
@@ -136,13 +150,19 @@ def test_updater(tmp_dir, dvc, server):
         text=True,
     )
 
-    match = re.search(r".*Spawned .*updater.* with pid (.*)", output, flags=re.M)
+    match = re.search(
+        r".*Spawned .*updater.* with pid (.*)", output, flags=re.MULTILINE
+    )
     assert match, "no match for the pid"
     pid = int(match.group(1).strip())
 
     with suppress(psutil.NoSuchProcess):
         psutil.Process(pid).wait(timeout=10)
-    assert f"Process {pid} exiting with 0" in logfile.read_text(encoding="utf8")
+
+    log_contents = logfile.read_text(encoding="utf8")
+    expected_line = (f"Process {pid} " if os.name != "nt" else "") + "exiting with 0"
+    assert expected_line in log_contents
+
     assert server.RequestHandlerClass.hits == {"GET": 1}
     # check that the file is saved correctly
     updater_file = Path(dvc.tmp_dir) / Updater.UPDATER_FILE

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/func/test_ls.py

@@ -2,10 +2,12 @@ import os
 import shutil
 import textwrap
 from operator import itemgetter
+from os.path import join
 
 import pytest
 
 from dvc.repo import Repo
+from dvc.repo.ls import ls_tree
 from dvc.scm import CloneError
 
 FS_STRUCTURE = {
@@ -222,7 +224,12 @@ def test_ls_repo_with_missed_path_dvc_only(tmp_dir, dvc, scm):
     tmp_dir.dvc_gen(DVC_STRUCTURE, commit="dvc")
 
     with pytest.raises(FileNotFoundError):
-        Repo.ls(os.fspath(tmp_dir), path="missed_path", recursive=True, dvc_only=True)
+        Repo.ls(
+            os.fspath(tmp_dir),
+            path="missed_path",
+            recursive=True,
+            dvc_only=True,
+        )
 
 
 def test_ls_repo_with_removed_dvc_dir(tmp_dir, dvc, scm):
@@ -467,20 +474,49 @@ def test_ls_granular(erepo_dir, M):
 
     entries = Repo.ls(os.fspath(erepo_dir), os.path.join("dir", "subdir"))
     assert entries == [
-        {"isout": True, "isdir": False, "isexec": False, "path": "bar", "size": 3},
-        {"isout": True, "isdir": False, "isexec": False, "path": "foo", "size": 3},
+        {
+            "isout": True,
+            "isdir": False,
+            "isexec": False,
+            "path": "bar",
+            "size": 3,
+            "md5": "37b51d194a7513e45b56f6524f2d51f2",
+        },
+        {
+            "isout": True,
+            "isdir": False,
+            "isexec": False,
+            "path": "foo",
+            "size": 3,
+            "md5": "acbd18db4cc2f85cedef654fccc4a4d8",
+        },
     ]
 
     entries = Repo.ls(os.fspath(erepo_dir), "dir")
     assert entries == [
-        {"isout": True, "isdir": False, "isexec": False, "path": "1", "size": 1},
-        {"isout": True, "isdir": False, "isexec": False, "path": "2", "size": 1},
+        {
+            "isout": True,
+            "isdir": False,
+            "isexec": False,
+            "path": "1",
+            "size": 1,
+            "md5": "c4ca4238a0b923820dcc509a6f75849b",
+        },
+        {
+            "isout": True,
+            "isdir": False,
+            "isexec": False,
+            "path": "2",
+            "size": 1,
+            "md5": "c81e728d9d4c2f636f067f89cc14862c",
+        },
         {
             "isout": True,
             "isdir": True,
             "isexec": False,
             "path": "subdir",
             "size": M.instance_of(int),
+            "md5": None,
         },
     ]
 
@@ -506,14 +542,42 @@ def test_ls_target(erepo_dir, use_scm):
         return Repo.ls(os.fspath(erepo_dir), path)
 
     assert _ls(os.path.join("dir", "1")) == [
-        {"isout": isout, "isdir": False, "isexec": False, "path": "1", "size": 1}
+        {
+            "isout": isout,
+            "isdir": False,
+            "isexec": False,
+            "path": "1",
+            "size": 1,
+            "md5": "c4ca4238a0b923820dcc509a6f75849b" if not use_scm else None,
+        }
     ]
     assert _ls(os.path.join("dir", "subdir", "foo")) == [
-        {"isout": isout, "isdir": False, "isexec": False, "path": "foo", "size": 3}
+        {
+            "isout": isout,
+            "isdir": False,
+            "isexec": False,
+            "path": "foo",
+            "size": 3,
+            "md5": "acbd18db4cc2f85cedef654fccc4a4d8" if not use_scm else None,
+        }
     ]
     assert _ls(os.path.join("dir", "subdir")) == [
-        {"isdir": False, "isexec": 0, "isout": isout, "path": "bar", "size": 3},
-        {"isdir": False, "isexec": 0, "isout": isout, "path": "foo", "size": 3},
+        {
+            "isdir": False,
+            "isexec": 0,
+            "isout": isout,
+            "path": "bar",
+            "size": 3,
+            "md5": "37b51d194a7513e45b56f6524f2d51f2" if not use_scm else None,
+        },
+        {
+            "isdir": False,
+            "isexec": 0,
+            "isout": isout,
+            "path": "foo",
+            "size": 3,
+            "md5": "acbd18db4cc2f85cedef654fccc4a4d8" if not use_scm else None,
+        },
     ]
 
 
@@ -576,6 +640,7 @@ def test_broken_symlink(tmp_dir, dvc, M):
             "isexec": False,
             "path": ".dvcignore",
             "size": M.instance_of(int),
+            "md5": None,
         },
         {
             "isout": False,
@@ -583,6 +648,7 @@ def test_broken_symlink(tmp_dir, dvc, M):
             "isexec": False,
             "path": "link",
             "size": 0,
+            "md5": None,
         },
     ]
 
@@ -614,14 +680,23 @@ def test_ls_broken_dir(tmp_dir, dvc, M):
             "isout": False,
             "path": ".dvcignore",
             "size": M.instance_of(int),
+            "md5": None,
+        },
+        {
+            "isdir": True,
+            "isexec": False,
+            "isout": True,
+            "path": "broken",
+            "size": 3,
+            "md5": "630bd47b538d2a513c7d267d07e0bc44.dir",
         },
-        {"isdir": True, "isexec": False, "isout": True, "path": "broken", "size": 3},
         {
             "isdir": False,
             "isexec": False,
             "isout": False,
             "path": "broken.dvc",
             "size": M.instance_of(int),
+            "md5": None,
         },
         {
             "isdir": True,
@@ -629,6 +704,7 @@ def test_ls_broken_dir(tmp_dir, dvc, M):
             "isout": True,
             "path": "dir",
             "size": M.instance_of(int),
+            "md5": "91aaa9bb58b657d623ef143b195a67e4.dir",
         },
         {
             "isdir": False,
@@ -636,14 +712,23 @@ def test_ls_broken_dir(tmp_dir, dvc, M):
             "isout": False,
             "path": "dir.dvc",
             "size": M.instance_of(int),
+            "md5": None,
+        },
+        {
+            "isdir": False,
+            "isexec": False,
+            "isout": True,
+            "path": "foo",
+            "size": 3,
+            "md5": "acbd18db4cc2f85cedef654fccc4a4d8",
         },
-        {"isdir": False, "isexec": False, "isout": True, "path": "foo", "size": 3},
         {
             "isdir": False,
             "isexec": False,
             "isout": False,
             "path": "foo.dvc",
             "size": M.instance_of(int),
+            "md5": None,
         },
     ]
 
@@ -652,3 +737,264 @@ def test_ls_broken_dir(tmp_dir, dvc, M):
 
     with pytest.raises(DataIndexDirError):
         Repo.ls(os.fspath(tmp_dir), recursive=True)
+
+
+def test_ls_maxdepth(tmp_dir, scm, dvc):
+    tmp_dir.scm_gen(FS_STRUCTURE, commit="init")
+    tmp_dir.dvc_gen(DVC_STRUCTURE, commit="dvc")
+
+    files = Repo.ls(os.fspath(tmp_dir), "structure.xml", maxdepth=0, recursive=True)
+    match_files(files, ((("structure.xml",), True),))
+
+    files = Repo.ls(os.fspath(tmp_dir), maxdepth=0, recursive=True)
+    match_files(files, (((os.curdir,), False),))
+
+    files = Repo.ls(os.fspath(tmp_dir), maxdepth=1, recursive=True)
+    match_files(
+        files,
+        (
+            ((".dvcignore",), False),
+            ((".gitignore",), False),
+            (("README.md",), False),
+            (("structure.xml.dvc",), False),
+            (("model",), False),
+            (("data",), False),
+            (("structure.xml",), True),
+        ),
+    )
+    files = Repo.ls(os.fspath(tmp_dir), maxdepth=2, recursive=True)
+    match_files(
+        files,
+        (
+            ((".dvcignore",), False),
+            ((".gitignore",), False),
+            (("README.md",), False),
+            ((join("data", "subcontent"),), False),
+            ((join("model", ".gitignore"),), False),
+            ((join("model", "people.csv"),), True),
+            ((join("model", "people.csv.dvc"),), False),
+            ((join("model", "script.py"),), False),
+            ((join("model", "train.py"),), False),
+            (("structure.xml",), True),
+            (("structure.xml.dvc",), False),
+        ),
+    )
+
+    files = Repo.ls(os.fspath(tmp_dir), maxdepth=3, recursive=True)
+    match_files(
+        files,
+        (
+            ((".dvcignore",), False),
+            ((".gitignore",), False),
+            (("README.md",), False),
+            ((join("data", "subcontent", ".gitignore"),), False),
+            ((join("data", "subcontent", "data.xml"),), True),
+            ((join("data", "subcontent", "data.xml.dvc"),), False),
+            ((join("data", "subcontent", "statistics"),), False),
+            ((join("model", ".gitignore"),), False),
+            ((join("model", "people.csv"),), True),
+            ((join("model", "people.csv.dvc"),), False),
+            ((join("model", "script.py"),), False),
+            ((join("model", "train.py"),), False),
+            ((join("structure.xml"),), True),
+            ((join("structure.xml.dvc"),), False),
+        ),
+    )
+
+    files = Repo.ls(os.fspath(tmp_dir), maxdepth=4, recursive=True)
+    match_files(
+        files,
+        (
+            ((".dvcignore",), False),
+            ((".gitignore",), False),
+            (("README.md",), False),
+            ((join("data", "subcontent", ".gitignore"),), False),
+            ((join("data", "subcontent", "data.xml"),), True),
+            ((join("data", "subcontent", "data.xml.dvc"),), False),
+            ((join("data", "subcontent", "statistics", ".gitignore"),), False),
+            ((join("data", "subcontent", "statistics", "data.csv"),), True),
+            ((join("data", "subcontent", "statistics", "data.csv.dvc"),), False),
+            ((join("model", ".gitignore"),), False),
+            ((join("model", "people.csv"),), True),
+            ((join("model", "people.csv.dvc"),), False),
+            ((join("model", "script.py"),), False),
+            ((join("model", "train.py"),), False),
+            (("structure.xml",), True),
+            (("structure.xml.dvc",), False),
+        ),
+    )
+
+
+def _simplify_tree(files):
+    ret = {}
+    for path, info in files.items():
+        if content := info.get("contents"):
+            ret[path] = _simplify_tree(content)
+        else:
+            ret[path] = None
+    return ret
+
+
+def test_ls_tree(M, tmp_dir, scm, dvc):
+    tmp_dir.scm_gen(FS_STRUCTURE, commit="init")
+    tmp_dir.dvc_gen(DVC_STRUCTURE, commit="dvc")
+
+    files = ls_tree(os.fspath(tmp_dir), "structure.xml")
+    assert _simplify_tree(files) == {"structure.xml": None}
+
+    files = ls_tree(os.fspath(tmp_dir))
+
+    expected = {
+        ".": {
+            ".dvcignore": None,
+            ".gitignore": None,
+            "README.md": None,
+            "data": {
+                "subcontent": {
+                    ".gitignore": None,
+                    "data.xml": None,
+                    "data.xml.dvc": None,
+                    "statistics": {
+                        ".gitignore": None,
+                        "data.csv": None,
+                        "data.csv.dvc": None,
+                    },
+                }
+            },
+            "model": {
+                ".gitignore": None,
+                "people.csv": None,
+                "people.csv.dvc": None,
+                "script.py": None,
+                "train.py": None,
+            },
+            "structure.xml": None,
+            "structure.xml.dvc": None,
+        }
+    }
+    assert _simplify_tree(files) == expected
+
+    files = ls_tree(os.fspath(tmp_dir), "model")
+    assert _simplify_tree(files) == {
+        "model": {
+            ".gitignore": None,
+            "people.csv": None,
+            "people.csv.dvc": None,
+            "script.py": None,
+            "train.py": None,
+        }
+    }
+
+
+def test_ls_tree_dvc_only(M, tmp_dir, scm, dvc):
+    tmp_dir.scm_gen(FS_STRUCTURE, commit="init")
+    tmp_dir.dvc_gen(DVC_STRUCTURE, commit="dvc")
+
+    files = ls_tree(os.fspath(tmp_dir), dvc_only=True)
+
+    expected = {
+        ".": {
+            "data": {
+                "subcontent": {"data.xml": None, "statistics": {"data.csv": None}}
+            },
+            "model": {"people.csv": None},
+            "structure.xml": None,
+        }
+    }
+    assert _simplify_tree(files) == expected
+
+
+def test_ls_tree_maxdepth(M, tmp_dir, scm, dvc):
+    tmp_dir.scm_gen(FS_STRUCTURE, commit="init")
+    tmp_dir.dvc_gen(DVC_STRUCTURE, commit="dvc")
+
+    files = ls_tree(os.fspath(tmp_dir), maxdepth=0)
+    assert _simplify_tree(files) == {".": None}
+
+    files = ls_tree(os.fspath(tmp_dir), maxdepth=1)
+    assert _simplify_tree(files) == {
+        ".": {
+            ".dvcignore": None,
+            ".gitignore": None,
+            "README.md": None,
+            "data": None,
+            "model": None,
+            "structure.xml": None,
+            "structure.xml.dvc": None,
+        }
+    }
+
+    files = ls_tree(os.fspath(tmp_dir), maxdepth=2)
+    assert _simplify_tree(files) == {
+        ".": {
+            ".dvcignore": None,
+            ".gitignore": None,
+            "README.md": None,
+            "data": {"subcontent": None},
+            "model": {
+                ".gitignore": None,
+                "people.csv": None,
+                "people.csv.dvc": None,
+                "script.py": None,
+                "train.py": None,
+            },
+            "structure.xml": None,
+            "structure.xml.dvc": None,
+        }
+    }
+
+    files = ls_tree(os.fspath(tmp_dir), maxdepth=3)
+    assert _simplify_tree(files) == {
+        ".": {
+            ".dvcignore": None,
+            ".gitignore": None,
+            "README.md": None,
+            "data": {
+                "subcontent": {
+                    ".gitignore": None,
+                    "data.xml": None,
+                    "data.xml.dvc": None,
+                    "statistics": None,
+                }
+            },
+            "model": {
+                ".gitignore": None,
+                "people.csv": None,
+                "people.csv.dvc": None,
+                "script.py": None,
+                "train.py": None,
+            },
+            "structure.xml": None,
+            "structure.xml.dvc": None,
+        }
+    }
+
+    files = ls_tree(os.fspath(tmp_dir), maxdepth=4)
+    assert _simplify_tree(files) == {
+        ".": {
+            ".dvcignore": None,
+            ".gitignore": None,
+            "README.md": None,
+            "data": {
+                "subcontent": {
+                    ".gitignore": None,
+                    "data.xml": None,
+                    "data.xml.dvc": None,
+                    "statistics": {
+                        ".gitignore": None,
+                        "data.csv": None,
+                        "data.csv.dvc": None,
+                    },
+                }
+            },
+            "model": {
+                ".gitignore": None,
+                "people.csv": None,
+                "people.csv.dvc": None,
+                "script.py": None,
+                "train.py": None,
+            },
+            "structure.xml": None,
+            "structure.xml.dvc": None,
+        }
+    }

tests/unit/command/ls/test_ls.py

@@ -1,7 +1,7 @@
 import json
 
 from dvc.cli import parse_args
-from dvc.commands.ls import CmdList
+from dvc.commands.ls import CmdList, show_tree
 
 
 def _test_cli(mocker, *args):
@@ -27,6 +27,7 @@ def test_list(mocker):
         config=None,
         remote=None,
         remote_config=None,
+        maxdepth=None,
     )
 
 
@@ -42,6 +43,7 @@ def test_list_recursive(mocker):
         config=None,
         remote=None,
         remote_config=None,
+        maxdepth=None,
     )
 
 
@@ -57,6 +59,7 @@ def test_list_git_ssh_rev(mocker):
         config=None,
         remote=None,
         remote_config=None,
+        maxdepth=None,
     )
 
 
@@ -73,6 +76,7 @@ def test_list_targets(mocker):
         config=None,
         remote=None,
         remote_config=None,
+        maxdepth=None,
     )
 
 
@@ -88,6 +92,7 @@ def test_list_outputs_only(mocker):
         config=None,
         remote=None,
         remote_config=None,
+        maxdepth=None,
     )
 
 
@@ -114,6 +119,65 @@ def test_list_config(mocker):
         config="myconfig",
         remote="myremote",
         remote_config={"k1": "v1", "k2": "v2"},
+        maxdepth=None,
+    )
+
+
+def test_list_level(mocker):
+    url = "local_dir"
+    m = _test_cli(mocker, url, None, "--level", "1")
+    m.assert_called_once_with(
+        url,
+        None,
+        recursive=False,
+        rev=None,
+        dvc_only=False,
+        config=None,
+        remote=None,
+        remote_config=None,
+        maxdepth=1,
+    )
+
+
+def test_list_tree(mocker):
+    url = "[email protected]:repo"
+    cli_args = parse_args(
+        [
+            "list",
+            url,
+            "local_dir",
+            "--rev",
+            "123",
+            "--tree",
+            "--show-hash",
+            "--size",
+            "--level",
+            "2",
+            "--dvc-only",
+            "--config",
+            "myconfig",
+            "--remote",
+            "myremote",
+            "--remote-config",
+            "k1=v1",
+            "k2=v2",
+        ]
+    )
+    assert cli_args.func == CmdList
+
+    cmd = cli_args.func(cli_args)
+    m = mocker.patch("dvc.repo.ls.ls_tree")
+
+    assert cmd.run() == 0
+    m.assert_called_once_with(
+        url,
+        "local_dir",
+        rev="123",
+        dvc_only=True,
+        config="myconfig",
+        remote="myremote",
+        remote_config={"k1": "v1", "k2": "v2"},
+        maxdepth=2,
     )
 
 
@@ -165,6 +229,268 @@ def test_show_colors(mocker, capsys, monkeypatch):
     ]
 
 
+def test_show_size(mocker, capsys):
+    cli_args = parse_args(["list", "local_dir", "--size"])
+    assert cli_args.func == CmdList
+    cmd = cli_args.func(cli_args)
+
+    result = [
+        {
+            "isdir": False,
+            "isexec": 0,
+            "isout": False,
+            "path": ".dvcignore",
+            "size": 100,
+        },
+        {
+            "isdir": False,
+            "isexec": 0,
+            "isout": False,
+            "path": ".gitignore",
+            "size": 200,
+        },
+        {
+            "isdir": False,
+            "isexec": 0,
+            "isout": False,
+            "path": "README.md",
+            "size": 100_000,
+        },
+    ]
+    mocker.patch("dvc.repo.Repo.ls", return_value=result)
+
+    assert cmd.run() == 0
+    out, _ = capsys.readouterr()
+    entries = out.splitlines()
+
+    assert entries == ["  100  .dvcignore", "  200  .gitignore", "97.7k  README.md"]
+
+
+def test_show_hash(mocker, capsys):
+    cli_args = parse_args(["list", "local_dir", "--show-hash"])
+    assert cli_args.func == CmdList
+    cmd = cli_args.func(cli_args)
+
+    result = [
+        {
+            "isdir": False,
+            "isexec": 0,
+            "isout": False,
+            "path": ".dvcignore",
+            "md5": "123",
+        },
+        {
+            "isdir": False,
+            "isexec": 0,
+            "isout": False,
+            "path": ".gitignore",
+            "md5": "456",
+        },
+        {
+            "isdir": False,
+            "isexec": 0,
+            "isout": False,
+            "path": "README.md",
+            "md5": "789",
+        },
+    ]
+    mocker.patch("dvc.repo.Repo.ls", return_value=result)
+
+    assert cmd.run() == 0
+    out, _ = capsys.readouterr()
+    entries = out.splitlines()
+
+    assert entries == ["123  .dvcignore", "456  .gitignore", "789  README.md"]
+
+
+def test_show_size_and_hash(mocker, capsys):
+    cli_args = parse_args(["list", "local_dir", "--size", "--show-hash"])
+    assert cli_args.func == CmdList
+    cmd = cli_args.func(cli_args)
+
+    result = [
+        {
+            "isdir": False,
+            "isexec": 0,
+            "isout": False,
+            "path": ".dvcignore",
+            "size": 100,
+            "md5": "123",
+        },
+        {
+            "isdir": False,
+            "isexec": 0,
+            "isout": False,
+            "path": ".gitignore",
+            "size": 200,
+            "md5": "456",
+        },
+        {
+            "isdir": False,
+            "isexec": 0,
+            "isout": False,
+            "path": "README.md",
+            "size": 100_000,
+            "md5": "789",
+        },
+    ]
+    mocker.patch("dvc.repo.Repo.ls", return_value=result)
+
+    assert cmd.run() == 0
+    out, _ = capsys.readouterr()
+    entries = out.splitlines()
+
+    assert entries == [
+        "  100  123  .dvcignore",
+        "  200  456  .gitignore",
+        "97.7k  789  README.md",
+    ]
+
+
+def test_show_tree(capsys):
+    entries = {
+        "data": {
+            "isout": True,
+            "isdir": True,
+            "isexec": False,
+            "size": 192,
+            "path": "data",
+            "md5": "3fb071066d5d5b282f56a0169340346d.dir",
+            "contents": {
+                "dir": {
+                    "isout": False,
+                    "isdir": True,
+                    "isexec": False,
+                    "size": 96,
+                    "path": "data/dir",
+                    "md5": None,
+                    "contents": {
+                        "subdir": {
+                            "isout": True,
+                            "isdir": True,
+                            "isexec": False,
+                            "size": 96,
+                            "md5": None,
+                            "path": "data/dir/subdir",
+                            "contents": {
+                                "foobar": {
+                                    "isout": True,
+                                    "isdir": False,
+                                    "isexec": False,
+                                    "size": 4,
+                                    "md5": "d3b07384d113edec49eaa6238ad5ff00",
+                                }
+                            },
+                        },
+                        "foo": {
+                            "isout": False,
+                            "isdir": False,
+                            "isexec": False,
+                            "size": 4,
+                            "path": "data/dir/foo",
+                            "md5": None,
+                        },
+                    },
+                },
+                "bar": {
+                    "isout": True,
+                    "isdir": False,
+                    "isexec": False,
+                    "size": 4,
+                    "path": "data/bar",
+                    "md5": "c157a79031e1c40f85931829bc5fc552",
+                },
+                "large-file": {
+                    "isout": False,
+                    "isdir": False,
+                    "isexec": False,
+                    "size": 1073741824,
+                    "path": "data/large-file",
+                    "md5": None,
+                },
+                "dir2": {
+                    "isout": True,
+                    "isdir": True,
+                    "isexec": False,
+                    "size": 96,
+                    "md5": None,
+                    "path": "data/dir2",
+                    "contents": {
+                        "foo": {
+                            "isout": True,
+                            "isdir": False,
+                            "isexec": False,
+                            "size": 4,
+                            "path": "data/dir2/foo",
+                            "md5": "d3b07384d113edec49eaa6238ad5ff00",
+                        }
+                    },
+                },
+            },
+        }
+    }
+
+    show_tree(entries, with_color=False)
+    out, _ = capsys.readouterr()
+    expected = """\
+data
+├── dir
+│   ├── subdir
+│   │   └── foobar
+│   └── foo
+├── bar
+├── large-file
+└── dir2
+    └── foo
+"""
+    assert out == expected
+
+    show_tree(entries, with_color=False, with_size=True)
+    out, _ = capsys.readouterr()
+    expected = """\
+  192  data
+   96  ├── dir
+   96  │   ├── subdir
+    4  │   │   └── foobar
+    4  │   └── foo
+    4  ├── bar
+1.00G  ├── large-file
+   96  └── dir2
+    4      └── foo
+"""
+    assert out == expected
+
+    show_tree(entries, with_color=False, with_hash=True)
+    out, _ = capsys.readouterr()
+    expected = """\
+3fb071066d5d5b282f56a0169340346d.dir  data
+-                                     ├── dir
+-                                     │   ├── subdir
+d3b07384d113edec49eaa6238ad5ff00      │   │   └── foobar
+-                                     │   └── foo
+c157a79031e1c40f85931829bc5fc552      ├── bar
+-                                     ├── large-file
+-                                     └── dir2
+d3b07384d113edec49eaa6238ad5ff00          └── foo
+"""
+    assert out == expected
+
+    show_tree(entries, with_color=False, with_hash=True, with_size=True)
+    out, _ = capsys.readouterr()
+    expected = """\
+  192  3fb071066d5d5b282f56a0169340346d.dir  data
+   96  -                                     ├── dir
+   96  -                                     │   ├── subdir
+    4  d3b07384d113edec49eaa6238ad5ff00      │   │   └── foobar
+    4  -                                     │   └── foo
+    4  c157a79031e1c40f85931829bc5fc552      ├── bar
+1.00G  -                                     ├── large-file
+   96  -                                     └── dir2
+    4  d3b07384d113edec49eaa6238ad5ff00          └── foo
+"""
+    assert out == expected
+
+
 def test_list_alias():
     cli_args = parse_args(["ls", "local_dir"])
     assert cli_args.func == CmdList

tests/unit/command/test_dag.py

@@ -133,7 +133,7 @@ def test_show_dot(repo):
         "\"stage: 'b.dvc'\" -> \"stage: '2'\";",
         "\"stage: 'b.dvc'\" -> \"stage: '3'\";",
         "\"stage: 'b.dvc'\";",
-        "strict digraph  {",
+        "strict digraph {",
         "}",
     ]
     actual = sorted(line.rstrip() for line in _show_dot(repo.index.graph).splitlines())
@@ -153,7 +153,7 @@ def test_show_dot_properly_escapes():
     )
 
     expected = {
-        "strict digraph  {",
+        "strict digraph {",
         '"data\\raw\\2.dvc";',
         '"prepare";',
         '"4";',

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"}},
+        },
+    }