deps: remove 3.11 checks for hydra; has 3.11 support now

Bumps [pytest-xdist](https://github.com/pytest-dev/pytest-xdist) from 3.0.2 to 3.1.0.
- [Release notes](https://github.com/pytest-dev/pytest-xdist/releases)
- [Changelog](https://github.com/pytest-dev/pytest-xdist/blob/master/CHANGELOG.rst)
- [Commits](https://github.com/pytest-dev/pytest-xdist/compare/v3.0.2...v3.1.0)

---
updated-dependencies:
- dependency-name: pytest-xdist
  dependency-type: direct:production
  update-type: version-update:semver-minor
...

Signed-off-by: dependabot[bot] <[email protected]>

dvc/commands/experiments/__init__.py

@@ -18,6 +18,7 @@ from dvc.commands.experiments import (
     queue_worker,
     remove,
     run,
+    save,
     show,
 )
 
@@ -34,6 +35,7 @@ SUB_COMMANDS = [
     queue_worker,
     remove,
     run,
+    save,
     show,
 ]
 

dvc/commands/experiments/ls.py

@@ -3,6 +3,7 @@ import logging
 
 from dvc.cli.command import CmdBase
 from dvc.cli.utils import append_doc_link
+from dvc.ui import ui
 
 logger = logging.getLogger(__name__)
 
@@ -16,20 +17,13 @@ class CmdExperimentsList(CmdBase):
             num=self.args.num,
             git_remote=self.args.git_remote,
         )
-        tags = self.repo.scm.describe(exps)
-        remained = {baseline for baseline, tag in tags.items() if tag is None}
-        base = "refs/heads"
-        ref_heads = self.repo.scm.describe(remained, base=base)
 
         for baseline in exps:
-            name = baseline[:7]
-            if tags[baseline] or ref_heads[baseline]:
-                name = tags[baseline] or ref_heads[baseline][len(base) + 1 :]
             if not name_only:
-                print(f"{name}:")
+                ui.write(f"{baseline}:")
             for exp_name in exps[baseline]:
                 indent = "" if name_only else "\t"
-                print(f"{indent}{exp_name}")
+                ui.write(f"{indent}{exp_name}")
 
         return 0
 

dvc/commands/experiments/save.py

@@ -0,0 +1,79 @@
+import argparse
+import logging
+
+from dvc.cli.command import CmdBase
+from dvc.cli.utils import append_doc_link
+from dvc.exceptions import DvcException
+from dvc.ui import ui
+
+logger = logging.getLogger(__name__)
+
+
+class CmdExperimentsSave(CmdBase):
+    def run(self):
+
+        try:
+            ref = self.repo.experiments.save(
+                name=self.args.name,
+                force=self.args.force,
+                include_untracked=self.args.include_untracked,
+            )
+        except DvcException:
+            logger.exception("failed to save experiment")
+            return 1
+
+        if self.args.json:
+            ui.write_json({"ref": ref})
+        else:
+            name = self.repo.experiments.get_exact_name([ref])[ref]
+            ui.write(f"Experiment has been saved as: {name}")
+            ui.write(
+                "\nTo promote an experiment to a Git branch run:\n\n"
+                "\tdvc exp branch <exp> <branch>\n"
+            )
+
+        return 0
+
+
+def add_parser(experiments_subparsers, parent_parser):
+    EXPERIMENTS_SAVE_HELP = "Save current workspace as a dvc experiment."
+    save_parser = experiments_subparsers.add_parser(
+        "save",
+        parents=[parent_parser],
+        description=append_doc_link(EXPERIMENTS_SAVE_HELP, "exp/save"),
+        help=EXPERIMENTS_SAVE_HELP,
+        formatter_class=argparse.RawDescriptionHelpFormatter,
+    )
+    save_parser.add_argument(
+        "-f",
+        "--force",
+        action="store_true",
+        default=False,
+        help="Save even if hash value for dependencies/outputs changed.",
+    )
+    save_parser.add_argument(
+        "--json",
+        "--show-json",
+        action="store_true",
+        default=False,
+        help="Show output in JSON format.",
+    )
+    save_parser.add_argument(
+        "-n",
+        "--name",
+        default=None,
+        help=(
+            "Human-readable experiment name. If not specified, a name will "
+            "be auto-generated."
+        ),
+        metavar="<name>",
+    )
+    save_parser.add_argument(
+        "-I",
+        "--include-untracked",
+        action="append",
+        default=[],
+        help="List of untracked paths to include in the experiment.",
+        metavar="<path>",
+    )
+    save_parser.set_defaults(func=CmdExperimentsSave)

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,6 +5,7 @@ from typing import TYPE_CHECKING, Iterable, Optional
 
 from funcy import cached_property
 
+from dvc.config import NoRemoteError, RemoteConfigError
 from dvc_data.hashfile.db import get_index
 
 if TYPE_CHECKING:
@@ -23,6 +24,14 @@ class Remote:
 
         self.worktree = config.pop("worktree", False)
         self.config = config
+        if self.worktree:
+            version_aware = self.config.get("version_aware")
+            if version_aware is False:
+                raise RemoteConfigError(
+                    "worktree remotes require version_aware cloud"
+                )
+            if version_aware is None:
+                self.fs.version_aware = True
 
     @cached_property
     def odb(self):
@@ -54,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")
 
@@ -88,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/env.py

@@ -4,6 +4,8 @@ DVC_PAGER = "DVC_PAGER"
 DVC_ROOT = "DVC_ROOT"
 DVCLIVE_RESUME = "DVCLIVE_RESUME"
 DVC_IGNORE_ISATTY = "DVC_IGNORE_ISATTY"
+DVC_EXP_BASELINE_REV = "DVC_EXP_BASELINE_REV"
+DVC_EXP_NAME = "DVC_EXP_NAME"
 DVC_EXP_GIT_REMOTE = "DVC_EXP_GIT_REMOTE"
 DVC_EXP_AUTO_PUSH = "DVC_EXP_AUTO_PUSH"
 DVC_NO_ANALYTICS = "DVC_NO_ANALYTICS"

dvc/render/converter/vega.py

@@ -59,13 +59,12 @@ def _verify_field(file2datapoints: Dict[str, List], filename: str, field: str):
     return
 
 
-def _get_x(properties: Dict, file2datapoints: Dict[str, List[Dict]]):
+def _get_xs(properties: Dict, file2datapoints: Dict[str, List[Dict]]):
     x = properties.get("x", None)
     if x is not None and isinstance(x, dict):
-        filename, field = first(_file_field(x))
-        _verify_field(file2datapoints, filename, field)
-        return filename, field
-    return None
+        for filename, field in _file_field(x):
+            _verify_field(file2datapoints, filename, field)
+            yield filename, field
 
 
 def _get_ys(properties, file2datapoints: Dict[str, List[Dict]]):
@@ -193,20 +192,28 @@ class VegaConverter(Converter):
 
         props_update = {}
 
-        x = _get_x(properties, file2datapoints)
+        xs = list(_get_xs(properties, file2datapoints))
 
         # assign "step" if no x provided
-        if not x:
+        if not xs:
             x_file, x_field = (
                 None,
                 INDEX_FIELD,
             )
         else:
-            x_file, x_field = x
+            x_file, x_field = xs[0]
         props_update["x"] = x_field
 
         ys = list(_get_ys(properties, file2datapoints))
 
+        num_xs = len(xs)
+        num_ys = len(ys)
+        if num_xs > 1 and num_xs != num_ys:
+            raise DvcException(
+                f"Cannot have different number of x and y data sources. Found "
+                f"{num_xs} x and {num_ys} y data sources."
+            )
+
         all_datapoints = []
         all_y_fields = {y_field for _, y_field in ys}
 
@@ -216,7 +223,9 @@ class VegaConverter(Converter):
         else:
             props_update["y"] = first(all_y_fields)
 
-        for y_file, y_field in ys:
+        for i, (y_file, y_field) in enumerate(ys):
+            if num_xs > 1:
+                x_file, x_field = xs[i]
             datapoints = deepcopy(file2datapoints.get(y_file, []))
 
             if props_update.get("y", None) == "dvc_inferred_y_value":
@@ -239,7 +248,7 @@ class VegaConverter(Converter):
                 except IndexError:
                     raise DvcException(
                         f"Cannot join '{x_field}' from '{x_file}' and "
-                        "'{y_field}' from '{y_file}'. "
+                        f"'{y_field}' from '{y_file}'. "
                         "They have to have same length."
                     )
 

dvc/repo/experiments/__init__.py

@@ -32,7 +32,7 @@ from .refs import (
     ExpRefInfo,
 )
 from .stash import ExpStashEntry
-from .utils import exp_refs_by_rev, scm_locked, unlocked_repo
+from .utils import exp_refs_by_rev, unlocked_repo
 
 logger = logging.getLogger(__name__)
 
@@ -50,18 +50,12 @@ class Experiments:
     )
 
     def __init__(self, repo):
-        from dvc.lock import make_lock
         from dvc.scm import NoSCMError
 
         if repo.config["core"].get("no_scm", False):
             raise NoSCMError
 
         self.repo = repo
-        self.scm_lock = make_lock(
-            os.path.join(self.repo.tmp_dir, "exp_scm_lock"),
-            tmp_dir=self.repo.tmp_dir,
-            hardlink_lock=repo.config["core"].get("hardlink_lock", False),
-        )
 
     @property
     def scm(self):
@@ -253,7 +247,6 @@ class Experiments:
         if self.scm.get_ref(str(exp_ref)):
             raise ExperimentExistsError(exp_ref.name)
 
-    @scm_locked
     def new(
         self,
         queue: BaseStashQueue,
@@ -279,7 +272,6 @@ class Experiments:
         except ExperimentExistsError as err:
             if not (kwargs.get("force", False) or kwargs.get("reset", False)):
                 raise err
-
         return queue.put(*args, **kwargs)
 
     def _resume_checkpoint(
@@ -487,6 +479,11 @@ class Experiments:
 
         return run(self.repo, *args, **kwargs)
 
+    def save(self, *args, **kwargs):
+        from dvc.repo.experiments.save import save
+
+        return save(self.repo, *args, **kwargs)
+
     def gc(self, *args, **kwargs):
         from dvc.repo.experiments.gc import gc
 

dvc/repo/experiments/executor/base.py

@@ -12,6 +12,7 @@ from typing import (
     Callable,
     Dict,
     Iterable,
+    List,
     NamedTuple,
     Optional,
     Tuple,
@@ -34,17 +35,7 @@ from ..exceptions import (
     ExperimentExistsError,
     UnchangedExperimentError,
 )
-from ..refs import (
-    EXEC_BASELINE,
-    EXEC_BRANCH,
-    EXEC_CHECKPOINT,
-    EXEC_HEAD,
-    EXEC_MERGE,
-    EXEC_NAMESPACE,
-    EXPS_NAMESPACE,
-    EXPS_STASH,
-    ExpRefInfo,
-)
+from ..refs import EXEC_BASELINE, EXEC_BRANCH, EXEC_CHECKPOINT, ExpRefInfo
 
 if TYPE_CHECKING:
     from queue import Queue
@@ -58,10 +49,6 @@ if TYPE_CHECKING:
 logger = logging.getLogger(__name__)
 
 
-EXEC_TMP_DIR = "exps"
-EXEC_PID_DIR = "run"
-
-
 class ExecutorResult(NamedTuple):
     exp_hash: Optional[str]
     ref_info: Optional["ExpRefInfo"]
@@ -172,6 +159,7 @@ class BaseExecutor(ABC):
     @abstractmethod
     def init_git(
         self,
+        repo: "Repo",
         scm: "Git",
         stash_rev: str,
         entry: "ExpStashEntry",
@@ -319,6 +307,7 @@ class BaseExecutor(ABC):
     def fetch_exps(
         self,
         dest_scm: "Git",
+        refs: List[str],
         force: bool = False,
         on_diverged: Callable[[str, bool], None] = None,
         **kwargs,
@@ -327,26 +316,19 @@ class BaseExecutor(ABC):
 
         Args:
             dest_scm: Destination Git instance.
+            refs: reference names to be fetched from the remotes.
             force: If True, diverged refs will be overwritten
             on_diverged: Callback in the form on_diverged(ref, is_checkpoint)
                 to be called when an experiment ref has diverged.
 
         Extra kwargs will be passed into the remote git client.
         """
-        from ..utils import iter_remote_refs
 
-        refs = []
-        has_checkpoint = False
-        for ref in iter_remote_refs(
-            dest_scm,
-            self.git_url,
-            base=EXPS_NAMESPACE,
-            **kwargs,
-        ):
-            if ref == EXEC_CHECKPOINT:
-                has_checkpoint = True
-            elif not ref.startswith(EXEC_NAMESPACE) and ref != EXPS_STASH:
-                refs.append(ref)
+        if EXEC_CHECKPOINT in refs:
+            refs.remove(EXEC_CHECKPOINT)
+            has_checkpoint = True
+        else:
+            has_checkpoint = False
 
         def on_diverged_ref(orig_ref: str, new_rev: str):
             if force:
@@ -755,21 +737,12 @@ class BaseExecutor(ABC):
             dvc_logger.setLevel(level)
 
     @contextmanager
-    def set_exec_refs(
-        self, scm: "Git", stash_rev: str, entry: "ExpStashEntry"
-    ):
+    def set_temp_refs(self, scm: "Git", temp_dict: Dict[str, str]):
         try:
-            # Executor will be initialized with an empty git repo that
-            # we populate by pushing:
-            #   EXEC_HEAD - the base commit for this experiment
-            #   EXEC_MERGE - the unmerged changes (from our stash)
-            #       to be reproduced
-            #   EXEC_BASELINE - the baseline commit for this experiment
-            scm.set_ref(EXEC_HEAD, entry.head_rev)
-            scm.set_ref(EXEC_MERGE, stash_rev)
-            scm.set_ref(EXEC_BASELINE, entry.baseline_rev)
+            for ref, rev in temp_dict.items():
+                scm.set_ref(ref, rev)
             yield
         finally:
-            for ref in (EXEC_HEAD, EXEC_MERGE, EXEC_BASELINE):
+            for ref in temp_dict:
                 if scm.get_ref(ref):
                     scm.remove_ref(ref)

dvc/repo/experiments/executor/local.py

@@ -2,11 +2,14 @@ import logging
 import os
 from contextlib import ExitStack
 from tempfile import mkdtemp
-from typing import TYPE_CHECKING, Optional
+from typing import TYPE_CHECKING, List, Optional
 
-from funcy import cached_property
+from funcy import cached_property, retry
 from scmrepo.exceptions import SCMError as _SCMError
+from shortuuid import uuid
 
+from dvc.exceptions import DvcException
+from dvc.lock import LockError
 from dvc.scm import SCM, GitMergeError
 from dvc.utils.fs import makedirs, remove
 
@@ -18,16 +21,19 @@ from ..refs import (
     EXEC_HEAD,
     EXEC_MERGE,
     EXEC_NAMESPACE,
+    EXPS_TEMP,
+    ExpRefInfo,
 )
-from .base import EXEC_TMP_DIR, BaseExecutor, TaskStatus
+from ..utils import EXEC_TMP_DIR, get_exp_rwlock
+from .base import BaseExecutor, ExecutorResult, TaskStatus
 
 if TYPE_CHECKING:
     from scmrepo.git import Git
 
     from dvc.repo import Repo
 
-    from ..refs import ExpRefInfo
     from ..stash import ExpStashEntry
+    from .base import ExecutorInfo
 
 logger = logging.getLogger(__name__)
 
@@ -67,8 +73,10 @@ class TempDirExecutor(BaseLocalExecutor):
     QUIET = True
     DEFAULT_LOCATION = "tempdir"
 
+    @retry(180, errors=LockError, timeout=1)
     def init_git(
         self,
+        repo: "Repo",
         scm: "Git",
         stash_rev: str,
         entry: "ExpStashEntry",
@@ -85,18 +93,43 @@ class TempDirExecutor(BaseLocalExecutor):
         if infofile:
             self.info.dump_json(infofile)
 
-        with self.set_exec_refs(scm, stash_rev, entry):
-            refspec = f"{EXEC_NAMESPACE}/"
-            push_refspec(scm, self.git_url, refspec, refspec)
+        temp_head = f"{EXPS_TEMP}/head-{uuid()}"
+        temp_merge = f"{EXPS_TEMP}/merge-{uuid()}"
+        temp_baseline = f"{EXPS_TEMP}/baseline-{uuid()}"
 
-        if branch:
-            push_refspec(scm, self.git_url, branch, branch)
-            self.scm.set_ref(EXEC_BRANCH, branch, symbolic=True)
-        elif self.scm.get_ref(EXEC_BRANCH):
-            self.scm.remove_ref(EXEC_BRANCH)
+        temp_ref_dict = {
+            temp_head: entry.head_rev,
+            temp_merge: stash_rev,
+            temp_baseline: entry.baseline_rev,
+        }
+        with get_exp_rwlock(
+            repo, writes=[temp_head, temp_merge, temp_baseline]
+        ), self.set_temp_refs(scm, temp_ref_dict):
+            # Executor will be initialized with an empty git repo that
+            # we populate by pushing:
+            #   EXEC_HEAD - the base commit for this experiment
+            #   EXEC_MERGE - the unmerged changes (from our stash)
+            #       to be reproduced
+            #   EXEC_BASELINE - the baseline commit for this experiment
+            refspec = [
+                (temp_head, EXEC_HEAD),
+                (temp_merge, EXEC_MERGE),
+                (temp_baseline, EXEC_BASELINE),
+            ]
+
+            if branch:
+                refspec.append((branch, branch))
+                with get_exp_rwlock(repo, reads=[branch]):
+                    push_refspec(scm, self.git_url, refspec)
+                self.scm.set_ref(EXEC_BRANCH, branch, symbolic=True)
+            else:
+                push_refspec(scm, self.git_url, refspec)
+                if self.scm.get_ref(EXEC_BRANCH):
+                    self.scm.remove_ref(EXEC_BRANCH)
+
+            if self.scm.get_ref(EXEC_CHECKPOINT):
+                self.scm.remove_ref(EXEC_CHECKPOINT)
 
-        if self.scm.get_ref(EXEC_CHECKPOINT):
-            self.scm.remove_ref(EXEC_CHECKPOINT)
         # checkout EXEC_HEAD and apply EXEC_MERGE on top of it without
         # committing
         head = EXEC_BRANCH if branch else EXEC_HEAD
@@ -167,8 +200,10 @@ class WorkspaceExecutor(BaseLocalExecutor):
         logger.debug("Init workspace executor in '%s'", root_dir)
         return executor
 
+    @retry(180, errors=LockError, timeout=1)
     def init_git(
         self,
+        repo: "Repo",
         scm: "Git",
         stash_rev: str,
         entry: "ExpStashEntry",
@@ -179,25 +214,26 @@ class WorkspaceExecutor(BaseLocalExecutor):
         if infofile:
             self.info.dump_json(infofile)
 
-        scm.set_ref(EXEC_HEAD, entry.head_rev)
-        scm.set_ref(EXEC_MERGE, stash_rev)
-        scm.set_ref(EXEC_BASELINE, entry.baseline_rev)
-        self._detach_stack.enter_context(
-            self.scm.detach_head(
-                self.scm.get_ref(EXEC_HEAD),
-                force=True,
-                client="dvc",
+        with get_exp_rwlock(repo, writes=[EXEC_NAMESPACE]):
+            scm.set_ref(EXEC_HEAD, entry.head_rev)
+            scm.set_ref(EXEC_MERGE, stash_rev)
+            scm.set_ref(EXEC_BASELINE, entry.baseline_rev)
+            self._detach_stack.enter_context(
+                self.scm.detach_head(
+                    self.scm.get_ref(EXEC_HEAD),
+                    force=True,
+                    client="dvc",
+                )
             )
-        )
-        merge_rev = self.scm.get_ref(EXEC_MERGE)
-        try:
-            self.scm.merge(merge_rev, squash=True, commit=False)
-        except _SCMError as exc:
-            raise GitMergeError(str(exc), scm=self.scm)
-        if branch:
-            self.scm.set_ref(EXEC_BRANCH, branch, symbolic=True)
-        elif scm.get_ref(EXEC_BRANCH):
-            self.scm.remove_ref(EXEC_BRANCH)
+            merge_rev = self.scm.get_ref(EXEC_MERGE)
+            try:
+                self.scm.merge(merge_rev, squash=True, commit=False)
+            except _SCMError as exc:
+                raise GitMergeError(str(exc), scm=self.scm)
+            if branch:
+                self.scm.set_ref(EXEC_BRANCH, branch, symbolic=True)
+            elif scm.get_ref(EXEC_BRANCH):
+                self.scm.remove_ref(EXEC_BRANCH)
 
     def init_cache(self, repo: "Repo", rev: str, run_cache: bool = True):
         pass
@@ -213,3 +249,57 @@ class WorkspaceExecutor(BaseLocalExecutor):
             checkpoint = self.scm.get_ref(EXEC_CHECKPOINT)
             if checkpoint and checkpoint != self._orig_checkpoint:
                 self.scm.set_ref(EXEC_APPLY, checkpoint)
+
+    @classmethod
+    def save(
+        cls,
+        info: "ExecutorInfo",
+        force: bool = False,
+        include_untracked: Optional[List[str]] = None,
+    ) -> ExecutorResult:
+        from dvc.repo import Repo
+
+        exp_hash: Optional[str] = None
+        exp_ref: Optional[ExpRefInfo] = None
+
+        dvc = Repo(os.path.join(info.root_dir, info.dvc_dir))
+        old_cwd = os.getcwd()
+        if info.wdir:
+            os.chdir(os.path.join(dvc.scm.root_dir, info.wdir))
+        else:
+            os.chdir(dvc.root_dir)
+
+        try:
+            stages = dvc.commit([], force=force)
+            exp_hash = cls.hash_exp(stages)
+            if include_untracked:
+                dvc.scm.add(include_untracked)
+            cls.commit(
+                dvc.scm,
+                exp_hash,
+                exp_name=info.name,
+                force=force,
+            )
+            ref: Optional[str] = dvc.scm.get_ref(EXEC_BRANCH, follow=False)
+            exp_ref = ExpRefInfo.from_ref(ref) if ref else None
+            untracked = dvc.scm.untracked_files()
+            if untracked:
+                logger.warning(
+                    "The following untracked files were present in "
+                    "the workspace before saving but "
+                    "will not be included in the experiment commit:\n"
+                    "\t%s",
+                    ", ".join(untracked),
+                )
+            info.result_hash = exp_hash
+            info.result_ref = ref
+            info.result_force = False
+            info.status = TaskStatus.SUCCESS
+        except DvcException:
+            info.status = TaskStatus.FAILED
+            raise
+        finally:
+            dvc.close()
+            os.chdir(old_cwd)
+
+        return ExecutorResult(ref, exp_ref, info.result_force)

dvc/repo/experiments/executor/ssh.py

@@ -9,6 +9,7 @@ from dvc_ssh import SSHFileSystem
 from funcy import first
 
 from ..refs import (
+    EXEC_BASELINE,
     EXEC_BRANCH,
     EXEC_CHECKPOINT,
     EXEC_HEAD,
@@ -119,6 +120,7 @@ class SSHExecutor(BaseExecutor):
 
     def init_git(
         self,
+        repo: "Repo",
         scm: "Git",
         stash_rev: str,
         entry: "ExpStashEntry",
@@ -147,12 +149,18 @@ class SSHExecutor(BaseExecutor):
             # (see https://github.com/iterative/dvc/issues/6508)
             kwargs = self._git_client_args(fs)
 
-            with self.set_exec_refs(scm, stash_rev, entry):
-                refspec = f"{EXEC_NAMESPACE}/"
-                push_refspec(scm, self.git_url, refspec, refspec, **kwargs)
+            ref_dict = {
+                EXEC_HEAD: entry.head_rev,
+                EXEC_MERGE: stash_rev,
+                EXEC_BASELINE: entry.baseline_rev,
+            }
+            with self.set_temp_refs(scm, ref_dict):
+                exec_namespace = f"{EXEC_NAMESPACE}/"
+                refspec = [(exec_namespace, exec_namespace)]
+                push_refspec(scm, self.git_url, refspec, **kwargs)
 
             if branch:
-                push_refspec(scm, self.git_url, branch, branch, **kwargs)
+                push_refspec(scm, self.git_url, [(branch, branch)], **kwargs)
                 self._ssh_cmd(fs, f"git symbolic-ref {EXEC_BRANCH} {branch}")
             else:
                 self._ssh_cmd(

dvc/repo/experiments/ls.py

@@ -6,7 +6,7 @@ from dvc.repo.scm_context import scm_context
 from dvc.scm import iter_revs
 from dvc.types import Optional
 
-from .utils import exp_refs, exp_refs_by_baseline
+from .utils import exp_refs_by_baseline
 
 logger = logging.getLogger(__name__)
 
@@ -20,19 +20,22 @@ def ls(
     num: int = 1,
     git_remote: Optional[str] = None,
 ):
-    results = defaultdict(list)
-    if all_commits:
-        gen = exp_refs(repo.scm, git_remote)
-        for info in gen:
-            results[info.baseline_sha].append(info.name)
-        return results
-
-    rev = rev or "HEAD"
-
-    revs = iter_revs(repo.scm, [rev], num)
-    rev_set = set(revs.keys())
+    rev_set = None
+    if not all_commits:
+        revs = iter_revs(repo.scm, [rev or "HEAD"], num)
+        rev_set = set(revs.keys())
     ref_info_dict = exp_refs_by_baseline(repo.scm, rev_set, git_remote)
-    for rev, ref_info_list in ref_info_dict.items():
-        results[rev] = [ref_info.name for ref_info in ref_info_list]
+
+    tags = repo.scm.describe(ref_info_dict.keys())
+    remained = {baseline for baseline, tag in tags.items() if tag is None}
+    base = "refs/heads"
+    ref_heads = repo.scm.describe(remained, base=base)
+
+    results = defaultdict(list)
+    for baseline in ref_info_dict:
+        name = baseline[:7]
+        if tags[baseline] or ref_heads[baseline]:
+            name = tags[baseline] or ref_heads[baseline][len(base) + 1 :]
+        results[name] = [info.name for info in ref_info_dict[baseline]]
 
     return results

dvc/repo/experiments/queue/base.py

@@ -20,22 +20,18 @@ from typing import (
 from funcy import cached_property, retry
 
 from dvc.dependency import ParamsDependency
-from dvc.env import DVCLIVE_RESUME
+from dvc.env import DVC_EXP_BASELINE_REV, DVC_EXP_NAME, DVCLIVE_RESUME
 from dvc.exceptions import DvcException
 from dvc.lock import LockError
 from dvc.ui import ui
 
 from ..exceptions import CheckpointExistsError, ExperimentExistsError
-from ..executor.base import (
-    EXEC_PID_DIR,
-    EXEC_TMP_DIR,
-    BaseExecutor,
-    ExecutorResult,
-)
+from ..executor.base import BaseExecutor, ExecutorResult
 from ..executor.local import WorkspaceExecutor
 from ..refs import ExpRefInfo
 from ..stash import ExpStash, ExpStashEntry
-from ..utils import exp_refs_by_rev, scm_locked
+from ..utils import EXEC_PID_DIR, EXEC_TMP_DIR, exp_refs_by_rev, get_exp_rwlock
+from .utils import get_remote_executor_refs
 
 if TYPE_CHECKING:
     from scmrepo.git import Git
@@ -400,7 +396,13 @@ class BaseStashQueue(ABC):
                             )
 
                     # save additional repro command line arguments
-                    run_env = {DVCLIVE_RESUME: "1"} if resume_rev else {}
+                    run_env = {
+                        DVC_EXP_BASELINE_REV: baseline_rev,
+                    }
+                    if name:
+                        run_env[DVC_EXP_NAME] = name
+                    if resume_rev:
+                        run_env[DVCLIVE_RESUME] = "1"
                     self._pack_args(*args, run_env=run_env, **kwargs)
 
                     # save experiment as a stash commit
@@ -555,28 +557,39 @@ class BaseStashQueue(ABC):
 
     @staticmethod
     @retry(180, errors=LockError, timeout=1)
-    @scm_locked
+    def get_stash_entry(
+        exp: "Experiments",
+        queue_entry: QueueEntry,
+    ) -> "ExpStashEntry":
+        stash = ExpStash(exp.scm, queue_entry.stash_ref)
+        stash_rev = queue_entry.stash_rev
+        with get_exp_rwlock(exp.repo, writes=[queue_entry.stash_ref]):
+            stash_entry = stash.stash_revs.get(
+                stash_rev,
+                ExpStashEntry(None, stash_rev, stash_rev, None, None),
+            )
+            if stash_entry.stash_index is not None:
+                stash.drop(stash_entry.stash_index)
+        return stash_entry
+
+    @classmethod
     def init_executor(
+        cls,
         exp: "Experiments",
         queue_entry: QueueEntry,
         executor_cls: Type[BaseExecutor] = WorkspaceExecutor,
         **kwargs,
     ) -> BaseExecutor:
-        scm = exp.scm
-        stash = ExpStash(scm, queue_entry.stash_ref)
-        stash_rev = queue_entry.stash_rev
-        stash_entry = stash.stash_revs.get(
-            stash_rev,
-            ExpStashEntry(None, stash_rev, stash_rev, None, None),
-        )
-        if stash_entry.stash_index is not None:
-            stash.drop(stash_entry.stash_index)
+        stash_entry = cls.get_stash_entry(exp, queue_entry)
+
         executor = executor_cls.from_stash_entry(
             exp.repo, stash_entry, **kwargs
         )
 
+        stash_rev = queue_entry.stash_rev
         infofile = exp.celery_queue.get_infofile_path(stash_rev)
         executor.init_git(
+            exp.repo,
             exp.repo.scm,
             stash_rev,
             stash_entry,
@@ -597,7 +610,6 @@ class BaseStashQueue(ABC):
 
     @staticmethod
     @retry(180, errors=LockError, timeout=1)
-    @scm_locked
     def collect_git(
         exp: "Experiments",
         executor: BaseExecutor,
@@ -611,16 +623,20 @@ class BaseStashQueue(ABC):
                 raise CheckpointExistsError(ref_info.name)
             raise ExperimentExistsError(ref_info.name)
 
-        for ref in executor.fetch_exps(
-            exp.scm,
-            force=exec_result.force,
-            on_diverged=on_diverged,
-        ):
-            exp_rev = exp.scm.get_ref(ref)
-            if exp_rev:
-                assert exec_result.exp_hash
-                logger.debug("Collected experiment '%s'.", exp_rev[:7])
-                results[exp_rev] = exec_result.exp_hash
+        refs = get_remote_executor_refs(exp.scm, executor.git_url)
+
+        with get_exp_rwlock(exp.repo, writes=refs):
+            for ref in executor.fetch_exps(
+                exp.scm,
+                refs,
+                force=exec_result.force,
+                on_diverged=on_diverged,
+            ):
+                exp_rev = exp.scm.get_ref(ref)
+                if exp_rev:
+                    assert exec_result.exp_hash
+                    logger.debug("Collected experiment '%s'.", exp_rev[:7])
+                    results[exp_rev] = exec_result.exp_hash
 
         return results
 

dvc/repo/experiments/queue/celery.py

@@ -23,7 +23,9 @@ from dvc.exceptions import DvcException
 from dvc.ui import ui
 
 from ..exceptions import UnresolvedQueueExpNamesError
-from ..executor.base import EXEC_TMP_DIR, ExecutorInfo, ExecutorResult
+from ..executor.base import ExecutorInfo, ExecutorResult
+from ..refs import CELERY_STASH
+from ..utils import EXEC_TMP_DIR, get_exp_rwlock
 from .base import BaseStashQueue, QueueDoneResult, QueueEntry, QueueGetResult
 from .exceptions import CannotKillTasksError
 from .tasks import run_exp
@@ -164,7 +166,8 @@ class LocalCeleryQueue(BaseStashQueue):
 
     def put(self, *args, **kwargs) -> QueueEntry:
         """Stash an experiment and add it to the queue."""
-        entry = self._stash_exp(*args, **kwargs)
+        with get_exp_rwlock(self.repo, writes=["workspace", CELERY_STASH]):
+            entry = self._stash_exp(*args, **kwargs)
         self.celery.signature(run_exp.s(entry.asdict())).delay()
         return entry
 

dvc/repo/experiments/queue/tasks.py

@@ -1,4 +1,4 @@
-from typing import Any, Dict
+from typing import TYPE_CHECKING, Any, Dict
 
 from celery import shared_task
 from celery.utils.log import get_task_logger
@@ -7,11 +7,15 @@ from ..executor.base import ExecutorInfo
 from ..executor.local import TempDirExecutor
 from .base import BaseStashQueue, QueueEntry
 
+if TYPE_CHECKING:
+    from ..executor.base import BaseExecutor
+
+
 logger = get_task_logger(__name__)
 
 
 @shared_task
-def setup_exp(entry_dict: Dict[str, Any]) -> TempDirExecutor:
+def setup_exp(entry_dict: Dict[str, Any]) -> "BaseExecutor":
     """Setup an experiment.
 
     Arguments:

dvc/repo/experiments/queue/tempdir.py

@@ -6,14 +6,13 @@ from funcy import cached_property, first
 
 from ..exceptions import ExpQueueEmptyError
 from ..executor.base import (
-    EXEC_PID_DIR,
-    EXEC_TMP_DIR,
     BaseExecutor,
     ExecutorInfo,
     ExecutorResult,
     TaskStatus,
 )
 from ..executor.local import TempDirExecutor
+from ..utils import EXEC_PID_DIR, EXEC_TMP_DIR
 from .base import BaseStashQueue, QueueEntry, QueueGetResult
 from .utils import fetch_running_exp_from_temp_dir
 from .workspace import WorkspaceQueue

dvc/repo/experiments/queue/utils.py

@@ -1,17 +1,40 @@
 import logging
-from typing import TYPE_CHECKING, Dict
+from typing import TYPE_CHECKING, Dict, List
 
 from scmrepo.exceptions import SCMError
 
 from ..executor.base import ExecutorInfo, TaskStatus
+from ..refs import EXEC_CHECKPOINT, EXEC_NAMESPACE, EXPS_NAMESPACE, EXPS_STASH
+from ..utils import get_exp_rwlock, iter_remote_refs
 
 logger = logging.getLogger(__name__)
 
 
 if TYPE_CHECKING:
+    from scmrepo.git import Git
+
     from .base import BaseStashQueue
 
 
+def get_remote_executor_refs(scm: "Git", remote_url: str) -> List[str]:
+    """Get result list refs from a remote repository
+
+    Args:
+        remote_url : remote executor's url
+    """
+    refs = []
+    for ref in iter_remote_refs(
+        scm,
+        remote_url,
+        base=EXPS_NAMESPACE,
+    ):
+        if ref == EXEC_CHECKPOINT or (
+            not ref.startswith(EXEC_NAMESPACE) and ref != EXPS_STASH
+        ):
+            refs.append(ref)
+    return refs
+
+
 def fetch_running_exp_from_temp_dir(
     queue: "BaseStashQueue", rev: str, fetch_refs: bool
 ) -> Dict[str, Dict]:
@@ -36,7 +59,7 @@ def fetch_running_exp_from_temp_dir(
         info = ExecutorInfo.from_dict(load_json(infofile))
     except OSError:
         return result
-    if info.status < TaskStatus.FAILED:
+    if info.status <= TaskStatus.RUNNING:
         result[rev] = info.asdict()
         if info.git_url and fetch_refs and info.status > TaskStatus.PREPARING:
 
@@ -45,15 +68,18 @@ def fetch_running_exp_from_temp_dir(
 
             executor = TempDirExecutor.from_info(info)
             try:
-                for ref in executor.fetch_exps(
-                    queue.scm,
-                    on_diverged=on_diverged,
-                ):
-                    logger.debug("Updated running experiment '%s'.", ref)
-                    last_rev = queue.scm.get_ref(ref)
-                    result[rev]["last"] = last_rev
-                    if last_rev:
-                        result[last_rev] = info.asdict()
+                refs = get_remote_executor_refs(queue.scm, executor.git_url)
+                with get_exp_rwlock(queue.repo, writes=refs):
+                    for ref in executor.fetch_exps(
+                        queue.scm,
+                        refs,
+                        on_diverged=on_diverged,
+                    ):
+                        logger.debug("Updated running experiment '%s'.", ref)
+                        last_rev = queue.scm.get_ref(ref)
+                        result[rev]["last"] = last_rev
+                        if last_rev:
+                            result[last_rev] = info.asdict()
             except (InvalidRemoteSCMRepo, SCMError):
                 # ignore stale info files
                 del result[rev]

dvc/repo/experiments/queue/workspace.py

@@ -21,7 +21,8 @@ from ..executor.base import (
     TaskStatus,
 )
 from ..executor.local import WorkspaceExecutor
-from ..refs import EXEC_BRANCH
+from ..refs import EXEC_BRANCH, WORKSPACE_STASH
+from ..utils import get_exp_rwlock
 from .base import BaseStashQueue, QueueDoneResult, QueueEntry, QueueGetResult
 
 if TYPE_CHECKING:
@@ -34,7 +35,8 @@ class WorkspaceQueue(BaseStashQueue):
     _EXEC_NAME: Optional[str] = "workspace"
 
     def put(self, *args, **kwargs) -> QueueEntry:
-        return self._stash_exp(*args, **kwargs)
+        with get_exp_rwlock(self.repo, writes=["workspace", WORKSPACE_STASH]):
+            return self._stash_exp(*args, **kwargs)
 
     def get(self) -> QueueGetResult:
         revs = self.stash.stash_revs

dvc/repo/experiments/refs.py

@@ -16,6 +16,7 @@ EXEC_BRANCH = f"{EXEC_NAMESPACE}/EXEC_BRANCH"
 EXEC_BASELINE = f"{EXEC_NAMESPACE}/EXEC_BASELINE"
 EXEC_HEAD = f"{EXEC_NAMESPACE}/EXEC_HEAD"
 EXEC_MERGE = f"{EXEC_NAMESPACE}/EXEC_MERGE"
+EXPS_TEMP = f"{EXPS_NAMESPACE}/temp"
 STASHES = {WORKSPACE_STASH, CELERY_STASH}
 
 
@@ -23,9 +24,7 @@ class ExpRefInfo:
 
     namespace = EXPS_NAMESPACE
 
-    def __init__(
-        self, baseline_sha: Optional[str] = None, name: Optional[str] = None
-    ):
+    def __init__(self, baseline_sha: str, name: Optional[str] = None):
         self.baseline_sha = baseline_sha
         self.name: str = name if name else ""
 
@@ -33,7 +32,7 @@ class ExpRefInfo:
         return "/".join(self.parts)
 
     def __repr__(self):
-        baseline = f"'{self.baseline_sha}'" if self.baseline_sha else "None"
+        baseline = f"'{self.baseline_sha}'"
         name = f"'{self.name}'" if self.name else "None"
         return f"ExpRefInfo(baseline_sha={baseline}, name={name})"
 
@@ -41,11 +40,7 @@ class ExpRefInfo:
     def parts(self):
         return (
             (self.namespace,)
-            + (
-                (self.baseline_sha[:2], self.baseline_sha[2:])
-                if self.baseline_sha
-                else ()
-            )
+            + ((self.baseline_sha[:2], self.baseline_sha[2:]))
             + ((self.name,) if self.name else ())
         )
 
@@ -54,14 +49,13 @@ class ExpRefInfo:
         try:
             parts = ref.split("/")
             if (
-                len(parts) < 2
-                or len(parts) == 3
+                len(parts) < 4
                 or len(parts) > 5
                 or "/".join(parts[:2]) != EXPS_NAMESPACE
             ):
                 raise InvalidExpRefError(ref)
         except ValueError:
             raise InvalidExpRefError(ref)
-        baseline_sha = parts[2] + parts[3] if len(parts) >= 4 else None
+        baseline_sha = parts[2] + parts[3]
         name = parts[4] if len(parts) == 5 else None
         return cls(baseline_sha, name)

dvc/repo/experiments/remove.py

@@ -152,8 +152,7 @@ def _remove_commited_exps(
                 push_refspec(
                     scm,
                     remote,
-                    None,
-                    str(ref_info),
+                    [(None, str(ref_info))],
                     progress=pbar.update_git,
                 )
     else:

dvc/repo/experiments/save.py

@@ -0,0 +1,44 @@
+import logging
+import os
+from typing import TYPE_CHECKING, List, Optional
+
+from funcy import first
+
+if TYPE_CHECKING:
+    from dvc.repo import Repo
+
+
+logger = logging.getLogger(__name__)
+
+
+def save(
+    repo: "Repo",
+    name: Optional[str] = None,
+    force: bool = False,
+    include_untracked: Optional[List[str]] = None,
+) -> Optional[str]:
+    """Save the current workspace status as an experiment.
+
+    Returns the saved experiment's SHAs.
+    """
+    queue = repo.experiments.workspace_queue
+    logger.debug("Saving workspace in %s", os.getcwd())
+
+    staged, _, _ = repo.scm.status(untracked_files="no")
+    if staged:
+        logger.warning(
+            "Your workspace contains staged Git changes which will be "
+            "unstaged before saving this experiment."
+        )
+        repo.scm.reset()
+
+    entry = repo.experiments.new(queue=queue, name=name, force=force)
+    executor = queue.init_executor(repo.experiments, entry)
+
+    save_result = executor.save(
+        executor.info, force=force, include_untracked=include_untracked
+    )
+    result = queue.collect_executor(repo.experiments, executor, save_result)
+
+    exp_rev = first(result)
+    return exp_rev

dvc/repo/experiments/utils.py

@@ -1,6 +1,9 @@
+import os
+import sys
 from collections import defaultdict
 from functools import wraps
 from typing import (
+    TYPE_CHECKING,
     Callable,
     Dict,
     Generator,
@@ -9,6 +12,7 @@ from typing import (
     Mapping,
     Optional,
     Set,
+    Tuple,
     Union,
 )
 
@@ -16,6 +20,7 @@ from scmrepo.git import Git
 
 from dvc.exceptions import InvalidArgumentError
 from dvc.repo.experiments.exceptions import AmbiguousExpRefInfo
+from dvc.rwlock import rwlock
 
 from .refs import (
     EXEC_APPLY,
@@ -28,18 +33,34 @@ from .refs import (
     ExpRefInfo,
 )
 
+if TYPE_CHECKING:
+    from dvc.repo import Repo
 
-def scm_locked(f):
-    # Lock the experiments workspace so that we don't try to perform two
-    # different sequences of git operations at once
-    @wraps(f)
-    def wrapper(exp, *args, **kwargs):
-        from dvc.scm import map_scm_exception
 
-        with map_scm_exception(), exp.scm_lock:
-            return f(exp, *args, **kwargs)
+EXEC_TMP_DIR = "exps"
+EXEC_PID_DIR = "run"
 
-    return wrapper
+
+def get_exp_rwlock(
+    repo: "Repo",
+    reads: Optional[List[str]] = None,
+    writes: Optional[List[str]] = None,
+):
+    reads = reads or []
+    writes = writes or []
+
+    cmd = " ".join(sys.argv)
+    path = os.path.join(repo.tmp_dir, EXEC_TMP_DIR)
+    repo.fs.makedirs(path, exist_ok=True)
+
+    return rwlock(
+        path,
+        repo.fs,
+        cmd,
+        reads,
+        writes,
+        repo.config["core"].get("hardlink_lock", False),
+    )
 
 
 def unlocked_repo(f):
@@ -79,13 +100,13 @@ def exp_refs_by_rev(scm: "Git", rev: str) -> Generator[ExpRefInfo, None, None]:
 
 
 def exp_refs_by_baseline(
-    scm: "Git", revs: Set[str], url: Optional[str] = None
+    scm: "Git", revs: Optional[Set[str]] = None, url: Optional[str] = None
 ) -> Mapping[str, List[ExpRefInfo]]:
     """Iterate over all experiment refs with the specified baseline."""
     all_exp_refs = exp_refs(scm, url)
     result = defaultdict(list)
     for ref in all_exp_refs:
-        if ref.baseline_sha in revs:
+        if revs is None or ref.baseline_sha in revs:
             result[ref.baseline_sha].append(ref)
     return result
 
@@ -108,8 +129,7 @@ def iter_remote_refs(
 def push_refspec(
     scm: "Git",
     url: str,
-    src: Optional[str],
-    dest: str,
+    push_list=List[Tuple[Optional[str], str]],
     force: bool = False,
     on_diverged: Optional[Callable[[str, str], bool]] = None,
     **kwargs,
@@ -119,20 +139,21 @@ def push_refspec(
 
     from ...scm import GitAuthError, SCMError
 
-    if not src:
-        refspecs = [f":{dest}"]
-    elif src.endswith("/"):
-        refspecs = []
-        dest = dest.rstrip("/") + "/"
-        for ref in scm.iter_refs(base=src):
-            refname = ref.split("/")[-1]
-            refspecs.append(f"{ref}:{dest}{refname}")
-    else:
-        if dest.endswith("/"):
-            refname = src.split("/")[-1]
-            refspecs = [f"{src}:{dest}/{refname}"]
+    refspecs = []
+    for src, dest in push_list:
+        if not src:
+            refspecs.append(f":{dest}")
+        elif src.endswith("/"):
+            dest = dest.rstrip("/") + "/"
+            for ref in scm.iter_refs(base=src):
+                refname = ref.split("/")[-1]
+                refspecs.append(f"{ref}:{dest}{refname}")
         else:
-            refspecs = [f"{src}:{dest}"]
+            if dest.endswith("/"):
+                refname = src.split("/")[-1]
+                refspecs.append(f"{src}:{dest}/{refname}")
+            else:
+                refspecs.append(f"{src}:{dest}")
 
     try:
         results = scm.push_refspecs(

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

@@ -11,6 +11,7 @@ from typing import (
     Optional,
     Set,
     Tuple,
+    Union,
 )
 
 from funcy import cached_property, nullcontext
@@ -67,14 +68,50 @@ class Index:
         if stages is not None:
             self.stages: List["Stage"] = stages
         self._collected_targets: Dict[int, List["StageInfo"]] = {}
+        self._metrics: Dict[str, List[str]] = {}
+        self._plots: Dict[str, Any] = {}
+        self._params: Dict[str, List[str]] = {}
 
     @cached_property
     def stages(self) -> List["Stage"]:  # pylint: disable=method-hidden
         # note that ideally we should be keeping this in a set as it is unique,
         # hashable and has no concept of orderliness on its own. But we depend
         # on this to be somewhat ordered for status/metrics/plots, etc.
+        return self._collect()
+
+    @cached_property
+    def _top_metrics(self):
+        self._collect()
+        return self._metrics
+
+    @cached_property
+    def _top_plots(self):
+        self._collect()
+        return self._plots
+
+    @cached_property
+    def _top_params(self):
+        self._collect()
+        return self._params
+
+    def _collect(self):
+        if "stages" in self.__dict__:
+            return self.stages
+
         onerror = self.repo.stage_collection_error_handler
-        return self.stage_collector.collect_repo(onerror=onerror)
+
+        # pylint: disable=protected-access
+        (
+            stages,
+            metrics,
+            plots,
+            params,
+        ) = self.stage_collector._collect_all_from_repo(onerror=onerror)
+        self.stages = stages
+        self._metrics = metrics
+        self._plots = plots
+        self._params = params
+        return stages
 
     def __repr__(self) -> str:
         from dvc.fs import LocalFileSystem
@@ -129,7 +166,7 @@ class Index:
             )
         )
 
-    def _collect_targets(
+    def collect_targets(
         self, targets: Optional["TargetType"], **kwargs: Any
     ) -> List["StageInfo"]:
         from itertools import chain
@@ -177,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)
@@ -273,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:
@@ -429,10 +466,10 @@ class IndexView:
         return prefixes
 
     @cached_property
-    def data(self) -> "Dict[str, DataIndexView]":
+    def data(self) -> Dict[str, Union["DataIndex", "DataIndexView"]]:
         from functools import partial
 
-        from dvc_data.index import view
+        from dvc_data.index import DataIndex, view
 
         def key_filter(workspace: str, key: "DataIndexKey"):
             try:
@@ -444,10 +481,14 @@ class IndexView:
             except KeyError:
                 return False
 
-        data = {}
+        data: Dict[str, Union["DataIndex", "DataIndexView"]] = {}
         for workspace, data_index in self._index.data.items():
-            data_index.load()
-            data[workspace] = view(data_index, partial(key_filter, workspace))
+            if self._stages:
+                data[workspace] = view(
+                    data_index, partial(key_filter, workspace)
+                )
+            else:
+                data[workspace] = DataIndex()
         return data
 
 

dvc/repo/metrics/show.py

@@ -9,7 +9,12 @@ from dvc.output import Output
 from dvc.repo import locked
 from dvc.repo.collect import StrPaths, collect
 from dvc.scm import NoSCMError
-from dvc.utils import error_handler, errored_revisions, onerror_collect
+from dvc.utils import (
+    as_posix,
+    error_handler,
+    errored_revisions,
+    onerror_collect,
+)
 from dvc.utils.collections import ensure_list
 from dvc.utils.serialize import load_path
 
@@ -28,6 +33,17 @@ def _to_fs_paths(metrics: List[Output]) -> StrPaths:
     return result
 
 
+def _collect_top_level_metrics(repo):
+    top_metrics = repo.index._top_metrics  # pylint: disable=protected-access
+    for dvcfile, metrics in top_metrics.items():
+        wdir = repo.fs.path.relpath(
+            repo.fs.path.parent(dvcfile), repo.root_dir
+        )
+        for file in metrics:
+            path = repo.fs.path.join(wdir, as_posix(file))
+            yield repo.fs.path.normpath(path)
+
+
 def _collect_metrics(repo, targets, revision, recursive):
     metrics, fs_paths = collect(
         repo,
@@ -94,6 +110,7 @@ def _read_metrics(repo, metrics, rev, onerror=None):
 
 def _gather_metrics(repo, targets, rev, recursive, onerror=None):
     metrics = _collect_metrics(repo, targets, rev, recursive)
+    metrics.extend(_collect_top_level_metrics(repo))
     return _read_metrics(repo, metrics, rev, onerror=onerror)
 
 

dvc/repo/params/show.py

@@ -20,7 +20,12 @@ from dvc.repo.collect import collect
 from dvc.scm import NoSCMError
 from dvc.stage import PipelineStage
 from dvc.ui import ui
-from dvc.utils import error_handler, errored_revisions, onerror_collect
+from dvc.utils import (
+    as_posix,
+    error_handler,
+    errored_revisions,
+    onerror_collect,
+)
 from dvc.utils.collections import ensure_list
 from dvc.utils.serialize import load_path
 
@@ -35,6 +40,17 @@ def _is_params(dep: "Output"):
     return isinstance(dep, ParamsDependency)
 
 
+def _collect_top_level_params(repo):
+    top_params = repo.index._top_params  # pylint: disable=protected-access
+    for dvcfile, params in top_params.items():
+        wdir = repo.fs.path.relpath(
+            repo.fs.path.parent(dvcfile), repo.root_dir
+        )
+        for file in params:
+            path = repo.fs.path.join(wdir, as_posix(file))
+            yield repo.fs.path.normpath(path)
+
+
 def _collect_configs(
     repo: "Repo", rev, targets=None, deps=False, stages=None
 ) -> Tuple[List["Output"], List[str]]:
@@ -176,6 +192,7 @@ def _gather_params(
     param_outs, params_fs_paths = _collect_configs(
         repo, rev, targets=targets, deps=deps, stages=stages
     )
+    params_fs_paths.extend(_collect_top_level_params(repo=repo))
     params = _read_params(
         repo,
         params=param_outs,

dvc/repo/plots/__init__.py

@@ -439,36 +439,33 @@ def _resolve_definitions(
 
 
 def _collect_pipeline_files(repo, targets: List[str], props, onerror=None):
-    from dvc.dvcfile import PipelineFile
-
     result: Dict[str, Dict] = {}
-    dvcfiles = {stage.dvcfile for stage in repo.index.stages}
-    for dvcfile in dvcfiles:
-        if isinstance(dvcfile, PipelineFile):
-            dvcfile_path = _relpath(repo.dvcfs, dvcfile.path)
-            dvcfile_defs = dvcfile.load().get("plots", {})
-            dvcfile_defs_dict: Dict[str, Union[Dict, None]] = {}
-            if isinstance(dvcfile_defs, list):
-                for elem in dvcfile_defs:
-                    if isinstance(elem, str):
-                        dvcfile_defs_dict[elem] = None
-                    else:
-                        k, v = list(elem.items())[0]
-                        dvcfile_defs_dict[k] = v
-            else:
-                dvcfile_defs_dict = dvcfile_defs
-            resolved = _resolve_definitions(
-                repo.dvcfs,
-                targets,
-                props,
-                dvcfile_path,
-                dvcfile_defs_dict,
-                onerror=onerror,
-            )
-            dpath.util.merge(
-                result,
-                {dvcfile_path: resolved},
-            )
+    top_plots = repo.index._top_plots  # pylint: disable=protected-access
+    for dvcfile, plots_def in top_plots.items():
+        dvcfile_path = _relpath(repo.dvcfs, dvcfile)
+        dvcfile_defs_dict: Dict[str, Union[Dict, None]] = {}
+        if isinstance(plots_def, list):
+            for elem in plots_def:
+                if isinstance(elem, str):
+                    dvcfile_defs_dict[elem] = None
+                else:
+                    k, v = list(elem.items())[0]
+                    dvcfile_defs_dict[k] = v
+        else:
+            dvcfile_defs_dict = plots_def
+
+        resolved = _resolve_definitions(
+            repo.dvcfs,
+            targets,
+            props,
+            dvcfile_path,
+            dvcfile_defs_dict,
+            onerror=onerror,
+        )
+        dpath.util.merge(
+            result,
+            {dvcfile_path: resolved},
+        )
     return result
 
 

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:
-        return push_worktree(self, _remote)
+    if _remote.worktree or _remote.fs.version_aware:
+        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/stage.py

@@ -23,6 +23,7 @@ from dvc.exceptions import (
     OutputNotFoundError,
 )
 from dvc.repo import lock_repo
+from dvc.ui import ui
 from dvc.utils import as_posix, parse_target, relpath
 
 logger = logging.getLogger(__name__)
@@ -139,7 +140,14 @@ class StageLoad:
         )
         with self.repo.scm_context:
             stage.dump(update_lock=update_lock)
-            stage.ignore_outs()
+            try:
+                stage.ignore_outs()
+            except FileNotFoundError as exc:
+                ui.warn(
+                    f"Could not create .gitignore entry in {exc.filename}."
+                    " DVC will attempt to create .gitignore entry again when"
+                    " the stage is run."
+                )
 
         return stage
 
@@ -502,3 +510,104 @@ class StageLoad:
 
     def collect_repo(self, onerror: Callable[[str, Exception], None] = None):
         return list(self._collect_repo(onerror))
+
+    def _load_file(self, path):
+        from dvc.dvcfile import Dvcfile
+        from dvc.stage.loader import SingleStageLoader, StageLoader
+
+        path = self._get_filepath(path)
+        dvcfile = Dvcfile(self.repo, path)
+        # `dvcfile.stages` is not cached
+        stages = dvcfile.stages  # type: ignore
+
+        if isinstance(stages, SingleStageLoader):
+            stages_ = [stages[None]]
+        else:
+            assert isinstance(stages, StageLoader)
+            keys = self._get_keys(stages)
+            stages_ = [stages[key] for key in keys]
+
+        return (
+            stages_,
+            stages.metrics_data,
+            stages.plots_data,
+            stages.params_data,
+        )
+
+    def _collect_all_from_repo(
+        self, onerror: Callable[[str, Exception], None] = None
+    ):
+        """Collects all of the stages present in the DVC repo.
+
+        Args:
+            onerror (optional): callable that will be called with two args:
+                the filepath whose collection failed and the exc instance.
+                It can report the error to continue with the collection
+                (and, skip failed ones), or raise the exception to abort
+                the collection.
+        """
+        from dvc.dvcfile import is_valid_filename
+        from dvc.fs import LocalFileSystem
+
+        scm = self.repo.scm
+        sep = self.fs.sep
+        outs: Set[str] = set()
+
+        is_local_fs = isinstance(self.fs, LocalFileSystem)
+
+        def is_ignored(path):
+            # apply only for the local fs
+            return is_local_fs and scm.is_ignored(path)
+
+        def is_dvcfile_and_not_ignored(root, file):
+            return is_valid_filename(file) and not is_ignored(
+                f"{root}{sep}{file}"
+            )
+
+        def is_out_or_ignored(root, directory):
+            dir_path = f"{root}{sep}{directory}"
+            # trailing slash needed to check if a directory is gitignored
+            return dir_path in outs or is_ignored(f"{dir_path}{sep}")
+
+        walk_iter = self.repo.dvcignore.walk(self.fs, self.repo.root_dir)
+        if logger.isEnabledFor(logging.TRACE):  # type: ignore[attr-defined]
+            walk_iter = log_walk(walk_iter)
+
+        stages = []
+        metrics = {}
+        plots = {}
+        params = {}
+
+        for root, dirs, files in walk_iter:
+            dvcfile_filter = partial(is_dvcfile_and_not_ignored, root)
+            for file in filter(dvcfile_filter, files):
+                file_path = self.fs.path.join(root, file)
+                try:
+                    (
+                        new_stages,
+                        new_metrics,
+                        new_plots,
+                        new_params,
+                    ) = self._load_file(file_path)
+                except DvcException as exc:
+                    if onerror:
+                        onerror(relpath(file_path), exc)
+                        continue
+                    raise
+
+                stages.extend(new_stages)
+                if new_metrics:
+                    metrics[file_path] = new_metrics
+                if new_plots:
+                    plots[file_path] = new_plots
+                if new_params:
+                    params[file_path] = new_params
+
+                outs.update(
+                    out.fspath
+                    for stage in new_stages
+                    for out in stage.outs
+                    if out.protocol == "local"
+                )
+            dirs[:] = [d for d in dirs if not is_out_or_ignored(root, d)]
+        return stages, metrics, plots, params

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,4 +1,6 @@
-from typing import TYPE_CHECKING, Optional
+import logging
+from functools import partial
+from typing import TYPE_CHECKING, Any, Dict, Iterable, Optional, Union, cast
 
 from dvc.fs.callbacks import Callback
 
@@ -7,14 +9,30 @@ 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,
     push: bool = False,
+    latest_only: bool = True,
     **kwargs,
 ) -> "IndexView":
     """Return view of data that can be stored in worktree remotes.
@@ -38,6 +56,10 @@ def worktree_view(
             or (push and not out.can_push)
         ):
             return False
+        # If we are not enforcing push to latest version and have a version
+        # for this out, we assume it still exists and can skip pushing it
+        if push and not latest_only and out.meta.version_id is not None:
+            return False
         return True
 
     return index.targets_view(
@@ -48,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
@@ -67,55 +94,193 @@ def fetch_worktree(repo: "Repo", remote: "Remote") -> int:
         return save(index, callback=cb)
 
 
-def push_worktree(repo: "Repo", remote: "Remote") -> int:
+def push_worktree(
+    repo: "Repo",
+    remote: "Remote",
+    targets: Optional["TargetType"] = None,
+    **kwargs,
+) -> int:
     from dvc_data.index import checkout
-    from dvc_data.index.save import build_tree
 
-    view = worktree_view(repo.index, push=True)
-    index = view.data["repo"]
-    total = len(index)
+    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_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:
+        diff_kwargs: Dict[str, Any] = {
+            "meta_only": True,
+            "meta_cmp_key": partial(_meta_checksum, remote.fs),
+        }
+    else:
+        diff_kwargs = {}
+
+    total = len(new_index)
     with Callback.as_tqdm_callback(
         unit="file", desc="Push", disable=total == 0
     ) as cb:
         cb.set_size(total)
         pushed = checkout(
-            index,
+            new_index,
             remote.path,
             remote.fs,
-            latest_only=False,
+            old=old_index,
+            delete=remote.worktree,
             callback=cb,
+            latest_only=remote.worktree,
+            **diff_kwargs,
         )
+    if pushed:
+        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 _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:
-            if not out.use_cache:
+            _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
-
-            if not out.is_in_repo:
-                continue
-
-            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)
-                    _, hash_info = old_tree.get(
-                        repo.fs.path.relparts(fs_path, out.fs_path)
+            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
-                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
-                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)
 
-    return pushed
+
+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/rwlock.py

@@ -4,6 +4,7 @@ import os
 from collections import defaultdict
 from contextlib import contextmanager
 
+import psutil
 from voluptuous import Invalid, Optional, Required, Schema
 
 from .exceptions import DvcException
@@ -72,35 +73,67 @@ def _infos_to_str(infos):
     )
 
 
-def _check_blockers(lock, info, *, mode, waiters):
+def _check_blockers(tmp_dir, lock, info, *, mode, waiters):
     from .lock import LockError
 
-    for waiter_path in waiters:
-        blockers = [
-            blocker
-            for path, infos in lock[mode].items()
-            if localfs.path.overlaps(waiter_path, path)
-            for blocker in (infos if isinstance(infos, list) else [infos])
-            if blocker != info
-        ]
+    non_existing_pid = set()
 
-        if not blockers:
+    blockers = []
+    to_release = defaultdict(list)
+    for path, infos in lock[mode].items():
+        for waiter_path in waiters:
+            if localfs.path.overlaps(waiter_path, path):
+                break
+        else:
             continue
 
+        infos = infos if isinstance(infos, list) else [infos]
+        for blocker in infos:
+            if blocker == info:
+                continue
+
+            pid = int(blocker["pid"])
+
+            if pid in non_existing_pid:
+                pass
+            elif psutil.pid_exists(pid):
+                blockers.append(blocker)
+                continue
+            else:
+                non_existing_pid.add(pid)
+                cmd = blocker["cmd"]
+                logger.warning(
+                    "Process '%s' with (Pid %s), in RWLock-file '%s'"
+                    " had been killed. Auto remove it from the lock file.",
+                    cmd,
+                    pid,
+                    relpath(path),
+                )
+            to_release[json.dumps(blocker, sort_keys=True)].append(path)
+
+    if to_release:
+        for info_json, path_list in to_release.items():
+            info = json.loads(info_json)
+            if mode == "read":
+                _release_read(lock, info, path_list)
+            elif mode == "write":
+                _release_write(lock, info, path_list)
+
+    if blockers:
         raise LockError(
-            "'{path}' is busy, it is being blocked by:\n"
-            "{blockers}\n"
+            f"'{waiter_path}' is busy, it is being blocked by:\n"
+            f"{_infos_to_str(blockers)}\n"
             "\n"
             "If there are no processes with such PIDs, you can manually "
-            "remove '.dvc/tmp/rwlock' and try again.".format(
-                path=waiter_path, blockers=_infos_to_str(blockers)
-            )
+            f"remove '{tmp_dir}/rwlock' and try again."
         )
 
 
 def _acquire_read(lock, info, paths):
     changes = []
 
+    lock["read"] = lock.get("read", defaultdict(list))
+
     for path in paths:
         readers = lock["read"][path]
         if info in readers:
@@ -115,6 +148,8 @@ def _acquire_read(lock, info, paths):
 def _acquire_write(lock, info, paths):
     changes = []
 
+    lock["write"] = lock.get("write", defaultdict(dict))
+
     for path in paths:
         if lock["write"][path] == info:
             continue
@@ -171,8 +206,10 @@ def rwlock(tmp_dir, fs, cmd, read, write, hardlink):
 
     with _edit_rwlock(tmp_dir, fs, hardlink) as lock:
 
-        _check_blockers(lock, info, mode="write", waiters=read + write)
-        _check_blockers(lock, info, mode="read", waiters=write)
+        _check_blockers(
+            tmp_dir, lock, info, mode="write", waiters=read + write
+        )
+        _check_blockers(tmp_dir, lock, info, mode="read", waiters=write)
 
         rchanges = _acquire_read(lock, info, read)
         wchanges = _acquire_write(lock, info, write)

dvc/schema.py

@@ -123,6 +123,8 @@ MULTI_STAGE_SCHEMA = {
     PLOTS: Any(SINGLE_PLOT_SCHEMA, [Any(str, SINGLE_PLOT_SCHEMA)]),
     STAGES: SINGLE_PIPELINE_STAGE_SCHEMA,
     VARS_KWD: VARS_SCHEMA,
+    StageParams.PARAM_PARAMS: [str],
+    StageParams.PARAM_METRICS: [str],
 }
 
 COMPILED_SINGLE_STAGE_SCHEMA = Schema(SINGLE_STAGE_SCHEMA)

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/stage/loader.py

@@ -24,6 +24,9 @@ class StageLoader(Mapping):
         self.dvcfile = dvcfile
         self.data = data or {}
         self.stages_data = self.data.get("stages", {})
+        self.metrics_data = self.data.get("metrics", [])
+        self.params_data = self.data.get("params", [])
+        self.plots_data = self.data.get("plots", {})
         self.repo = self.dvcfile.repo
 
         lockfile_data = lockfile_data or {}
@@ -171,6 +174,9 @@ class SingleStageLoader(Mapping):
         self.dvcfile = dvcfile
         self.stage_data = stage_data or {}
         self.stage_text = stage_text
+        self.metrics_data = []
+        self.params_data = []
+        self.plots_data = {}
 
     def __getitem__(self, item):
         if item:

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

@@ -50,11 +50,11 @@ dependencies = [
     "rich>=10.13.0",
     "pyparsing>=2.4.7",
     "typing-extensions>=3.7.4",
-    "scmrepo==0.1.3",
-    "dvc-render==0.0.14",
+    "scmrepo==0.1.4",
+    "dvc-render==0.0.15",
     "dvc-task==0.1.6",
     "dvclive>=1.0",
-    "dvc-data==0.28.0",
+    "dvc-data==0.28.4",
     "dvc-http==2.27.2",
     "hydra-core>=1.1.0",
     "iterative-telemetry==0.0.6",
@@ -73,7 +73,7 @@ classifiers = [
 azure = ["dvc-azure==2.20.5"]
 gdrive = ["dvc-gdrive==2.19.1"]
 gs = ["dvc-gs==2.20.0"]
-hdfs = ["dvc-hdfs==2.19.0; python_version < '3.11'"]
+hdfs = ["dvc-hdfs==2.19.0"]
 oss = ["dvc-oss==2.19.0"]
 s3 = ["dvc-s3==2.21.0"]
 ssh = ["dvc-ssh==2.20.0"]
@@ -90,7 +90,7 @@ 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
@@ -98,12 +98,10 @@ tests = [
     "flaky==3.7.0",
     "pytest-timeout==2.1.0",
     "filelock==3.8.0",
-    # optional dependencies
-    "dvclive>=1.0",
-    "pywin32>=225; sys_platform == 'win32'",
     "beautifulsoup4==4.11.1",
+    "pywin32>=225; sys_platform == 'win32'",  # optional dependency
     # pylint requirements
-    "pylint==2.15.5",
+    "pylint==2.15.7",
     # 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,
@@ -111,9 +109,9 @@ tests = [
     "pylint-plugin-utils==0.7",
     # type-checking
     "mypy==0.990",
-    "types-requests>=2.27.15",
-    "types-tabulate>=0.8.6",
-    "types-toml>=0.10.4",
+    "types-requests",
+    "types-tabulate",
+    "types-toml",
 ]
 all = ["dvc[azure,gdrive,gs,hdfs,oss,s3,ssh,webdav,webhdfs]"]
 dev = ["dvc[azure,gdrive,gs,hdfs,oss,s3,ssh,webdav,webhdfs,tests]"]
@@ -163,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/executor/test_ssh.py

@@ -38,7 +38,7 @@ def test_init_from_stash(tmp_dir, scm, dvc, machine_instance, mocker):
 
 @pytest.mark.needs_internet
 @pytest.mark.parametrize("cloud", [pytest.lazy_fixture("git_ssh")])
-def test_init_git(tmp_dir, scm, cloud, mocker):
+def test_init_git(tmp_dir, dvc, scm, cloud, mocker):
     tmp_dir.scm_gen({"foo": "foo", "dir": {"bar": "bar"}}, commit="init")
     baseline_rev = scm.get_rev()
     tmp_dir.gen("foo", "stashed")
@@ -59,7 +59,7 @@ def test_init_git(tmp_dir, scm, cloud, mocker):
         fs_factory=partial(_ssh_factory, cloud),
     )
     infofile = str((root_url / "foo.run").path)
-    executor.init_git(scm, rev, mock, infofile=infofile)
+    executor.init_git(dvc, scm, rev, mock, infofile=infofile)
     assert root_url.path == executor._repo_abspath
 
     fs = cloud._ssh

tests/func/experiments/test_experiments.py

@@ -2,12 +2,14 @@ import itertools
 import logging
 import os
 import stat
+from textwrap import dedent
 
 import pytest
 from funcy import first
 
 from dvc.dvcfile import PIPELINE_FILE
 from dvc.exceptions import ReproductionError
+from dvc.repo.experiments.exceptions import ExperimentExistsError
 from dvc.repo.experiments.queue.base import BaseStashQueue
 from dvc.repo.experiments.utils import exp_refs_by_rev
 from dvc.scm import resolve_rev
@@ -43,8 +45,6 @@ def test_new_simple(tmp_dir, scm, dvc, exp_stage, mocker, name, workspace):
 
 
 def test_experiment_exists(tmp_dir, scm, dvc, exp_stage, mocker, workspace):
-    from dvc.repo.experiments.exceptions import ExperimentExistsError
-
     dvc.experiments.run(
         exp_stage.addressing,
         name="foo",
@@ -417,7 +417,7 @@ def test_packed_args_exists(tmp_dir, scm, dvc, exp_stage, caplog):
         assert "Temporary DVC file" in caplog.text
 
 
-def _prepare_experiments(tmp_dir, scm, dvc, exp_stage):
+def test_list(tmp_dir, scm, dvc, exp_stage):
     baseline_a = scm.get_rev()
     results = dvc.experiments.run(exp_stage.addressing, params=["foo=2"])
     exp_a = first(results)
@@ -428,54 +428,29 @@ def _prepare_experiments(tmp_dir, scm, dvc, exp_stage):
     ref_info_b = first(exp_refs_by_rev(scm, exp_b))
 
     tmp_dir.scm_gen("new", "new", commit="new")
-    baseline_c = scm.get_rev()
     results = dvc.experiments.run(exp_stage.addressing, params=["foo=4"])
     exp_c = first(results)
     ref_info_c = first(exp_refs_by_rev(scm, exp_c))
 
-    return baseline_a, baseline_c, ref_info_a, ref_info_b, ref_info_c
-
-
-def test_list(tmp_dir, scm, dvc, exp_stage):
-    (
-        baseline_a,
-        baseline_c,
-        ref_info_a,
-        ref_info_b,
-        ref_info_c,
-    ) = _prepare_experiments(tmp_dir, scm, dvc, exp_stage)
-
-    assert dvc.experiments.ls() == {baseline_c: [ref_info_c.name]}
+    assert dvc.experiments.ls() == {"master": [ref_info_c.name]}
 
     exp_list = dvc.experiments.ls(rev=ref_info_a.baseline_sha)
     assert {key: set(val) for key, val in exp_list.items()} == {
-        baseline_a: {ref_info_a.name, ref_info_b.name}
+        baseline_a[:7]: {ref_info_a.name, ref_info_b.name}
     }
 
     exp_list = dvc.experiments.ls(all_commits=True)
     assert {key: set(val) for key, val in exp_list.items()} == {
-        baseline_a: {ref_info_a.name, ref_info_b.name},
-        baseline_c: {ref_info_c.name},
+        baseline_a[:7]: {ref_info_a.name, ref_info_b.name},
+        "master": {ref_info_c.name},
     }
 
-
-def test_list_cli(tmp_dir, scm, dvc, capsys, exp_stage):
-    from dvc.cli import main
-
-    baseline_a, _, ref_info_a, ref_info_b, ref_info_c = _prepare_experiments(
-        tmp_dir, scm, dvc, exp_stage
-    )
-
-    # Make sure that we prioritize the current branch name
     scm.checkout("branch", True)
-
-    capsys.readouterr()
-    assert main(["exp", "list", "-A"]) == 0
-    cap = capsys.readouterr()
-    assert set(cap.out.split()) == set(
-        ["branch:", baseline_a[:7] + ":"]
-        + [ref_info_a.name, ref_info_b.name, ref_info_c.name]
-    )
+    exp_list = dvc.experiments.ls(all_commits=True)
+    assert {key: set(val) for key, val in exp_list.items()} == {
+        baseline_a[:7]: {ref_info_a.name, ref_info_b.name},
+        "branch": {ref_info_c.name},
+    }
 
 
 def test_subdir(tmp_dir, scm, dvc, workspace):
@@ -707,3 +682,29 @@ def test_experiments_workspace_not_log_exception(caplog, dvc, scm):
             dvc.experiments.run()
 
     assert not caplog.text
+
+
+def test_run_env(tmp_dir, dvc, scm):
+    dump_run_env = dedent(
+        """\
+        import os
+        from dvc.env import DVC_EXP_BASELINE_REV, DVC_EXP_NAME
+        with open("DVC_EXP_BASELINE_REV", "w") as f:
+            f.write(os.environ.get(DVC_EXP_BASELINE_REV, ""))
+        with open("DVC_EXP_NAME", "w") as f:
+            f.write(os.environ.get(DVC_EXP_NAME, ""))
+        """
+    )
+    (tmp_dir / "dump_run_env.py").write_text(dump_run_env)
+    baseline = scm.get_rev()
+    dvc.stage.add(
+        cmd="python dump_run_env.py",
+        name="run_env",
+    )
+    dvc.experiments.run()
+    assert (tmp_dir / "DVC_EXP_BASELINE_REV").read_text().strip() == baseline
+    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
+    assert (tmp_dir / "DVC_EXP_NAME").read_text().strip() == "foo"

tests/func/experiments/test_remote.py

@@ -151,7 +151,6 @@ def test_list_remote(tmp_dir, scm, dvc, git_downstream, exp_stage, use_url):
     ref_info_b = first(exp_refs_by_rev(scm, exp_b))
 
     tmp_dir.scm_gen("new", "new", commit="new")
-    baseline_c = scm.get_rev()
     results = dvc.experiments.run(exp_stage.addressing, params=["foo=4"])
     exp_c = first(results)
     ref_info_c = first(exp_refs_by_rev(scm, exp_c))
@@ -165,13 +164,13 @@ def test_list_remote(tmp_dir, scm, dvc, git_downstream, exp_stage, use_url):
     git_downstream.tmp_dir.scm.fetch_refspecs(remote, ["master:master"])
     exp_list = downstream_exp.ls(rev=baseline_a, git_remote=remote)
     assert {key: set(val) for key, val in exp_list.items()} == {
-        baseline_a: {ref_info_a.name, ref_info_b.name}
+        baseline_a[:7]: {ref_info_a.name, ref_info_b.name}
     }
 
     exp_list = downstream_exp.ls(all_commits=True, git_remote=remote)
     assert {key: set(val) for key, val in exp_list.items()} == {
-        baseline_a: {ref_info_a.name, ref_info_b.name},
-        baseline_c: {ref_info_c.name},
+        baseline_a[:7]: {ref_info_a.name, ref_info_b.name},
+        "master": {ref_info_c.name},
     }
 
 

tests/func/experiments/test_save.py

@@ -0,0 +1,85 @@
+from contextlib import nullcontext
+
+import pytest
+from funcy import first
+
+from dvc.repo.experiments.exceptions import ExperimentExistsError
+from dvc.repo.experiments.utils import exp_refs_by_rev
+from dvc.scm import resolve_rev
+from dvc.stage.exceptions import StageCommitError
+
+
+@pytest.mark.parametrize("name", (None, "test"))
+def test_exp_save(tmp_dir, dvc, scm, exp_stage, name):
+    baseline = scm.get_rev()
+
+    exp = dvc.experiments.save(name=name)
+    ref_info = first(exp_refs_by_rev(scm, exp))
+    assert ref_info and ref_info.baseline_sha == baseline
+
+    exp_name = name if name else ref_info.name
+    assert dvc.experiments.get_exact_name([exp])[exp] == exp_name
+    assert resolve_rev(scm, exp_name) == exp
+
+
+@pytest.mark.parametrize(
+    ("force", "expected_raises"),
+    (
+        (False, pytest.raises(StageCommitError)),
+        (True, nullcontext()),
+    ),
+)
+def test_exp_save_force(tmp_dir, dvc, scm, exp_stage, force, expected_raises):
+    with open(tmp_dir / "copy.py", "a", encoding="utf-8") as fh:
+        fh.write("\n# dummy change")
+
+    with expected_raises:
+        dvc.experiments.save(force=force)
+
+
+def test_exp_save_overwrite_experiment(tmp_dir, dvc, scm, exp_stage):
+    dvc.experiments.save(name="dummy")
+
+    with open(tmp_dir / "copy.py", "a", encoding="utf-8") as fh:
+        fh.write("\n# dummy change")
+
+    with pytest.raises(ExperimentExistsError):
+        dvc.experiments.save(name="dummy")
+
+    dvc.experiments.save(name="dummy", force=True)
+
+
+def test_exp_save_after_commit(tmp_dir, dvc, scm, exp_stage):
+    baseline = scm.get_rev()
+    dvc.experiments.save(name="exp-1")
+
+    tmp_dir.scm_gen({"new_file": "new_file"}, commit="new baseline")
+    dvc.experiments.save(name="exp-2")
+
+    all_exps = dvc.experiments.ls(all_commits=True)
+    assert all_exps[baseline[:7]] == ["exp-1"]
+    assert all_exps["master"] == ["exp-2"]
+
+
+def test_exp_save_with_staged_changes(tmp_dir, dvc, scm):
+    tmp_dir.gen({"new_file": "new_file"})
+    scm.add("new_file")
+
+    dvc.experiments.save(name="exp")
+
+    _, _, unstaged = scm.status()
+    assert "new_file" in unstaged
+
+
+def test_exp_save_include_untracked(tmp_dir, dvc, scm, exp_stage):
+    new_file = tmp_dir / "new_file"
+    for i in range(2):
+        new_file.write_text(f"exp-{i}")
+        dvc.experiments.save(name=f"exp-{i}", include_untracked=["new_file"])
+
+    _, _, unstaged = scm.status()
+    assert "new_file" in unstaged
+    assert new_file.read_text() == f"exp-{i}"
+
+    dvc.experiments.apply("exp-0")
+    assert new_file.read_text() == "exp-0"

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

@@ -10,15 +10,17 @@ from scmrepo.exceptions import SCMError
 
 from dvc.cli import main
 from dvc.repo.experiments.executor.base import (
-    EXEC_PID_DIR,
-    EXEC_TMP_DIR,
     BaseExecutor,
     ExecutorInfo,
     TaskStatus,
 )
 from dvc.repo.experiments.queue.base import QueueEntry
 from dvc.repo.experiments.refs import CELERY_STASH, ExpRefInfo
-from dvc.repo.experiments.utils import exp_refs_by_rev
+from dvc.repo.experiments.utils import (
+    EXEC_PID_DIR,
+    EXEC_TMP_DIR,
+    exp_refs_by_rev,
+)
 from dvc.utils import relpath
 from dvc.utils.fs import makedirs
 from dvc.utils.serialize import YAMLFileCorruptedError

tests/func/metrics/test_diff.py

@@ -1,6 +1,10 @@
 import json
+from os.path import join
+
+import pytest
 
 from dvc.cli import main
+from dvc.utils import relpath
 
 
 def test_metrics_diff_simple(tmp_dir, scm, dvc, run_copy_metrics):
@@ -206,3 +210,28 @@ def test_metrics_diff_non_metrics(tmp_dir, scm, dvc):
     assert result == {
         "some_file.yaml": {"foo": {"old": 1, "new": 3, "diff": 2}}
     }
+
+
+@pytest.mark.parametrize(
+    "dvcfile, metrics_file",
+    [
+        ("dvc.yaml", "my_metrics.yaml"),
+        ("dir/dvc.yaml", "my_metrics.yaml"),
+        ("dir/dvc.yaml", join("..", "my_metrics.yaml")),
+    ],
+)
+def test_diff_top_level_metrics(tmp_dir, dvc, scm, dvcfile, metrics_file):
+    directory = (tmp_dir / dvcfile).parent
+    directory.mkdir(exist_ok=True)
+    (tmp_dir / dvcfile).dump({"metrics": [metrics_file]})
+
+    metrics_file = directory / metrics_file
+    metrics_file.dump({"foo": 3})
+    scm.add_commit([metrics_file, tmp_dir / dvcfile], message="add metrics")
+
+    metrics_file.dump({"foo": 5})
+    assert dvc.metrics.diff() == {
+        relpath(directory / metrics_file): {
+            "foo": {"diff": 2, "new": 5, "old": 3}
+        }
+    }

tests/func/params/test_diff.py

@@ -1,3 +1,5 @@
+from os.path import join
+
 import pytest
 
 from dvc.utils import relpath
@@ -241,3 +243,28 @@ def test_diff_without_targets_specified(tmp_dir, dvc, scm, file):
             "y": {"new": "100", "old": None},
         }
     }
+
+
+@pytest.mark.parametrize(
+    "dvcfile, params_file",
+    [
+        ("dvc.yaml", "my_params.yaml"),
+        ("dir/dvc.yaml", "my_params.yaml"),
+        ("dir/dvc.yaml", join("..", "my_params.yaml")),
+    ],
+)
+def test_diff_top_level_params(tmp_dir, dvc, scm, dvcfile, params_file):
+    directory = (tmp_dir / dvcfile).parent
+    directory.mkdir(exist_ok=True)
+    (tmp_dir / dvcfile).dump({"params": [params_file]})
+
+    params_file = directory / params_file
+    params_file.dump({"foo": 3})
+    scm.add_commit([params_file, tmp_dir / dvcfile], message="add params")
+
+    params_file.dump({"foo": 5})
+    assert dvc.params.diff() == {
+        relpath(directory / params_file): {
+            "foo": {"diff": 2, "new": 5, "old": 3}
+        }
+    }

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

tests/unit/command/test_experiments.py

@@ -16,6 +16,7 @@ from dvc.commands.experiments.pull import CmdExperimentsPull
 from dvc.commands.experiments.push import CmdExperimentsPush
 from dvc.commands.experiments.remove import CmdExperimentsRemove
 from dvc.commands.experiments.run import CmdExperimentsRun
+from dvc.commands.experiments.save import CmdExperimentsSave
 from dvc.commands.experiments.show import CmdExperimentsShow, show_experiments
 from dvc.exceptions import InvalidArgumentError
 from dvc.repo import Repo
@@ -934,3 +935,17 @@ def test_show_experiments_pcp(tmp_dir, mocker):
 
     assert kwargs["output_path"] == str(tmp_dir / "dvc_plots" / "index.html")
     assert kwargs["color_by"] == "Experiment"
+
+
+def test_experiments_save(dvc, scm, mocker):
+    cli_args = parse_args(["exp", "save", "--name", "exp-name", "--force"])
+    assert cli_args.func == CmdExperimentsSave
+
+    cmd = cli_args.func(cli_args)
+    m = mocker.patch("dvc.repo.experiments.save.save", return_value="acabb")
+
+    assert cmd.run() == 0
+
+    m.assert_called_once_with(
+        cmd.repo, name="exp-name", force=True, include_untracked=[]
+    )

tests/unit/command/test_status.py

@@ -81,7 +81,7 @@ def test_status_empty(dvc, mocker, capsys):
 
     cmd = cli_args.func(cli_args)
 
-    spy = mocker.spy(cmd.repo.stage, "_collect_repo")
+    spy = mocker.spy(cmd.repo.stage, "_collect_all_from_repo")
 
     assert cmd.run() == 0
 
@@ -108,7 +108,10 @@ def test_status_up_to_date(dvc, mocker, capsys, cloud_opts, expected_message):
     mocker.patch.dict(cmd.repo.config, {"core": {"remote": "default"}})
     mocker.patch.object(cmd.repo, "status", autospec=True, return_value={})
     mocker.patch.object(
-        cmd.repo.stage, "_collect_repo", return_value=[object()], autospec=True
+        cmd.repo.stage,
+        "_collect_all_from_repo",
+        return_value=[[object()], {}, {}, {}],
+        autospec=True,
     )
 
     assert cmd.run() == 0

tests/unit/render/test_vega_converter.py

@@ -2,6 +2,7 @@ from collections import OrderedDict
 
 import pytest
 
+from dvc.exceptions import DvcException
 from dvc.render import VERSION_FIELD
 from dvc.render.converter.vega import VegaConverter, _lists
 
@@ -235,6 +236,44 @@ def test_finding_lists(dictionary, expected_result):
             },
             id="multi_source_json",
         ),
+        pytest.param(
+            {
+                "f": {"metric": [{"v": 1, "v2": 0.1}, {"v": 2, "v2": 0.2}]},
+                "f2": {"metric": [{"v": 3, "v2": 0.3}]},
+            },
+            {"x": "v", "y": {"f": "v2", "f2": "v2"}},
+            [
+                {
+                    "v": 1,
+                    "v2": 0.1,
+                    VERSION_FIELD: {
+                        "revision": "r",
+                        "filename": "f",
+                        "field": "v2",
+                    },
+                },
+                {
+                    "v": 2,
+                    "v2": 0.2,
+                    VERSION_FIELD: {
+                        "revision": "r",
+                        "filename": "f",
+                        "field": "v2",
+                    },
+                },
+                {
+                    "v": 3,
+                    "v2": 0.3,
+                    VERSION_FIELD: {
+                        "revision": "r",
+                        "filename": "f2",
+                        "field": "v2",
+                    },
+                },
+            ],
+            {"x": "v", "y": "v2", "x_label": "v", "y_label": "v2"},
+            id="multi_file_json",
+        ),
     ],
 )
 def test_convert(
@@ -250,6 +289,47 @@ def test_convert(
     assert resolved_properties == expected_properties
 
 
+@pytest.mark.parametrize(
+    "input_data,properties",
+    [
+        pytest.param(
+            {
+                "f": {
+                    "metric": [
+                        {"v": 1},
+                        {"v": 2},
+                    ]
+                },
+                "f2": {"metric": [{"v2": 0.1}]},
+            },
+            {"x": {"f": "v"}, "y": {"f2": "v2"}},
+            id="unequal_datapoints",
+        ),
+        pytest.param(
+            {
+                "f": {
+                    "metric": [
+                        {"v": 1, "v2": 0.1},
+                        {"v": 2, "v2": 0.2},
+                    ]
+                },
+                "f2": {
+                    "metric": [
+                        {"v": 3, "v2": 0.3},
+                    ]
+                },
+            },
+            {"x": {"f": "v", "f2": "v3"}, "y": {"f": "v2"}},
+            id="unequal_x_y",
+        ),
+    ],
+)
+def test_convert_fail(input_data, properties):
+    converter = VegaConverter("f", input_data, properties)
+    with pytest.raises(DvcException):
+        converter.flat_datapoints("r")
+
+
 @pytest.mark.parametrize(
     "properties,label",
     [

tests/unit/test_rwlock.py

@@ -1,3 +1,4 @@
+import json
 import os
 
 import pytest
@@ -105,3 +106,37 @@ def test_broken_rwlock(tmp_path):
     with pytest.raises(RWLockFileCorruptedError):
         with _edit_rwlock(dir_path, localfs, False):
             pass
+
+
+@pytest.mark.parametrize("return_value", [True, False])
+def test_corrupted_rwlock(tmp_path, mocker, return_value):
+    dir_path = os.fspath(tmp_path)
+    path = tmp_path / "rwlock"
+
+    foo = "foo"
+    bar = "bar"
+    cmd_foo = "cmd_foo"
+    cmd_bar = "cmd_bar"
+    mocker.patch("psutil.pid_exists", return_value=return_value)
+
+    corrupted_rwlock = {
+        "write": {foo: {"pid": 1234, "cmd": cmd_foo}},
+        "read": {
+            foo: [{"pid": 5555, "cmd": cmd_foo}],
+            bar: [
+                {"pid": 6666, "cmd": cmd_bar},
+                {"pid": 7777, "cmd": cmd_bar},
+            ],
+        },
+    }
+
+    path.write_text(json.dumps(corrupted_rwlock), encoding="utf-8")
+
+    if return_value:
+        with pytest.raises(LockError):
+            with rwlock(dir_path, localfs, "cmd_other", [], [foo, bar], False):
+                pass
+    else:
+        with rwlock(dir_path, localfs, "cmd_other", [], [foo, bar], False):
+            pass
+        assert path.read_text() == """{"read": {}}"""