bump DVCLive lower bound

* Bump dvclive lower bound

* Update pyproject.toml

Co-authored-by: Dave Berenbaum <[email protected]>

Co-authored-by: Dave Berenbaum <[email protected]>

dvc/commands/update.py

@@ -31,7 +31,8 @@ class CmdUpdate(CmdBase):
 def add_parser(subparsers, parent_parser):
     UPDATE_HELP = (
         "Update data artifact imported (via dvc import or dvc import-url) "
-        "from an external DVC repository or URL."
+        "from an external DVC repository or URL, or update data from a "
+        "DVC worktree remote."
     )
     update_parser = subparsers.add_parser(
         "update",

dvc/data_cloud.py

@@ -5,7 +5,7 @@ from typing import TYPE_CHECKING, Iterable, Optional
 
 from funcy import cached_property
 
-from dvc.config import RemoteConfigError
+from dvc.config import NoRemoteError, RemoteConfigError
 from dvc_data.hashfile.db import get_index
 
 if TYPE_CHECKING:
@@ -63,8 +63,6 @@ class DataCloud:
         name: Optional[str] = None,
         command: str = "<command>",
     ) -> "Remote":
-        from dvc.config import NoRemoteError
-
         if not name:
             name = self.repo.config["core"].get("remote")
 
@@ -97,6 +95,10 @@ class DataCloud:
         command: str = "<command>",
     ) -> "HashFileDB":
         remote = self.get_remote(name=name, command=command)
+        if remote.fs.version_aware or remote.worktree:
+            raise NoRemoteError(
+                f"'{command}' is unsupported for cloud versioned remotes"
+            )
         return remote.odb
 
     def _log_missing(self, status: "CompareStatusResult"):

dvc/repo/__init__.py

@@ -536,11 +536,16 @@ class Repo:
         self.scm.close()
         self.state.close()
 
+    def _reset_cached_indecies(self):
+        self.__dict__.pop("index", None)
+        self.__dict__.pop("dvcignore", None)
+        self.__dict__.pop("dvcfs", None)
+        self.__dict__.pop("datafs", None)
+
     def _reset(self):
         self.state.close()
         self.scm._reset()  # pylint: disable=protected-access
-        self.__dict__.pop("index", None)
-        self.__dict__.pop("dvcignore", None)
+        self._reset_cached_indecies()
 
     def __enter__(self):
         return self

dvc/repo/brancher.py

@@ -77,7 +77,7 @@ def brancher(  # noqa: E302
         from dvc.fs import GitFileSystem
 
         for sha, names in found_revs.items():
-            self.__dict__.pop("index", None)
+            self._reset_cached_indecies()  # pylint: disable=W0212
             self.fs = GitFileSystem(scm=scm, rev=sha)
             self.root_dir = self.fs.path.join("/", *repo_root_parts)
 

dvc/repo/experiments/__init__.py

@@ -32,7 +32,7 @@ from .refs import (
     ExpRefInfo,
 )
 from .stash import ExpStashEntry
-from .utils import exp_refs_by_rev, unlocked_repo
+from .utils import check_ref_format, exp_refs_by_rev, unlocked_repo
 
 logger = logging.getLogger(__name__)
 
@@ -236,17 +236,6 @@ class Experiments:
             "\tdvc exp branch <exp> <branch>\n"
         )
 
-    def _validate_new_ref(self, exp_ref: ExpRefInfo):
-        from .utils import check_ref_format
-
-        if not exp_ref.name:
-            return
-
-        check_ref_format(self.scm, exp_ref)
-
-        if self.scm.get_ref(str(exp_ref)):
-            raise ExperimentExistsError(exp_ref.name)
-
     def new(
         self,
         queue: BaseStashQueue,
@@ -265,13 +254,15 @@ class Experiments:
 
         name = kwargs.get("name", None)
         baseline_sha = kwargs.get("baseline_rev") or self.repo.scm.get_rev()
-        exp_ref = ExpRefInfo(baseline_sha=baseline_sha, name=name)
 
-        try:
-            self._validate_new_ref(exp_ref)
-        except ExperimentExistsError as err:
-            if not (kwargs.get("force", False) or kwargs.get("reset", False)):
-                raise err
+        if name:
+            exp_ref = ExpRefInfo(baseline_sha=baseline_sha, name=name)
+            check_ref_format(self.scm, exp_ref)
+            force = kwargs.get("force", False)
+            reset = kwargs.get("reset", False)
+            if self.scm.get_ref(str(exp_ref)) and not (force or reset):
+                raise ExperimentExistsError(exp_ref.name)
+
         return queue.put(*args, **kwargs)
 
     def _resume_checkpoint(

dvc/repo/experiments/queue/base.py

@@ -30,7 +30,13 @@ from ..executor.base import BaseExecutor, ExecutorResult
 from ..executor.local import WorkspaceExecutor
 from ..refs import ExpRefInfo
 from ..stash import ExpStash, ExpStashEntry
-from ..utils import EXEC_PID_DIR, EXEC_TMP_DIR, exp_refs_by_rev, get_exp_rwlock
+from ..utils import (
+    EXEC_PID_DIR,
+    EXEC_TMP_DIR,
+    exp_refs_by_rev,
+    get_exp_rwlock,
+    get_random_exp_name,
+)
 from .utils import get_remote_executor_refs
 
 if TYPE_CHECKING:
@@ -281,7 +287,7 @@ class BaseStashQueue(ABC):
                 output.
         """
 
-    def _stash_exp(
+    def _stash_exp(  # noqa: C901
         self,
         *args,
         params: Optional[Dict[str, List[str]]] = None,
@@ -399,8 +405,9 @@ class BaseStashQueue(ABC):
                     run_env = {
                         DVC_EXP_BASELINE_REV: baseline_rev,
                     }
-                    if name:
-                        run_env[DVC_EXP_NAME] = name
+                    if not name:
+                        name = get_random_exp_name(self.scm, baseline_rev)
+                    run_env[DVC_EXP_NAME] = name
                     if resume_rev:
                         run_env[DVCLIVE_RESUME] = "1"
                     self._pack_args(*args, run_env=run_env, **kwargs)

dvc/repo/fetch.py

@@ -1,6 +1,6 @@
 import logging
 from contextlib import suppress
-from typing import TYPE_CHECKING, Optional
+from typing import TYPE_CHECKING, Optional, Sequence
 
 from dvc.config import NoRemoteError
 from dvc.exceptions import DownloadError
@@ -9,6 +9,7 @@ from dvc.fs import Schemes
 from . import locked
 
 if TYPE_CHECKING:
+    from dvc.cloud import Remote
     from dvc.repo import Repo
     from dvc.types import TargetType
     from dvc_data.hashfile.transfer import TransferResult
@@ -45,18 +46,19 @@ def fetch(
             remote is configured
     """
     from dvc.repo.imports import save_imports
-    from dvc.repo.worktree import fetch_worktree
     from dvc_data.hashfile.transfer import TransferResult
 
     if isinstance(targets, str):
         targets = [targets]
 
+    worktree_remote: Optional["Remote"] = None
     with suppress(NoRemoteError):
         _remote = self.cloud.get_remote(name=remote)
-        if _remote.worktree:
-            return fetch_worktree(self, _remote)
+        if _remote.worktree or _remote.fs.version_aware:
+            worktree_remote = _remote
 
     failed_count = 0
+    transferred_count = 0
 
     try:
         if run_cache:
@@ -67,22 +69,35 @@ def fetch(
     no_remote_msg: Optional[str] = None
     result = TransferResult(set(), set())
     try:
-        d, f = _fetch(
-            self,
-            targets,
-            all_branches=all_branches,
-            all_tags=all_tags,
-            all_commits=all_commits,
-            with_deps=with_deps,
-            force=True,
-            remote=remote,
-            jobs=jobs,
-            recursive=recursive,
-            revs=revs,
-            odb=odb,
-        )
-        result.transferred.update(d)
-        result.failed.update(f)
+        if worktree_remote is not None:
+            transferred_count += _fetch_worktree(
+                self,
+                worktree_remote,
+                revs=revs,
+                all_branches=all_branches,
+                all_tags=all_tags,
+                all_commits=all_commits,
+                targets=targets,
+                with_deps=with_deps,
+                recursive=recursive,
+            )
+        else:
+            d, f = _fetch(
+                self,
+                targets,
+                all_branches=all_branches,
+                all_tags=all_tags,
+                all_commits=all_commits,
+                with_deps=with_deps,
+                force=True,
+                remote=remote,
+                jobs=jobs,
+                recursive=recursive,
+                revs=revs,
+                odb=odb,
+            )
+            result.transferred.update(d)
+            result.failed.update(f)
     except NoRemoteError as exc:
         no_remote_msg = str(exc)
 
@@ -108,7 +123,8 @@ def fetch(
             logger.error(no_remote_msg)
         raise DownloadError(failed_count)
 
-    return len(result.transferred)
+    transferred_count += len(result.transferred)
+    return transferred_count
 
 
 def _fetch(
@@ -155,3 +171,26 @@ def _fetch(
             result.transferred.update(d)
             result.failed.update(f)
     return result
+
+
+def _fetch_worktree(
+    repo: "Repo",
+    remote: "Remote",
+    revs: Optional[Sequence[str]] = None,
+    all_branches: bool = False,
+    all_tags: bool = False,
+    all_commits: bool = False,
+    targets: Optional["TargetType"] = None,
+    **kwargs,
+) -> int:
+    from dvc.repo.worktree import fetch_worktree
+
+    downloaded = 0
+    for _ in repo.brancher(
+        revs=revs,
+        all_branches=all_branches,
+        all_tags=all_tags,
+        all_commits=all_commits,
+    ):
+        downloaded += fetch_worktree(repo, remote, targets=targets, **kwargs)
+    return downloaded

dvc/repo/index.py

@@ -166,7 +166,7 @@ class Index:
             )
         )
 
-    def _collect_targets(
+    def collect_targets(
         self, targets: Optional["TargetType"], **kwargs: Any
     ) -> List["StageInfo"]:
         from itertools import chain
@@ -214,7 +214,7 @@ class Index:
         """
         stage_infos = [
             stage_info
-            for stage_info in self._collect_targets(targets, **kwargs)
+            for stage_info in self.collect_targets(targets, **kwargs)
             if not stage_filter or stage_filter(stage_info.stage)
         ]
         return IndexView(self, stage_infos, outs_filter=outs_filter)
@@ -310,7 +310,7 @@ class Index:
         from collections import defaultdict
 
         used: "ObjectContainer" = defaultdict(set)
-        pairs = self._collect_targets(
+        pairs = self.collect_targets(
             targets, recursive=recursive, with_deps=with_deps
         )
         for stage, filter_info in pairs:
@@ -484,7 +484,6 @@ class IndexView:
         data: Dict[str, Union["DataIndex", "DataIndexView"]] = {}
         for workspace, data_index in self._index.data.items():
             if self._stages:
-                data_index.load()
                 data[workspace] = view(
                     data_index, partial(key_filter, workspace)
                 )

dvc/repo/push.py

@@ -1,11 +1,14 @@
-from typing import TYPE_CHECKING, Optional
+from typing import TYPE_CHECKING, Optional, Sequence
 
-from dvc.exceptions import UploadError
+from dvc.exceptions import InvalidArgumentError, UploadError
 
 from ..utils import glob_targets
 from . import locked
 
 if TYPE_CHECKING:
+    from dvc.cloud import Remote
+    from dvc.repo import Repo
+    from dvc.types import TargetType
     from dvc_objects.db.base import ObjectDB
 
 
@@ -26,55 +29,89 @@ def push(
     odb: Optional["ObjectDB"] = None,
     include_imports=False,
 ):
-    from dvc.repo.worktree import push_worktree
 
+    worktree_remote: Optional["Remote"] = None
     _remote = self.cloud.get_remote(name=remote)
     if _remote.worktree or _remote.fs.version_aware:
-        return push_worktree(self, _remote)
+        worktree_remote = _remote
 
+    pushed = 0
     used_run_cache = (
         self.stage_cache.push(remote, odb=odb) if run_cache else []
     )
+    pushed += len(used_run_cache)
 
     if isinstance(targets, str):
         targets = [targets]
 
     expanded_targets = glob_targets(targets, glob=glob)
 
-    used = self.used_objs(
-        expanded_targets,
-        all_branches=all_branches,
-        all_tags=all_tags,
-        all_commits=all_commits,
-        with_deps=with_deps,
-        force=True,
-        remote=remote,
-        jobs=jobs,
-        recursive=recursive,
-        used_run_cache=used_run_cache,
-        revs=revs,
-        push=True,
-    )
-
-    pushed = len(used_run_cache)
-    if odb:
-        all_ids = set()
-        for dest_odb, obj_ids in used.items():
-            if not include_imports and dest_odb and dest_odb.read_only:
-                continue
-            all_ids.update(obj_ids)
-        result = self.cloud.push(all_ids, jobs, remote=remote, odb=odb)
-        if result.failed:
-            raise UploadError(len(result.failed))
-        pushed += len(result.transferred)
+    if worktree_remote is not None:
+        pushed += _push_worktree(
+            self,
+            worktree_remote,
+            revs=revs,
+            all_branches=all_branches,
+            all_tags=all_tags,
+            all_commits=all_commits,
+            targets=expanded_targets,
+            with_deps=with_deps,
+            recursive=recursive,
+        )
     else:
-        for dest_odb, obj_ids in used.items():
-            if dest_odb and dest_odb.read_only:
-                continue
-            result = self.cloud.push(
-                obj_ids, jobs, remote=remote, odb=odb or dest_odb
-            )
+        used = self.used_objs(
+            expanded_targets,
+            all_branches=all_branches,
+            all_tags=all_tags,
+            all_commits=all_commits,
+            with_deps=with_deps,
+            force=True,
+            remote=remote,
+            jobs=jobs,
+            recursive=recursive,
+            used_run_cache=used_run_cache,
+            revs=revs,
+            push=True,
+        )
+
+        if odb:
+            all_ids = set()
+            for dest_odb, obj_ids in used.items():
+                if not include_imports and dest_odb and dest_odb.read_only:
+                    continue
+                all_ids.update(obj_ids)
+            result = self.cloud.push(all_ids, jobs, remote=remote, odb=odb)
             if result.failed:
                 raise UploadError(len(result.failed))
             pushed += len(result.transferred)
+        else:
+            for dest_odb, obj_ids in used.items():
+                if dest_odb and dest_odb.read_only:
+                    continue
+                result = self.cloud.push(
+                    obj_ids, jobs, remote=remote, odb=odb or dest_odb
+                )
+                if result.failed:
+                    raise UploadError(len(result.failed))
+                pushed += len(result.transferred)
     return pushed
+
+
+def _push_worktree(
+    repo: "Repo",
+    remote: "Remote",
+    revs: Optional[Sequence[str]] = None,
+    all_branches: bool = False,
+    all_tags: bool = False,
+    all_commits: bool = False,
+    targets: Optional["TargetType"] = None,
+    **kwargs,
+) -> int:
+    from dvc.repo.worktree import push_worktree
+
+    if revs or all_branches or all_tags or all_commits:
+        raise InvalidArgumentError(
+            "Multiple rev push is unsupported for cloud versioned remotes"
+        )
+
+    return push_worktree(repo, remote, targets=targets, **kwargs)

dvc/repo/update.py

@@ -1,7 +1,13 @@
+from typing import TYPE_CHECKING, List
+
 from dvc.exceptions import InvalidArgumentError
+from dvc.stage.exceptions import StageUpdateError
 
 from . import locked
 
+if TYPE_CHECKING:
+    from dvc.repo.stage import StageInfo
+
 
 @locked
 def update(
@@ -15,6 +21,7 @@ def update(
     jobs=None,
 ):
     from ..dvcfile import Dvcfile
+    from .worktree import update_worktree_stages
 
     if not targets:
         targets = [None]
@@ -28,15 +35,21 @@ def update(
         )
 
     if not to_remote and remote:
-        raise InvalidArgumentError(
-            "--remote can't be used without --to-remote"
-        )
+        if not self.cloud.get_remote(name=remote).worktree:
+            raise InvalidArgumentError(
+                "--remote can't be used without --to-remote"
+            )
 
-    stages = set()
-    for target in targets:
-        stages.update(self.stage.collect(target, recursive=recursive))
+    import_stages = set()
+    other_stage_infos: List["StageInfo"] = []
 
-    for stage in stages:
+    for stage_info in self.index.collect_targets(targets, recursive=recursive):
+        if stage_info.stage.is_import:
+            import_stages.add(stage_info.stage)
+        else:
+            other_stage_infos.append(stage_info)
+
+    for stage in import_stages:
         stage.update(
             rev,
             to_remote=to_remote,
@@ -47,4 +60,29 @@ def update(
         dvcfile = Dvcfile(self, stage.path)
         dvcfile.dump(stage)
 
+    if other_stage_infos:
+        remote_obj = self.cloud.get_remote(name=remote)
+        if not remote_obj.worktree:
+            raise StageUpdateError(other_stage_infos[0].stage.relpath)
+        if rev:
+            raise InvalidArgumentError(
+                "--rev can't be used with worktree update"
+            )
+        if no_download:
+            raise InvalidArgumentError(
+                "--no-download can't be used with worktree update"
+            )
+        if to_remote:
+            raise InvalidArgumentError(
+                "--to-remote can't be used with worktree update"
+            )
+        update_worktree_stages(
+            self,
+            other_stage_infos,
+            remote_obj,
+        )
+
+    stages = import_stages | {
+        stage_info.stage for stage_info in other_stage_infos
+    }
     return list(stages)

dvc/repo/worktree.py

@@ -1,5 +1,6 @@
 import logging
-from typing import TYPE_CHECKING, Any, Dict, Optional
+from functools import partial
+from typing import TYPE_CHECKING, Any, Dict, Iterable, Optional, Union, cast
 
 from dvc.fs.callbacks import Callback
 
@@ -8,13 +9,25 @@ if TYPE_CHECKING:
     from dvc.index import Index, IndexView
     from dvc.output import Output
     from dvc.repo import Repo
+    from dvc.repo.stage import StageInfo
     from dvc.stage import Stage
     from dvc.types import TargetType
     from dvc_data.hashfile.meta import Meta
+    from dvc_data.hashfile.tree import Tree
+    from dvc_data.index import DataIndex, DataIndexView
+    from dvc_objects.fs.base import FileSystem
 
 logger = logging.getLogger(__name__)
 
 
+# for files, if our version's checksum (etag) matches the latest remote
+# checksum, we do not need to push, even if the version IDs don't match
+def _meta_checksum(fs: "FileSystem", meta: "Meta") -> Any:
+    if not meta or meta.isdir:
+        return meta
+    return getattr(meta, fs.PARAM_CHECKSUM)
+
+
 def worktree_view(
     index: "Index",
     targets: Optional["TargetType"] = None,
@@ -57,10 +70,15 @@ def worktree_view(
     )
 
 
-def fetch_worktree(repo: "Repo", remote: "Remote") -> int:
+def fetch_worktree(
+    repo: "Repo",
+    remote: "Remote",
+    targets: Optional["TargetType"] = None,
+    **kwargs,
+) -> int:
     from dvc_data.index import save
 
-    view = worktree_view(repo.index)
+    view = worktree_view(repo.index, targets=targets, **kwargs)
     index = view.data["repo"]
     for key, entry in index.iteritems():
         entry.fs = remote.fs
@@ -76,31 +94,34 @@ def fetch_worktree(repo: "Repo", remote: "Remote") -> int:
         return save(index, callback=cb)
 
 
-def push_worktree(repo: "Repo", remote: "Remote") -> int:
-    from dvc_data.index import build, checkout
+def push_worktree(
+    repo: "Repo",
+    remote: "Remote",
+    targets: Optional["TargetType"] = None,
+    **kwargs,
+) -> int:
+    from dvc_data.index import checkout
 
-    view = worktree_view(repo.index, push=True, latest_only=remote.worktree)
+    view = worktree_view(
+        repo.index,
+        push=True,
+        targets=targets,
+        latest_only=remote.worktree,
+        **kwargs,
+    )
     new_index = view.data["repo"]
     if remote.worktree:
-        logger.debug("Indexing latest worktree for '%s'", remote.path)
-        old_index = build(remote.path, remote.fs)
+        logger.debug("indexing latest worktree for '%s'", remote.path)
+        old_index = _build_worktree_index(repo, remote, view)
         logger.debug("Pushing worktree changes to '%s'", remote.path)
     else:
         old_index = None
         logger.debug("Pushing version-aware files to '%s'", remote.path)
 
     if remote.worktree:
-
-        # for files, if our version's checksum (etag) matches the latest remote
-        # checksum, we do not need to push, even if the version IDs don't match
-        def _checksum(meta: "Meta") -> Any:
-            if not meta or meta.isdir:
-                return meta
-            return getattr(meta, remote.fs.PARAM_CHECKSUM)
-
         diff_kwargs: Dict[str, Any] = {
             "meta_only": True,
-            "meta_cmp_key": _checksum,
+            "meta_cmp_key": partial(_meta_checksum, remote.fs),
         }
     else:
         diff_kwargs = {}
@@ -121,41 +142,145 @@ def push_worktree(repo: "Repo", remote: "Remote") -> int:
             **diff_kwargs,
         )
     if pushed:
-        _update_pushed_meta(repo, view)
+        for stage in view.stages:
+            for out in stage.outs:
+                workspace, _key = out.index_key
+                _update_out_meta(out, repo.index.data[workspace])
+            stage.dvcfile.dump(stage, with_files=True, update_pipeline=False)
     return pushed
 
 
-def _update_pushed_meta(repo: "Repo", view: "IndexView"):
+def _build_worktree_index(
+    repo: "Repo", remote: "Remote", view: "IndexView"
+) -> "DataIndex":
+    from dvc_data.index import DataIndex
+    from dvc_data.index.build import build_entries
+
+    index = DataIndex()
+    for out in view.outs:
+        _workspace, key = out.index_key
+        parts = out.fs.path.relparts(out.fs_path, repo.root_dir)
+        path = remote.fs.path.join(remote.path, *parts)
+        for entry in build_entries(path, remote.fs):
+            if not entry.key or entry.key == ("",):
+                entry.key = key
+            else:
+                entry.key = key + entry.key
+            index.add(entry)
+    return index
+
+
+def _update_out_meta(
+    out: "Output", index: Union["DataIndex", "DataIndexView"]
+):
     from dvc_data.index.save import build_tree
 
+    _, key = out.index_key
+    entry = index[key]
+    repo = out.stage.repo
+    if out.isdir():
+        old_tree = cast("Tree", out.get_obj())
+        entry.hash_info = old_tree.hash_info
+        entry.meta = out.meta
+        for subkey, entry in index.iteritems(key):
+            if entry.meta.isdir:
+                continue
+            fs_path = repo.fs.path.join(repo.root_dir, *subkey)
+            meta, hash_info = old_tree.get(
+                repo.fs.path.relparts(fs_path, out.fs_path)
+            )
+            entry.hash_info = hash_info
+            if meta.version_id is not None:
+                # preserve existing version IDs for unchanged files in
+                # this dir (entry will have the latest remote version
+                # ID after checkout)
+                entry.meta = meta
+        tree_meta, new_tree = build_tree(index, key)
+        out.obj = new_tree
+        out.hash_info = new_tree.hash_info
+        out.meta = tree_meta
+    else:
+        if entry.hash_info:
+            out.hash_info = entry.hash_info
+        if out.meta.version_id is None:
+            out.meta = entry.meta
+
+
+def update_worktree_stages(
+    repo: "Repo",
+    stage_infos: Iterable["StageInfo"],
+    remote: "Remote",
+):
+    from dvc.repo.index import IndexView
+    from dvc_data.index import build
+
+    def outs_filter(out: "Output") -> bool:
+        return out.is_in_repo and out.use_cache and out.can_push
+
+    view = IndexView(
+        repo.index,
+        stage_infos,
+        outs_filter=outs_filter,
+    )
+    local_index = view.data["repo"]
+    logger.debug("indexing latest worktree for '%s'", remote.path)
+    remote_index = build(remote.path, remote.fs)
     for stage in view.stages:
         for out in stage.outs:
-            workspace, key = out.index_key
-            index = repo.index.data[workspace]
-            entry = index[key]
-            if out.isdir():
-                old_tree = out.get_obj()
-                entry.hash_info = old_tree.hash_info
-                entry.meta = out.meta
-                for subkey, entry in index.iteritems(key):
-                    if entry.meta.isdir:
-                        continue
-                    fs_path = repo.fs.path.join(repo.root_dir, *subkey)
-                    meta, hash_info = old_tree.get(
-                        repo.fs.path.relparts(fs_path, out.fs_path)
+            _workspace, key = out.index_key
+            if key not in remote_index:
+                logger.warning(
+                    "Could not update '%s', it does not exist in the remote",
+                    out,
+                )
+                continue
+            entry = remote_index[key]
+            if entry.meta and entry.meta.isdir:
+                if not any(
+                    subkey != key and subentry.meta and not subentry.meta.isdir
+                    for subkey, subentry in remote_index.iteritems(key)
+                ):
+                    logger.warning(
+                        "Could not update '%s', directory is empty in the "
+                        "remote",
+                        out,
                     )
-                    entry.hash_info = hash_info
-                    if meta.version_id is not None:
-                        # preserve existing version IDs for unchanged files in
-                        # this dir (entry will have the latest remote version
-                        # ID after checkout)
-                        entry.meta = meta
-                tree_meta, new_tree = build_tree(index, key)
-                out.obj = new_tree
-                out.hash_info = new_tree.hash_info
-                out.meta = tree_meta
-            else:
-                out.hash_info = entry.hash_info
-                if out.meta.version_id is None:
-                    out.meta = entry.meta
+                    continue
+            _fetch_out_changes(out, local_index, remote_index, remote)
+        stage.save(merge_versioned=True)
+        for out in stage.outs:
+            _update_out_meta(out, remote_index)
         stage.dvcfile.dump(stage, with_files=True, update_pipeline=False)
+
+
+def _fetch_out_changes(
+    out: "Output",
+    local_index: "DataIndex",
+    remote_index: "DataIndex",
+    remote: "Remote",
+):
+    from dvc_data.index import DataIndex, checkout
+
+    _, key = out.index_key
+    old = DataIndex()
+    new = DataIndex()
+    for _, entry in local_index.iteritems(key):
+        old.add(entry)
+    for _, entry in remote_index.iteritems(key):
+        new.add(entry)
+    total = len(new)
+    with Callback.as_tqdm_callback(
+        unit="file", desc=f"Updating '{out}'", disable=total == 0
+    ) as cb:
+        cb.set_size(total)
+        checkout(
+            new,
+            out.repo.root_dir,
+            out.fs,
+            old=old,
+            delete=True,
+            update_meta=False,
+            meta_only=True,
+            meta_cmp_key=partial(_meta_checksum, remote.fs),
+            callback=cb,
+        )

dvc/stage/cache.py

@@ -267,11 +267,15 @@ class StageCache:
         return ret
 
     def push(self, remote: Optional[str], odb: Optional["ObjectDB"] = None):
-        dest_odb = odb or self.repo.cloud.get_remote_odb(remote)
+        dest_odb = odb or self.repo.cloud.get_remote_odb(
+            remote, "push --run-cache"
+        )
         return self.transfer(self.repo.odb.local, dest_odb)
 
     def pull(self, remote: Optional[str], odb: Optional["ObjectDB"] = None):
-        odb = odb or self.repo.cloud.get_remote_odb(remote)
+        odb = odb or self.repo.cloud.get_remote_odb(
+            remote, "fetch --run-cache"
+        )
         return self.transfer(odb, self.repo.odb.local)
 
     def get_used_objs(self, used_run_cache, *args, **kwargs):

dvc/utils/hydra.py

@@ -1,27 +1,9 @@
-import sys
 from pathlib import Path
 from typing import TYPE_CHECKING, List
 
-try:
-    from hydra import compose, initialize_config_dir
-    from hydra._internal.config_loader_impl import ConfigLoaderImpl
-    from hydra._internal.core_plugins.basic_sweeper import BasicSweeper
-    from hydra.core.override_parser.overrides_parser import OverridesParser
-    from hydra.core.override_parser.types import ValueType
-    from hydra.errors import ConfigCompositionException, OverrideParseException
-    from omegaconf import OmegaConf
-
-    hydra_compatible = True
-except ValueError:
-    if sys.version_info >= (3, 11):
-        hydra_compatible = False
-    else:
-        raise
-
-from dvc.exceptions import DvcException, InvalidArgumentError
+from dvc.exceptions import InvalidArgumentError
 
 from .collections import merge_dicts, remove_missing_keys, to_omegaconf
-from .serialize import DUMPERS, MODIFIERS
 
 if TYPE_CHECKING:
     from dvc.types import StrPath
@@ -46,10 +28,10 @@ def compose_and_dump(
     .. _Hydra Override:
         https://hydra.cc/docs/advanced/override_grammar/basic/
     """
-    if not hydra_compatible:
-        raise DvcException(
-            "hydra functionality is not supported in Python >= 3.11"
-        )
+    from hydra import compose, initialize_config_dir
+    from omegaconf import OmegaConf
+
+    from .serialize import DUMPERS
 
     with initialize_config_dir(config_dir, version_base=None):
         cfg = compose(config_name=config_name, overrides=overrides)
@@ -71,10 +53,11 @@ def apply_overrides(path: "StrPath", overrides: List[str]) -> None:
     .. _Hydra Override:
         https://hydra.cc/docs/next/advanced/override_grammar/basic/
     """
-    if not hydra_compatible:
-        raise DvcException(
-            "hydra functionality is not supported in Python >= 3.11"
-        )
+    from hydra._internal.config_loader_impl import ConfigLoaderImpl
+    from hydra.errors import ConfigCompositionException, OverrideParseException
+    from omegaconf import OmegaConf
+
+    from .serialize import MODIFIERS
 
     suffix = Path(path).suffix.lower()
 
@@ -101,10 +84,7 @@ def apply_overrides(path: "StrPath", overrides: List[str]) -> None:
 
 
 def to_hydra_overrides(path_overrides):
-    if not hydra_compatible:
-        raise DvcException(
-            "hydra functionality is not supported in Python >= 3.11"
-        )
+    from hydra.core.override_parser.overrides_parser import OverridesParser
 
     parser = OverridesParser.create()
     return parser.parse_overrides(overrides=path_overrides)
@@ -117,10 +97,8 @@ def dict_product(dicts):
 
 
 def get_hydra_sweeps(path_overrides):
-    if not hydra_compatible:
-        raise DvcException(
-            "hydra functionality is not supported in Python >= 3.11"
-        )
+    from hydra._internal.core_plugins.basic_sweeper import BasicSweeper
+    from hydra.core.override_parser.types import ValueType
 
     path_sweeps = {}
     for path, overrides in path_overrides.items():

pyproject.toml

@@ -51,10 +51,10 @@ dependencies = [
     "pyparsing>=2.4.7",
     "typing-extensions>=3.7.4",
     "scmrepo==0.1.4",
-    "dvc-render==0.0.14",
-    "dvc-task==0.1.6",
-    "dvclive>=1.0",
-    "dvc-data==0.28.3",
+    "dvc-render==0.0.15",
+    "dvc-task==0.1.8",
+    "dvclive>=1.2.2",
+    "dvc-data==0.28.4",
     "dvc-http==2.27.2",
     "hydra-core>=1.1.0",
     "iterative-telemetry==0.0.6",
@@ -90,18 +90,18 @@ tests = [
     # Test requirements
     "pytest==7.2.0",
     "pytest-cov==4.0.0",
-    "pytest-xdist==3.0.2",
+    "pytest-xdist==3.1.0",
     "pytest-mock==3.10.0",
     "pytest-lazy-fixture==0.6.3",
     # https://github.com/docker/docker-py/issues/2902
     "pytest-docker==0.11.0; python_version < '3.10' or sys_platform != 'win32'",
     "flaky==3.7.0",
     "pytest-timeout==2.1.0",
-    "filelock==3.8.0",
+    "filelock==3.8.2",
     "beautifulsoup4==4.11.1",
     "pywin32>=225; sys_platform == 'win32'",  # optional dependency
     # pylint requirements
-    "pylint==2.15.7",
+    "pylint==2.15.8",
     # we use this to suppress pytest-related false positives in our tests.
     "pylint-pytest==1.1.2",
     # we use this to suppress some messages in tests, eg: foo/bar naming,
@@ -161,7 +161,7 @@ line_length = 79
 
 [tool.pytest.ini_options]
 log_level = "debug"
-addopts = "-ra -p no:hydra_pytest --cov-config pyproject.toml"
+addopts = "-ra --cov-config pyproject.toml"
 markers = [
     "needs_internet: Might need network access for the tests",
     "vscode: Tests verifying contract between DVC and VSCode plugin",

tests/conftest.py

@@ -250,18 +250,6 @@ def run_copy_metrics(tmp_dir, copy_script):
     return run
 
 
-@pytest.fixture(autouse=True)
-def mock_hydra_conf(mocker):
-    if sys.version_info < (3, 11):
-        return
-
-    # `hydra.conf` fails to import in 3.11, it raises ValueError due to changes
-    # in dataclasses. See https://github.com/python/cpython/pull/29867.
-    # NOTE: using sentinel here so that any imports from `hydra.conf`
-    # return a mock.
-    sys.modules["hydra.conf"] = mocker.sentinel
-
-
 @pytest.fixture(autouse=True)
 def gc_collect_on_dvc_close_on_win_311(mocker):
     if sys.version_info < (3, 11) and os.name != "nt":

tests/func/experiments/test_experiments.py

@@ -684,7 +684,7 @@ def test_experiments_workspace_not_log_exception(caplog, dvc, scm):
     assert not caplog.text
 
 
-def test_run_env(tmp_dir, dvc, scm):
+def test_run_env(tmp_dir, dvc, scm, mocker):
     dump_run_env = dedent(
         """\
         import os
@@ -703,7 +703,7 @@ def test_run_env(tmp_dir, dvc, scm):
     )
     dvc.experiments.run()
     assert (tmp_dir / "DVC_EXP_BASELINE_REV").read_text().strip() == baseline
-    assert (tmp_dir / "DVC_EXP_NAME").read_text().strip() == ""
+    assert (tmp_dir / "DVC_EXP_NAME").read_text().strip() != ""
 
     dvc.experiments.run(name="foo")
     assert (tmp_dir / "DVC_EXP_BASELINE_REV").read_text().strip() == baseline

tests/func/experiments/test_set_params.py

@@ -1,5 +1,3 @@
-import sys
-
 import pytest
 
 from dvc.exceptions import InvalidArgumentError
@@ -21,18 +19,7 @@ def test_modify_params(params_repo, dvc, changes, expected):
         assert fobj.read().strip() == expected
 
 
-@pytest.mark.parametrize(
-    "hydra_enabled",
-    [
-        pytest.param(
-            True,
-            marks=pytest.mark.skipif(
-                sys.version_info >= (3, 11), reason="unsupported on 3.11"
-            ),
-        ),
-        False,
-    ],
-)
+@pytest.mark.parametrize("hydra_enabled", [True, False])
 @pytest.mark.parametrize(
     "config_dir,config_name",
     [
@@ -93,12 +80,7 @@ def test_hydra_compose_and_dump(
     "hydra_enabled,overrides,expected",
     [
         (
-            pytest.param(
-                True,
-                marks=pytest.mark.xfail(
-                    sys.version_info >= (3, 11), reason="unsupported on 3.11"
-                ),
-            ),
+            True,
             ["db=mysql,postgresql"],
             [
                 {"params.yaml": ["db=mysql"]},

tests/func/experiments/test_show.py

@@ -220,6 +220,7 @@ def test_show_failed_experiment(tmp_dir, scm, dvc, failed_exp_stage):
                 "msg": "Experiment run failed.",
                 "type": "",
             },
+            "name": ANY,
         }
     }
 

tests/func/experiments/test_utils.py

@@ -0,0 +1,29 @@
+from funcy import first
+
+
+def test_generate_random_exp_name(tmp_dir, dvc, scm, exp_stage, mocker):
+    mocker.patch(
+        "dvc.repo.experiments.utils.random.choice",
+        side_effect=[
+            0,
+            0,
+            0,
+            0,
+            1,
+            1,
+            0,
+            0,
+        ],
+    )
+
+    ref = first(dvc.experiments.run(exp_stage.addressing, params=["foo=1"]))
+    assert dvc.experiments.get_exact_name([ref])[ref] == "0-0"
+
+    # Causes 1 retry
+    ref = first(dvc.experiments.run(exp_stage.addressing, params=["foo=2"]))
+    assert dvc.experiments.get_exact_name([ref])[ref] == "1-1"
+
+    tmp_dir.scm_gen({"foo": "bar"}, commit="foo")
+    # Can use same name because of different baseline_rev
+    ref = first(dvc.experiments.run(exp_stage.addressing, params=["foo=1"]))
+    assert dvc.experiments.get_exact_name([ref])[ref] == "0-0"

tests/func/utils/test_hydra.py

@@ -1,5 +1,3 @@
-import sys
-
 import pytest
 
 from dvc.exceptions import InvalidArgumentError
@@ -93,16 +91,16 @@ from dvc.exceptions import InvalidArgumentError
 def test_apply_overrides(tmp_dir, suffix, overrides, expected):
     from dvc.utils.hydra import apply_overrides
 
-    if suffix == "toml":
-        if overrides in [
-            ["foo=baz"],
-            ["foo.0=bar"],
-            ["foo=baz", "goo=bar"],
-            ["lorem=null"],
-        ]:
-            # TOML dumper breaks when overriding a list/dict with other type
-            # or when handling `null` values.
-            pytest.xfail()
+    if suffix == "toml" and overrides in [
+        ["foo=baz"],
+        ["foo.0=bar"],
+        ["foo=baz", "goo=bar"],
+        ["lorem=null"],
+    ]:
+        pytest.skip(
+            "TOML dumper breaks when overriding a list/dict with other type or"
+            " when handling `null` values."
+        )
 
     params_file = tmp_dir / f"params.{suffix}"
     params_file.dump(
@@ -140,7 +138,6 @@ def hydra_setup(tmp_dir, config_dir, config_name):
     return str(config_dir)
 
 
-@pytest.mark.skipif(sys.version_info >= (3, 11), reason="unsupported on 3.11")
 @pytest.mark.parametrize("suffix", ["yaml", "toml", "json"])
 @pytest.mark.parametrize(
     "overrides,expected",
@@ -180,7 +177,6 @@ def test_compose_and_dump(tmp_dir, suffix, overrides, expected):
     assert output_file.parse() == expected
 
 
-@pytest.mark.skipif(sys.version_info >= (3, 11), reason="unsupported on 3.11")
 def test_compose_and_dump_yaml_handles_string(tmp_dir):
     """Regression test for 8583"""
     from dvc.utils.hydra import compose_and_dump