exp save: use -m for --message short option

* repro --pull: don't fail for cloud-versioned remotes

* raise RunCacheNotSupported for cloud-versioned remotes

.pre-commit-config.yaml

@@ -32,7 +32,7 @@ repos:
     hooks:
       - id: isort
   - repo: https://github.com/astral-sh/ruff-pre-commit
-    rev: 'v0.0.290'
+    rev: 'v0.0.291'
     hooks:
       - id: ruff
         args: [--fix, --exit-non-zero-on-fix]

dvc/api/__init__.py

@@ -1,5 +1,6 @@
 from dvc.fs.dvc import _DVCFileSystem as DVCFileSystem
 
+from .artifacts import artifacts_show
 from .data import open  # pylint: disable=redefined-builtin
 from .data import get_url, read
 from .experiments import exp_save, exp_show
@@ -10,6 +11,7 @@ __all__ = [
     "all_branches",
     "all_commits",
     "all_tags",
+    "artifacts_show",
     "exp_save",
     "exp_show",
     "get_url",

dvc/api/artifacts.py

@@ -0,0 +1,47 @@
+from typing import Any, Dict, Optional
+
+from dvc.repo import Repo
+
+
+def artifacts_show(
+    name: str,
+    version: Optional[str] = None,
+    stage: Optional[str] = None,
+    repo: Optional[str] = None,
+) -> Dict[str, str]:
+    """
+    Return path and Git revision for an artifact in a DVC project.
+
+    The resulting path and revision can be used in conjunction with other dvc.api
+    calls to open and read the artifact.
+
+    Args:
+        name (str): name of the artifact to open.
+        version (str, optional): version of the artifact to open. Defaults to
+            the latest version.
+        stage (str, optional): name of the model registry stage.
+        repo: (str, optional): path or URL for the DVC repo.
+
+    Returns:
+        Dictionary of the form:
+            {
+                "rev": ...,
+                "path": ...,
+            }
+
+    Raises:
+        dvc.exceptions.ArtifactNotFoundError: The specified artifact was not found in
+            the repo.
+    """
+    if version and stage:
+        raise ValueError("Artifact version and stage are mutually exclusive.")
+
+    repo_kwargs: Dict[str, Any] = {
+        "subrepos": True,
+        "uninitialized": True,
+    }
+    with Repo.open(repo, **repo_kwargs) as _repo:
+        rev = _repo.artifacts.get_rev(name, version=version, stage=stage)
+        with _repo.switch(rev):
+            path = _repo.artifacts.get_path(name)
+        return {"rev": rev, "path": path}

dvc/cli/parser.py

@@ -7,6 +7,7 @@ from functools import lru_cache
 from dvc import __version__
 from dvc.commands import (
     add,
+    artifacts,
     cache,
     check_ignore,
     checkout,
@@ -89,6 +90,7 @@ COMMANDS = [
     experiments,
     check_ignore,
     data,
+    artifacts,
 ]
 
 

dvc/commands/artifacts.py

@@ -0,0 +1,133 @@
+import argparse
+import logging
+
+from dvc.cli import completion
+from dvc.cli.command import CmdBaseNoRepo
+from dvc.cli.utils import DictAction, append_doc_link, fix_subparsers
+from dvc.exceptions import DvcException
+
+logger = logging.getLogger(__name__)
+
+
+class CmdArtifactsGet(CmdBaseNoRepo):
+    def run(self):
+        from dvc.repo.artifacts import Artifacts
+        from dvc.scm import CloneError
+        from dvc.ui import ui
+
+        try:
+            count, out = Artifacts.get(
+                self.args.url,
+                name=self.args.name,
+                version=self.args.rev,
+                stage=self.args.stage,
+                force=self.args.force,
+                config=self.args.config,
+                remote=self.args.remote,
+                remote_config=self.args.remote_config,
+                out=self.args.out,
+            )
+            ui.write(f"Downloaded {count} file(s) to '{out}'")
+            return 0
+        except CloneError:
+            logger.exception("failed to get '%s'", self.args.name)
+            return 1
+        except DvcException:
+            logger.exception(
+                "failed to get '%s' from '%s'", self.args.name, self.args.url
+            )
+            return 1
+
+
+def add_parser(subparsers, parent_parser):
+    ARTIFACTS_HELP = "DVC model registry artifact commands."
+
+    artifacts_parser = subparsers.add_parser(
+        "artifacts",
+        parents=[parent_parser],
+        description=append_doc_link(ARTIFACTS_HELP, "artifacts"),
+        help=ARTIFACTS_HELP,
+        formatter_class=argparse.RawDescriptionHelpFormatter,
+    )
+    artifacts_subparsers = artifacts_parser.add_subparsers(
+        dest="cmd",
+        help="Use `dvc artifacts CMD --help` to display command-specific help.",
+    )
+    fix_subparsers(artifacts_subparsers)
+
+    ARTIFACTS_GET_HELP = "Download an artifact from a DVC project."
+    get_parser = artifacts_subparsers.add_parser(
+        "get",
+        parents=[parent_parser],
+        description=append_doc_link(ARTIFACTS_GET_HELP, "artifacts/get"),
+        help=ARTIFACTS_HELP,
+        formatter_class=argparse.RawDescriptionHelpFormatter,
+    )
+    get_parser.add_argument("url", help="Location of DVC repository to download from")
+    get_parser.add_argument(
+        "name", help="Name of artifact in the repository"
+    ).complete = completion.FILE
+    get_parser.add_argument(
+        "--rev",
+        nargs="?",
+        help="Artifact version",
+        metavar="<version>",
+    )
+    get_parser.add_argument(
+        "--stage",
+        nargs="?",
+        help="Artifact stage",
+        metavar="<stage>",
+    )
+    get_parser.add_argument(
+        "-o",
+        "--out",
+        nargs="?",
+        help="Destination path to download artifact to",
+        metavar="<path>",
+    ).complete = completion.DIR
+    get_parser.add_argument(
+        "-j",
+        "--jobs",
+        type=int,
+        help=(
+            "Number of jobs to run simultaneously. "
+            "The default value is 4 * cpu_count(). "
+        ),
+        metavar="<number>",
+    )
+    get_parser.add_argument(
+        "-f",
+        "--force",
+        action="store_true",
+        default=False,
+        help="Override local file or folder if exists.",
+    )
+    get_parser.add_argument(
+        "--config",
+        type=str,
+        help=(
+            "Path to a config file that will be merged with the config "
+            "in the target repository."
+        ),
+    )
+    get_parser.add_argument(
+        "--remote",
+        type=str,
+        help=(
+            "Remote name to set as a default in the target repository "
+            "(only applicable when downloading from DVC remote)."
+        ),
+    )
+    get_parser.add_argument(
+        "--remote-config",
+        type=str,
+        nargs="*",
+        action=DictAction,
+        help=(
+            "Remote config options to merge with a remote's config (default or one "
+            "specified by '--remote') in the target repository (only applicable "
+            "when downloading from DVC remote)."
+        ),
+    )
+    get_parser.set_defaults(func=CmdArtifactsGet)

dvc/commands/data_sync.py

@@ -14,6 +14,10 @@ class CmdDataBase(CmdBase):
         from dvc.utils.humanize import get_summary
 
         default_msg = "Everything is up to date."
+
+        if not self.args.remote and not self.repo.config["core"].get("remote"):
+            ui.warn("No remote provided and no default remote set.")
+
         ui.write(get_summary(stats.items()) or default_msg)
 
 

dvc/commands/experiments/apply.py

@@ -4,21 +4,13 @@ import logging
 from dvc.cli import completion
 from dvc.cli.command import CmdBase
 from dvc.cli.utils import append_doc_link
-from dvc.ui import ui
 
 logger = logging.getLogger(__name__)
 
 
 class CmdExperimentsApply(CmdBase):
     def run(self):
-        if not self.args.force:
-            ui.write(
-                "The --no-force option is deprecated and will be removed in a future"
-                " DVC release. To revert the result of 'exp apply', run:\n"
-                "\n\tgit reset --hard\n"
-                "\tgit stash apply refs/exps/apply/stash\n"
-            )
-        self.repo.experiments.apply(self.args.experiment)
+        self.repo.experiments.apply(self.args.experiment, force=self.args.force)
 
         return 0
 
@@ -33,10 +25,11 @@ def add_parser(experiments_subparsers, parent_parser):
         formatter_class=argparse.RawDescriptionHelpFormatter,
     )
     experiments_apply_parser.add_argument(
-        "--no-force",
-        action="store_false",
-        dest="force",
-        help="Fail if this command would overwrite conflicting changes.",
+        "-f",
+        "--force",
+        action="store_true",
+        default=False,
+        help="Do not prompt when removing working directory files.",
     )
     experiments_apply_parser.add_argument(
         "experiment", help="Experiment to be applied."

dvc/commands/experiments/exec_run.py

@@ -49,7 +49,7 @@ def add_parser(experiments_subparsers, parent_parser):
         ),
     )
     exec_run_parser.add_argument(
-        "-M",
+        "-m",
         "--message",
         type=str,
         default=None,

dvc/commands/experiments/rename.py

@@ -55,7 +55,7 @@ def add_parser(experiments_subparsers, parent_parser):
     )
     experiments_rename_parser.add_argument(
         "name",
-        help="Name of new experiment.",
+        help="New name for the experiment.",
         metavar="<name>",
     )
     experiments_rename_parser.add_argument(

dvc/commands/experiments/save.py

@@ -72,7 +72,7 @@ def add_parser(experiments_subparsers, parent_parser):
         metavar="<path>",
     )
     save_parser.add_argument(
-        "-M",
+        "-m",
         "--message",
         type=str,
         default=None,

dvc/commands/gc.py

@@ -81,6 +81,7 @@ class CmdGC(CmdBase):
             rev=self.args.rev,
             num=self.args.num,
             not_in_remote=self.args.not_in_remote,
+            dry=self.args.dry,
         )
         return 0
 
@@ -213,4 +214,10 @@ def add_parser(subparsers, parent_parser):
         ),
         metavar="<paths>",
     )
+    gc_parser.add_argument(
+        "--dry",
+        action="store_true",
+        default=False,
+        help=("Only print what would get removed without actually removing."),
+    )
     gc_parser.set_defaults(func=CmdGC)

dvc/data_cloud.py

@@ -135,7 +135,7 @@ class DataCloud:
 
         remote = self.get_remote(name=name, command=command)
         if remote.fs.version_aware or remote.worktree:
-            raise NoRemoteError(
+            raise RemoteConfigError(
                 f"'{command}' is unsupported for cloud versioned remotes"
             )
         if hash_name in LEGACY_HASH_NAMES:

dvc/exceptions.py

@@ -1,6 +1,6 @@
 """Exceptions raised by the dvc."""
 import errno
-from typing import TYPE_CHECKING, Dict, List, Set
+from typing import TYPE_CHECKING, Dict, List, Optional, Set
 
 from dvc.utils import format_link
 
@@ -336,3 +336,24 @@ class CacheLinkError(DvcException):
 class PrettyDvcException(DvcException):
     def __pretty_exc__(self, **kwargs):
         """Print prettier exception message."""
+
+
+class ArtifactNotFoundError(DvcException):
+    """Thrown if an artifact is not found in the DVC repo.
+
+    Args:
+        name (str): artifact name.
+    """
+
+    def __init__(
+        self,
+        name: str,
+        version: Optional[str] = None,
+        stage: Optional[str] = None,
+    ):
+        self.name = name
+        self.version = version
+        self.stage = stage
+
+        desc = f" @ {stage or version}" if (stage or version) else ""
+        super().__init__(f"Unable to find artifact '{name}{desc}'")

dvc/fs/__init__.py

@@ -49,7 +49,9 @@ known_implementations.update(
 # pylint: enable=unused-import
 
 
-def download(fs: "FileSystem", fs_path: str, to: str, jobs: Optional[int] = None):
+def download(
+    fs: "FileSystem", fs_path: str, to: str, jobs: Optional[int] = None
+) -> int:
     with Callback.as_tqdm_callback(
         desc=f"Downloading {fs.path.name(fs_path)}",
         unit="files",
@@ -63,7 +65,8 @@ def download(fs: "FileSystem", fs_path: str, to: str, jobs: Optional[int] = None
                 if not path.endswith(fs.path.flavour.sep)
             ]
             if not from_infos:
-                return localfs.makedirs(to, exist_ok=True)
+                localfs.makedirs(to, exist_ok=True)
+                return 0
             to_infos = [
                 localfs.path.join(to, *fs.path.relparts(info, fs_path))
                 for info in from_infos
@@ -82,6 +85,7 @@ def download(fs: "FileSystem", fs_path: str, to: str, jobs: Optional[int] = None
             callback=cb,
             batch_size=jobs,
         )
+        return len(to_infos)
 
 
 def parse_external_url(url, config=None):

dvc/repo/__init__.py

@@ -577,6 +577,26 @@ class Repo:
 
         return DVCFileSystem(repo=self, subrepos=self.subrepos, **self._fs_conf)
 
+    @cached_property
+    def _btime(self):
+        if not self.tmp_dir:
+            return None
+
+        # Not all python versions/filesystems/platforms provide creation
+        # time (st_birthtime, stx_btime, etc), so we use our own dummy
+        # file and its mtime instead.
+        path = os.path.join(self.tmp_dir, "btime")
+
+        try:
+            with open(path, "x"):  # pylint: disable=unspecified-encoding
+                pass
+        except FileNotFoundError:
+            return None
+        except FileExistsError:
+            pass
+
+        return os.path.getmtime(path)
+
     @cached_property
     def site_cache_dir(self) -> str:
         import getpass
@@ -609,12 +629,17 @@ class Repo:
         # components were changed (useful to prevent newer dvc versions from
         # using older broken cache). Please reset this back to 0 if other parts
         # of the token components are changed.
-        salt = 0
+        salt = 1
+
+        # NOTE: This helps us avoid accidentally reusing cache for repositories
+        # that just happened to be at the same path as old deleted ones.
+        btime = self._btime or getattr(os.stat(root_dir), "st_birthtime", None)
 
         md5 = hashlib.md5(  # noqa: S324  # nosec B324, B303
             str(
                 (
                     root_dir,
+                    btime,
                     getpass.getuser(),
                     version_tuple[0],
                     salt,

dvc/repo/artifacts.py

@@ -1,42 +1,61 @@
 import logging
-import re
+import os
+import posixpath
 from pathlib import Path
-from typing import Dict, Optional
+from typing import TYPE_CHECKING, Any, Dict, List, Optional, Tuple, Union
 
 from dvc.annotations import Artifact
 from dvc.dvcfile import PROJECT_FILE
-from dvc.exceptions import InvalidArgumentError
-from dvc.repo import Repo
-from dvc.utils import relpath
+from dvc.exceptions import (
+    ArtifactNotFoundError,
+    DvcException,
+    FileExistsLocallyError,
+    InvalidArgumentError,
+)
+from dvc.utils import relpath, resolve_output
+from dvc.utils.objects import cached_property
 from dvc.utils.serialize import modify_yaml
 
+if TYPE_CHECKING:
+    from gto.tag import Tag as GTOTag
+    from scmrepo.git import GitTag
+
+    from dvc.repo import Repo
+    from dvc.scm import Git
+
 logger = logging.getLogger(__name__)
 
 
-# Constants are taken from GTO.
-# When we make it a dependency, we can import them instead
-SEPARATOR_IN_NAME = ":"
-DIRNAME = r"[a-z0-9-_./]+"
-NAME = r"[a-z0-9]([a-z0-9-/]*[a-z0-9])?"
-NAME_RE = re.compile(f"^{NAME}$")
-FULLNAME = f"((?P<dirname>{DIRNAME}){SEPARATOR_IN_NAME})?(?P<name>{NAME})"
-FULLNAME_RE = re.compile(f"^{FULLNAME}$")
+def check_name_format(name: str) -> None:
+    from gto.constants import assert_name_is_valid
+    from gto.exceptions import ValidationError
+
+    try:
+        assert_name_is_valid(name)
+    except ValidationError as exc:
+        raise InvalidArgumentError(
+            f"Can't use '{name}' as artifact name (ID)."
+        ) from exc
 
 
 def name_is_compatible(name: str) -> bool:
-    return bool(NAME_RE.search(name))
+    """
+    Only needed by DVCLive per iterative/dvclive#715
+    Will be removed in future release.
+    """
+    from gto.constants import assert_name_is_valid
+    from gto.exceptions import ValidationError
 
-
-def check_name_format(name: str) -> None:
-    if not name_is_compatible(name):
-        raise InvalidArgumentError(
-            f"Can't use '{name}' as artifact name (ID)."
-            " You can use letters and numbers, and use '-' as separator"
-            " (but not at the start or end)."
-        )
+    try:
+        assert_name_is_valid(name)
+        return True
+    except ValidationError:
+        return False
 
 
 def check_for_nested_dvc_repo(dvcfile: Path):
+    from dvc.repo import Repo
+
     if dvcfile.is_absolute():
         raise InvalidArgumentError("Use relative path to dvc.yaml.")
     path = dvcfile.parent
@@ -48,11 +67,36 @@ def check_for_nested_dvc_repo(dvcfile: Path):
         path = path.parent
 
 
+def _reformat_name(name: str) -> str:
+    from gto.constants import SEPARATOR_IN_NAME, fullname_re
+
+    # NOTE: DVC accepts names like
+    #   path/to/dvc.yaml:artifact_name
+    # but Studio/GTO tags are generated with
+    #   path/to:artifact_name
+    m = fullname_re.match(name)
+    if m and m.group("dirname"):
+        group = m.group("dirname").rstrip(SEPARATOR_IN_NAME)
+        dirname, basename = posixpath.split(group)
+        if basename == PROJECT_FILE:
+            name = f"{dirname}{SEPARATOR_IN_NAME}{m.group('name')}"
+    return name
+
+
 class Artifacts:
     def __init__(self, repo: "Repo") -> None:
         self.repo = repo
 
+    @cached_property
+    def scm(self) -> Optional["Git"]:
+        from dvc.scm import Git
+
+        if isinstance(self.repo.scm, Git):
+            return self.repo.scm
+        return None
+
     def read(self) -> Dict[str, Dict[str, Artifact]]:
+        """Read artifacts from dvc.yaml."""
         artifacts: Dict[str, Dict[str, Artifact]] = {}
         for (
             dvcfile,
@@ -69,6 +113,7 @@ class Artifacts:
         return artifacts
 
     def add(self, name: str, artifact: Artifact, dvcfile: Optional[str] = None):
+        """Add artifact to dvc.yaml."""
         with self.repo.scm_context(quiet=True):
             check_name_format(name)
             dvcyaml = Path(dvcfile or PROJECT_FILE)
@@ -85,3 +130,219 @@ class Artifacts:
             self.repo.scm_context.track_file(dvcfile)
 
         return artifacts.get(name)
+
+    def get_rev(
+        self, name: str, version: Optional[str] = None, stage: Optional[str] = None
+    ):
+        """Return revision containing the given artifact."""
+        from gto.base import sort_versions
+        from gto.tag import find, parse_tag
+
+        assert not (version and stage)
+        name = _reformat_name(name)
+        tags: List["GitTag"] = find(
+            name=name, version=version, stage=stage, scm=self.scm
+        )
+        if not tags:
+            raise ArtifactNotFoundError(name, version=version, stage=stage)
+        if version or stage:
+            return tags[-1].target
+        gto_tags: List["GTOTag"] = sort_versions(parse_tag(tag) for tag in tags)
+        return gto_tags[0].tag.target
+
+    def get_path(self, name: str):
+        """Return repo fspath for the given artifact."""
+        from gto.constants import SEPARATOR_IN_NAME, fullname_re
+
+        name = _reformat_name(name)
+        m = fullname_re.match(name)
+        if not m:
+            raise ArtifactNotFoundError(name)
+        dirname = m.group("dirname")
+        if dirname:
+            dirname = dirname.rstrip(SEPARATOR_IN_NAME)
+        dvcyaml = os.path.join(dirname, PROJECT_FILE) if dirname else PROJECT_FILE
+        artifact_name = m.group("name")
+        try:
+            artifact = self.read()[dvcyaml][artifact_name]
+        except KeyError as exc:
+            raise ArtifactNotFoundError(name) from exc
+        return os.path.join(dirname, artifact.path) if dirname else artifact.path
+
+    def download(
+        self,
+        name: str,
+        version: Optional[str] = None,
+        stage: Optional[str] = None,
+        out: Optional[str] = None,
+        force: bool = False,
+        jobs: Optional[int] = None,
+    ) -> Tuple[int, str]:
+        """Download the specified artifact."""
+        from dvc.fs import download as fs_download
+
+        logger.debug("Trying to download artifact '%s' via DVC", name)
+        rev = self.get_rev(name, version=version, stage=stage)
+        with self.repo.switch(rev):
+            path = self.get_path(name)
+            out = resolve_output(path, out, force=force)
+            fs = self.repo.dvcfs
+            fs_path = fs.from_os_path(path)
+            count = fs_download(
+                fs,
+                fs_path,
+                os.path.abspath(out),
+                jobs=jobs,
+            )
+        return count, out
+
+    @staticmethod
+    def _download_studio(
+        repo_url: str,
+        name: str,
+        version: Optional[str] = None,
+        stage: Optional[str] = None,
+        out: Optional[str] = None,
+        force: bool = False,
+        jobs: Optional[int] = None,
+        studio_config: Optional[Dict[str, Any]] = None,
+        **kwargs,
+    ) -> Tuple[int, str]:
+        from dvc_studio_client.model_registry import get_download_uris
+
+        from dvc.fs import Callback, HTTPFileSystem, generic, localfs
+
+        logger.debug("Trying to download artifact '%s' via studio", name)
+        out = out or os.getcwd()
+        to_infos: List[str] = []
+        from_infos: List[str] = []
+        if studio_config is None:
+            studio_config = {}
+        studio_config["repo_url"] = repo_url
+        try:
+            kwargs["dvc_studio_config"] = studio_config
+            for path, url in get_download_uris(
+                repo_url,
+                name,
+                version=version,
+                stage=stage,
+                **kwargs,
+            ).items():
+                to_info = localfs.path.join(out, path)
+                if localfs.exists(to_info) and not force:
+                    hint = "\nTo override it, re-run with '--force'."
+                    raise FileExistsLocallyError(  # noqa: TRY301
+                        relpath(to_info), hint=hint
+                    )
+                to_infos.append(to_info)
+                from_infos.append(url)
+        except DvcException:
+            raise
+        except Exception as exc:  # noqa: BLE001, pylint: disable=broad-except
+            raise DvcException(
+                f"Failed to download artifact '{name}' via Studio"
+            ) from exc
+        fs = HTTPFileSystem()
+        jobs = jobs or fs.jobs
+        with Callback.as_tqdm_callback(
+            desc=f"Downloading '{name}' from '{repo_url}'",
+            unit="files",
+        ) as cb:
+            cb.set_size(len(from_infos))
+            generic.copy(
+                fs, from_infos, localfs, to_infos, callback=cb, batch_size=jobs
+            )
+
+        return len(to_infos), relpath(localfs.path.commonpath(to_infos))
+
+    @classmethod
+    def get(  # noqa: C901, PLR0913
+        cls,
+        url: str,
+        name: str,
+        version: Optional[str] = None,
+        stage: Optional[str] = None,
+        config: Optional[Union[str, Dict[str, Any]]] = None,
+        remote: Optional[str] = None,
+        remote_config: Optional[Union[str, Dict[str, Any]]] = None,
+        out: Optional[str] = None,
+        force: bool = False,
+        jobs: Optional[int] = None,
+    ):
+        from dvc.config import Config
+        from dvc.repo import Repo
+
+        if version and stage:
+            raise InvalidArgumentError(
+                "Artifact version and stage are mutually exclusive."
+            )
+
+        # NOTE: We try to download the artifact up to three times
+        # 1. via studio with studio config loaded from environment
+        # 2. via studio with studio config loaded from DVC repo 'studio'
+        #    section + environment
+        # 3. via DVC remote
+
+        name = _reformat_name(name)
+        saved_exc: Optional[Exception] = None
+        try:
+            logger.trace("Trying studio-only config")  # type: ignore[attr-defined]
+            return cls._download_studio(
+                url,
+                name,
+                version=version,
+                stage=stage,
+                out=out,
+                force=force,
+                jobs=jobs,
+            )
+        except FileExistsLocallyError:
+            raise
+        except Exception as exc:  # noqa: BLE001, pylint: disable=broad-except
+            saved_exc = exc
+
+        if config and not isinstance(config, dict):
+            config = Config.load_file(config)
+        with Repo.open(
+            url=url,
+            subrepos=True,
+            uninitialized=True,
+            config=config,
+            remote=remote,
+            remote_config=remote_config,
+        ) as repo:
+            logger.trace("Trying repo [studio] config")  # type: ignore[attr-defined]
+            studio_config = dict(repo.config.get("studio"))
+            try:
+                return cls._download_studio(
+                    url,
+                    name,
+                    version=version,
+                    stage=stage,
+                    out=out,
+                    force=force,
+                    jobs=jobs,
+                    dvc_studio_config=studio_config,
+                )
+            except FileExistsLocallyError:
+                raise
+            except Exception as exc:  # noqa: BLE001, pylint: disable=broad-except
+                saved_exc = exc
+
+            try:
+                return repo.artifacts.download(
+                    name,
+                    version=version,
+                    stage=stage,
+                    out=out,
+                    force=force,
+                    jobs=jobs,
+                )
+            except FileExistsLocallyError:
+                raise
+            except Exception as exc:  # noqa: BLE001, pylint: disable=broad-except
+                if saved_exc:
+                    logger.exception(str(saved_exc), exc_info=saved_exc.__cause__)
+                raise DvcException(
+                    f"Failed to download artifact '{name}' via DVC remote"
+                ) from exc

dvc/repo/checkout.py

@@ -140,7 +140,8 @@ def checkout(  # noqa: C901
 
     with ui.progress(
         unit="entry",
-        desc="Building data index",
+        desc="Building workspace index",
+        leave=True,
     ) as pb:
         old = build_data_index(
             view,
@@ -155,6 +156,7 @@ def checkout(  # noqa: C901
     with ui.progress(
         desc="Comparing indexes",
         unit="entry",
+        leave=True,
     ) as pb:
         diff = compare(old, new, relink=relink, delete=True, callback=pb.as_callback())
 
@@ -175,7 +177,8 @@ def checkout(  # noqa: C901
 
     with ui.progress(
         unit="file",
-        desc="Checkout",
+        desc="Applying changes",
+        leave=True,
     ) as pb:
         apply(
             diff,

dvc/repo/diff.py

@@ -22,7 +22,7 @@ def _hash(entry):
     return None
 
 
-def _diff(old, new, with_missing=False):
+def _diff(old, new, data_keys, with_missing=False):
     from dvc_data.index.diff import ADD, DELETE, MODIFY, RENAME
     from dvc_data.index.diff import diff as idiff
 
@@ -34,12 +34,23 @@ def _diff(old, new, with_missing=False):
         "not in cache": [],
     }
 
+    def meta_cmp_key(meta):
+        if not meta:
+            return meta
+        return meta.isdir
+
     for change in idiff(
         old,
         new,
         with_renames=True,
-        hash_only=True,
+        meta_cmp_key=meta_cmp_key,
+        roots=data_keys,
     ):
+        if (change.old and change.old.isdir and not change.old.hash_info) or (
+            change.new and change.new.isdir and not change.new.hash_info
+        ):
+            continue
+
         if change.typ == ADD:
             ret["added"].append(
                 {
@@ -116,6 +127,7 @@ def diff(
         b_rev = "workspace"
         with_missing = True
 
+    data_keys = set()
     for rev in self.brancher(revs=[a_rev, b_rev]):
         if rev == "workspace" and b_rev != "workspace":
             # brancher always returns workspace, but we only need to compute
@@ -132,6 +144,8 @@ def diff(
             recursive=recursive,
         )
 
+        data_keys.update(view.data_keys.get("repo", set()))
+
         if rev == "workspace":
             from .index import build_data_index
 
@@ -165,4 +179,4 @@ def diff(
         new = indexes[b_rev]
 
     with ui.status("Calculating diff"):
-        return _diff(old, new, with_missing=with_missing)
+        return _diff(old, new, data_keys, with_missing=with_missing)

dvc/repo/experiments/executor/base.py

@@ -295,7 +295,7 @@ class BaseExecutor(ABC):
             stages = dvc.commit([], force=True, relink=False)
             exp_hash = cls.hash_exp(stages)
             if include_untracked:
-                dvc.scm.add(include_untracked)
+                dvc.scm.add(include_untracked, force=True)  # type: ignore[call-arg]
             cls.commit(
                 dvc.scm,  # type: ignore[arg-type]
                 exp_hash,

dvc/repo/fetch.py

@@ -2,6 +2,7 @@ import logging
 from typing import List, Tuple
 
 from dvc.exceptions import DownloadError
+from dvc.ui import ui
 from dvc_data.index import DataIndex, FileStorage
 
 from . import locked
@@ -106,7 +107,6 @@ def fetch(  # noqa: C901, PLR0913
     """
     from fsspec.utils import tokenize
 
-    from dvc.fs.callbacks import Callback
     from dvc_data.index.fetch import collect
     from dvc_data.index.fetch import fetch as ifetch
 
@@ -140,29 +140,31 @@ def fetch(  # noqa: C901, PLR0913
 
     cache_key = ("fetch", tokenize(sorted(indexes.keys())))
 
-    with Callback.as_tqdm_callback(
+    with ui.progress(
         desc="Collecting",
         unit="entry",
-    ) as cb:
+        leave=True,
+    ) as pb:
         data = collect(
             indexes.values(),
             "remote",
             cache_index=self.data_index,
             cache_key=cache_key,
-            callback=cb,
+            callback=pb.as_callback(),
         )
     data, unversioned_count = _log_unversioned(data)
     failed_count += unversioned_count
 
-    with Callback.as_tqdm_callback(
+    with ui.progress(
         desc="Fetching",
-        unit="file",
-    ) as cb:
+        bar_format="{desc}",
+        leave=True,
+    ) as pb:
         try:
             fetch_transferred, fetch_failed = ifetch(
                 data,
                 jobs=jobs,
-                callback=cb,
+                callback=pb.as_callback(),
             )  # pylint: disable=assignment-from-no-return
         finally:
             for fs_index in data:

dvc/repo/gc.py

@@ -67,6 +67,7 @@ def gc(  # noqa: PLR0913, C901
     rev: Optional[str] = None,
     num: Optional[int] = None,
     not_in_remote: bool = False,
+    dry: bool = False,
 ):
     # require `workspace` to be true to come into effect.
     # assume `workspace` to be enabled if any of `all_tags`, `all_commits`,
@@ -128,8 +129,10 @@ def gc(  # noqa: PLR0913, C901
     for scheme, odb in self.cache.by_scheme():
         if not odb:
             continue
-        removed = ogc(odb, used_obj_ids, jobs=jobs)
-        if not removed:
+        num_removed = ogc(odb, used_obj_ids, jobs=jobs, dry=dry)
+        if num_removed:
+            logger.info("Removed %d objects from %s cache.", num_removed, scheme)
+        else:
             logger.info("No unused '%s' cache to remove.", scheme)
 
     if not cloud:
@@ -137,9 +140,10 @@ def gc(  # noqa: PLR0913, C901
 
     for remote_odb, obj_ids in odb_to_obj_ids.items():
         assert remote_odb is not None
-        removed = ogc(remote_odb, obj_ids, jobs=jobs)
-        if removed:
+        num_removed = ogc(remote_odb, obj_ids, jobs=jobs, dry=dry)
+        if num_removed:
             get_index(remote_odb).clear()
+            logger.info("Removed %d objects from remote.", num_removed)
         else:
             logger.info("No unused cache to remove from remote.")
 

dvc/repo/index.py

@@ -1,5 +1,6 @@
 import logging
 import time
+from collections import defaultdict
 from functools import partial
 from typing import (
     TYPE_CHECKING,
@@ -319,6 +320,22 @@ class Index:
         for stage in self.stages:
             yield from stage.outs
 
+    @cached_property
+    def out_data_keys(self) -> Dict[str, Set["DataIndexKey"]]:
+        by_workspace: Dict[str, Set["DataIndexKey"]] = defaultdict(set)
+
+        by_workspace["repo"] = set()
+        by_workspace["local"] = set()
+
+        for out in self.outs:
+            if not out.use_cache:
+                continue
+
+            ws, key = out.index_key
+            by_workspace[ws].add(key)
+
+        return dict(by_workspace)
+
     @property
     def decorated_outs(self) -> Iterator["Output"]:
         for output in self.outs:
@@ -359,8 +376,6 @@ class Index:
 
     @cached_property
     def data_keys(self) -> Dict[str, Set["DataIndexKey"]]:
-        from collections import defaultdict
-
         by_workspace: Dict[str, Set["DataIndexKey"]] = defaultdict(set)
 
         by_workspace["repo"] = set()
@@ -377,8 +392,6 @@ class Index:
 
     @cached_property
     def metric_keys(self) -> Dict[str, Set["DataIndexKey"]]:
-        from collections import defaultdict
-
         from .metrics.show import _collect_top_level_metrics
 
         by_workspace: Dict[str, Set["DataIndexKey"]] = defaultdict(set)
@@ -400,8 +413,6 @@ class Index:
 
     @cached_property
     def plot_keys(self) -> Dict[str, Set["DataIndexKey"]]:
-        from collections import defaultdict
-
         by_workspace: Dict[str, Set["DataIndexKey"]] = defaultdict(set)
 
         by_workspace["repo"] = set()
@@ -520,8 +531,6 @@ class Index:
         jobs: Optional[int] = None,
         push: bool = False,
     ) -> "ObjectContainer":
-        from collections import defaultdict
-
         used: "ObjectContainer" = defaultdict(set)
         pairs = self.collect_targets(targets, recursive=recursive, with_deps=with_deps)
         for stage, filter_info in pairs:
@@ -640,9 +649,23 @@ class IndexView:
         yield from {out for (out, _) in self._filtered_outs}
 
     @cached_property
-    def _data_prefixes(self) -> Dict[str, "_DataPrefixes"]:
-        from collections import defaultdict
+    def out_data_keys(self) -> Dict[str, Set["DataIndexKey"]]:
+        by_workspace: Dict[str, Set["DataIndexKey"]] = defaultdict(set)
 
+        by_workspace["repo"] = set()
+        by_workspace["local"] = set()
+
+        for out in self.outs:
+            if not out.use_cache:
+                continue
+
+            ws, key = out.index_key
+            by_workspace[ws].add(key)
+
+        return dict(by_workspace)
+
+    @cached_property
+    def _data_prefixes(self) -> Dict[str, "_DataPrefixes"]:
         prefixes: Dict[str, "_DataPrefixes"] = defaultdict(
             lambda: _DataPrefixes(set(), set())
         )
@@ -660,8 +683,6 @@ class IndexView:
 
     @cached_property
     def data_keys(self) -> Dict[str, Set["DataIndexKey"]]:
-        from collections import defaultdict
-
         ret: Dict[str, Set["DataIndexKey"]] = defaultdict(set)
 
         for out, filter_info in self._filtered_outs:
@@ -714,7 +735,7 @@ class IndexView:
         return data
 
 
-def build_data_index(  # noqa: C901
+def build_data_index(  # noqa: C901, PLR0912
     index: Union["Index", "IndexView"],
     path: str,
     fs: "FileSystem",
@@ -777,14 +798,6 @@ def build_data_index(  # noqa: C901
             data.add(entry)
             callback.relative_update(1)
 
-        if compute_hash:
-            tree_meta, tree = build_tree(data, key, name=hash_name)
-            out_entry.meta = tree_meta
-            out_entry.hash_info = tree.hash_info
-            out_entry.loaded = True
-            data.add(out_entry)
-            callback.relative_update(1)
-
     for key in parents:
         parent_path = fs.path.join(path, *key)
         if not fs.exists(parent_path):
@@ -793,6 +806,23 @@ def build_data_index(  # noqa: C901
         data.add(direntry)
         callback.relative_update(1)
 
+    if compute_hash:
+        out_keys = index.out_data_keys.get(workspace, set())
+        data_keys = index.data_keys.get(workspace, set())
+        for key in data_keys.intersection(out_keys):
+            hash_name = _get_entry_hash_name(index, workspace, key)
+
+            out_entry = data.get(key)
+            if not out_entry or not out_entry.isdir:
+                continue
+
+            tree_meta, tree = build_tree(data, key, name=hash_name)
+            out_entry.meta = tree_meta
+            out_entry.hash_info = tree.hash_info
+            out_entry.loaded = True
+            data.add(out_entry)
+            callback.relative_update(1)
+
     return data
 
 
@@ -801,10 +831,14 @@ def _get_entry_hash_name(
 ) -> str:
     from dvc_data.hashfile.hash import DEFAULT_ALGORITHM
 
-    try:
-        src_entry = index.data[workspace][key]
+    for idx in reversed(range(len(key) + 1)):
+        prefix = key[:idx]
+        try:
+            src_entry = index.data[workspace][prefix]
+        except KeyError:
+            continue
+
         if src_entry.hash_info and src_entry.hash_info.name:
             return src_entry.hash_info.name
-    except KeyError:
-        pass
+
     return DEFAULT_ALGORITHM

dvc/stage/cache.py

@@ -6,6 +6,7 @@ from typing import TYPE_CHECKING, List, Optional, Tuple
 from funcy import first
 
 from dvc import fs
+from dvc.config import RemoteConfigError
 from dvc.exceptions import CollectCacheError, DvcException
 from dvc.utils import dict_sha256, relpath
 
@@ -99,10 +100,16 @@ class StageCache:
         if not os.path.exists(cache_dir):
             return None
 
-        for value in os.listdir(cache_dir):
-            cache = self._load_cache(key, value)
-            if cache:
-                return cache
+        newest_entry = first(
+            sorted(
+                os.listdir(cache_dir),
+                key=lambda f: os.path.getmtime(os.path.join(cache_dir, f)),
+                reverse=True,
+            )
+        )
+        cache = self._load_cache(key, newest_entry)
+        if cache:
+            return cache
 
         return None
 
@@ -268,15 +275,21 @@ 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, "push --run-cache", hash_name="md5-dos2unix"
-        )
+        try:
+            dest_odb = odb or self.repo.cloud.get_remote_odb(
+                remote, "push --run-cache", hash_name="md5-dos2unix"
+            )
+        except RemoteConfigError as e:
+            raise RunCacheNotSupported(e) from e
         return self.transfer(self.repo.cache.legacy, dest_odb)
 
     def pull(self, remote: Optional[str], odb: Optional["ObjectDB"] = None):
-        odb = odb or self.repo.cloud.get_remote_odb(
-            remote, "fetch --run-cache", hash_name="md5-dos2unix"
-        )
+        try:
+            odb = odb or self.repo.cloud.get_remote_odb(
+                remote, "fetch --run-cache", hash_name="md5-dos2unix"
+            )
+        except RemoteConfigError as e:
+            raise RunCacheNotSupported(e) from e
         return self.transfer(odb, self.repo.cache.legacy)
 
     def get_used_objs(self, used_run_cache, *args, **kwargs):

dvc/utils/__init__.py

@@ -248,7 +248,7 @@ def env2bool(var, undefined=False):
     return bool(re.search("1|y|yes|true", var, flags=re.I))
 
 
-def resolve_output(inp, out, force=False):
+def resolve_output(inp: str, out: Optional[str], force=False) -> str:
     from urllib.parse import urlparse
 
     from dvc.exceptions import FileExistsLocallyError

pyproject.toml

@@ -1,6 +1,6 @@
 [build-system]
 build-backend = "setuptools.build_meta"
-requires = ["setuptools>=61", "setuptools_scm[toml]>=7,<8"]
+requires = ["setuptools>=61", "setuptools_scm[toml]>=7"]
 
 [project]
 name = "dvc"
@@ -34,16 +34,17 @@ dependencies = [
     "configobj>=5.0.6",
     "distro>=1.3",
     "dpath<3,>=2.1.0",
-    "dvc-data>=2.16.3,<2.17.0",
+    "dvc-data>=2.18.1,<2.19.0",
     "dvc-http>=2.29.0",
     "dvc-render>=0.3.1,<1",
-    "dvc-studio-client>=0.9.2,<1",
+    "dvc-studio-client>=0.13.0,<1",
     "dvc-task>=0.3.0,<1",
     "flatten_dict<1,>=0.4.1",
     # https://github.com/iterative/dvc/issues/9654
     "flufl.lock>=5,<8",
     "funcy>=1.14",
     "grandalf<1,>=0.7",
+    "gto>=1.4.0,<2",
     "hydra-core>=1.1",
     "iterative-telemetry>=0.0.7",
     "networkx>=2.5",
@@ -76,7 +77,7 @@ gs = ["dvc-gs==2.22.1"]
 hdfs = ["dvc-hdfs==2.19"]
 lint = [
     "mypy==1.5.1",
-    "pylint==2.17.5",
+    "pylint==2.17.6",
     "types-colorama",
     "types-psutil",
     "types-pyinstaller",

tests/func/artifacts/test_artifacts.py

@@ -5,8 +5,8 @@ from copy import deepcopy
 import pytest
 
 from dvc.annotations import Artifact
-from dvc.exceptions import InvalidArgumentError
-from dvc.repo.artifacts import name_is_compatible
+from dvc.exceptions import ArtifactNotFoundError, InvalidArgumentError
+from dvc.repo.artifacts import check_name_format
 from dvc.utils.strictyaml import YAMLSyntaxError, YAMLValidationError
 
 dvcyaml = {
@@ -43,7 +43,7 @@ def test_artifacts_read_subdir(tmp_dir, dvc):
 
 def test_artifacts_read_bad_name(tmp_dir, dvc, caplog):
     bad_name_dvcyaml = deepcopy(dvcyaml)
-    bad_name_dvcyaml["artifacts"]["bad_name"] = {"type": "model", "path": "bad.pkl"}
+    bad_name_dvcyaml["artifacts"]["_bad_name_"] = {"type": "model", "path": "bad.pkl"}
 
     (tmp_dir / "dvc.yaml").dump(bad_name_dvcyaml)
 
@@ -54,7 +54,7 @@ def test_artifacts_read_bad_name(tmp_dir, dvc, caplog):
 
     with caplog.at_level(logging.WARNING):
         assert tmp_dir.dvc.artifacts.read() == {"dvc.yaml": artifacts}
-        assert "Can't use 'bad_name' as artifact name (ID)" in caplog.text
+        assert "Can't use '_bad_name_' as artifact name (ID)" in caplog.text
 
 
 def test_artifacts_add_subdir(tmp_dir, dvc):
@@ -160,7 +160,7 @@ def test_artifacts_read_fails_on_id_duplication(tmp_dir, dvc):
     ],
 )
 def test_name_is_compatible(name):
-    assert name_is_compatible(name)
+    check_name_format(name)
 
 
 @pytest.mark.parametrize(
@@ -172,7 +172,6 @@ def test_name_is_compatible(name):
         "###",
         "@@@",
         "a model",
-        "a_model",
         "-model",
         "model-",
         "model@1",
@@ -181,4 +180,34 @@ def test_name_is_compatible(name):
     ],
 )
 def test_name_is_compatible_fails(name):
-    assert not name_is_compatible(name)
+    with pytest.raises(InvalidArgumentError):
+        check_name_format(name)
+
+
+def test_get_rev(tmp_dir, dvc, scm):
+    scm.tag("[email protected]#1", annotated=True, message="foo")
+    scm.tag("[email protected]#1", annotated=True, message="foo")
+    scm.tag("myart#dev#1", annotated=True, message="foo")
+    rev = scm.get_rev()
+
+    assert dvc.artifacts.get_rev("myart") == rev
+    assert dvc.artifacts.get_rev("myart", version="v1.0.0") == rev
+    assert dvc.artifacts.get_rev("subdir:myart", version="v2.0.0") == rev
+    assert dvc.artifacts.get_rev("subdir/dvc.yaml:myart", version="v2.0.0") == rev
+    with pytest.raises(ArtifactNotFoundError):
+        dvc.artifacts.get_rev("myart", version="v3.0.0")
+    with pytest.raises(ArtifactNotFoundError):
+        dvc.artifacts.get_rev("myart", stage="prod")
+
+
+def test_get_path(tmp_dir, dvc):
+    (tmp_dir / "dvc.yaml").dump(dvcyaml)
+    subdir = tmp_dir / "subdir"
+    subdir.mkdir()
+    (subdir / "dvc.yaml").dump(dvcyaml)
+
+    assert dvc.artifacts.get_path("myart") == "myart.pkl"
+    assert dvc.artifacts.get_path("subdir:myart") == os.path.join("subdir", "myart.pkl")
+    assert dvc.artifacts.get_path("subdir/dvc.yaml:myart") == os.path.join(
+        "subdir", "myart.pkl"
+    )

tests/func/experiments/test_save.py

@@ -156,6 +156,18 @@ def test_untracked_dvclock_is_included_in_exp(tmp_dir, scm, dvc):
     assert fs.exists("dvc.lock")
 
 
+def test_exp_save_include_untracked_force(tmp_dir, dvc, scm):
+    setup_stage(tmp_dir, dvc, scm)
+
+    new_file = tmp_dir / "new_file"
+    new_file.write_text("new_file")
+    dvc.scm.ignore(new_file)
+    exp = dvc.experiments.save(include_untracked=["new_file"])
+
+    fs = scm.get_fs(exp)
+    assert fs.exists("new_file")
+
+
 def test_exp_save_custom_message(tmp_dir, dvc, scm):
     setup_stage(tmp_dir, dvc, scm)
 

tests/func/test_diff.py

@@ -668,3 +668,59 @@ def test_rename_multiple_files_same_hashes(tmp_dir, scm, dvc):
             },
         ],
     }
+
+
+def test_diff_granular(tmp_dir, dvc, scm):
+    tmp_dir.gen(
+        {
+            "dir": {
+                "data": {
+                    "subdir": {"subfoo": "subfoo", "subbar": "subbar"},
+                    "foo": "foo",
+                    "bar": "bar",
+                },
+            },
+        }
+    )
+
+    dvc.add(os.path.join("dir", "data"))
+    scm.add(os.path.join("dir", "data.dvc"))
+    scm.add(os.path.join("dir", ".gitignore"))
+    scm.commit("data")
+
+    assert dvc.diff() == {}
+
+    (tmp_dir / "dir" / "data" / "subdir" / "new").write_text("new")
+
+    assert dvc.diff() == {
+        "added": [
+            {
+                "hash": "22af645d1859cb5ca6da0c484f1f37ea",
+                "path": os.path.join("dir", "data", "subdir", "new"),
+            }
+        ],
+        "deleted": [],
+        "modified": [
+            {
+                "hash": {
+                    "new": "efa5b20d5f935dcc5555b26db6e19b76.dir",
+                    "old": "1aca2c799df82929bbdd976557975546.dir",
+                },
+                "path": os.path.join("dir", "data", ""),
+            }
+        ],
+        "not in cache": [],
+        "renamed": [],
+    }
+    assert dvc.diff(targets=[os.path.join("dir", "data", "subdir")]) == {
+        "added": [
+            {
+                "hash": "22af645d1859cb5ca6da0c484f1f37ea",
+                "path": os.path.join("dir", "data", "subdir", "new"),
+            }
+        ],
+        "deleted": [],
+        "modified": [],
+        "not in cache": [],
+        "renamed": [],
+    }

tests/func/test_gc.py

@@ -36,14 +36,14 @@ def test_gc_api(dvc, good_and_bad_cache):
     dvc.gc(workspace=True)
     odb = dvc.cache.local
     good_cache, bad_cache = good_and_bad_cache
-    assert set(odb.oids_exist([*good_cache, *bad_cache])) == good_cache
+    assert set(odb.all()) == good_cache
 
 
 def test_gc_cli(dvc, good_and_bad_cache):
     assert main(["gc", "-wf"]) == 0
     odb = dvc.cache.local
     good_cache, bad_cache = good_and_bad_cache
-    assert set(odb.oids_exist([*good_cache, *bad_cache])) == good_cache
+    assert set(odb.all()) == good_cache
 
 
 def test_gc_branches_tags(tmp_dir, dvc, scm):
@@ -423,3 +423,19 @@ def test_gc_cloud_remote_field(tmp_dir, scm, dvc, mocker, make_remote):
     mocked_remove = mocker.spy(LocalFileSystem, "remove")
     dvc.gc(workspace=True, cloud=True)
     assert len(mocked_remove.mock_calls) == 2  # local and other_remote
+
+
+def test_gc_dry(dvc, good_and_bad_cache):
+    dvc.gc(workspace=True, dry=True)
+    odb = dvc.cache.local
+    good_cache, bad_cache = good_and_bad_cache
+    assert set(odb.all()) != good_cache
+
+
+def test_gc_logging(caplog, dvc, good_and_bad_cache):
+    with caplog.at_level(logging.INFO, logger="dvc"):
+        dvc.gc(workspace=True)
+
+    assert "Removed 3 objects from repo cache." in caplog.text
+    assert "No unused 'local' cache to remove." in caplog.text
+    assert "No unused 'legacy' cache to remove." in caplog.text

tests/func/test_run_cache.py

@@ -1,8 +1,10 @@
 import os
 
 import pytest
+from funcy import first
 
 from dvc.dvcfile import LOCK_FILE
+from dvc.stage.cache import RunCacheNotSupported, _get_stage_hash
 from dvc.utils.fs import remove
 
 
@@ -99,6 +101,7 @@ def test_memory_for_multiple_runs_of_same_stage(tmp_dir, dvc, run_copy, mocker):
     from dvc.stage import run as _run
 
     mock_restore = mocker.spy(dvc.stage_cache, "restore")
+    mocker.spy(dvc.stage_cache, "_load_cache")
     mock_run = mocker.spy(_run, "cmd_run")
 
     (tmp_dir / "bar").unlink()
@@ -123,6 +126,46 @@ def test_memory_for_multiple_runs_of_same_stage(tmp_dir, dvc, run_copy, mocker):
     assert (tmp_dir / LOCK_FILE).exists()
 
 
+def test_newest_entry_is_loaded_for_non_deterministic_stage(tmp_dir, dvc, mocker):
+    tmp_dir.gen("foo", "foo")
+    assert not os.path.exists(dvc.stage_cache.cache_dir)
+
+    dvc.stage.add(
+        name="non-deterministic",
+        cmd='python -c "from time import time; print(time())" > bar',
+        deps=["foo"],
+        outs=["bar"],
+    )
+
+    for i in range(4):
+        (stage,) = dvc.reproduce("non-deterministic", force=True)
+        assert _recurse_count_files(dvc.stage_cache.cache_dir) == i + 1
+
+    key = _get_stage_hash(stage)
+    cache_dir = dvc.stage_cache._get_cache_dir(key)
+    old_entries = os.listdir(cache_dir)
+
+    (stage,) = dvc.reproduce("non-deterministic", force=True)
+    newest_output = (tmp_dir / "bar").read_text()
+    newest_entry = first(e for e in os.listdir(cache_dir) if e not in old_entries)
+
+    from dvc.stage import run as _run
+
+    mock_restore = mocker.spy(dvc.stage_cache, "restore")
+    mock_load = mocker.spy(dvc.stage_cache, "_load_cache")
+    mock_run = mocker.spy(_run, "cmd_run")
+
+    (tmp_dir / "bar").unlink()
+    (tmp_dir / LOCK_FILE).unlink()
+    (stage,) = dvc.reproduce("non-deterministic")
+
+    assert (tmp_dir / LOCK_FILE).exists()
+    assert (tmp_dir / "bar").read_text() == newest_output
+    mock_run.assert_not_called()
+    mock_restore.assert_called_once_with(stage, dry=False)
+    mock_load.assert_called_with(key, newest_entry)
+
+
 def test_memory_runs_of_multiple_stages(tmp_dir, dvc, run_copy, mocker):
     tmp_dir.gen("foo", "foo")
     assert not os.path.exists(dvc.stage_cache.cache_dir)
@@ -186,3 +229,15 @@ def test_restore_pull(tmp_dir, dvc, run_copy, mocker, local_remote):
     assert (tmp_dir / "bar").exists()
     assert not (tmp_dir / "foo").unlink()
     assert (tmp_dir / LOCK_FILE).exists()
+
+
+def test_push_pull_unsupported(tmp_dir, dvc, mocker, run_copy):
+    tmp_dir.gen("foo", "foo")
+    run_copy("foo", "bar", name="copy-foo-bar")
+    mocker.patch.object(
+        dvc.cloud, "get_remote_odb", side_effect=RunCacheNotSupported("foo")
+    )
+    with pytest.raises(RunCacheNotSupported):
+        dvc.push(run_cache=True)
+    with pytest.raises(RunCacheNotSupported):
+        dvc.pull(run_cache=True)

tests/unit/command/test_experiments.py

@@ -27,7 +27,19 @@ def test_experiments_apply(dvc, scm, mocker):
 
     assert cmd.run() == 0
 
-    m.assert_called_once_with(cmd.repo, "exp_rev")
+    m.assert_called_once_with(cmd.repo, "exp_rev", force=False)
+
+
+def test_experiments_apply_force(dvc, scm, mocker):
+    cli_args = parse_args(["experiments", "apply", "exp_rev", "--force"])
+    assert cli_args.func == CmdExperimentsApply
+
+    cmd = cli_args.func(cli_args)
+    m = mocker.patch("dvc.repo.experiments.apply.apply", return_value={})
+
+    assert cmd.run() == 0
+
+    m.assert_called_once_with(cmd.repo, "exp_rev", force=True)
 
 
 def test_experiments_diff(dvc, scm, mocker):
@@ -467,6 +479,24 @@ def test_experiments_save(dvc, scm, mocker):
     )
 
 
+def test_experiments_save_message(dvc, scm, mocker):
+    cli_args = parse_args(["exp", "save", "-m", "custom commit message"])
+    assert cli_args.func == CmdExperimentsSave
+
+    cmd = cli_args.func(cli_args)
+    m = mocker.patch("dvc.repo.experiments.save.save", return_value="acabb")
+
+    assert cmd.run() == 0
+
+    m.assert_called_once_with(
+        cmd.repo,
+        name=None,
+        force=False,
+        include_untracked=[],
+        message="custom commit message",
+    )
+
+
 def test_experiments_clean(dvc, scm, mocker):
     cli_args = parse_args(["experiments", "clean"])
     assert cli_args.func == CmdExperimentsClean

tests/unit/command/test_gc.py

@@ -22,6 +22,7 @@ def test_(dvc, scm, mocker):
             "--force",
             "--jobs",
             "3",
+            "--dry",
             "--projects",
             "project1",
             "project2",
@@ -49,6 +50,7 @@ def test_(dvc, scm, mocker):
         rev=None,
         num=None,
         not_in_remote=False,
+        dry=True,
     )
 
     cli_args = parse_args(["gc"])