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

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

* raise RunCacheNotSupported for cloud-versioned remotes

.github/workflows/build.yaml

@@ -1,20 +1,22 @@
-name: Build package
+name: Build and upload package
 on:
   push:
     branches:
     - main
   release:
-    types: [released, prereleased]
+    types:
+    - published
   workflow_dispatch:
 
+env:
+  FORCE_COLOR: 1
+
 permissions:
   contents: read
 
 jobs:
-  pip:
-    permissions:
-      contents: read
-      id-token: write
+  build:
+    name: Build & Verify package
     runs-on: ubuntu-latest
     steps:
     - uses: actions/checkout@v4
@@ -26,24 +28,65 @@ jobs:
         python-version: 3.8
 
     - run: python -m pip install --upgrade pip setuptools_scm build twine
+
     - name: Force version for Test PyPI upload
       if: ${{ !startsWith(github.ref, 'refs/tags') }}
       run: echo version=$(python -m setuptools_scm | awk -F+ '{print $1}' | tail -1) >> $GITHUB_ENV
 
-    - run: |
-        echo 'PKG = "pip"'>dvc/_build.py
-        python -m build
+    - run: echo 'PKG = "pip"'>dvc/_build.py
+    - run: python -m build
       env:
-        SETUPTOOLS_SCM_PRETEND_VERSION: ${{ env.version }}
+        SETUPTOOLS_SCM_PRETEND_VERSION_FOR_DVC: ${{ env.version }}
     - run: twine check --strict dist/*
 
-    - name: Publish to PyPI
-      if: github.event_name == 'release'
-      uses: pypa/gh-action-pypi-publish@release/v1
+    - uses: actions/upload-artifact@v3
+      with:
+        name: Packages
+        path: dist/
 
-    - name: Publish to Test PyPI
-      if: ${{ github.event_name == 'release' || (github.event_name == 'push' && github.ref == 'refs/heads/main') }}
+  test-pypi-publish:
+    name: Publish dev package to test.pypi.org
+    runs-on: ubuntu-latest
+    needs: build
+    if: ${{ github.event.action == 'published' || (github.repository == 'iterative/dvc' && github.event_name == 'push' && github.ref == 'refs/heads/main') }}
+
+    environment:
+      name: test-pypi
+      url: https://test.pypi.org/p/dvc
+
+    permissions:
+      id-token: write
+
+    steps:
+    - uses: actions/download-artifact@v3
+      with:
+        name: Packages
+        path: dist
+
+    - name: Upload package to Test PyPI
       uses: pypa/gh-action-pypi-publish@release/v1
       with:
         repository-url: https://test.pypi.org/legacy/
         skip-existing: true
+
+  pypi-publish:
+    name: Publish released package to pypi.org
+    runs-on: ubuntu-latest
+    needs: build
+    if: github.event.action == 'published'
+
+    environment:
+      name: pypi
+      url: https://pypi.org/p/dvc
+
+    permissions:
+      id-token: write
+
+    steps:
+    - uses: actions/download-artifact@v3
+      with:
+        name: Packages
+        path: dist
+
+    - name: Upload package to PyPI
+      uses: pypa/gh-action-pypi-publish@release/v1

.pre-commit-config.yaml

@@ -32,7 +32,7 @@ repos:
     hooks:
       - id: isort
   - repo: https://github.com/astral-sh/ruff-pre-commit
-    rev: 'v0.0.288'
+    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/cachemgr.py

@@ -36,6 +36,7 @@ def _get_odb(
 class CacheManager:
     CACHE_DIR = "cache"
     FILES_DIR = "files"
+    FS_DIR = "fs"
 
     def __init__(self, repo):
         self._repo = repo
@@ -73,6 +74,21 @@ class CacheManager:
         legacy_odb = _get_odb(repo, settings, hash_name="md5-dos2unix", **kwargs)
         self._odb["legacy"] = legacy_odb
 
+    @property
+    def fs_cache(self):
+        """Filesystem-based cache.
+
+        Currently used as a temporary location to download files that we don't
+        yet have a regular oid (e.g. md5) for.
+        """
+        from dvc_data.index import FileStorage
+
+        return FileStorage(
+            key=(),
+            fs=self.local.fs,
+            path=self.local.fs.path.join(self.default_local_cache_dir, self.FS_DIR),
+        )
+
     def _init_odb(self, schemes):
         for scheme in schemes:
             remote = self.config.get(scheme)

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

@@ -12,6 +12,7 @@ from dvc.commands.experiments import (
     push,
     queue_worker,
     remove,
+    rename,
     run,
     save,
     show,
@@ -28,6 +29,7 @@ SUB_COMMANDS = [
     push,
     queue_worker,
     remove,
+    rename,
     run,
     save,
     show,

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

@@ -0,0 +1,68 @@
+import argparse
+import logging
+
+from dvc.cli.command import CmdBase
+from dvc.cli.utils import append_doc_link
+from dvc.exceptions import InvalidArgumentError
+from dvc.ui import ui
+
+logger = logging.getLogger(__name__)
+
+
+class CmdExperimentsRename(CmdBase):
+    def run(self):
+        from dvc.utils import humanize
+
+        if not (self.args.experiment and self.args.name):
+            raise InvalidArgumentError(
+                "An experiment to rename and a new experiment name are required."
+            )
+        renamed = self.repo.experiments.rename(
+            exp_name=self.args.experiment,
+            new_name=self.args.name,
+            git_remote=self.args.git_remote,
+            force=self.args.force,
+        )
+        if renamed:
+            ui.write(f"Renamed experiments: {humanize.join(map(repr, renamed))}")
+        else:
+            ui.write("No experiments to rename.")
+
+        return 0
+
+
+def add_parser(experiments_subparsers, parent_parser):
+    EXPERIMENTS_RENAME_HELP = "Rename experiments."
+    experiments_rename_parser = experiments_subparsers.add_parser(
+        "rename",
+        parents=[parent_parser],
+        description=append_doc_link(EXPERIMENTS_RENAME_HELP, "exp/rename"),
+        help=EXPERIMENTS_RENAME_HELP,
+        formatter_class=argparse.RawDescriptionHelpFormatter,
+    )
+    rename_group = experiments_rename_parser.add_mutually_exclusive_group()
+    rename_group.add_argument(
+        "-g",
+        "--git-remote",
+        metavar="<git_remote>",
+        help="Name or URL of the Git remote to rename the experiment from",
+    )
+    experiments_rename_parser.add_argument(
+        "experiment",
+        help="Experiment to rename.",
+        nargs="?",
+        metavar="<experiment>",
+    )
+    experiments_rename_parser.add_argument(
+        "name",
+        help="New name for the experiment.",
+        metavar="<name>",
+    )
+    experiments_rename_parser.add_argument(
+        "-f",
+        "--force",
+        action="store_true",
+        default=False,
+        help="Replace experiment if it already exists.",
+    )
+    experiments_rename_parser.set_defaults(func=CmdExperimentsRename)

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

@@ -44,7 +44,6 @@ from .utils.fs import path_isin
 if TYPE_CHECKING:
     from dvc_data.hashfile.obj import HashFile
     from dvc_data.index import DataIndexKey
-    from dvc_objects.db import ObjectDB
 
     from .ignore import DvcIgnoreFilter
 
@@ -1161,10 +1160,11 @@ class Output:
             return obj.filter(prefix)
         return obj
 
-    def get_used_objs(  # noqa: C901
+    def get_used_objs(  # noqa: C901, PLR0911
         self, **kwargs
-    ) -> Dict[Optional["ObjectDB"], Set["HashInfo"]]:
+    ) -> Dict[Optional["HashFileDB"], Set["HashInfo"]]:
         """Return filtered set of used object IDs for this out."""
+        from dvc.cachemgr import LEGACY_HASH_NAMES
 
         if not self.use_cache:
             return {}
@@ -1203,13 +1203,17 @@ class Output:
             return {}
 
         if self.remote:
-            remote = self.repo.cloud.get_remote_odb(
+            remote_odb = self.repo.cloud.get_remote_odb(
                 name=self.remote, hash_name=self.hash_name
             )
-        else:
-            remote = None
-
-        return {remote: self._named_obj_ids(obj)}
+            other_odb = self.repo.cloud.get_remote_odb(
+                name=self.remote,
+                hash_name=(
+                    "md5" if self.hash_name in LEGACY_HASH_NAMES else "md5-dos2unix"
+                ),
+            )
+            return {remote_odb: self._named_obj_ids(obj), other_odb: set()}
+        return {None: self._named_obj_ids(obj)}
 
     def _named_obj_ids(self, obj):
         name = str(self)

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

@@ -395,6 +395,11 @@ class Experiments:
 
         return remove(self.repo, *args, **kwargs)
 
+    def rename(self, *args, **kwargs):
+        from dvc.repo.experiments.rename import rename
+
+        return rename(self.repo, *args, **kwargs)
+
     def clean(self, *args, **kwargs):
         from dvc.repo.experiments.clean import clean
 

dvc/repo/experiments/exceptions.py

@@ -24,8 +24,6 @@ class ExperimentExistsError(DvcException):
             "Experiment conflicts with existing experiment "
             f"'{name}'. To overwrite the existing experiment run:\n\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

@@ -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/experiments/rename.py

@@ -0,0 +1,60 @@
+import logging
+from typing import TYPE_CHECKING, Dict, List, Optional, Union
+
+from dvc.repo.experiments.exceptions import (
+    ExperimentExistsError,
+    UnresolvedExpNamesError,
+)
+from dvc.repo.experiments.utils import resolve_name
+from dvc.scm import Git
+
+from .refs import ExpRefInfo
+
+if TYPE_CHECKING:
+    from dvc.repo import Repo
+
+logger = logging.getLogger(__name__)
+
+
+def rename(
+    repo: "Repo",
+    new_name: str,
+    exp_name: Union[str, None] = None,
+    git_remote: Optional[str] = None,
+    force: bool = False,
+) -> Union[List[str], None]:
+    renamed: List[str] = []
+    remained: List[str] = []
+    assert isinstance(repo.scm, Git)
+
+    if exp_name == new_name:
+        return None
+
+    if exp_name:
+        results: Dict[str, Union[ExpRefInfo, None]] = resolve_name(
+            scm=repo.scm, exp_names=exp_name, git_remote=git_remote
+        )
+        for name, result in results.items():
+            if result is None:
+                remained.append(name)
+                continue
+
+            new_ref = ExpRefInfo(baseline_sha=result.baseline_sha, name=new_name)
+            if repo.scm.get_ref(str(new_ref)) and not force:
+                raise ExperimentExistsError(new_name)
+
+            _rename_exp(scm=repo.scm, ref_info=result, new_name=new_name)
+            renamed.append(name)
+
+    if remained:
+        raise UnresolvedExpNamesError(remained, git_remote=git_remote)
+
+    return renamed
+
+
+def _rename_exp(scm: "Git", ref_info: "ExpRefInfo", new_name: str):
+    rev = scm.get_ref(str(ref_info))
+    scm.remove_ref(str(ref_info))
+    ref_info.name = new_name
+    scm.set_ref(str(ref_info), rev)
+    return new_name

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

@@ -1,5 +1,5 @@
 import logging
-from typing import TYPE_CHECKING, List, Optional, Set, Tuple
+from typing import TYPE_CHECKING, List, Optional
 
 from dvc.exceptions import InvalidArgumentError
 
@@ -7,8 +7,7 @@ from . import locked
 
 if TYPE_CHECKING:
     from dvc.repo import Repo
-    from dvc_data.hashfile.hash_info import HashInfo
-    from dvc_objects.db import ObjectDB
+    from dvc.repo.index import ObjectContainer
 
 logger = logging.getLogger(__name__)
 
@@ -33,22 +32,25 @@ def _validate_args(**kwargs):
         raise InvalidArgumentError("`--num` can only be used alongside `--rev`")
 
 
-def _used_obj_ids_not_in_remote(repo, default_remote, jobs, remote_odb_to_obj_ids):
+def _used_obj_ids_not_in_remote(
+    remote_odb_to_obj_ids: "ObjectContainer", jobs: Optional[int] = None
+):
     used_obj_ids = set()
     remote_oids = set()
-    for remote_odb, obj_ids in remote_odb_to_obj_ids:
-        if remote_odb is None:
-            remote_odb = repo.cloud.get_remote_odb(default_remote, "gc --not-in-remote")
-
+    for remote_odb, obj_ids in remote_odb_to_obj_ids.items():
+        assert remote_odb
         remote_oids.update(
-            remote_odb.list_oids_exists({x.value for x in obj_ids}, jobs=jobs)
+            remote_odb.list_oids_exists(
+                {x.value for x in obj_ids if x.value},
+                jobs=jobs,
+            )
         )
         used_obj_ids.update(obj_ids)
     return {obj for obj in used_obj_ids if obj.value not in remote_oids}
 
 
 @locked
-def gc(  # noqa: PLR0912, PLR0913, C901
+def gc(  # noqa: PLR0913, C901
     self: "Repo",
     all_branches: bool = False,
     cloud: bool = False,
@@ -65,6 +67,7 @@ def gc(  # noqa: PLR0912, 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`,
@@ -92,8 +95,7 @@ def gc(  # noqa: PLR0912, PLR0913, C901
         repos = []
     all_repos = [Repo(path) for path in repos]
 
-    odb_to_obj_ids: List[Tuple[Optional["ObjectDB"], Set["HashInfo"]]] = []
-
+    odb_to_obj_ids: "ObjectContainer" = {}
     with ExitStack() as stack:
         for repo in all_repos:
             stack.enter_context(repo.lock)
@@ -112,33 +114,50 @@ def gc(  # noqa: PLR0912, PLR0913, C901
                 revs=[rev] if rev else None,
                 num=num or 1,
             ).items():
-                odb_to_obj_ids.append((odb, obj_ids))
-
-    if not odb_to_obj_ids:
-        odb_to_obj_ids = [(None, set())]
+                if odb not in odb_to_obj_ids:
+                    odb_to_obj_ids[odb] = set()
+                odb_to_obj_ids[odb].update(obj_ids)
 
+    if cloud or not_in_remote:
+        _merge_remote_obj_ids(self, remote, odb_to_obj_ids)
     if not_in_remote:
-        used_obj_ids = _used_obj_ids_not_in_remote(self, remote, jobs, odb_to_obj_ids)
+        used_obj_ids = _used_obj_ids_not_in_remote(odb_to_obj_ids, jobs=jobs)
     else:
         used_obj_ids = set()
-        for _, obj_ids in odb_to_obj_ids:
-            used_obj_ids.update(obj_ids)
+        used_obj_ids.update(*odb_to_obj_ids.values())
 
     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:
         return
 
-    for remote_odb, obj_ids in odb_to_obj_ids:
-        if remote_odb is None:
-            remote_odb = repo.cloud.get_remote_odb(remote, "gc -c")
-        removed = ogc(remote_odb, obj_ids, jobs=jobs)
-        if removed:
+    for remote_odb, obj_ids in odb_to_obj_ids.items():
+        assert remote_odb is not None
+        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.")
+
+
+def _merge_remote_obj_ids(
+    repo: "Repo", remote: Optional[str], used_objs: "ObjectContainer"
+):
+    # Merge default remote used objects with remote-per-output used objects
+    default_obj_ids = used_objs.pop(None, set())
+    remote_odb = repo.cloud.get_remote_odb(remote, "gc -c", hash_name="md5")
+    if remote_odb not in used_objs:
+        used_objs[remote_odb] = set()
+    used_objs[remote_odb].update(default_obj_ids)
+    legacy_odb = repo.cloud.get_remote_odb(remote, "gc -c", hash_name="md5-dos2unix")
+    if legacy_odb not in used_objs:
+        used_objs[legacy_odb] = set()
+    used_objs[legacy_odb].update(default_obj_ids)

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,
@@ -33,14 +34,14 @@ if TYPE_CHECKING:
     from dvc.repo.stage import StageInfo
     from dvc.stage import Stage
     from dvc.types import TargetType
+    from dvc_data.hashfile.db import HashFileDB
     from dvc_data.hashfile.hash_info import HashInfo
     from dvc_data.index import DataIndex, DataIndexKey, DataIndexView
-    from dvc_objects.db import ObjectDB
     from dvc_objects.fs.base import FileSystem
 
 
 logger = logging.getLogger(__name__)
-ObjectContainer = Dict[Optional["ObjectDB"], Set["HashInfo"]]
+ObjectContainer = Dict[Optional["HashFileDB"], Set["HashInfo"]]
 
 
 def log_walk(seq):
@@ -186,12 +187,13 @@ def _load_storage_from_out(storage_map, key, out):
             from fsspec.utils import tokenize
 
             # partial import
+            fs_cache = out.repo.cache.fs_cache
             storage_map.add_cache(
                 FileStorage(
                     key,
-                    out.cache.fs,
-                    out.cache.fs.path.join(
-                        out.cache.path, "fs", dep.fs.protocol, tokenize(dep.fs_path)
+                    fs_cache.fs,
+                    fs_cache.fs.path.join(
+                        fs_cache.path, dep.fs.protocol, tokenize(dep.fs_path)
                     ),
                 )
             )
@@ -318,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:
@@ -358,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()
@@ -376,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)
@@ -399,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()
@@ -457,6 +469,9 @@ class Index:
             if not out.use_cache:
                 continue
 
+            if not out.is_in_repo:
+                continue
+
             ws, key = out.index_key
             if ws not in by_workspace:
                 by_workspace[ws] = index.view((*prefix, ws))
@@ -516,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:
@@ -636,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())
         )
@@ -656,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:
@@ -710,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",
@@ -773,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):
@@ -789,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
 
 
@@ -797,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/__init__.py

@@ -48,6 +48,7 @@ if TYPE_CHECKING:
     from dvc.output import Output
     from dvc.repo import Repo
     from dvc.types import StrPath
+    from dvc_data.hashfile.db import HashFileDB
     from dvc_data.hashfile.hash_info import HashInfo
     from dvc_objects.db import ObjectDB
 
@@ -732,7 +733,7 @@ class Stage(params.StageParams):
 
     def get_used_objs(
         self, *args, **kwargs
-    ) -> Dict[Optional["ObjectDB"], Set["HashInfo"]]:
+    ) -> Dict[Optional["HashFileDB"], Set["HashInfo"]]:
         """Return set of object IDs used by this stage."""
         if self.is_partial_import and not self.is_repo_import:
             return {}

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

dvc/utils/studio.py

@@ -1,5 +1,5 @@
 import logging
-from typing import TYPE_CHECKING, Any, Dict, List, Optional, cast
+from typing import TYPE_CHECKING, Any, Dict, List, Optional
 from urllib.parse import urljoin
 
 import requests
@@ -73,7 +73,7 @@ def notify_refs(
             logger.warning("failed to notify Studio: %s", msg.lower())
             return {}
 
-        r = cast("Response", e.response)
+        r = e.response
         d = ignore(Exception, default={})(r.json)()
         status = r.status_code
         if detail := d.get("detail"):

pyproject.toml

@@ -34,16 +34,17 @@ dependencies = [
     "configobj>=5.0.6",
     "distro>=1.3",
     "dpath<3,>=2.1.0",
-    "dvc-data>=2.16.0,<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_rename.py

@@ -0,0 +1,33 @@
+import pytest
+
+from dvc.repo.experiments.exceptions import (
+    ExperimentExistsError,
+    UnresolvedExpNamesError,
+)
+from dvc.repo.experiments.utils import exp_refs_by_names
+
+
+def test_rename_experiment_by_name(scm, dvc, exp_stage):
+    dvc.experiments.run(exp_stage.addressing, name="test-name", params=["foo=1"])
+    old_ref = exp_refs_by_names(scm, {"test-name"})
+    dvc.experiments.rename("new-name", "test-name")
+    new_ref = exp_refs_by_names(scm, {"new-name"})
+    assert scm.get_ref(str(old_ref["test-name"][0])) is None
+    assert scm.get_ref(str(new_ref["new-name"][0])) is not None
+    with pytest.raises(UnresolvedExpNamesError):
+        dvc.experiments.rename("new-name", "foo")
+
+
+def test_same_name(dvc, exp_stage):
+    dvc.experiments.run(exp_stage.addressing, name="same-name", params=["foo=1"])
+    assert dvc.experiments.rename("same-name", "same-name") is None
+
+
+def test_existing_name(dvc, exp_stage):
+    dvc.experiments.run(exp_stage.addressing, name="first-name", params=["foo=1"])
+    dvc.experiments.run(exp_stage.addressing, name="second-name", params=["foo=2"])
+
+    with pytest.raises(ExperimentExistsError):
+        dvc.experiments.rename("second-name", "first-name")
+
+    dvc.experiments.rename("second-name", "first-name", force=True)

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

@@ -9,6 +9,7 @@ from dvc.commands.experiments.ls import CmdExperimentsList
 from dvc.commands.experiments.pull import CmdExperimentsPull
 from dvc.commands.experiments.push import CmdExperimentsPush
 from dvc.commands.experiments.remove import CmdExperimentsRemove
+from dvc.commands.experiments.rename import CmdExperimentsRename
 from dvc.commands.experiments.run import CmdExperimentsRun
 from dvc.commands.experiments.save import CmdExperimentsSave
 from dvc.commands.experiments.show import CmdExperimentsShow
@@ -26,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):
@@ -418,6 +431,40 @@ def test_experiments_remove_invalid(dvc, scm, mocker, capsys, caplog):
     )
 
 
+def test_experiments_rename_flag(dvc, scm, mocker, capsys, caplog):
+    cli_args = parse_args(
+        [
+            "experiments",
+            "rename",
+            "--git-remote",
+            "myremote",
+            "exp-123",
+            "exp-234",
+        ]
+    )
+    assert cli_args.func == CmdExperimentsRename
+    cmd = cli_args.func(cli_args)
+    m = mocker.patch("dvc.repo.experiments.rename.rename", return_value={})
+    assert cmd.run() == 0
+    m.assert_called_once_with(
+        cmd.repo,
+        exp_name="exp-123",
+        new_name="exp-234",
+        git_remote="myremote",
+        force=False,
+    )
+
+
+def test_experiments_rename_invalid(dvc, scm, mocker, capsys, caplog):
+    cmd = CmdExperimentsRename(parse_args(["exp", "rename", "exp-1"]))
+    with pytest.raises(InvalidArgumentError) as excinfo:
+        cmd.run()
+    assert (
+        str(excinfo.value)
+        == "An experiment to rename and a new experiment name are required."
+    )
+
+
 def test_experiments_save(dvc, scm, mocker):
     cli_args = parse_args(["exp", "save", "--name", "exp-name", "--force"])
     assert cli_args.func == CmdExperimentsSave
@@ -432,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"])