ls: compute relparts once for every dir

Cuts down `dvc ls . data/mnist/dataset --recursive` for me from ~25 to ~12
seconds.

`relparts` is an expensive operation as it calls `relpath` that calls `abspath`,
so we want to avoid calling it as much as possible.

.github/workflows/tests.yaml

@@ -9,8 +9,6 @@ on:
   workflow_dispatch:
 
 env:
-  DVC_TEST: "true"
-  HOMEBREW_NO_AUTO_UPDATE: 1
   SHELL: /bin/bash
 
 concurrency:
@@ -71,8 +69,11 @@ jobs:
         pip install --upgrade pip wheel
         pip install -e ".[dev]"
     - name: run tests
-      timeout-minutes: 20
-      run: pytest -nauto --durations 100 --cov --cov-report=xml --cov-report=term -k "${{ matrix.pytest-filter }}"
+      timeout-minutes: 30
+      run: >
+        pytest -nauto --timeout=300 --durations=100
+        --cov --cov-report=xml --cov-report=term
+        -k "${{ matrix.pytest-filter }}"
     - name: upload coverage report
       uses: codecov/codecov-action@v3
       with:

dvc/commands/experiments/pull.py

@@ -68,7 +68,7 @@ def add_parser(experiments_subparsers, parent_parser):
         "-f",
         "--force",
         action="store_true",
-        help="Replace local experiment already exists.",
+        help="Replace local experiment if it already exists.",
     )
     experiments_pull_parser.add_argument(
         "--no-cache",

dvc/commands/experiments/save.py

@@ -49,7 +49,7 @@ def add_parser(experiments_subparsers, parent_parser):
         "--force",
         action="store_true",
         default=False,
-        help="Save even if hash value for dependencies/outputs changed.",
+        help="Replace experiment if it already exists.",
     )
     save_parser.add_argument(
         "--json",

dvc/commands/get.py

@@ -4,6 +4,7 @@ import logging
 from dvc.cli import completion
 from dvc.cli.command import CmdBaseNoRepo
 from dvc.exceptions import DvcException
+from dvc.scm import CloneError
 
 from ..cli.utils import append_doc_link
 
@@ -44,6 +45,9 @@ class CmdGet(CmdBaseNoRepo):
                 jobs=self.args.jobs,
             )
             return 0
+        except CloneError:
+            logger.exception(f"failed to get '{self.args.path}'")
+            return 1
         except DvcException:
             logger.exception(
                 "failed to get '{}' from '{}'".format(

dvc/commands/imp.py

@@ -5,6 +5,7 @@ from dvc.cli import completion
 from dvc.cli.command import CmdBase
 from dvc.cli.utils import append_doc_link
 from dvc.exceptions import DvcException
+from dvc.scm import CloneError
 
 logger = logging.getLogger(__name__)
 
@@ -26,6 +27,9 @@ class CmdImport(CmdBase):
                 meta=self.args.meta,
                 jobs=self.args.jobs,
             )
+        except CloneError:
+            logger.exception(f"failed to import '{self.args.path}'")
+            return 1
         except DvcException:
             logger.exception(
                 "failed to import '{}' from '{}'.".format(

dvc/commands/queue/kill.py

@@ -11,13 +11,18 @@ class CmdQueueKill(CmdBase):
     """Kill exp task in queue."""
 
     def run(self):
-        self.repo.experiments.celery_queue.kill(revs=self.args.task)
+        self.repo.experiments.celery_queue.kill(
+            revs=self.args.task, force=self.args.force
+        )
 
         return 0
 
 
 def add_parser(queue_subparsers, parent_parser):
-    QUEUE_KILL_HELP = "Kill actively running experiment queue tasks."
+    QUEUE_KILL_HELP = (
+        "Gracefully interrupt running experiment queue tasks "
+        "(equivalent to Ctrl-C)"
+    )
     queue_kill_parser = queue_subparsers.add_parser(
         "kill",
         parents=[parent_parser],
@@ -25,6 +30,13 @@ def add_parser(queue_subparsers, parent_parser):
         help=QUEUE_KILL_HELP,
         formatter_class=argparse.RawDescriptionHelpFormatter,
     )
+    queue_kill_parser.add_argument(
+        "-f",
+        "--force",
+        action="store_true",
+        default=False,
+        help="Forcefully and immediately kill running experiment queue tasks",
+    )
     queue_kill_parser.add_argument(
         "task",
         nargs="*",

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

@@ -97,6 +97,8 @@ HTTP_COMMON = {
     "ask_password": Bool,
     "ssl_verify": Any(Bool, str),
     "method": str,
+    "connect_timeout": All(Coerce(float), Range(0, min_included=True)),
+    "read_timeout": All(Coerce(float), Range(0, min_included=True)),
     Optional("verify", default=False): Bool,
 }
 WEBDAV_COMMON = {

dvc/data_cloud.py

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

dvc/dvcfile.py

@@ -1,7 +1,18 @@
 import contextlib
 import logging
 import os
-from typing import TYPE_CHECKING, Any, Callable, Tuple, TypeVar, Union
+from typing import (
+    TYPE_CHECKING,
+    Any,
+    Callable,
+    Dict,
+    List,
+    Tuple,
+    TypeVar,
+    Union,
+)
+
+from funcy import cached_property
 
 from dvc.exceptions import DvcException
 from dvc.parsing.versions import LOCKFILE_VERSION, SCHEMA_KWD
@@ -180,6 +191,10 @@ class SingleStageFile(FileMixin):
     from dvc.schema import COMPILED_SINGLE_STAGE_SCHEMA as SCHEMA
     from dvc.stage.loader import SingleStageLoader as LOADER
 
+    metrics: List[str] = []
+    plots: Dict[str, Any] = {}
+    params: List[str] = []
+
     @property
     def stage(self):
         data, raw = self._load()
@@ -223,6 +238,11 @@ class PipelineFile(FileMixin):
     def _lockfile(self):
         return Lockfile(self.repo, os.path.splitext(self.path)[0] + ".lock")
 
+    def _reset(self):
+        self.__dict__.pop("contents", None)
+        self.__dict__.pop("lockfile_contents", None)
+        self.__dict__.pop("resolver", None)
+
     def dump(
         self, stage, update_pipeline=True, update_lock=True, **kwargs
     ):  # pylint: disable=arguments-differ
@@ -278,11 +298,36 @@ class PipelineFile(FileMixin):
             "PipelineFile has multiple stages. Please specify it's name."
         )
 
+    @cached_property
+    def contents(self):
+        return self._load()[0]
+
+    @cached_property
+    def lockfile_contents(self):
+        return self._lockfile.load()
+
+    @cached_property
+    def resolver(self):
+        from .parsing import DataResolver
+
+        wdir = self.repo.fs.path.parent(self.path)
+        return DataResolver(self.repo, wdir, self.contents)
+
     @property
     def stages(self):
-        data, _ = self._load()
-        lockfile_data = self._lockfile.load()
-        return self.LOADER(self, data, lockfile_data)
+        return self.LOADER(self, self.contents, self.lockfile_contents)
+
+    @property
+    def metrics(self):
+        return self.contents.get("metrics", [])
+
+    @property
+    def plots(self):
+        return self.contents.get("plots", {})
+
+    @property
+    def params(self):
+        return self.contents.get("params", [])
 
     def remove(self, force=False):
         if not force:

dvc/fs/dvc.py

@@ -276,43 +276,38 @@ class _DVCFileSystem(AbstractFileSystem):  # pylint:disable=abstract-method
         key = self._get_key_from_relative(path)
         repo, dvc_fs, subkey = self._get_subrepo_info(key)
 
-        names = set()
+        dvc_infos = {}
         if dvc_fs:
             with suppress(FileNotFoundError):
                 dvc_path = _get_dvc_path(dvc_fs, subkey)
-                for entry in dvc_fs.ls(dvc_path, detail=False):
-                    names.add(dvc_fs.path.name(entry))
+                for info in dvc_fs.ls(dvc_path, detail=True):
+                    dvc_infos[dvc_fs.path.name(info["name"])] = info
 
+        fs_infos = {}
         ignore_subrepos = kwargs.get("ignore_subrepos", True)
         if not dvc_only:
             fs = self.repo.fs
             fs_path = self._from_key(key)
             try:
-                for entry in repo.dvcignore.ls(
-                    fs, fs_path, detail=False, ignore_subrepos=ignore_subrepos
+                for info in repo.dvcignore.ls(
+                    fs, fs_path, detail=True, ignore_subrepos=ignore_subrepos
                 ):
-                    names.add(fs.path.name(entry))
+                    fs_infos[fs.path.name(info["name"])] = info
             except (FileNotFoundError, NotADirectoryError):
                 pass
 
         dvcfiles = kwargs.get("dvcfiles", False)
-        if not dvcfiles:
-            names = (name for name in names if not _is_dvc_file(name))
 
         infos = []
         paths = []
+        names = set(dvc_infos.keys()) | set(fs_infos.keys())
         for name in names:
-            entry_path = self.path.join(path, name)
-            entry_key = key + (name,)
-            try:
-                info = self._info(
-                    entry_key,
-                    entry_path,
-                    ignore_subrepos=ignore_subrepos,
-                    check_ignored=False,
-                )
-            except FileNotFoundError:
+            if not dvcfiles and _is_dvc_file(name):
                 continue
+
+            entry_path = self.path.join(path, name)
+            info = _merge_info(repo, fs_infos.get(name), dvc_infos.get(name))
+            info["name"] = entry_path
             infos.append(info)
             paths.append(entry_path)
 

dvc/render/converter/vega.py

@@ -126,25 +126,30 @@ class VegaConverter(Converter):
                     break
 
     def _infer_x_y(self):
-        def _infer_files(from_name, to_name):
-            from_value = self.properties.get(from_name, None)
-            to_value = self.properties.get(to_name, None)
+        x = self.properties.get("x", None)
+        y = self.properties.get("y", None)
 
-            if isinstance(to_value, str):
-                self.inferred_properties[to_name] = {}
-                if isinstance(from_value, dict):
-                    for file in from_value.keys():
-                        self.inferred_properties[to_name][file] = to_value
-                else:
-                    self.inferred_properties[to_name][self.plot_id] = to_value
+        # Infer x.
+        if isinstance(x, str):
+            self.inferred_properties["x"] = {}
+            # If multiple y files, duplicate x for each file.
+            if isinstance(y, dict):
+                for file, fields in y.items():
+                    # Duplicate x for each y.
+                    if isinstance(fields, list):
+                        self.inferred_properties["x"][file] = [x] * len(fields)
+                    else:
+                        self.inferred_properties["x"][file] = x
+            # Otherwise use plot ID as file.
             else:
-                self.inferred_properties[to_name] = to_value
+                self.inferred_properties["x"][self.plot_id] = x
 
-        _infer_files("y", "x")
-        _infer_files("x", "y")
-
-        if self.inferred_properties.get("y", None) is None:
+        # Infer y.
+        if y is None:
             self._infer_y_from_data()
+        # If y files not provided, use plot ID as file.
+        elif not isinstance(y, dict):
+            self.inferred_properties["y"] = {self.plot_id: y}
 
     def _find_datapoints(self):
         result = {}

dvc/repo/__init__.py

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

dvc/repo/brancher.py

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

dvc/repo/experiments/__init__.py

@@ -32,7 +32,7 @@ from .refs import (
     ExpRefInfo,
 )
 from .stash import ExpStashEntry
-from .utils import exp_refs_by_rev, unlocked_repo
+from .utils import check_ref_format, exp_refs_by_rev, unlocked_repo
 
 logger = logging.getLogger(__name__)
 
@@ -180,8 +180,7 @@ class Experiments:
         if not entries:
             return results
 
-        # TODO: re-enable --jobs concurrency
-        self.celery_queue.spawn_worker()
+        self.celery_queue.start_workers(count=kwargs.get("jobs", 1))
         failed = []
         try:
             ui.write(
@@ -236,17 +235,6 @@ class Experiments:
             "\tdvc exp branch <exp> <branch>\n"
         )
 
-    def _validate_new_ref(self, exp_ref: ExpRefInfo):
-        from .utils import check_ref_format
-
-        if not exp_ref.name:
-            return
-
-        check_ref_format(self.scm, exp_ref)
-
-        if self.scm.get_ref(str(exp_ref)):
-            raise ExperimentExistsError(exp_ref.name)
-
     def new(
         self,
         queue: BaseStashQueue,
@@ -265,13 +253,14 @@ class Experiments:
 
         name = kwargs.get("name", None)
         baseline_sha = kwargs.get("baseline_rev") or self.repo.scm.get_rev()
-        exp_ref = ExpRefInfo(baseline_sha=baseline_sha, name=name)
 
-        try:
-            self._validate_new_ref(exp_ref)
-        except ExperimentExistsError as err:
-            if not (kwargs.get("force", False) or kwargs.get("reset", False)):
-                raise err
+        if name:
+            exp_ref = ExpRefInfo(baseline_sha=baseline_sha, name=name)
+            check_ref_format(self.scm, exp_ref)
+            force = kwargs.get("force", False)
+            if self.scm.get_ref(str(exp_ref)) and not force:
+                raise ExperimentExistsError(exp_ref.name)
+
         return queue.put(*args, **kwargs)
 
     def _resume_checkpoint(

dvc/repo/experiments/apply.py

@@ -16,9 +16,10 @@ from .executor.base import BaseExecutor
 from .refs import EXEC_APPLY
 
 if TYPE_CHECKING:
-    from scmrepo import Git
+    from scmrepo.git import Git
 
     from dvc.repo import Repo
+    from dvc.repo.experiments import Experiments
 
 logger = logging.getLogger(__name__)
 
@@ -31,20 +32,33 @@ def apply(repo: "Repo", rev: str, force: bool = True, **kwargs):
     from dvc.repo.checkout import checkout as dvc_checkout
     from dvc.scm import GitMergeError, RevError, resolve_rev
 
-    exps = repo.experiments
+    exps: "Experiments" = repo.experiments
+
+    is_stash: bool = False
 
     try:
         exp_rev = resolve_rev(repo.scm, rev)
         exps.check_baseline(exp_rev)
-    except (RevError, BaselineMismatchError) as exc:
+    except RevError as exc:
+        (
+            exp_ref_info,
+            queue_entry,
+        ) = exps.celery_queue.get_ref_and_entry_by_names(rev)[rev]
+        baseline_sha = repo.scm.get_rev()
+        if exp_ref_info:
+            if baseline_sha != exp_ref_info.baseline_sha:
+                raise InvalidExpRevError(rev)
+            exp_rev = repo.scm.get_ref(str(exp_ref_info))
+        elif queue_entry:
+            if queue_entry.baseline_rev != baseline_sha:
+                raise InvalidExpRevError(rev)
+            exp_rev = queue_entry.stash_rev
+            is_stash = True
+        else:
+            raise InvalidExpRevError(rev) from exc
+    except BaselineMismatchError as exc:
         raise InvalidExpRevError(rev) from exc
 
-    stash_rev = exp_rev in exps.stash_revs
-    if not stash_rev and not exps.get_branch_by_rev(
-        exp_rev, allow_multiple=True
-    ):
-        raise InvalidExpRevError(exp_rev)
-
     # NOTE: we don't use scmrepo's stash_workspace() here since we need
     # finer control over the merge behavior when we unstash everything
     with _apply_workspace(repo, rev, force):
@@ -55,7 +69,7 @@ def apply(repo: "Repo", rev: str, force: bool = True, **kwargs):
 
     repo.scm.reset()
 
-    if stash_rev:
+    if is_stash:
         args_path = os.path.join(repo.tmp_dir, BaseExecutor.PACKED_ARGS_FILE)
         if os.path.exists(args_path):
             remove(args_path)

dvc/repo/experiments/exceptions.py

@@ -6,12 +6,6 @@ if TYPE_CHECKING:
     from .refs import ExpRefInfo
 
 
-class UnchangedExperimentError(DvcException):
-    def __init__(self, rev):
-        super().__init__(f"Experiment unchanged from '{rev[:7]}'.")
-        self.rev = rev
-
-
 class BaselineMismatchError(DvcException):
     def __init__(self, rev, expected):
         if hasattr(rev, "hexsha"):
@@ -25,13 +19,13 @@ class BaselineMismatchError(DvcException):
 
 
 class ExperimentExistsError(DvcException):
-    def __init__(self, name: str):
+    def __init__(self, name: str, command: str = "run"):
         msg = (
             "Experiment conflicts with existing experiment "
             f"'{name}'. To overwrite the existing experiment run:\n\n"
-            "\tdvc exp run -f ...\n\n"
-            "To run this experiment with a different name run:\n\n"
-            f"\tdvc exp run -n <new_name> ...\n"
+            f"\tdvc exp {command} -f ...\n\n"
+            f"To {command} this experiment with a different name run:\n\n"
+            f"\tdvc exp {command} -n <new_name> ...\n"
         )
         super().__init__(msg)
         self.name = name

dvc/repo/experiments/executor/base.py

@@ -30,11 +30,7 @@ from dvc.ui import ui
 from dvc.utils import dict_sha256, env2bool, relpath
 from dvc.utils.fs import remove
 
-from ..exceptions import (
-    CheckpointExistsError,
-    ExperimentExistsError,
-    UnchangedExperimentError,
-)
+from ..exceptions import CheckpointExistsError, ExperimentExistsError
 from ..refs import EXEC_BASELINE, EXEC_BRANCH, EXEC_CHECKPOINT, ExpRefInfo
 
 if TYPE_CHECKING:
@@ -496,7 +492,6 @@ class BaseExecutor(ABC):
                     info,
                     stages,
                     exp_hash,
-                    checkpoint_reset,
                     auto_push,
                     git_remote,
                     repro_force,
@@ -517,29 +512,20 @@ class BaseExecutor(ABC):
         info,
         stages,
         exp_hash,
-        checkpoint_reset,
         auto_push,
         git_remote,
         repro_force,
     ) -> Tuple[Optional[str], Optional["ExpRefInfo"], bool]:
-        try:
-            is_checkpoint = any(stage.is_checkpoint for stage in stages)
-            if is_checkpoint and checkpoint_reset:
-                # For reset checkpoint stages, we need to force
-                # overwriting existing checkpoint refs even though
-                # repro may not have actually been run with --force
-                repro_force = True
-            cls.commit(
-                dvc.scm,
-                exp_hash,
-                exp_name=info.name,
-                force=repro_force,
-                checkpoint=is_checkpoint,
-            )
-            if auto_push:
-                cls._auto_push(dvc, dvc.scm, git_remote)
-        except UnchangedExperimentError:
-            pass
+        is_checkpoint = any(stage.is_checkpoint for stage in stages)
+        cls.commit(
+            dvc.scm,
+            exp_hash,
+            exp_name=info.name,
+            force=repro_force,
+            checkpoint=is_checkpoint,
+        )
+        if auto_push:
+            cls._auto_push(dvc, dvc.scm, git_remote)
         ref: Optional[str] = dvc.scm.get_ref(EXEC_BRANCH, follow=False)
         exp_ref: Optional["ExpRefInfo"] = (
             ExpRefInfo.from_ref(ref) if ref else None
@@ -650,18 +636,15 @@ class BaseExecutor(ABC):
         unchanged: Iterable["PipelineStage"],
         stages: Iterable["PipelineStage"],
     ):
-        try:
-            exp_hash = cls.hash_exp(list(stages) + list(unchanged))
-            exp_rev = cls.commit(
-                scm, exp_hash, exp_name=name, force=force, checkpoint=True
-            )
+        exp_hash = cls.hash_exp(list(stages) + list(unchanged))
+        exp_rev = cls.commit(
+            scm, exp_hash, exp_name=name, force=force, checkpoint=True
+        )
 
-            if env2bool(DVC_EXP_AUTO_PUSH):
-                git_remote = os.getenv(DVC_EXP_GIT_REMOTE)
-                cls._auto_push(dvc, scm, git_remote)
-            ui.write(f"Checkpoint experiment iteration '{exp_rev[:7]}'.")
-        except UnchangedExperimentError:
-            pass
+        if env2bool(DVC_EXP_AUTO_PUSH):
+            git_remote = os.getenv(DVC_EXP_GIT_REMOTE)
+            cls._auto_push(dvc, scm, git_remote)
+        ui.write(f"Checkpoint experiment iteration '{exp_rev[:7]}'.")
 
     @classmethod
     def commit(
@@ -676,7 +659,6 @@ class BaseExecutor(ABC):
         rev = scm.get_rev()
         if not scm.is_dirty(untracked_files=False):
             logger.debug("No changes to commit")
-            raise UnchangedExperimentError(rev)
 
         check_conflict = False
         branch = scm.get_ref(EXEC_BRANCH, follow=False)

dvc/repo/experiments/executor/local.py

@@ -2,13 +2,12 @@ import logging
 import os
 from contextlib import ExitStack
 from tempfile import mkdtemp
-from typing import TYPE_CHECKING, List, Optional
+from typing import TYPE_CHECKING, Optional
 
 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
@@ -25,7 +24,7 @@ from ..refs import (
     ExpRefInfo,
 )
 from ..utils import EXEC_TMP_DIR, get_exp_rwlock
-from .base import BaseExecutor, ExecutorResult, TaskStatus
+from .base import BaseExecutor, TaskStatus
 
 if TYPE_CHECKING:
     from scmrepo.git import Git
@@ -33,7 +32,6 @@ if TYPE_CHECKING:
     from dvc.repo import Repo
 
     from ..stash import ExpStashEntry
-    from .base import ExecutorInfo
 
 logger = logging.getLogger(__name__)
 
@@ -249,57 +247,3 @@ 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/queue/base.py

@@ -30,7 +30,13 @@ from ..executor.base import BaseExecutor, ExecutorResult
 from ..executor.local import WorkspaceExecutor
 from ..refs import ExpRefInfo
 from ..stash import ExpStash, ExpStashEntry
-from ..utils import EXEC_PID_DIR, EXEC_TMP_DIR, exp_refs_by_rev, get_exp_rwlock
+from ..utils import (
+    EXEC_PID_DIR,
+    EXEC_TMP_DIR,
+    exp_refs_by_rev,
+    get_exp_rwlock,
+    get_random_exp_name,
+)
 from .utils import get_remote_executor_refs
 
 if TYPE_CHECKING:
@@ -80,6 +86,11 @@ class QueueDoneResult(NamedTuple):
     result: Optional[ExecutorResult]
 
 
+class ExpRefAndQueueEntry(NamedTuple):
+    exp_ref_info: Optional["ExpRefInfo"]
+    queue_entry: Optional["QueueEntry"]
+
+
 class BaseStashQueue(ABC):
     """Naive Git-stash based experiment queue.
 
@@ -281,7 +292,7 @@ class BaseStashQueue(ABC):
                 output.
         """
 
-    def _stash_exp(
+    def _stash_exp(  # noqa: C901
         self,
         *args,
         params: Optional[Dict[str, List[str]]] = None,
@@ -399,8 +410,9 @@ class BaseStashQueue(ABC):
                     run_env = {
                         DVC_EXP_BASELINE_REV: baseline_rev,
                     }
-                    if name:
-                        run_env[DVC_EXP_NAME] = name
+                    if not name:
+                        name = get_random_exp_name(self.scm, baseline_rev)
+                    run_env[DVC_EXP_NAME] = name
                     if resume_rev:
                         run_env[DVCLIVE_RESUME] = "1"
                     self._pack_args(*args, run_env=run_env, **kwargs)

dvc/repo/experiments/queue/celery.py

@@ -12,6 +12,7 @@ from typing import (
     NamedTuple,
     Optional,
     Set,
+    Union,
 )
 
 from celery.result import AsyncResult
@@ -26,7 +27,13 @@ from ..exceptions import UnresolvedQueueExpNamesError
 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 .base import (
+    BaseStashQueue,
+    ExpRefAndQueueEntry,
+    QueueDoneResult,
+    QueueEntry,
+    QueueGetResult,
+)
 from .exceptions import CannotKillTasksError
 from .tasks import run_exp
 from .utils import fetch_running_exp_from_temp_dir
@@ -36,6 +43,8 @@ if TYPE_CHECKING:
     from dvc_task.proc.manager import ProcessManager
     from dvc_task.worker import TemporaryWorker
 
+    from ..refs import ExpRefInfo
+
 logger = logging.getLogger(__name__)
 
 
@@ -57,8 +66,6 @@ class LocalCeleryQueue(BaseStashQueue):
 
     CELERY_DIR = "celery"
 
-    _shutdown_task_ids: Set[str] = set()
-
     @cached_property
     def wdir(self) -> str:
         return os.path.join(self.repo.tmp_dir, EXEC_TMP_DIR, self.CELERY_DIR)
@@ -116,7 +123,7 @@ class LocalCeleryQueue(BaseStashQueue):
             timeout=10,
         )
 
-    def spawn_worker(self, num: int = 1):
+    def _spawn_worker(self, num: int = 1):
         """spawn one single worker to process to queued tasks.
 
         Argument:
@@ -159,7 +166,7 @@ class LocalCeleryQueue(BaseStashQueue):
             if node_name in active_worker:
                 logger.debug(f"Exp queue worker {node_name} already exist")
                 continue
-            self.spawn_worker(num)
+            self._spawn_worker(num)
             started += 1
 
         return started
@@ -300,12 +307,15 @@ class LocalCeleryQueue(BaseStashQueue):
         return running_task_ids
 
     def _try_to_kill_tasks(
-        self, to_kill: Dict[QueueEntry, str]
+        self, to_kill: Dict[QueueEntry, str], force: bool
     ) -> Dict[QueueEntry, str]:
         fail_to_kill_entries: Dict[QueueEntry, str] = {}
         for queue_entry, rev in to_kill.items():
             try:
-                self.proc.kill(queue_entry.stash_rev)
+                if force:
+                    self.proc.kill(queue_entry.stash_rev)
+                else:
+                    self.proc.interrupt(queue_entry.stash_rev)
                 logger.debug(f"Task {rev} had been killed.")
             except ProcessLookupError:
                 fail_to_kill_entries[queue_entry] = rev
@@ -333,36 +343,45 @@ class LocalCeleryQueue(BaseStashQueue):
         if remained_revs:
             raise CannotKillTasksError(remained_revs)
 
-    def kill(self, revs: Collection[str]) -> None:
+    def _kill_entries(self, entries: Dict[QueueEntry, str], force: bool):
+        logger.debug(
+            "Found active tasks: '%s' to kill",
+            list(entries.values()),
+        )
+        inactive_entries: Dict[QueueEntry, str] = self._try_to_kill_tasks(
+            entries, force
+        )
+
+        if inactive_entries:
+            self._mark_inactive_tasks_failure(inactive_entries)
+
+    def kill(self, revs: Collection[str], force: bool = False) -> None:
         name_dict: Dict[
             str, Optional[QueueEntry]
         ] = self.match_queue_entry_by_name(set(revs), self.iter_active())
 
-        to_kill: Dict[QueueEntry, str] = {}
         missing_revs: List[str] = []
+        to_kill: Dict[QueueEntry, str] = {}
         for rev, queue_entry in name_dict.items():
             if queue_entry is None:
                 missing_revs.append(rev)
             else:
                 to_kill[queue_entry] = rev
+
         if missing_revs:
             raise UnresolvedQueueExpNamesError(missing_revs)
 
-        fail_to_kill_entries: Dict[QueueEntry, str] = self._try_to_kill_tasks(
-            to_kill
-        )
-
-        if fail_to_kill_entries:
-            self._mark_inactive_tasks_failure(fail_to_kill_entries)
+        if to_kill:
+            self._kill_entries(to_kill, force)
 
     def shutdown(self, kill: bool = False):
         self.celery.control.shutdown()
         if kill:
-            for _, task_entry in self._iter_active_tasks():
-                try:
-                    self.proc.kill(task_entry.stash_rev)
-                except ProcessLookupError:
-                    continue
+            to_kill: Dict[QueueEntry, str] = {}
+            for entry in self.iter_active():
+                to_kill[entry] = entry.name or entry.stash_rev
+            if to_kill:
+                self._kill_entries(to_kill, True)
 
     def follow(
         self,
@@ -440,3 +459,31 @@ class LocalCeleryQueue(BaseStashQueue):
                 )
             )
         return result
+
+    def get_ref_and_entry_by_names(
+        self,
+        exp_names: Union[str, List[str]],
+        git_remote: Optional[str] = None,
+    ) -> Dict[str, ExpRefAndQueueEntry]:
+        """Find finished ExpRefInfo or queued or failed QueueEntry by name"""
+        from ..utils import resolve_name
+
+        if isinstance(exp_names, str):
+            exp_names = [exp_names]
+        results: Dict[str, ExpRefAndQueueEntry] = {}
+
+        exp_ref_match: Dict[str, Optional["ExpRefInfo"]] = resolve_name(
+            self.scm, exp_names, git_remote
+        )
+        if not git_remote:
+            queue_entry_match: Dict[
+                str, Optional["QueueEntry"]
+            ] = self.match_queue_entry_by_name(
+                exp_names, self.iter_queued(), self.iter_done()
+            )
+
+        for exp_name in exp_names:
+            exp_ref = exp_ref_match[exp_name]
+            queue_entry = None if git_remote else queue_entry_match[exp_name]
+            results[exp_name] = ExpRefAndQueueEntry(exp_ref, queue_entry)
+        return results

dvc/repo/experiments/queue/tempdir.py

@@ -1,9 +1,12 @@
 import logging
 import os
+from collections import defaultdict
 from typing import TYPE_CHECKING, Dict, Generator, Optional
 
 from funcy import cached_property, first
 
+from dvc.exceptions import DvcException
+
 from ..exceptions import ExpQueueEmptyError
 from ..executor.base import (
     BaseExecutor,
@@ -90,6 +93,51 @@ class TempDirQueue(WorkspaceQueue):
                     None,
                 )
 
+    def _reproduce_entry(
+        self, entry: QueueEntry, executor: BaseExecutor
+    ) -> Dict[str, Dict[str, str]]:
+        from dvc.stage.monitor import CheckpointKilledError
+
+        results: Dict[str, Dict[str, str]] = defaultdict(dict)
+        exec_name = self._EXEC_NAME or entry.stash_rev
+        infofile = self.get_infofile_path(exec_name)
+        try:
+            rev = entry.stash_rev
+            exec_result = executor.reproduce(
+                info=executor.info,
+                rev=rev,
+                infofile=infofile,
+                log_level=logger.getEffectiveLevel(),
+                log_errors=True,
+            )
+            if not exec_result.exp_hash:
+                raise DvcException(
+                    f"Failed to reproduce experiment '{rev[:7]}'"
+                )
+            if exec_result.ref_info:
+                results[rev].update(
+                    self.collect_executor(
+                        self.repo.experiments, executor, exec_result
+                    )
+                )
+        except CheckpointKilledError:
+            results[rev].update(
+                self.collect_executor(
+                    self.repo.experiments, executor, exec_result
+                )
+            )
+
+            return results
+        except DvcException:
+            raise
+        except Exception as exc:
+            raise DvcException(
+                f"Failed to reproduce experiment '{rev[:7]}'"
+            ) from exc
+        finally:
+            executor.cleanup(infofile)
+        return results
+
     @staticmethod
     def collect_executor(
         exp: "Experiments",

dvc/repo/experiments/queue/workspace.py

@@ -22,7 +22,7 @@ from ..executor.base import (
 )
 from ..executor.local import WorkspaceExecutor
 from ..refs import EXEC_BRANCH, WORKSPACE_STASH
-from ..utils import get_exp_rwlock
+from ..utils import get_exp_rwlock, to_studio_params
 from .base import BaseStashQueue, QueueDoneResult, QueueEntry, QueueGetResult
 
 if TYPE_CHECKING:
@@ -96,6 +96,8 @@ class WorkspaceQueue(BaseStashQueue):
     def _reproduce_entry(
         self, entry: QueueEntry, executor: BaseExecutor
     ) -> Dict[str, Dict[str, str]]:
+        from dvc_studio_client.post_live_metrics import post_live_metrics
+
         from dvc.stage.monitor import CheckpointKilledError
 
         results: Dict[str, Dict[str, str]] = defaultdict(dict)
@@ -103,6 +105,13 @@ class WorkspaceQueue(BaseStashQueue):
         infofile = self.get_infofile_path(exec_name)
         try:
             rev = entry.stash_rev
+            post_live_metrics(
+                "start",
+                executor.info.baseline_rev,
+                executor.info.name,
+                "dvc",
+                params=to_studio_params(self.repo.params.show()),
+            )
             exec_result = executor.reproduce(
                 info=executor.info,
                 rev=rev,
@@ -130,6 +139,13 @@ class WorkspaceQueue(BaseStashQueue):
                 f"Failed to reproduce experiment '{rev[:7]}'"
             ) from exc
         finally:
+            post_live_metrics(
+                "done",
+                executor.info.baseline_rev,
+                executor.info.name,
+                "dvc",
+                experiment_rev=first(results.get(rev, {})),
+            )
             executor.cleanup(infofile)
         return results
 

dvc/repo/experiments/remove.py

@@ -1,19 +1,12 @@
 import logging
-from typing import (
-    TYPE_CHECKING,
-    Dict,
-    Iterable,
-    List,
-    NamedTuple,
-    Optional,
-    Union,
-)
+from typing import TYPE_CHECKING, Dict, Iterable, List, Optional, Union
 
 from dvc.repo import locked
 from dvc.repo.scm_context import scm_context
 from dvc.scm import iter_revs
 
 from .exceptions import UnresolvedExpNamesError
+from .queue.base import QueueEntry
 from .refs import ExpRefInfo
 from .utils import exp_refs, exp_refs_by_baseline, push_refspec
 
@@ -22,58 +15,12 @@ if TYPE_CHECKING:
     from dvc.repo.experiments.queue.celery import LocalCeleryQueue
     from dvc.scm import Git
 
-    from .queue.base import QueueEntry
+    from .queue.base import ExpRefAndQueueEntry
 
 
 logger = logging.getLogger(__name__)
 
 
-class ExpRefAndQueueEntry(NamedTuple):
-    exp_ref_list: List["ExpRefInfo"]
-    queue_entry_list: List["QueueEntry"]
-    removed: List["str"]
-
-
-def _get_ref_and_entry_by_names(
-    exp_names: Union[str, List[str]],
-    scm: "Git",
-    celery_queue: "LocalCeleryQueue",
-    git_remote: Optional[str],
-) -> ExpRefAndQueueEntry:
-    from .utils import resolve_name
-
-    exp_ref_list: List["ExpRefInfo"] = []
-    queue_entry_list: List["QueueEntry"] = []
-    removed: List[str] = []
-    if isinstance(exp_names, str):
-        exp_names = [exp_names]
-    exp_ref_match: Dict[str, Optional["ExpRefInfo"]] = resolve_name(
-        scm, exp_names, git_remote
-    )
-    if not git_remote:
-        queue_entry_match: Dict[
-            str, Optional["QueueEntry"]
-        ] = celery_queue.match_queue_entry_by_name(
-            exp_names, celery_queue.iter_queued(), celery_queue.iter_done()
-        )
-
-    remained = []
-    for exp_name in exp_names:
-        exp_ref = exp_ref_match[exp_name]
-        queue_entry = None if git_remote else queue_entry_match[exp_name]
-        if exp_ref or queue_entry:
-            if exp_ref:
-                exp_ref_list.append(exp_ref)
-            if queue_entry:
-                queue_entry_list.append(queue_entry)
-            removed.append(exp_name)
-        else:
-            remained.append(exp_name)
-    if remained:
-        raise UnresolvedExpNamesError(remained)
-    return ExpRefAndQueueEntry(exp_ref_list, queue_entry_list, removed)
-
-
 @locked
 @scm_context
 def remove(
@@ -103,12 +50,22 @@ def remove(
     exp_ref_list: List["ExpRefInfo"] = []
     queue_entry_list: List["QueueEntry"] = []
     if exp_names:
-        result = _get_ref_and_entry_by_names(
-            exp_names, repo.scm, celery_queue, git_remote
-        )
-        removed.extend(result.removed)
-        exp_ref_list.extend(result.exp_ref_list)
-        queue_entry_list.extend(result.queue_entry_list)
+        results: Dict[
+            str, "ExpRefAndQueueEntry"
+        ] = celery_queue.get_ref_and_entry_by_names(exp_names, git_remote)
+        remained: List[str] = []
+        for name, result in results.items():
+            if not result.exp_ref_info and not result.queue_entry:
+                remained.append(name)
+                continue
+            removed.append(name)
+            if result.exp_ref_info:
+                exp_ref_list.append(result.exp_ref_info)
+            if result.queue_entry:
+                queue_entry_list.append(result.queue_entry)
+
+        if remained:
+            raise UnresolvedExpNamesError(remained)
     elif rev:
         exp_ref_dict = _resolve_exp_by_baseline(repo, rev, num, git_remote)
         removed.extend(exp_ref_dict.keys())

dvc/repo/experiments/save.py

@@ -2,7 +2,9 @@ import logging
 import os
 from typing import TYPE_CHECKING, List, Optional
 
-from funcy import first
+from .exceptions import ExperimentExistsError
+from .refs import ExpRefInfo
+from .utils import check_ref_format, get_random_exp_name
 
 if TYPE_CHECKING:
     from dvc.repo import Repo
@@ -11,6 +13,32 @@ if TYPE_CHECKING:
 logger = logging.getLogger(__name__)
 
 
+def _save_experiment(
+    repo: "Repo",
+    baseline_rev: str,
+    force: bool,
+    name: Optional[str],
+    include_untracked: Optional[List[str]],
+) -> str:
+    repo.commit([], force=True)
+
+    name = name or get_random_exp_name(repo.scm, baseline_rev)
+    ref_info = ExpRefInfo(baseline_rev, name)
+    check_ref_format(repo.scm.dulwich, ref_info)
+    ref = str(ref_info)
+    if repo.scm.get_ref(ref) and not force:
+        raise ExperimentExistsError(ref_info.name, command="save")
+
+    repo.scm.add([], update=True)
+    if include_untracked:
+        repo.scm.add(include_untracked)
+    repo.scm.commit(f"dvc: commit experiment {name}", no_verify=True)
+    exp_rev = repo.scm.get_rev()
+    repo.scm.set_ref(ref, exp_rev, old_ref=None)
+
+    return exp_rev
+
+
 def save(
     repo: "Repo",
     name: Optional[str] = None,
@@ -21,24 +49,32 @@ def save(
 
     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:
+    _, _, untracked = repo.scm.status()
+    if include_untracked:
+        untracked = [
+            file for file in untracked if file not in include_untracked
+        ]
+    if untracked:
         logger.warning(
-            "Your workspace contains staged Git changes which will be "
-            "unstaged before saving this experiment."
+            "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),
         )
-        repo.scm.reset()
 
-    entry = repo.experiments.new(queue=queue, name=name, force=force)
-    executor = queue.init_executor(repo.experiments, entry)
+    with repo.scm.detach_head(client="dvc") as orig_head:
+        with repo.scm.stash_workspace() as workspace:
+            try:
+                if workspace is not None:
+                    repo.scm.stash.apply(workspace)
 
-    save_result = executor.save(
-        executor.info, force=force, include_untracked=include_untracked
-    )
-    result = queue.collect_executor(repo.experiments, executor, save_result)
+                exp_rev = _save_experiment(
+                    repo, orig_head, force, name, include_untracked
+                )
+            finally:
+                repo.scm.reset(hard=True)
 
-    exp_rev = first(result)
     return exp_rev

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

@@ -1,4 +1,3 @@
-from contextlib import suppress
 from typing import (
     TYPE_CHECKING,
     Any,
@@ -14,9 +13,7 @@ from typing import (
     Union,
 )
 
-from funcy import cached_property, nullcontext
-
-from dvc.utils import dict_md5
+from funcy import cached_property
 
 if TYPE_CHECKING:
     from networkx import DiGraph
@@ -25,9 +22,10 @@ if TYPE_CHECKING:
     from dvc.dependency import Dependency, ParamsDependency
     from dvc.fs import FileSystem
     from dvc.output import Output
+    from dvc.repo import Repo
     from dvc.repo.stage import StageInfo, StageLoad
     from dvc.stage import Stage
-    from dvc.types import StrPath, TargetType
+    from dvc.types import TargetType
     from dvc_data.hashfile.hash_info import HashInfo
     from dvc_data.index import DataIndex, DataIndexKey, DataIndexView
     from dvc_objects.db import ObjectDB
@@ -89,6 +87,21 @@ class Index:
         self._collect()
         return self._plots
 
+    @cached_property
+    def _plot_sources(self):
+        from dvc.repo.plots import _collect_pipeline_files
+
+        sources = []
+        for data in _collect_pipeline_files(self.repo, [], {}).values():
+            for plot_id, props in data.get("data", {}).items():
+                if isinstance(props.get("y"), dict):
+                    sources.extend(props["y"])
+                    if isinstance(props.get("x"), dict):
+                        sources.extend(props["x"])
+                else:
+                    sources.append(plot_id)
+        return sources
+
     @cached_property
     def _top_params(self):
         self._collect()
@@ -121,38 +134,6 @@ class Index:
             rev = self.repo.get_rev()[:7]
         return f"Index({self.repo}, fs@{rev})"
 
-    def __len__(self) -> int:
-        return len(self.stages)
-
-    def __contains__(self, stage: "Stage") -> bool:
-        # as we are keeping stages inside a list, it might be slower.
-        return stage in self.stages
-
-    def __iter__(self) -> Iterator["Stage"]:
-        yield from self.stages
-
-    def __getitem__(self, item: str) -> "Stage":
-        """Get a stage by its addressing attribute."""
-        for stage in self:
-            if stage.addressing == item:
-                return stage
-        raise KeyError(f"{item} - available stages are {self.stages}")
-
-    def filter(self, filter_fn: Callable[["Stage"], bool]) -> "Index":
-        stages_it = filter(filter_fn, self)
-        return Index(self.repo, self.fs, stages=list(stages_it))
-
-    def slice(self, path: "StrPath") -> "Index":
-        from dvc.utils import relpath
-        from dvc.utils.fs import path_isin
-
-        target_path = relpath(path, self.repo.root_dir)
-
-        def is_stage_inside_path(stage: "Stage") -> bool:
-            return path_isin(stage.path_in_repo, target_path)
-
-        return self.filter(is_stage_inside_path)
-
     @staticmethod
     def _hash_targets(
         targets: Iterable[Optional[str]],
@@ -166,7 +147,7 @@ class Index:
             )
         )
 
-    def _collect_targets(
+    def collect_targets(
         self, targets: Optional["TargetType"], **kwargs: Any
     ) -> List["StageInfo"]:
         from itertools import chain
@@ -214,14 +195,14 @@ 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)
 
     @property
     def outs(self) -> Iterator["Output"]:
-        for stage in self:
+        for stage in self.stages:
             yield from stage.outs
 
     @property
@@ -244,7 +225,7 @@ class Index:
 
     @property
     def deps(self) -> Iterator["Dependency"]:
-        for stage in self:
+        for stage in self.stages:
             yield from stage.deps
 
     @property
@@ -310,7 +291,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:
@@ -338,56 +319,10 @@ class Index:
     def add(self, stage: "Stage") -> "Index":
         return self.update([stage])
 
-    def remove(
-        self, stage: "Stage", ignore_not_existing: bool = False
-    ) -> "Index":
-        stages = self._discard_stage(
-            stage, ignore_not_existing=ignore_not_existing
-        )
-        return Index(self.repo, self.fs, stages=stages)
-
-    def discard(self, stage: "Stage") -> "Index":
-        return self.remove(stage, ignore_not_existing=True)
-
-    def difference(self, stages: Iterable["Stage"]) -> "Index":
-        stages_set = set(self.stages) - set(stages)
-        return Index(self.repo, self.fs, stages=list(stages_set))
-
-    def _discard_stage(
-        self, stage: "Stage", ignore_not_existing: bool = False
-    ) -> List["Stage"]:
-        stages = self.stages[:]
-        ctx = suppress(ValueError) if ignore_not_existing else nullcontext()
-        with ctx:
-            stages.remove(stage)
-        return stages
-
     def check_graph(self) -> None:
         if not getattr(self.repo, "_skip_graph_checks", False):
             self.graph  # pylint: disable=pointless-statement
 
-    def dumpd(self) -> Dict[str, Dict]:
-        def dump(stage: "Stage"):
-            key = stage.path_in_repo
-            try:
-                key += ":" + stage.name  # type: ignore[attr-defined]
-            except AttributeError:
-                pass
-            return key, stage.dumpd()
-
-        return dict(dump(stage) for stage in self)
-
-    @cached_property
-    def identifier(self) -> str:
-        """Unique identifier for the index.
-
-        We can use this to optimize and skip opening some indices
-        eg: on push/pull/fetch/gc --all-commits.
-
-        Currently, it is unique to the platform (windows vs posix).
-        """
-        return dict_md5(self.dumpd())
-
 
 class _DataPrefixes(NamedTuple):
     explicit: Set["DataIndexKey"]
@@ -410,30 +345,13 @@ class IndexView:
         self._stages = list({stage for stage, _ in stage_infos})
         self._outs_filter = outs_filter
 
-    def __len__(self) -> int:
-        return len(self._stages)
-
-    def __contains__(self, stage: "Stage") -> bool:
-        # as we are keeping stages inside a list, it might be slower.
-        return stage in self._stages
-
-    def __iter__(self) -> Iterator["Stage"]:
-        yield from self._stages
-
-    def __getitem__(self, item: str) -> "Stage":
-        """Get a stage by its addressing attribute."""
-        for stage in self:
-            if stage.addressing == item:
-                return stage
-        raise KeyError(f"{item} - available stages are {self.stages}")
-
     @property
     def stages(self):
         return self._stages
 
     @property
     def deps(self) -> Iterator["Dependency"]:
-        for stage in self:
+        for stage in self.stages:
             yield from stage.deps
 
     @property
@@ -484,36 +402,9 @@ class IndexView:
         data: Dict[str, Union["DataIndex", "DataIndexView"]] = {}
         for workspace, data_index in self._index.data.items():
             if self._stages:
-                data_index.load()
                 data[workspace] = view(
                     data_index, partial(key_filter, workspace)
                 )
             else:
                 data[workspace] = DataIndex()
         return data
-
-
-if __name__ == "__main__":
-    import logging
-
-    from funcy import log_durations
-
-    from dvc.logger import setup
-    from dvc.repo import Repo
-
-    setup(level=logging.TRACE)  # type: ignore[attr-defined]
-
-    repo = Repo()
-    index = Index(repo, repo.fs)
-    print(index)
-    with log_durations(print, "collecting stages"):
-        # pylint: disable=pointless-statement
-        print("no of stages", len(index.stages))
-    with log_durations(print, "building graph"):
-        index.graph  # pylint: disable=pointless-statement
-    with log_durations(print, "calculating hash"):
-        print(index.identifier)
-    with log_durations(print, "updating"):
-        index2 = index.update(index.stages)
-    with log_durations(print, "calculating hash"):
-        print(index2.identifier)

dvc/repo/ls.py

@@ -70,11 +70,12 @@ def _ls(
         if not recursive:
             files.update(dirs)
 
+        parts = fs.path.relparts(root, fs_path)
+        if parts == (".",):
+            parts = ()
+
         for name, entry in files.items():
-            entry_fs_path = fs.path.join(root, name)
-            relparts = fs.path.relparts(entry_fs_path, fs_path)
-            name = os.path.join(*relparts)
-            infos[name] = entry
+            infos[os.path.join(*parts, name)] = entry
 
         if not recursive:
             break

dvc/repo/push.py

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

dvc/repo/reproduce.py

@@ -1,13 +1,14 @@
 import logging
-import typing
 from functools import partial
+from typing import TYPE_CHECKING, Iterator
 
 from dvc.exceptions import DvcException, ReproductionError
 from dvc.repo.scm_context import scm_context
+from dvc.stage.exceptions import CheckpointKilledError
 
 from . import locked
 
-if typing.TYPE_CHECKING:
+if TYPE_CHECKING:
     from dvc.stage import Stage
 
     from . import Repo
@@ -58,7 +59,7 @@ def _dump_stage(stage):
     dvcfile.dump(stage, update_pipeline=False)
 
 
-def _get_stage_files(stage: "Stage") -> typing.Iterator[str]:
+def _get_stage_files(stage: "Stage") -> Iterator[str]:
     yield stage.dvcfile.relpath
     for dep in stage.deps:
         if (
@@ -183,7 +184,8 @@ def _reproduce_stages(
     # `ret` is used to add a cosmetic newline.
     ret = []
     checkpoint_func = kwargs.pop("checkpoint_func", None)
-    for stage in steps:
+
+    for i, stage in enumerate(steps):
         if ret:
             logger.info("")
 
@@ -192,8 +194,6 @@ def _reproduce_stages(
                 _repro_callback, checkpoint_func, unchanged
             )
 
-        from dvc.stage.monitor import CheckpointKilledError
-
         try:
             ret = _reproduce_stage(stage, **kwargs)
 
@@ -210,7 +210,18 @@ def _reproduce_stages(
             if ret:
                 result.extend(ret)
         except CheckpointKilledError:
-            raise
+            result.append(stage)
+            logger.warning(
+                "Checkpoint stage '%s' was interrupted remaining stages in "
+                "pipeline will not be reproduced.",
+                stage.addressing,
+            )
+            logger.warning(
+                "skipped stages '%s'",
+                ", ".join(s.addressing for s in steps[i + 1 :]),
+            )
+
+            break
         except Exception as exc:
             raise ReproductionError(stage.addressing) from exc
 

dvc/repo/stage.py

@@ -369,7 +369,7 @@ class StageLoad:
             glob: Use `target` as a pattern to match stages in a file.
         """
         if not target:
-            return list(graph) if graph else list(self.repo.index)
+            return list(graph) if graph else self.repo.index.stages
 
         if recursive and self.fs.isdir(target):
             from dvc.repo.graph import collect_inside_path
@@ -410,7 +410,7 @@ class StageLoad:
             (see `collect()` for other arguments)
         """
         if not target:
-            return [StageInfo(stage) for stage in self.repo.index]
+            return [StageInfo(stage) for stage in self.repo.index.stages]
 
         target = as_posix(target)
 
@@ -517,7 +517,6 @@ class StageLoad:
 
         path = self._get_filepath(path)
         dvcfile = Dvcfile(self.repo, path)
-        # `dvcfile.stages` is not cached
         stages = dvcfile.stages  # type: ignore
 
         if isinstance(stages, SingleStageLoader):
@@ -526,13 +525,7 @@ class StageLoad:
             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,
-        )
+        return stages_, dvcfile
 
     def _collect_all_from_repo(
         self, onerror: Callable[[str, Exception], None] = None
@@ -583,12 +576,7 @@ class StageLoad:
             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)
+                    new_stages, dvcfile = self._load_file(file_path)
                 except DvcException as exc:
                     if onerror:
                         onerror(relpath(file_path), exc)
@@ -596,11 +584,11 @@ class StageLoad:
                     raise
 
                 stages.extend(new_stages)
-                if new_metrics:
+                if new_metrics := dvcfile.metrics:
                     metrics[file_path] = new_metrics
-                if new_plots:
+                if new_plots := dvcfile.plots:
                     plots[file_path] = new_plots
-                if new_params:
+                if new_params := dvcfile.params:
                     params[file_path] = new_params
 
                 outs.update(

dvc/repo/update.py

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

dvc/repo/worktree.py

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

dvc/scm.py

@@ -147,7 +147,7 @@ def clone(url: str, to_path: str, **kwargs):
                 fetch_all_exps(git, url, progress=pbar.update_git)
             return git
         except InternalCloneError as exc:
-            raise CloneError(str(exc))
+            raise CloneError("SCM error") from exc
 
 
 def resolve_rev(scm: "Git", rev: str) -> str:

dvc/stage/__init__.py

@@ -775,6 +775,7 @@ class PipelineStage(Stage):
         return f"{super().addressing}:{self.name}"
 
     def reload(self):
+        self.dvcfile._reset()  # pylint: disable=protected-access
         return self.dvcfile.stages[self.name]
 
     def _status_stage(self, ret):

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

@@ -9,6 +9,14 @@ class StageCmdFailedError(DvcException):
         super().__init__(msg)
 
 
+class CheckpointKilledError(DvcException):
+    def __init__(self, cmd, status=None):
+        msg = f"failed to finish: {cmd}"
+        if status is not None:
+            msg += f", exited with {status}"
+        super().__init__(msg)
+
+
 class StageFileDoesNotExistError(DvcException):
     DVC_IGNORED = "is dvc-ignored"
     DOES_NOT_EXIST = "does not exist"

dvc/stage/imports.py

@@ -21,22 +21,20 @@ def update_import(
     stage, rev=None, to_remote=False, remote=None, no_download=None, jobs=None
 ):
     stage.deps[0].update(rev=rev)
-    outs = stage.outs
-    deps = stage.deps
 
     frozen = stage.frozen
     stage.frozen = False
+    changed = stage.changed()
 
-    if stage.outs:
-        stage.outs[0].clear()
     try:
         if to_remote:
             _update_import_on_remote(stage, remote, jobs)
         else:
             stage.reproduce(no_download=no_download, jobs=jobs)
     finally:
-        if deps == stage.deps:
-            stage.outs = outs
+        if no_download and changed:
+            # Avoid retaining stale information
+            stage.outs[0].clear()
         stage.frozen = frozen
 
 

dvc/stage/loader.py

@@ -6,7 +6,7 @@ from itertools import chain
 from funcy import cached_property, get_in, lcat, once, project
 
 from dvc import dependency, output
-from dvc.parsing import FOREACH_KWD, JOIN, DataResolver, EntryNotFound
+from dvc.parsing import FOREACH_KWD, JOIN, EntryNotFound
 from dvc.parsing.versions import LOCKFILE_VERSION
 from dvc_data.hashfile.hash_info import HashInfo
 from dvc_data.hashfile.meta import Meta
@@ -20,13 +20,16 @@ logger = logging.getLogger(__name__)
 
 
 class StageLoader(Mapping):
-    def __init__(self, dvcfile, data, lockfile_data=None):
+    def __init__(
+        self,
+        dvcfile,
+        data,
+        lockfile_data=None,
+    ):
         self.dvcfile = dvcfile
+        self.resolver = self.dvcfile.resolver
         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 {}
@@ -36,11 +39,6 @@ class StageLoader(Mapping):
         else:
             self._lockfile_data = lockfile_data.get("stages", {})
 
-    @cached_property
-    def resolver(self):
-        wdir = self.repo.fs.path.parent(self.dvcfile.path)
-        return DataResolver(self.repo, wdir, self.data)
-
     @cached_property
     def lockfile_data(self):
         if not self._lockfile_data:
@@ -174,9 +172,6 @@ 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/stage/monitor.py

@@ -7,7 +7,7 @@ from dataclasses import dataclass
 from typing import TYPE_CHECKING, Callable, List
 
 from dvc.stage.decorators import relock_repo
-from dvc.stage.exceptions import StageCmdFailedError
+from dvc.stage.exceptions import CheckpointKilledError
 
 if TYPE_CHECKING:
     from dvc.stage import Stage
@@ -16,17 +16,13 @@ if TYPE_CHECKING:
 logger = logging.getLogger(__name__)
 
 
-class CheckpointKilledError(StageCmdFailedError):
-    pass
-
-
 @dataclass
 class MonitorTask:
     stage: "Stage"
     execute: Callable
     proc: subprocess.Popen
     done: threading.Event = threading.Event()
-    killed: threading.Event = threading.Event()
+    updated: threading.Event = threading.Event()
 
     @property
     def name(self) -> str:
@@ -77,7 +73,7 @@ class CheckpointTask(MonitorTask):
 class Monitor:
     AWAIT: float = 1.0
 
-    def __init__(self, tasks: List[MonitorTask]):
+    def __init__(self, tasks: List[CheckpointTask]):
         self.done = threading.Event()
         self.tasks = tasks
         self.monitor_thread = threading.Thread(
@@ -113,6 +109,7 @@ class Monitor:
                 if os.path.exists(task.signal_path):
                     try:
                         task.execute()
+                        task.updated.set()
                     except Exception:  # pylint: disable=broad-except
                         logger.exception(
                             "Error running '%s' task, '%s' will be aborted",
@@ -120,11 +117,11 @@ class Monitor:
                             task.stage,
                         )
                         Monitor.kill(task.proc)
-                        task.killed.set()
                     finally:
                         logger.debug(
                             "Removing signal file for '%s' task", task.name
                         )
                         os.remove(task.signal_path)
+
             if done.wait(Monitor.AWAIT):
                 return

dvc/stage/run.py

@@ -3,13 +3,17 @@ import os
 import signal
 import subprocess
 import threading
+from typing import TYPE_CHECKING, List
 
-from dvc.stage.monitor import Monitor
+from dvc.stage.monitor import CheckpointTask, Monitor
 from dvc.utils import fix_env
 
 from .decorators import unlocked_repo
 from .exceptions import StageCmdFailedError
 
+if TYPE_CHECKING:
+    from dvc.stage import Stage
+
 logger = logging.getLogger(__name__)
 
 
@@ -74,14 +78,15 @@ def get_executable():
     return (os.getenv("SHELL") or "/bin/sh") if os.name != "nt" else None
 
 
-def _run(stage, executable, cmd, checkpoint_func, **kwargs):
+def _run(stage: "Stage", executable, cmd, checkpoint_func, **kwargs):
+    # pylint: disable=protected-access
     main_thread = isinstance(
         threading.current_thread(),
-        threading._MainThread,  # pylint: disable=protected-access
+        threading._MainThread,  # type: ignore[attr-defined]
     )
+    old_handler = None
 
     exec_cmd = _make_cmd(executable, cmd)
-    old_handler = None
 
     try:
         p = subprocess.Popen(exec_cmd, **kwargs)
@@ -98,7 +103,7 @@ def _run(stage, executable, cmd, checkpoint_func, **kwargs):
 
         if p.returncode != 0:
             for t in tasks:
-                if t.killed.is_set():
+                if t.updated.is_set():
                     raise t.error_cls(cmd, p.returncode)
             raise StageCmdFailedError(cmd, p.returncode)
     finally:
@@ -106,12 +111,10 @@ def _run(stage, executable, cmd, checkpoint_func, **kwargs):
             signal.signal(signal.SIGINT, old_handler)
 
 
-def _get_monitor_tasks(stage, checkpoint_func, proc):
+def _get_monitor_tasks(stage, checkpoint_func, proc) -> List[CheckpointTask]:
 
     result = []
     if checkpoint_func:
-        from .monitor import CheckpointTask
-
         result.append(CheckpointTask(stage, checkpoint_func, proc))
 
     return result

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,14 +50,15 @@ dependencies = [
     "rich>=10.13.0",
     "pyparsing>=2.4.7",
     "typing-extensions>=3.7.4",
-    "scmrepo==0.1.4",
-    "dvc-render==0.0.14",
-    "dvc-task==0.1.6",
-    "dvclive>=1.0",
-    "dvc-data==0.28.3",
-    "dvc-http==2.27.2",
+    "scmrepo==0.1.5",
+    "dvc-render==0.0.17",
+    "dvc-task==0.1.9",
+    "dvclive>=1.2.2",
+    "dvc-data==0.28.5",
+    "dvc-http",
     "hydra-core>=1.1.0",
     "iterative-telemetry==0.0.6",
+    "dvc-studio-client>=0.1.1"
 ]
 dynamic = ["version"]
 classifiers = [
@@ -90,25 +91,25 @@ tests = [
     # Test requirements
     "pytest==7.2.0",
     "pytest-cov==4.0.0",
-    "pytest-xdist==3.0.2",
+    "pytest-xdist==3.1.0",
     "pytest-mock==3.10.0",
     "pytest-lazy-fixture==0.6.3",
     # https://github.com/docker/docker-py/issues/2902
     "pytest-docker==0.11.0; python_version < '3.10' or sys_platform != 'win32'",
     "flaky==3.7.0",
     "pytest-timeout==2.1.0",
-    "filelock==3.8.0",
+    "filelock==3.9.0",
     "beautifulsoup4==4.11.1",
     "pywin32>=225; sys_platform == 'win32'",  # optional dependency
     # pylint requirements
-    "pylint==2.15.7",
+    "pylint==2.15.9",
     # 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,
     # and, protected method calls in our tests
     "pylint-plugin-utils==0.7",
     # type-checking
-    "mypy==0.990",
+    "mypy==0.991",
     "types-requests",
     "types-tabulate",
     "types-toml",
@@ -161,7 +162,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",

scripts/build-requirements.txt

@@ -1,3 +1,3 @@
 # NOTE: using strict version due to
 # https://github.com/iterative/dvc/issues/7949
-PyInstaller==5.6.2
+PyInstaller==5.7.0

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

@@ -0,0 +1,14 @@
+import json
+
+import pytest
+
+pytestmark = pytest.mark.xfail(
+    raises=(
+        # looks like dvc-task is not saving json atomically
+        json.JSONDecodeError,
+        # needs investigation, maybe a file locking issue?
+        PermissionError,
+    ),
+    strict=False,
+    reason="See https://github.com/iterative/dvc/issues/8570",
+)

tests/func/experiments/conftest.py

@@ -3,6 +3,7 @@ import pytest
 from tests.unit.repo.experiments.conftest import (  # noqa, pylint disable=unused-argument
     checkpoint_stage,
     exp_stage,
+    failed_checkpoint_stage,
     failed_exp_stage,
     test_queue,
 )

tests/func/experiments/test_checkpoints.py

@@ -90,14 +90,11 @@ def test_resume_checkpoint(
 def test_reset_checkpoint(
     tmp_dir, scm, dvc, checkpoint_stage, caplog, workspace
 ):
-    dvc.experiments.run(
-        checkpoint_stage.addressing, name="foo", tmp_dir=not workspace
-    )
+    dvc.experiments.run(checkpoint_stage.addressing, tmp_dir=not workspace)
 
     results = dvc.experiments.run(
         checkpoint_stage.addressing,
         params=["foo=2"],
-        name="foo",
         tmp_dir=not workspace,
         reset=True,
     )
@@ -228,7 +225,8 @@ def test_auto_push_during_iterations(
     ref_info = first(exp_refs_by_rev(scm, exp))
     assert git_upstream.tmp_dir.scm.get_ref(str(ref_info)) == exp
 
-    assert auto_push_spy.call_count == 2
+    # Assert 3 pushes: 2 checkpoints and final commit
+    assert auto_push_spy.call_count == 3
     assert auto_push_spy.call_args[0][2] == remote
 
 
@@ -275,3 +273,23 @@ def test_auto_push_self_remote(
             "automatically be pushed to the default DVC remote (if any) "
             "on each experiment commit." in caplog.text
         )
+
+
+def test_tmp_dir_failed_checkpoint(
+    tmp_dir, scm, dvc, failed_checkpoint_stage, caplog
+):
+    baseline = scm.get_rev()
+
+    results = dvc.experiments.run(
+        failed_checkpoint_stage.addressing, params=["foo=2"], tmp_dir=True
+    )
+    exp = first(results)
+
+    result = dvc.experiments.show()[baseline]
+    # Assert 4 rows: baseline, 2 checkpoints, and final commit
+    assert len(result) == 4
+    assert exp in result
+    assert (
+        "Checkpoint stage 'failed-checkpoint-file' was interrupted remaining "
+        "stages in pipeline will not be reproduced." in caplog.text
+    )

tests/func/experiments/test_experiments.py

@@ -4,6 +4,8 @@ import os
 import stat
 from textwrap import dedent
 
+import dulwich.errors
+import dulwich.file
 import pytest
 from funcy import first
 
@@ -11,6 +13,7 @@ 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.refs import CELERY_STASH
 from dvc.repo.experiments.utils import exp_refs_by_rev
 from dvc.scm import resolve_rev
 from dvc.stage.exceptions import StageFileDoesNotExistError
@@ -114,12 +117,12 @@ def test_apply(tmp_dir, scm, dvc, exp_stage):
     exp_a = first(results)
 
     results = dvc.experiments.run(
-        exp_stage.addressing, params=["foo=3"], tmp_dir=True
+        exp_stage.addressing, params=["foo=3"], tmp_dir=True, name="foo"
     )
     exp_b = first(results)
 
     with pytest.raises(InvalidArgumentError):
-        dvc.experiments.apply("foo")
+        dvc.experiments.apply("bar")
 
     dvc.experiments.apply(exp_a)
     assert (tmp_dir / "params.yaml").read_text().strip() == "foo: 2"
@@ -131,11 +134,42 @@ def test_apply(tmp_dir, scm, dvc, exp_stage):
         assert (tmp_dir / "params.yaml").read_text().strip() == "foo: 2"
         assert (tmp_dir / "metrics.yaml").read_text().strip() == "foo: 2"
 
-    dvc.experiments.apply(exp_b)
+    dvc.experiments.apply("foo")
     assert (tmp_dir / "params.yaml").read_text().strip() == "foo: 3"
     assert (tmp_dir / "metrics.yaml").read_text().strip() == "foo: 3"
 
 
+@pytest.mark.xfail(
+    raises=(
+        dulwich.errors.CommitError,
+        dulwich.file.FileLocked,
+    ),
+    strict=False,
+    reason="See https://github.com/iterative/dvc/issues/8570",
+)
+def test_apply_failed(tmp_dir, scm, dvc, failed_exp_stage, test_queue):
+    from dvc.exceptions import InvalidArgumentError
+
+    dvc.experiments.run(
+        failed_exp_stage.addressing, params=["foo=2"], queue=True
+    )
+    exp_rev = dvc.experiments.scm.resolve_rev(f"{CELERY_STASH}@{{0}}")
+
+    dvc.experiments.run(
+        failed_exp_stage.addressing, params=["foo=3"], queue=True, name="foo"
+    )
+    dvc.experiments.run(run_all=True)
+
+    with pytest.raises(InvalidArgumentError):
+        dvc.experiments.apply("bar")
+
+    dvc.experiments.apply(exp_rev)
+    assert (tmp_dir / "params.yaml").read_text().strip() == "foo: 2"
+
+    dvc.experiments.apply("foo")
+    assert (tmp_dir / "params.yaml").read_text().strip() == "foo: 3"
+
+
 def test_apply_queued(tmp_dir, scm, dvc, exp_stage):
     from dvc.repo.experiments.exceptions import ApplyConflictError
 
@@ -183,7 +217,6 @@ def test_apply_untracked(tmp_dir, scm, dvc, exp_stage):
 
 
 def test_get_baseline(tmp_dir, scm, dvc, exp_stage):
-    from dvc.repo.experiments.refs import CELERY_STASH
 
     init_rev = scm.get_rev()
     assert dvc.experiments.get_baseline(init_rev) is None
@@ -684,7 +717,7 @@ def test_experiments_workspace_not_log_exception(caplog, dvc, scm):
     assert not caplog.text
 
 
-def test_run_env(tmp_dir, dvc, scm):
+def test_run_env(tmp_dir, dvc, scm, mocker):
     dump_run_env = dedent(
         """\
         import os
@@ -703,8 +736,15 @@ def test_run_env(tmp_dir, dvc, scm):
     )
     dvc.experiments.run()
     assert (tmp_dir / "DVC_EXP_BASELINE_REV").read_text().strip() == baseline
-    assert (tmp_dir / "DVC_EXP_NAME").read_text().strip() == ""
+    assert (tmp_dir / "DVC_EXP_NAME").read_text().strip() != ""
 
     dvc.experiments.run(name="foo")
     assert (tmp_dir / "DVC_EXP_BASELINE_REV").read_text().strip() == baseline
     assert (tmp_dir / "DVC_EXP_NAME").read_text().strip() == "foo"
+
+
+def test_experiment_unchanged(tmp_dir, scm, dvc, exp_stage):
+    dvc.experiments.run(exp_stage.addressing)
+    dvc.experiments.run(exp_stage.addressing)
+
+    assert len(dvc.experiments.ls()["master"]) == 2

tests/func/experiments/test_save.py

@@ -1,16 +1,27 @@
-from contextlib import nullcontext
-
 import pytest
 from funcy import first
 
-from dvc.repo.experiments.exceptions import ExperimentExistsError
+from dvc.repo.experiments.exceptions import (
+    ExperimentExistsError,
+    InvalidArgumentError,
+)
 from dvc.repo.experiments.utils import exp_refs_by_rev
 from dvc.scm import resolve_rev
-from dvc.stage.exceptions import StageCommitError
+
+
+@pytest.fixture
+def modified_exp_stage(exp_stage, tmp_dir):
+    with open(tmp_dir / "copy.py", "a", encoding="utf-8") as fh:
+        fh.write("\n# dummy change")
+    yield
+
+
+def test_exp_save_unchanged(tmp_dir, dvc, scm, exp_stage):
+    dvc.experiments.save()
 
 
 @pytest.mark.parametrize("name", (None, "test"))
-def test_exp_save(tmp_dir, dvc, scm, exp_stage, name):
+def test_exp_save(tmp_dir, dvc, scm, exp_stage, name, modified_exp_stage):
     baseline = scm.get_rev()
 
     exp = dvc.experiments.save(name=name)
@@ -22,39 +33,39 @@ def test_exp_save(tmp_dir, dvc, scm, exp_stage, 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")
+def test_exp_save_overwrite_experiment(
+    tmp_dir, dvc, scm, exp_stage, modified_exp_stage
+):
+    name = "dummy"
+    dvc.experiments.save(name=name)
 
     with pytest.raises(ExperimentExistsError):
-        dvc.experiments.save(name="dummy")
+        dvc.experiments.save(name=name)
 
-    dvc.experiments.save(name="dummy", force=True)
+    dvc.experiments.save(name=name, force=True)
+
+
+@pytest.mark.parametrize(
+    "name",
+    (
+        "invalid/name",
+        "invalid..name",
+        "invalid~name",
+        "invalid?name",
+        "invalidname.",
+    ),
+)
+def test_exp_save_invalid_name(tmp_dir, dvc, scm, exp_stage, name):
+    with pytest.raises(InvalidArgumentError):
+        dvc.experiments.save(name=name, force=True)
 
 
 def test_exp_save_after_commit(tmp_dir, dvc, scm, exp_stage):
     baseline = scm.get_rev()
-    dvc.experiments.save(name="exp-1")
+    dvc.experiments.save(name="exp-1", force=True)
 
     tmp_dir.scm_gen({"new_file": "new_file"}, commit="new baseline")
-    dvc.experiments.save(name="exp-2")
+    dvc.experiments.save(name="exp-2", force=True)
 
     all_exps = dvc.experiments.ls(all_commits=True)
     assert all_exps[baseline[:7]] == ["exp-1"]

tests/func/experiments/test_set_params.py

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

tests/func/experiments/test_show.py

@@ -220,6 +220,7 @@ def test_show_failed_experiment(tmp_dir, scm, dvc, failed_exp_stage):
                 "msg": "Experiment run failed.",
                 "type": "",
             },
+            "name": ANY,
         }
     }
 
@@ -244,7 +245,8 @@ def test_show_checkpoint(
     exp_rev = first(results)
 
     results = dvc.experiments.show()[baseline_rev]
-    assert len(results) == checkpoint_stage.iterations + 1
+    # Assert 4 rows: baseline, 2 checkpoints, and final commit
+    assert len(results) == checkpoint_stage.iterations + 2
 
     checkpoints = []
     for rev, exp in results.items():
@@ -688,14 +690,14 @@ def test_show_only_changed(tmp_dir, dvc, scm, capsys, copy_script):
     params_file = tmp_dir / "params.yaml"
     params_data = {
         "foo": 1,
-        "bar": 1,
+        "goobar": 1,
     }
     (tmp_dir / params_file).dump(params_data)
 
     dvc.run(
         cmd="python copy.py params.yaml metrics.yaml",
         metrics_no_cache=["metrics.yaml"],
-        params=["foo", "bar"],
+        params=["foo", "goobar"],
         name="copy-file",
         deps=["copy.py"],
     )
@@ -716,18 +718,18 @@ def test_show_only_changed(tmp_dir, dvc, scm, capsys, copy_script):
     capsys.readouterr()
     assert main(["exp", "show"]) == 0
     cap = capsys.readouterr()
-    assert "bar" in cap.out
+    assert "goobar" in cap.out
 
     capsys.readouterr()
     assert main(["exp", "show", "--only-changed"]) == 0
     cap = capsys.readouterr()
-    assert "bar" not in cap.out
+    assert "goobar" not in cap.out
 
     capsys.readouterr()
     assert main(["exp", "show", "--only-changed", "--keep=.*bar"]) == 0
     cap = capsys.readouterr()
-    assert "params.yaml:bar" in cap.out
-    assert "metrics.yaml:bar" in cap.out
+    assert "params.yaml:goobar" in cap.out
+    assert "metrics.yaml:goobar" in cap.out
 
 
 def test_show_parallel_coordinates(
@@ -1048,7 +1050,8 @@ def test_show_checkpoint_error(tmp_dir, scm, dvc, checkpoint_stage, mocker):
     exp_ref = str(first(exp_refs_by_rev(scm, exp_rev)))
 
     results = dvc.experiments.show()[baseline_rev]
-    assert len(results) == checkpoint_stage.iterations + 1
+    # Assert 4 rows: baseline, 2 checkpoints, and final commit
+    assert len(results) == checkpoint_stage.iterations + 2
 
     checkpoints = {}
     for rev in results:

tests/func/experiments/test_utils.py

@@ -0,0 +1,67 @@
+from dvc_studio_client import env
+from funcy import first
+
+
+def test_generate_random_exp_name(tmp_dir, dvc, scm, exp_stage, mocker):
+    mocker.patch(
+        "dvc.repo.experiments.utils.random.choice",
+        side_effect=[
+            0,
+            0,
+            0,
+            0,
+            1,
+            1,
+            0,
+            0,
+        ],
+    )
+
+    ref = first(dvc.experiments.run(exp_stage.addressing, params=["foo=1"]))
+    assert dvc.experiments.get_exact_name([ref])[ref] == "0-0"
+
+    # Causes 1 retry
+    ref = first(dvc.experiments.run(exp_stage.addressing, params=["foo=2"]))
+    assert dvc.experiments.get_exact_name([ref])[ref] == "1-1"
+
+    tmp_dir.scm_gen({"foo": "bar"}, commit="foo")
+    # Can use same name because of different baseline_rev
+    ref = first(dvc.experiments.run(exp_stage.addressing, params=["foo=1"]))
+    assert dvc.experiments.get_exact_name([ref])[ref] == "0-0"
+
+
+def test_post_to_studio(tmp_dir, dvc, scm, exp_stage, mocker, monkeypatch):
+    valid_response = mocker.MagicMock()
+    valid_response.status_code = 200
+    mocked_post = mocker.patch("requests.post", return_value=valid_response)
+
+    monkeypatch.setenv(env.STUDIO_ENDPOINT, "https://0.0.0.0")
+    monkeypatch.setenv(env.STUDIO_REPO_URL, "STUDIO_REPO_URL")
+    monkeypatch.setenv(env.STUDIO_TOKEN, "STUDIO_TOKEN")
+
+    baseline_sha = scm.get_rev()
+    exp_rev = first(
+        dvc.experiments.run(exp_stage.addressing, params=["foo=1"])
+    )
+    name = dvc.experiments.get_exact_name([exp_rev])[exp_rev]
+    assert mocked_post.call_count == 2
+
+    start_call, done_call = mocked_post.call_args_list
+
+    assert start_call.kwargs["json"] == {
+        "type": "start",
+        "repo_url": "STUDIO_REPO_URL",
+        "baseline_sha": baseline_sha,
+        "name": name,
+        "params": {"params.yaml": {"foo": 1}},
+        "client": "dvc",
+    }
+
+    assert done_call.kwargs["json"] == {
+        "type": "done",
+        "repo_url": "STUDIO_REPO_URL",
+        "baseline_sha": baseline_sha,
+        "name": name,
+        "client": "dvc",
+        "experiment_rev": exp_rev,
+    }

tests/func/test_ls.py

@@ -587,4 +587,10 @@ def test_broken_symlink(tmp_dir, dvc):
             "isexec": False,
             "path": ".dvcignore",
         },
+        {
+            "isout": False,
+            "isdir": False,
+            "isexec": False,
+            "path": "link",
+        },
     ]

tests/func/test_repo_index.py

@@ -1,12 +1,10 @@
-import os
 from itertools import chain
 
 import pytest
 from pygtrie import Trie
 
 from dvc.repo.index import Index
-from dvc.stage import PipelineStage, Stage
-from dvc.utils import relpath
+from dvc.stage import Stage
 
 
 def test_index(tmp_dir, scm, dvc, run_copy):
@@ -17,15 +15,11 @@ def test_index(tmp_dir, scm, dvc, run_copy):
     index = Index(dvc)
     assert index.fs == dvc.fs
 
-    assert len(index) == len(index.stages) == 2
-    assert set(index.stages) == set(index) == {stage1, stage2}
-    assert stage1 in index
-    assert stage2 in index
+    assert set(index.stages) == {stage1, stage2}
 
     assert index.outs_graph
     assert index.graph
     assert isinstance(index.outs_trie, Trie)
-    assert index.identifier
     index.check_graph()
 
 
@@ -42,41 +36,6 @@ def test_repr(tmp_dir, scm, dvc):
     assert repr(Index(dvc)) == f"Index({dvc}, fs@{rev[:7]})"
 
 
-def test_filter_index(tmp_dir, dvc, run_copy):
-    tmp_dir.dvc_gen("foo", "foo")
-    stage2 = run_copy("foo", "bar", name="copy-foo-bar")
-
-    def filter_pipeline(stage):
-        return bool(stage.cmd)
-
-    filtered_index = Index(dvc).filter(filter_pipeline)
-    assert list(filtered_index) == [stage2]
-
-
-def test_slice_index(tmp_dir, dvc):
-    tmp_dir.gen({"dir1": {"foo": "foo"}, "dir2": {"bar": "bar"}})
-    with (tmp_dir / "dir1").chdir():
-        (stage1,) = dvc.add("foo")
-    with (tmp_dir / "dir2").chdir():
-        (stage2,) = dvc.add("bar")
-
-    index = Index(dvc)
-
-    sliced = index.slice("dir1")
-    assert set(sliced) == {stage1}
-    assert sliced.stages is not index.stages  # sanity check
-
-    sliced = index.slice(tmp_dir / "dir1")
-    assert set(sliced) == {stage1}
-
-    sliced = index.slice("dir2")
-    assert set(sliced) == {stage2}
-
-    with (tmp_dir / "dir1").chdir():
-        sliced = index.slice(relpath(tmp_dir / "dir2"))
-        assert set(sliced) == {stage2}
-
-
 def outputs_equal(actual, expected):
     actual, expected = list(actual), list(expected)
 
@@ -147,7 +106,7 @@ def test_add_update(dvc):
     dup_stage2 = Stage(dvc, path="path2")
     dup_index = index.update([dup_stage1, dup_stage2])
     assert not index.stages
-    assert len(new_index) == 1
+    assert len(new_index.stages) == 1
     assert new_index.stages[0] is new_stage
     assert set(map(id, dup_index.stages)) == {id(dup_stage1), id(dup_stage2)}
 
@@ -165,67 +124,6 @@ def assert_index_equal(first, second, strict=True, ordered=True):
         ), "Index is not strictly equal"
 
 
-def test_discard_remove(dvc):
-    stage = Stage(dvc, path="path1")
-    index = Index(dvc, stages=[stage])
-
-    assert list(index.discard(Stage(dvc, "path2"))) == list(index)
-    new_index = index.discard(stage)
-    assert len(new_index) == 0
-
-    with pytest.raises(ValueError):
-        index.remove(Stage(dvc, "path2"))
-    assert index.stages == [stage]
-    assert list(index.remove(stage)) == []
-
-
-def test_difference(dvc):
-    stages = [Stage(dvc, path=f"path{i}") for i in range(10)]
-    index = Index(dvc, stages=stages)
-
-    new_index = index.difference([*stages[:5], Stage(dvc, path="path100")])
-    assert index.stages == stages
-    assert set(new_index) == set(stages[5:])
-
-
-def test_dumpd(dvc):
-    stages = [
-        PipelineStage(dvc, "dvc.yaml", name="stage1"),
-        Stage(dvc, "path"),
-    ]
-    index = Index(dvc, stages=stages)
-    assert index.dumpd() == {"dvc.yaml:stage1": {}, "path": {}}
-    assert index.identifier == "d43da84e9001540c26abf2bf4541c275"
-
-
-def test_unique_identifier(tmp_dir, dvc, scm, run_copy):
-    dvc.scm_context.autostage = True
-    tmp_dir.dvc_gen("foo", "foo")
-    run_copy("foo", "bar", name="copy-foo-bar")
-
-    revs = []
-    n_commits = 5
-    for i in range(n_commits):
-        # create a few empty commits
-        scm.commit(f"commit {i}")
-        revs.append(scm.get_rev())
-    assert len(set(revs)) == n_commits  # the commit revs should be unique
-
-    ids = []
-    for _ in dvc.brancher(revs=revs):
-        index = Index(dvc)
-        assert index.stages
-        ids.append(index.identifier)
-
-    # we get "workspace" as well from the brancher by default
-    assert len(revs) + 1 == len(ids)
-    possible_ids = {
-        True: "dd9cf3a3dc2caac5835e64f258ac2d7b",
-        False: "b446f3cdc35539884dcc474d90ac6286",
-    }
-    assert set(ids) == {possible_ids[os.name == "posix"]}
-
-
 def test_skip_graph_checks(dvc, mocker):
     # See https://github.com/iterative/dvc/issues/2671 for more info
     mock_build_graph = mocker.spy(Index.graph, "fget")
@@ -293,18 +191,6 @@ def test_used_objs(tmp_dir, scm, dvc, run_copy, rev):
     }
 
 
-def test_getitem(tmp_dir, dvc, run_copy):
-    (stage1,) = tmp_dir.dvc_gen("foo", "foo")
-    stage2 = run_copy("foo", "bar", name="copy-foo-bar")
-
-    index = Index(dvc)
-    assert index["foo.dvc"] == stage1
-    assert index["copy-foo-bar"] == stage2
-
-    with pytest.raises(KeyError):
-        _ = index["no-valid-stage-name"]
-
-
 def test_view_granular_dir(tmp_dir, scm, dvc, run_copy):
     tmp_dir.dvc_gen(
         {"dir": {"subdir": {"in-subdir": "in-subdir"}, "in-dir": "in-dir"}},

tests/func/test_stage.py

@@ -317,6 +317,8 @@ def test_stage_remove_pipeline_stage(tmp_dir, dvc, run_copy):
 
     with dvc.lock:
         stage.remove()
+
+    dvc_file._reset()
     assert stage.name not in dvc_file.stages
     assert "copy-bar-foobar" in dvc_file.stages
 

tests/func/test_update.py

@@ -2,6 +2,7 @@ import os
 
 import pytest
 
+from dvc.dependency import base
 from dvc.dvcfile import Dvcfile
 from dvc.exceptions import InvalidArgumentError
 from dvc.testing.tmp_dir import make_subrepo
@@ -156,6 +157,47 @@ def test_update_before_and_after_dvc_init(tmp_dir, dvc, git_dir):
     assert dvc.status([stage.path]) == {}
 
 
+def test_update_unchanged(tmp_dir, dvc, erepo_dir, mocker):
+    with erepo_dir.chdir():
+        erepo_dir.dvc_gen("file", "file content", commit="add file")
+
+    assert (erepo_dir / "file").exists()
+    stage = dvc.imp(os.fspath(erepo_dir), "file")
+
+    spy = mocker.spy(base, "fs_download")
+    dvc.update([stage.path])
+
+    assert not spy.called
+
+
+@pytest.mark.parametrize("outs_exist", (False, True))
+def test_update_no_download(tmp_dir, dvc, erepo_dir, outs_exist, mocker):
+    with erepo_dir.chdir():
+        erepo_dir.dvc_gen("file", "file content", commit="add file")
+        initial_rev = erepo_dir.scm.get_rev()
+
+    stage = dvc.imp(os.fspath(erepo_dir), "file", no_download=not outs_exist)
+
+    assert stage.deps[0].def_repo["rev_lock"] == initial_rev
+
+    dst = tmp_dir / "file"
+    assert dst.exists() is outs_exist
+
+    with erepo_dir.chdir():
+        erepo_dir.dvc_gen("file", "updated file content", commit="update file")
+        new_rev = erepo_dir.scm.get_rev()
+
+    updated_stage = dvc.update([stage.path], rev=new_rev, no_download=True)[0]
+    assert not dst.exists()
+
+    assert updated_stage.deps[0].def_repo["rev_lock"] == new_rev
+
+    # output must have no information since no_download=True
+    out = updated_stage.outs[0]
+    assert out.hash_info.value is None
+    assert out.meta.size is None
+
+
 def test_update_import_url(tmp_dir, dvc, workspace):
     workspace.gen("file", "file content")
 
@@ -176,17 +218,20 @@ def test_update_import_url(tmp_dir, dvc, workspace):
     assert dst.read_text() == "updated file content"
 
 
-def test_update_import_url_no_download(tmp_dir, dvc, workspace):
+@pytest.mark.parametrize("outs_exist", (False, True))
+def test_update_import_url_no_download(
+    tmp_dir, dvc, workspace, outs_exist, mocker
+):
     workspace.gen("file", "file content")
 
     dst = tmp_dir / "imported_file"
     stage = dvc.imp_url(
-        "remote://workspace/file", os.fspath(dst), no_download=True
+        "remote://workspace/file", os.fspath(dst), no_download=not outs_exist
     )
-    hash_info = stage.deps[0].hash_info
 
-    assert not dst.is_file()
-    assert stage.deps[0].hash_info.value == "d10b4c3ff123b26dc068d43a8bef2d23"
+    assert dst.exists() is outs_exist
+    hash_info = stage.deps[0].hash_info
+    assert hash_info.value == "d10b4c3ff123b26dc068d43a8bef2d23"
 
     workspace.gen("file", "updated file content")
 
@@ -197,12 +242,25 @@ def test_update_import_url_no_download(tmp_dir, dvc, workspace):
     assert updated_hash_info != hash_info
     assert updated_hash_info.value == "6ffba511ce3aa40b8231d1b1f8c5fba5"
 
+    # output must have no information since no_download=True
     out = updated_stage.outs[0]
     assert out.hash_info.value is None
     assert out.hash_info.name is None
     assert out.meta.size is None
 
 
+def test_update_import_url_unchanged(tmp_dir, dvc, workspace, mocker):
+    workspace.gen("file", "file content")
+
+    dst = tmp_dir / "imported_file"
+    stage = dvc.imp_url("remote://workspace/file", os.fspath(dst))
+
+    spy = mocker.spy(base, "fs_download")
+
+    dvc.update([stage.path])
+    assert not spy.called
+
+
 def test_update_rev(tmp_dir, dvc, scm, git_dir):
     with git_dir.chdir():
         git_dir.scm_gen({"foo": "foo"}, commit="first")

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

@@ -92,6 +92,7 @@ def test_experiments_kill(dvc, scm, mocker):
         [
             "queue",
             "kill",
+            "--force",
             "exp1",
             "exp2",
         ]
@@ -105,7 +106,7 @@ def test_experiments_kill(dvc, scm, mocker):
     )
 
     assert cmd.run() == 0
-    m.assert_called_once_with(revs=["exp1", "exp2"])
+    m.assert_called_once_with(revs=["exp1", "exp2"], force=True)
 
 
 def test_experiments_start(dvc, scm, mocker):
@@ -114,7 +115,7 @@ def test_experiments_start(dvc, scm, mocker):
 
     cmd = cli_args.func(cli_args)
     m = mocker.patch(
-        "dvc.repo.experiments.queue.celery.LocalCeleryQueue.spawn_worker",
+        "dvc.repo.experiments.queue.celery.LocalCeleryQueue._spawn_worker",
     )
 
     assert cmd.run() == 0

tests/unit/remote/test_http.py

@@ -1,91 +0,0 @@
-import ssl
-
-from dvc.fs import HTTPFileSystem
-
-
-def test_public_auth_method(dvc):
-    config = {
-        "url": "http://example.com/",
-        "path": "file.html",
-        "user": "",
-        "password": "",
-    }
-
-    fs = HTTPFileSystem(**config)
-
-    assert "auth" not in fs.fs_args["client_kwargs"]
-    assert "headers" not in fs.fs_args
-
-
-def test_basic_auth_method(dvc):
-    user = "username"
-    password = "password"
-    config = {
-        "url": "http://example.com/",
-        "path": "file.html",
-        "auth": "basic",
-        "user": user,
-        "password": password,
-    }
-
-    fs = HTTPFileSystem(**config)
-
-    assert fs.fs_args["client_kwargs"]["auth"].login == user
-    assert fs.fs_args["client_kwargs"]["auth"].password == password
-
-
-def test_custom_auth_method(dvc):
-    header = "Custom-Header"
-    password = "password"
-    config = {
-        "url": "http://example.com/",
-        "path": "file.html",
-        "auth": "custom",
-        "custom_auth_header": header,
-        "password": password,
-    }
-
-    fs = HTTPFileSystem(**config)
-
-    headers = fs.fs_args["headers"]
-    assert header in headers
-    assert headers[header] == password
-
-
-def test_ssl_verify_disable(dvc):
-    config = {
-        "url": "http://example.com/",
-        "path": "file.html",
-        "ssl_verify": False,
-    }
-
-    fs = HTTPFileSystem(**config)
-    assert not fs.fs_args["client_kwargs"]["connector"]._ssl
-
-
-def test_ssl_verify_custom_cert(dvc, mocker):
-    mocker.patch("ssl.SSLContext.load_verify_locations")
-    config = {
-        "url": "http://example.com/",
-        "path": "file.html",
-        "ssl_verify": "/path/to/custom/cabundle.pem",
-    }
-
-    fs = HTTPFileSystem(**config)
-
-    assert isinstance(
-        fs.fs_args["client_kwargs"]["connector"]._ssl, ssl.SSLContext
-    )
-
-
-def test_http_method(dvc):
-    config = {
-        "url": "http://example.com/",
-        "path": "file.html",
-    }
-
-    fs = HTTPFileSystem(**config, method="PUT")
-    assert fs.upload_method == "PUT"
-
-    fs = HTTPFileSystem(**config, method="POST")
-    assert fs.upload_method == "POST"

tests/unit/render/test_vega_converter.py

@@ -274,6 +274,111 @@ def test_finding_lists(dictionary, expected_result):
             {"x": "v", "y": "v2", "x_label": "v", "y_label": "v2"},
             id="multi_file_json",
         ),
+        pytest.param(
+            {"f": {"metric": [{"v": 1, "v2": 0.1}, {"v": 2, "v2": 0.2}]}},
+            {"y": ["v", "v2"]},
+            [
+                {
+                    "dvc_inferred_y_value": 1,
+                    "v": 1,
+                    "v2": 0.1,
+                    "step": 0,
+                    VERSION_FIELD: {
+                        "revision": "r",
+                        "filename": "f",
+                        "field": "v",
+                    },
+                },
+                {
+                    "dvc_inferred_y_value": 2,
+                    "v": 2,
+                    "v2": 0.2,
+                    "step": 1,
+                    VERSION_FIELD: {
+                        "revision": "r",
+                        "filename": "f",
+                        "field": "v",
+                    },
+                },
+                {
+                    "dvc_inferred_y_value": 0.1,
+                    "v": 1,
+                    "v2": 0.1,
+                    "step": 0,
+                    VERSION_FIELD: {
+                        "revision": "r",
+                        "filename": "f",
+                        "field": "v2",
+                    },
+                },
+                {
+                    "dvc_inferred_y_value": 0.2,
+                    "v": 2,
+                    "v2": 0.2,
+                    "step": 1,
+                    VERSION_FIELD: {
+                        "revision": "r",
+                        "filename": "f",
+                        "field": "v2",
+                    },
+                },
+            ],
+            {
+                "x": "step",
+                "y": "dvc_inferred_y_value",
+                "x_label": "step",
+                "y_label": "y",
+            },
+            id="y_list",
+        ),
+        pytest.param(
+            {
+                "f": {"metric": [{"v": 1, "v2": 0.1, "v3": 0.01}]},
+                "f2": {"metric": [{"v": 1, "v2": 0.1}]},
+            },
+            {"y": {"f": ["v2", "v3"], "f2": ["v2"]}, "x": "v"},
+            [
+                {
+                    "dvc_inferred_y_value": 0.1,
+                    "v": 1,
+                    "v2": 0.1,
+                    "v3": 0.01,
+                    VERSION_FIELD: {
+                        "revision": "r",
+                        "filename": "f",
+                        "field": "v2",
+                    },
+                },
+                {
+                    "dvc_inferred_y_value": 0.01,
+                    "v": 1,
+                    "v2": 0.1,
+                    "v3": 0.01,
+                    VERSION_FIELD: {
+                        "revision": "r",
+                        "filename": "f",
+                        "field": "v3",
+                    },
+                },
+                {
+                    "dvc_inferred_y_value": 0.1,
+                    "v": 1,
+                    "v2": 0.1,
+                    VERSION_FIELD: {
+                        "revision": "r",
+                        "filename": "f2",
+                        "field": "v2",
+                    },
+                },
+            ],
+            {
+                "x": "v",
+                "y": "dvc_inferred_y_value",
+                "x_label": "v",
+                "y_label": "y",
+            },
+            id="multi_source_y_single_x",
+        ),
     ],
 )
 def test_convert(

tests/unit/repo/experiments/conftest.py

@@ -30,15 +30,23 @@ CHECKPOINT_SCRIPT_FORMAT = dedent(
     shutil.copyfile({}, {})
 
     if os.getenv("DVC_CHECKPOINT"):
-        for _ in range(checkpoint_iterations):
+        for index in range(checkpoint_iterations):
             value += 1
+            {}
             with open(checkpoint_file, "w") as fobj:
                 fobj.write(str(value))
             make_checkpoint()
 """
 )
 CHECKPOINT_SCRIPT = CHECKPOINT_SCRIPT_FORMAT.format(
-    "sys.argv[1]", "sys.argv[2]", "sys.argv[3]", "sys.argv[4]"
+    "sys.argv[1]", "sys.argv[2]", "sys.argv[3]", "sys.argv[4]", ""
+)
+FAILED_CHECKPOINT_SCRIPT = CHECKPOINT_SCRIPT_FORMAT.format(
+    "sys.argv[1]",
+    "sys.argv[2]",
+    "sys.argv[3]",
+    "sys.argv[4]",
+    "if index == (checkpoint_iterations - 2): raise Exception",
 )
 
 
@@ -90,6 +98,30 @@ def checkpoint_stage(tmp_dir, scm, dvc, mocker):
     return stage
 
 
+@pytest.fixture
+def failed_checkpoint_stage(tmp_dir, scm, dvc, mocker):
+    mocker.patch("dvc.stage.run.Monitor.AWAIT", 0.01)
+
+    tmp_dir.gen("checkpoint.py", FAILED_CHECKPOINT_SCRIPT)
+    tmp_dir.gen("params.yaml", "foo: 1")
+    stage = dvc.run(
+        cmd=(
+            f"python checkpoint.py foo {DEFAULT_ITERATIONS+2} "
+            "params.yaml metrics.yaml"
+        ),
+        metrics_no_cache=["metrics.yaml"],
+        params=["foo"],
+        checkpoints=["foo"],
+        deps=["checkpoint.py"],
+        no_exec=True,
+        name="failed-checkpoint-file",
+    )
+    scm.add(["dvc.yaml", "checkpoint.py", "params.yaml", ".gitignore"])
+    scm.commit("init")
+    stage.iterations = DEFAULT_ITERATIONS
+    return stage
+
+
 @pytest.fixture
 def failed_exp_stage(tmp_dir, scm, dvc, copy_script):
     tmp_dir.gen("params.yaml", "foo: 1")
@@ -132,7 +164,7 @@ def test_queue(tmp_dir, dvc, scm, mocker) -> LocalCeleryQueue:
     import celery
 
     queue = dvc.experiments.celery_queue
-    mocker.patch.object(queue, "spawn_worker")
+    mocker.patch.object(queue, "_spawn_worker")
 
     f = partial(
         _thread_worker,

tests/unit/repo/experiments/queue/test_celery.py

@@ -29,27 +29,25 @@ def test_shutdown(test_queue, mocker):
 
 def test_shutdown_with_kill(test_queue, mocker):
 
-    sig = _foo.s()
-    mock_entry = mocker.Mock(stash_rev=_foo.name)
-
-    result = sig.freeze()
+    mock_entry_foo = mocker.Mock(stash_rev="af12de")
+    mock_entry_foo.name = "foo"
+    mock_entry_bar = mocker.Mock(stash_rev="bar")
+    mock_entry_bar.name = None
 
     shutdown_spy = mocker.patch("celery.app.control.Control.shutdown")
     mocker.patch.object(
         test_queue,
-        "_iter_active_tasks",
-        return_value=[(result, mock_entry)],
+        "iter_active",
+        return_value=[mock_entry_foo, mock_entry_bar],
     )
-    kill_spy = mocker.patch.object(test_queue.proc, "kill")
+    kill_spy = mocker.patch.object(test_queue, "_kill_entries")
 
     test_queue.shutdown(kill=True)
 
-    sig.delay()
-
-    assert result.get() == "foo"
-    assert result.id not in test_queue._shutdown_task_ids
-    kill_spy.assert_called_once_with(mock_entry.stash_rev)
     shutdown_spy.assert_called_once()
+    kill_spy.assert_called_once_with(
+        {mock_entry_foo: "foo", mock_entry_bar: "bar"}, True
+    )
 
 
 def test_post_run_after_kill(test_queue):
@@ -80,7 +78,8 @@ def test_post_run_after_kill(test_queue):
     assert result_foo.get(timeout=10) == "foo"
 
 
-def test_celery_queue_kill(test_queue, mocker):
+@pytest.mark.parametrize("force", [True, False])
+def test_celery_queue_kill(test_queue, mocker, force):
 
     mock_entry_foo = mocker.Mock(stash_rev="foo")
     mock_entry_bar = mocker.Mock(stash_rev="bar")
@@ -139,13 +138,13 @@ def test_celery_queue_kill(test_queue, mocker):
 
     kill_mock = mocker.patch.object(
         test_queue.proc,
-        "kill",
+        "kill" if force else "interrupt",
         side_effect=mocker.MagicMock(side_effect=kill_function),
     )
     with pytest.raises(
         CannotKillTasksError, match="Task 'foobar' is initializing,"
     ):
-        test_queue.kill(["bar", "foo", "foobar"])
+        test_queue.kill(["bar", "foo", "foobar"], force=force)
     assert kill_mock.called_once_with(mock_entry_foo.stash_rev)
     assert kill_mock.called_once_with(mock_entry_bar.stash_rev)
     assert kill_mock.called_once_with(mock_entry_foobar.stash_rev)

tests/unit/repo/experiments/test_remove.py

@@ -11,6 +11,7 @@ def test_remove_done_tasks(dvc, test_queue, scm, mocker):
     ref_info_dict = {}
     for name in success_test:
         ref_info_dict[name] = mocker.Mock()
+        ref_info_dict[name].name = name
     for name in failed_test:
         ref_info_dict[name] = None
 

tests/unit/repo/experiments/test_utils.py

@@ -32,7 +32,19 @@ def test_resolve_exp_ref(tmp_dir, scm, git_upstream, name_only, use_url):
 
 
 @pytest.mark.parametrize(
-    "name,result", [("name", True), ("group/name", False), ("na me", False)]
+    "name,result",
+    [
+        ("name", True),
+        ("group/name", False),
+        ("na me", False),
+        ("invalid/.name", False),
+        ("@", pytest.param(False, marks=pytest.mark.xfail)),
+        (":", False),
+        ("^", False),
+        ("*", False),
+        ("~", False),
+        ("?", False),
+    ],
 )
 def test_run_check_ref_format(scm, name, result):
 

tests/unit/stage/test_loader_pipeline_file.py

@@ -247,11 +247,11 @@ def test_load_stage_wdir_and_path_correctly(dvc, stage_data, lock_data):
 
 def test_load_stage_mapping(dvc, stage_data, lock_data):
     dvcfile = Dvcfile(dvc, PIPELINE_FILE)
-    loader = StageLoader(
-        dvcfile, {"stages": {"stage": stage_data}}, {"stage": lock_data}
-    )
-    assert len(loader) == 1
-    assert "stage" in loader
-    assert "stage1" not in loader
-    assert loader.keys() == {"stage"}
-    assert isinstance(loader["stage"], PipelineStage)
+    dvcfile.contents = {"stages": {"stage": stage_data}}
+    dvcfile.lockfile_contents = {"stage": lock_data}
+
+    assert len(dvcfile.stages) == 1
+    assert "stage" in dvcfile.stages
+    assert "stage1" not in dvcfile.stages
+    assert dvcfile.stages.keys() == {"stage"}
+    assert isinstance(dvcfile.stages["stage"], PipelineStage)

tests/unit/stage/test_stage.py

@@ -55,18 +55,19 @@ def test_path_conversion(dvc):
     assert stage.dumpd()["wdir"] == "../.."
 
 
-def test_stage_update(mocker):
+def test_stage_update(dvc, mocker):
     dep = RepoDependency({"url": "example.com"}, None, "dep_path")
     mocker.patch.object(dep, "update", return_value=None)
 
-    stage = Stage(None, "path", deps=[dep])
+    stage = Stage(dvc, "path", deps=[dep])
     reproduce = mocker.patch.object(stage, "reproduce")
     is_repo_import = mocker.patch(
         __name__ + ".Stage.is_repo_import", new_callable=mocker.PropertyMock
     )
 
     is_repo_import.return_value = True
-    stage.update()
+    with dvc.lock:
+        stage.update()
     assert reproduce.called_once_with()
 
     is_repo_import.return_value = False