Repository: iterative/dvc (python)

Issue:
`params_show`: locking issues
`params_show` requires the lock, which breaks in multi-gpu/multi-node training. Need to reconsider whether it's needed and/or recommend another method for reading params, at least in multi-gpu/multi-node scenarios (see https://dvc.org/doc/dvclive/ml-frameworks/pytorch#distributeddataparallel).

Pull request description:
drop lock from params_show and metrics_show
Not sure whether this is safe and what the ramifications are, but wanted to raise it and see if there is a good reason for locking here. Closes #10050.


Patch:
diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml
index 1def6d319..298a7bb58 100644
--- a/.pre-commit-config.yaml
+++ b/.pre-commit-config.yaml
@@ -22,7 +22,7 @@ repos:
       - id: sort-simple-yaml
       - id: trailing-whitespace
   - repo: https://github.com/astral-sh/ruff-pre-commit
-    rev: 'v0.1.7'
+    rev: 'v0.1.8'
     hooks:
       - id: ruff
         args: [--fix, --exit-non-zero-on-fix]
diff --git a/dvc/commands/gc.py b/dvc/commands/gc.py
index 55984fe56..9bb0f8ded 100644
--- a/dvc/commands/gc.py
+++ b/dvc/commands/gc.py
@@ -86,6 +86,7 @@ class CmdGC(CmdBase):
             num=self.args.num,
             not_in_remote=self.args.not_in_remote,
             dry=self.args.dry,
+            skip_failed=self.args.skip_failed,
         )
         return 0
 
@@ -188,6 +189,12 @@ def add_parser(subparsers, parent_parser):
         help="Remote storage to collect garbage in",
         metavar="<name>",
     )
+    gc_parser.add_argument(
+        "--skip-failed",
+        action="store_true",
+        default=False,
+        help="Skip revisions that fail when collected.",
+    )
     gc_parser.add_argument(
         "-f",
         "--force",
diff --git a/dvc/dependency/param.py b/dvc/dependency/param.py
index e30439018..311998ff6 100644
--- a/dvc/dependency/param.py
+++ b/dvc/dependency/param.py
@@ -54,12 +54,15 @@ def read_param_file(
                 continue
         return ret
 
+    from copy import deepcopy
+
     from dpath import merge
+    from funcy import distinct
 
-    for key_path in key_paths:
+    for key_path in distinct(key_paths):
         merge(
             ret,
-            dpath.search(config, key_path, separator="."),
+            deepcopy(dpath.search(config, key_path, separator=".")),
             separator=".",
         )
     return ret
diff --git a/dvc/exceptions.py b/dvc/exceptions.py
index cc74310eb..c79ee09b4 100644
--- a/dvc/exceptions.py
+++ b/dvc/exceptions.py
@@ -357,3 +357,15 @@ class ArtifactNotFoundError(DvcException):
 
         desc = f" @ {stage or version}" if (stage or version) else ""
         super().__init__(f"Unable to find artifact '{name}{desc}'")
+
+
+class RevCollectionError(DvcException):
+    """Thrown if a revision failed to be collected.
+
+    Args:
+        rev (str): revision that failed (or "workspace").
+    """
+
+    def __init__(self, rev):
+        self.rev = rev
+        super().__init__(f"Failed to collect '{rev}'")
diff --git a/dvc/repo/__init__.py b/dvc/repo/__init__.py
index 66559a52f..e0e6c0666 100644
--- a/dvc/repo/__init__.py
+++ b/dvc/repo/__init__.py
@@ -12,7 +12,12 @@ from typing import (
     Union,
 )
 
-from dvc.exceptions import NotDvcRepoError, OutputNotFoundError
+from dvc.exceptions import (
+    DvcException,
+    NotDvcRepoError,
+    OutputNotFoundError,
+    RevCollectionError,
+)
 from dvc.ignore import DvcIgnoreFilter
 from dvc.log import logger
 from dvc.utils.objects import cached_property
@@ -380,6 +385,7 @@ class Repo:
             except KeyError:
                 pass
         self.data_index.commit()
+        self.data_index.close()
         self._reset()
 
     def get_data_index_entry(
@@ -492,6 +498,7 @@ class Repo:
         revs=None,
         num=1,
         push: bool = False,
+        skip_failed: bool = False,
     ):
         """Get the stages related to the given target and collect
         the `info` of its outputs.
@@ -510,7 +517,7 @@ class Repo:
         """
         used = defaultdict(set)
 
-        for _ in self.brancher(
+        for rev in self.brancher(
             revs=revs,
             all_branches=all_branches,
             all_tags=all_tags,
@@ -519,17 +526,23 @@ class Repo:
             commit_date=commit_date,
             num=num,
         ):
-            for odb, objs in self.index.used_objs(
-                targets,
-                remote=remote,
-                force=force,
-                jobs=jobs,
-                recursive=recursive,
-                with_deps=with_deps,
-                push=push,
-            ).items():
-                used[odb].update(objs)
-
+            try:
+                for odb, objs in self.index.used_objs(
+                    targets,
+                    remote=remote,
+                    force=force,
+                    jobs=jobs,
+                    recursive=recursive,
+                    with_deps=with_deps,
+                    push=push,
+                ).items():
+                    used[odb].update(objs)
+            except DvcException as exc:
+                rev = rev or "workspace"
+                if skip_failed:
+                    logger.warning("Failed to collect '%s', skipping", rev)
+                else:
+                    raise RevCollectionError(rev) from exc
         if used_run_cache:
             for odb, objs in self.stage_cache.get_used_objs(
                 used_run_cache, remote=remote, force=force, jobs=jobs
diff --git a/dvc/repo/gc.py b/dvc/repo/gc.py
index 5b1f9fc4f..682d05a3f 100644
--- a/dvc/repo/gc.py
+++ b/dvc/repo/gc.py
@@ -68,6 +68,7 @@ def gc(  # noqa: PLR0913, C901
     num: Optional[int] = None,
     not_in_remote: bool = False,
     dry: bool = False,
+    skip_failed: bool = False,
 ):
     # require `workspace` to be true to come into effect.
     # assume `workspace` to be enabled if any of `all_tags`, `all_commits`,
@@ -113,6 +114,7 @@ def gc(  # noqa: PLR0913, C901
                 jobs=jobs,
                 revs=[rev] if rev else None,
                 num=num or 1,
+                skip_failed=skip_failed,
             ).items():
                 if odb not in odb_to_obj_ids:
                     odb_to_obj_ids[odb] = set()
diff --git a/dvc/repo/index.py b/dvc/repo/index.py
index 76e2ce3a4..b3dfb47ee 100644
--- a/dvc/repo/index.py
+++ b/dvc/repo/index.py
@@ -129,10 +129,25 @@ def _load_data_from_outs(index, prefix, outs):
         for key_len in range(1, len(key)):
             parents.add((ws, key[:key_len]))
 
+        loaded = None
+        if out.files:
+            loaded = True
+            for okey, ometa, ohi in out.get_obj():
+                for key_len in range(1, len(okey)):
+                    parents.add((ws, (*key, *okey[:key_len])))
+
+                fkey = (*key, *okey)
+                index[(*prefix, ws, *fkey)] = DataIndexEntry(
+                    key=fkey,
+                    meta=ometa,
+                    hash_info=ohi,
+                )
+
         entry = DataIndexEntry(
             key=key,
             meta=out.meta,
             hash_info=out.hash_info,
+            loaded=loaded,
         )
 
         if (
diff --git a/dvc/repo/metrics/show.py b/dvc/repo/metrics/show.py
index a2f7229b5..3f2ff7f37 100644
--- a/dvc/repo/metrics/show.py
+++ b/dvc/repo/metrics/show.py
@@ -17,7 +17,6 @@ from funcy import ldistinct
 from scmrepo.exceptions import SCMError
 
 from dvc.log import logger
-from dvc.repo import locked
 from dvc.scm import NoSCMError
 from dvc.utils import as_posix, expand_paths
 from dvc.utils.collections import ensure_list
@@ -185,7 +184,6 @@ def _hide_workspace(
     return res
 
 
-@locked
 def show(
     repo: "Repo",
     targets: Optional[List[str]] = None,
diff --git a/dvc/repo/params/show.py b/dvc/repo/params/show.py
index ea5ceb013..5105a52cc 100644
--- a/dvc/repo/params/show.py
+++ b/dvc/repo/params/show.py
@@ -5,7 +5,6 @@ from typing import TYPE_CHECKING, Any, Dict, Iterator, List, Optional, Tuple, Un
 
 from dvc.dependency.param import ParamsDependency, read_param_file
 from dvc.log import logger
-from dvc.repo import locked
 from dvc.repo.metrics.show import FileResult, Result
 from dvc.stage import PipelineStage
 from dvc.utils import as_posix, expand_paths
@@ -158,7 +157,6 @@ def _gather_params(
     return data
 
 
-@locked
 def show(
     repo: "Repo",
     targets: Optional[List[str]] = None,
diff --git a/dvc/repo/push.py b/dvc/repo/push.py
index ae42610d4..0124af791 100644
--- a/dvc/repo/push.py
+++ b/dvc/repo/push.py
@@ -6,21 +6,44 @@ from dvc.ui import ui
 from . import locked
 
 
+def _rebuild(idx, path, fs, cb):
+    from dvc_data.index import DataIndex, DataIndexEntry, Meta
+
+    new = DataIndex()
+    items = list(idx.items())
+
+    cb.set_size(len(items))
+    for key, entry in items:
+        if entry.meta and entry.meta.isdir:
+            meta = Meta(isdir=True)
+        else:
+            try:
+                meta = Meta.from_info(fs.info(fs.join(path, *key)), fs.protocol)
+            except FileNotFoundError:
+                meta = None
+
+        if meta:
+            new.add(DataIndexEntry(key=key, meta=meta))
+
+        cb.relative_update(1)
+
+    return new
+
+
 def _update_meta(index, **kwargs):
-    from dvc.repo.index import build_data_index
     from dvc.repo.worktree import _merge_push_meta, worktree_view_by_remotes
 
     stages = set()
     for remote_name, idx in worktree_view_by_remotes(index, push=True, **kwargs):
         remote = index.repo.cloud.get_remote(remote_name)
 
-        with ui.progress("Collecting", unit="entry") as pb:
-            new = build_data_index(
-                idx,
-                remote.path,
-                remote.fs,
-                callback=pb.as_callback(),
-            )
+        with ui.progress(
+            desc=f"Collecting {remote.path} on {remote.fs.protocol}",
+            unit="entry",
+            leave=True,
+        ) as pb:
+            cb = pb.as_callback()
+            new = _rebuild(idx.data["repo"], remote.path, remote.fs, cb)
 
         for out in idx.outs:
             if not remote.fs.version_aware:
@@ -100,6 +123,7 @@ def push(  # noqa: PLR0913
     with ui.progress(
         desc="Collecting",
         unit="entry",
+        leave=True,
     ) as pb:
         data = collect(
             [idx.data["repo"] for idx in indexes.values()],
@@ -114,7 +138,8 @@ def push(  # noqa: PLR0913
     try:
         with ui.progress(
             desc="Pushing",
-            unit="file",
+            bar_format="{desc}",
+            leave=True,
         ) as pb:
             push_transferred, push_failed = ipush(
                 data,
diff --git a/pyproject.toml b/pyproject.toml
index 41b28d8c0..ab97301ad 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -35,7 +35,7 @@ dependencies = [
     "configobj>=5.0.6",
     "distro>=1.3",
     "dpath<3,>=2.1.0",
-    "dvc-data>=3,<3.1",
+    "dvc-data>=3.3,<3.4",
     "dvc-http>=2.29.0",
     "dvc-render>=1.0.0,<2",
     "dvc-studio-client>=0.17.1,<1",
@@ -77,7 +77,7 @@ gdrive = ["dvc-gdrive>=3,<4"]
 gs = ["dvc-gs>=3,<4"]
 hdfs = ["dvc-hdfs>=3,<4"]
 lint = [
-    "mypy==1.7.1",
+    "mypy==1.8.0",
     "pandas-stubs",
     "types-colorama",
     "types-psutil",


Test patch:
diff --git a/dvc/testing/cloud.py b/dvc/testing/cloud.py
index f1f13a0ec..377f237a9 100644
--- a/dvc/testing/cloud.py
+++ b/dvc/testing/cloud.py
@@ -32,6 +32,14 @@ class Cloud(ABC):
     def write_bytes(self, contents):
         raise NotImplementedError
 
+    @abstractmethod
+    def unlink(self, missing_ok: bool = False) -> None:
+        pass
+
+    @abstractmethod
+    def rmdir(self, recursive: bool = True) -> None:
+        pass
+
     def read_text(self, encoding=None, errors=None):
         if not encoding:
             encoding = locale.getpreferredencoding(False)
diff --git a/dvc/testing/remote_tests.py b/dvc/testing/remote_tests.py
index 4294b2bd1..b47d6e1f0 100644
--- a/dvc/testing/remote_tests.py
+++ b/dvc/testing/remote_tests.py
@@ -171,7 +171,7 @@ class TestRemoteVersionAware:
         (stage,) = tmp_dir.dvc_gen("foo", "foo")
         run_copy("foo", "foo_copy", name="copy")
 
-        dvc.push()
+        assert dvc.push()
         assert (remote_version_aware / "foo").read_text() == "foo"
         assert (remote_version_aware / "foo_copy").read_text() == "foo"
         foo_dvc = (tmp_dir / "foo.dvc").read_text()
@@ -185,26 +185,26 @@ class TestRemoteVersionAware:
         remove(tmp_dir / "foo")
         remove(tmp_dir / "foo_copy")
 
-        dvc.pull()
+        assert dvc.pull()
         assert (tmp_dir / "foo").read_text() == "foo"
         assert (tmp_dir / "foo_copy").read_text() == "foo"
         assert (tmp_dir / "foo.dvc").read_text() == foo_dvc
         assert (tmp_dir / "dvc.lock").read_text() == dvc_lock
 
-        dvc.push()
+        assert not dvc.push()
         assert (remote_version_aware / "foo").read_text() == "foo"
         assert (remote_version_aware / "foo_copy").read_text() == "foo"
         assert (tmp_dir / "foo.dvc").read_text() == foo_dvc
         assert (tmp_dir / "dvc.lock").read_text() == dvc_lock
 
         dvc.reproduce()
-        dvc.push()
+        assert not dvc.push()
         assert (remote_version_aware / "foo").read_text() == "foo"
         assert (remote_version_aware / "foo_copy").read_text() == "foo"
         assert (tmp_dir / "foo.dvc").read_text() == foo_dvc
         assert (tmp_dir / "dvc.lock").read_text() == dvc_lock
 
-    def test_dir(self, tmp_dir, dvc, run_copy, remote_version_aware):
+    def test_dir(self, tmp_dir, dvc, run_copy, remote_version_aware):  # noqa: PLR0915
         (stage,) = tmp_dir.dvc_gen(
             {
                 "data_dir": {
@@ -215,7 +215,8 @@ class TestRemoteVersionAware:
             }
         )
 
-        dvc.push()
+        assert not dvc.fetch()
+        assert dvc.push()
 
         data_dir_dvc = (tmp_dir / "data_dir.dvc").read_text()
         assert "files" in data_dir_dvc
@@ -230,7 +231,7 @@ class TestRemoteVersionAware:
         remove(dvc.cache.local.path)
         remove(tmp_dir / "data_dir")
 
-        dvc.pull()
+        assert dvc.pull()
         assert (tmp_dir / "data_dir" / "data").read_text() == "data"
         assert (
             tmp_dir / "data_dir" / "data_sub_dir" / "data_sub"
@@ -240,20 +241,47 @@ class TestRemoteVersionAware:
         run_copy("data_dir", "data_dir_copy", name="copy")
         dvc_lock = (tmp_dir / "dvc.lock").read_text()
 
-        dvc.push()
+        assert dvc.push()
+        assert (remote_version_aware / "data_dir").exists()
+        assert (remote_version_aware / "data_dir" / "data").exists()
+        assert (remote_version_aware / "data_dir_copy").exists()
+        assert (remote_version_aware / "data_dir_copy" / "data").exists()
         assert (tmp_dir / "data_dir.dvc").read_text() == data_dir_dvc
         assert (tmp_dir / "dvc.lock").read_text() != dvc_lock
         dvc_lock = (tmp_dir / "dvc.lock").read_text()
 
-        dvc.push()
+        assert not dvc.push()
+        assert (remote_version_aware / "data_dir").exists()
+        assert (remote_version_aware / "data_dir" / "data").exists()
+        assert (remote_version_aware / "data_dir_copy").exists()
+        assert (remote_version_aware / "data_dir_copy" / "data").exists()
         assert (tmp_dir / "data_dir.dvc").read_text() == data_dir_dvc
         assert (tmp_dir / "dvc.lock").read_text() == dvc_lock
 
         dvc.cache.local.clear()
         remove(tmp_dir / "data_dir")
-        dvc.push()
+        remove(tmp_dir / "data_dir_copy")
+        assert not dvc.push()
         assert (remote_version_aware / "data_dir").exists()
         assert (remote_version_aware / "data_dir" / "data").exists()
+        assert (remote_version_aware / "data_dir_copy").exists()
+        assert (remote_version_aware / "data_dir_copy" / "data").exists()
+        assert (tmp_dir / "data_dir.dvc").read_text() == data_dir_dvc
+        assert (tmp_dir / "dvc.lock").read_text() == dvc_lock
+
+        (remote_version_aware / "data_dir").rmdir()
+        (remote_version_aware / "data_dir_copy").rmdir()
+        assert not (remote_version_aware / "data_dir").exists()
+        assert not (remote_version_aware / "data_dir_copy").exists()
+        assert dvc.pull()
+        assert (tmp_dir / "data_dir" / "data").read_text() == "data"
+        assert (
+            tmp_dir / "data_dir" / "data_sub_dir" / "data_sub"
+        ).read_text() == "data_sub"
+        assert (tmp_dir / "data_dir_copy" / "data").read_text() == "data"
+        assert (
+            tmp_dir / "data_dir_copy" / "data_sub_dir" / "data_sub"
+        ).read_text() == "data_sub"
 
 
 class TestRemoteWorktree:
diff --git a/tests/func/test_gc.py b/tests/func/test_gc.py
index f1e28465a..af42c0a1a 100644
--- a/tests/func/test_gc.py
+++ b/tests/func/test_gc.py
@@ -7,7 +7,7 @@ import textwrap
 import pytest
 
 from dvc.cli import main
-from dvc.exceptions import CollectCacheError, InvalidArgumentError
+from dvc.exceptions import CollectCacheError, InvalidArgumentError, RevCollectionError
 from dvc.fs import LocalFileSystem
 from dvc.utils.fs import remove
 from dvc_data.hashfile.db.local import LocalHashFileDB
@@ -111,8 +111,9 @@ def test_gc_no_dir_cache(tmp_dir, dvc):
 
     remove(dir_stage.outs[0].cache_path)
 
-    with pytest.raises(CollectCacheError):
+    with pytest.raises(RevCollectionError) as exc:
         dvc.gc(workspace=True)
+    assert type(exc.value.__cause__) == CollectCacheError
 
     assert _count_files(dvc.cache.local.path) == 4
     dvc.gc(force=True, workspace=True)
@@ -439,3 +440,13 @@ def test_gc_logging(caplog, dvc, good_and_bad_cache):
     assert "Removed 3 objects from repo cache." in caplog.text
     assert "No unused 'local' cache to remove." in caplog.text
     assert "No unused 'legacy' cache to remove." in caplog.text
+
+
+def test_gc_skip_failed(tmp_dir, dvc):
+    with open("dvc.yaml", mode="w") as f:
+        f.write("\ninvalid")
+
+    with pytest.raises(RevCollectionError):
+        dvc.gc(force=True, workspace=True)
+
+    dvc.gc(force=True, workspace=True, skip_failed=True)
diff --git a/tests/unit/command/test_gc.py b/tests/unit/command/test_gc.py
index 5df832767..98102a580 100644
--- a/tests/unit/command/test_gc.py
+++ b/tests/unit/command/test_gc.py
@@ -26,6 +26,7 @@ def test_(dvc, scm, mocker):
             "--projects",
             "project1",
             "project2",
+            "--skip-failed",
         ]
     )
     assert cli_args.func == CmdGC
@@ -51,6 +52,7 @@ def test_(dvc, scm, mocker):
         num=None,
         not_in_remote=False,
         dry=True,
+        skip_failed=True,
     )
 
     cli_args = parse_args(["gc"])