show hydra dumped results in debug mode (#9460)

* show hydra dumped results in debug mode

* update hydra composition trace msg

.github/mergify.yml

@@ -0,0 +1,9 @@
+pull_request_rules:
+  - name: backport patches to 2.x
+    conditions:
+      - base=main
+      - label=backport
+    actions:
+      backport:
+        branches:
+          - "2.x"

.github/workflows/codeql.yml

@@ -2,7 +2,7 @@ name: "CodeQL"
 
 on:
   push:
-    branches: ["main"]
+    branches: ["main", "2.x"]
   pull_request:
     branches: ["main"]
   schedule:

.github/workflows/packages.yaml

@@ -73,6 +73,8 @@ jobs:
 
   pip:
     runs-on: ubuntu-20.04
+    permissions:
+      id-token: write
     steps:
     - uses: actions/checkout@v3
       with:
@@ -100,13 +102,10 @@ jobs:
     - name: Publish packages to PyPI
       if: github.event_name == 'release'
       uses: pypa/gh-action-pypi-publish@release/v1
-      with:
-        password: ${{ secrets.PYPI_TOKEN }}
 
     - name: Publish to Test PyPI
       if: ${{ github.event_name == 'release' || (github.event_name == 'push' && github.ref == 'refs/heads/main') }}
       uses: pypa/gh-action-pypi-publish@release/v1
       with:
-        password: ${{ secrets.TEST_PYPI_TOKEN }}
         repository_url: https://test.pypi.org/legacy/
         skip_existing: true

.github/workflows/tests.yaml

@@ -2,7 +2,7 @@ name: Tests
 
 on:
   push:
-    branches: [main]
+    branches: [main, 2.x]
   pull_request:
   schedule:
     - cron: '5 1 * * *'  # every day at 01:05
@@ -42,6 +42,8 @@ jobs:
     - uses: pre-commit/[email protected]
     - name: run tests
       timeout-minutes: 40
+      env:
+        PYTHONUTF8: 1
       run: >
         pytest -n=logical --timeout=300 --durations=100
         --cov --cov-report=xml --cov-report=term

dvc/annotations.py

@@ -28,9 +28,21 @@ class Annotation:
 
 
 @dataclass
-class Artifact(Annotation):
+class Artifact:
     PARAM_PATH: ClassVar[str] = "path"
-    path: Optional[str] = None
+    PARAM_DESC: ClassVar[str] = "desc"
+    PARAM_TYPE: ClassVar[str] = "type"
+    PARAM_LABELS: ClassVar[str] = "labels"
+    PARAM_META: ClassVar[str] = "meta"
+
+    path: str
+    desc: Optional[str] = None
+    type: Optional[str] = None  # noqa: A003
+    labels: List[str] = field(default_factory=list)
+    meta: Dict[str, Any] = field(default_factory=dict)
+
+    def to_dict(self) -> Dict[str, str]:
+        return compact(asdict(self))
 
 
 ANNOTATION_FIELDS = [field.name for field in fields(Annotation)]

dvc/api/__init__.py

@@ -2,11 +2,16 @@ from dvc.fs.dvc import _DVCFileSystem as DVCFileSystem
 
 from .data import open  # pylint: disable=redefined-builtin
 from .data import get_url, read
-from .experiments import exp_save, make_checkpoint
+from .experiments import exp_save, exp_show, make_checkpoint
+from .scm import all_branches, all_commits, all_tags
 from .show import metrics_show, params_show
 
 __all__ = [
+    "all_branches",
+    "all_commits",
+    "all_tags",
     "exp_save",
+    "exp_show",
     "get_url",
     "make_checkpoint",
     "open",

dvc/api/experiments.py

@@ -1,10 +1,13 @@
 import builtins
 import os
 from time import sleep
-from typing import List, Optional
+from typing import Dict, List, Optional, Union
+
+from rich.text import Text
 
 from dvc.env import DVC_CHECKPOINT, DVC_ROOT
 from dvc.repo import Repo
+from dvc.repo.experiments.show import tabulate
 from dvc.stage.monitor import CheckpointTask
 
 
@@ -69,3 +72,77 @@ def exp_save(
         return repo.experiments.save(
             name=name, force=force, include_untracked=include_untracked
         )
+
+
+def _postprocess(exp_rows):
+    for exp_row in exp_rows:
+        for k, v in exp_row.items():
+            if isinstance(v, Text):
+                v_str = str(v)
+                try:
+                    exp_row[k] = float(v_str)
+                except ValueError:
+                    exp_row[k] = v_str
+
+            if not exp_row[k]:
+                exp_row[k] = None
+
+    return exp_rows
+
+
+def exp_show(
+    repo: Optional[str] = None,
+    revs: Optional[Union[str, List[str]]] = None,
+    num: int = 1,
+    param_deps: bool = False,
+    force: bool = False,
+) -> List[Dict]:
+    """Get DVC experiments tracked in `repo`.
+
+    Without arguments, this function will retrieve all experiments derived from
+    the Git `HEAD`.
+
+    See the options below to customize the experiments retrieved.
+
+    Args:
+        repo (str, optional): location of the DVC repository.
+            Defaults to the current project (found by walking up from the
+            current working directory tree).
+            It can be a URL or a file system path.
+            Both HTTP and SSH protocols are supported for online Git repos
+            (e.g. [user@]server:project.git).
+        revs (Union[str, List[str]], optional): Git revision(s) (e.g. branch,
+            tag, SHA commit) to use as a reference point to start listing
+            experiments.
+            Defaults to `None`, which will use `HEAD` as starting point.
+        num (int, optional): show experiments from the last `num` commits
+            (first parents) starting from the `revs` baseline.
+            Give a negative value to include all first-parent commits (similar
+            to `git log -n`).
+            Defaults to 1.
+        param_deps (bool, optional): include only parameters that are stage
+            dependencies.
+            Defaults to `False`.
+        force (bool, optional): force re-collection of experiments instead of
+            loading from internal experiments cache.
+            DVC caches `exp_show` data for completed experiments to improve
+            performance of subsequent calls.
+            When `force` is specified, DVC will reload all experiment data and
+            ignore any previously cached results.
+            Defaults to `False`.
+
+    Returns:
+        List[Dict]: Each item in the list will contain a dictionary with
+            the info for an individual experiment.
+            See Examples below.
+    """
+    with Repo.open(repo) as _repo:
+        experiments = _repo.experiments.show(
+            revs=revs,
+            num=num,
+            param_deps=param_deps,
+            force=force,
+        )
+        td, _ = tabulate(experiments, fill_value=None)
+
+        return _postprocess(td.as_dict())

dvc/api/scm.py

@@ -0,0 +1,54 @@
+from typing import List, Optional
+
+from dvc.repo import Repo
+
+
+def all_branches(repo: Optional[str] = None) -> List[str]:
+    """Get all Git branches in a DVC repository.
+
+    Args:
+        repo (str, optional): location of the DVC repository.
+            Defaults to the current project (found by walking up from the
+            current working directory tree).
+            It can be a URL or a file system path.
+            Both HTTP and SSH protocols are supported for online Git repos
+            (e.g. [user@]server:project.git).
+    Returns:
+        List[str]: Names of the Git branches.
+    """
+    with Repo.open(repo) as _repo:
+        return _repo.scm.list_branches()
+
+
+def all_commits(repo: Optional[str] = None) -> List[str]:
+    """Get all Git commits in a DVC repository.
+
+    Args:
+        repo (str, optional): location of the DVC repository.
+            Defaults to the current project (found by walking up from the
+            current working directory tree).
+            It can be a URL or a file system path.
+            Both HTTP and SSH protocols are supported for online Git repos
+            (e.g. [user@]server:project.git).
+    Returns:
+        List[str]: SHAs of the Git commits.
+    """
+    with Repo.open(repo) as _repo:
+        return _repo.scm.list_all_commits()
+
+
+def all_tags(repo: Optional[str] = None) -> List[str]:
+    """Get all Git tags in a DVC repository.
+
+    Args:
+        repo (str, optional): location of the DVC repository.
+            Defaults to the current project (found by walking up from the
+            current working directory tree).
+            It can be a URL or a file system path.
+            Both HTTP and SSH protocols are supported for online Git repos
+            (e.g. [user@]server:project.git).
+    Returns:
+        List[str]: Names of the Git tags.
+    """
+    with Repo.open(repo) as _repo:
+        return _repo.scm.list_tags()

dvc/cli/__init__.py

@@ -245,7 +245,7 @@ def main(argv=None):  # noqa: C901, PLR0912, PLR0915
     finally:
         logger.setLevel(outer_log_level)
 
-        from dvc.external_repo import clean_repos
+        from dvc.repo.open_repo import clean_repos
 
         # Remove cached repos in the end of the call, these are anonymous
         # so won't be reused by any other subsequent run anyway.

dvc/commands/add.py

@@ -62,8 +62,11 @@ class CmdAdd(CmdBase):
                 remote=self.args.remote,
                 to_remote=self.args.to_remote,
                 jobs=self.args.jobs,
+                force=self.args.force,
             )
-
+        except FileNotFoundError:
+            logger.exception("")
+            return 1
         except DvcException:
             logger.exception("")
             return 1
@@ -140,6 +143,13 @@ def add_parser(subparsers, parent_parser):
         ),
         metavar="<number>",
     )
+    parser.add_argument(
+        "-f",
+        "--force",
+        action="store_true",
+        default=False,
+        help="Override local file or folder if exists.",
+    )
 
     _add_annotating_args(parser)
     parser.add_argument(

dvc/commands/data.py

@@ -117,6 +117,7 @@ class CmdDataStatus(CmdBase):
             status = self.repo.data_status(
                 granular=self.args.granular,
                 untracked_files=self.args.untracked_files,
+                not_in_remote=self.args.not_in_remote,
                 remote_refresh=self.args.remote_refresh,
             )
 
@@ -244,10 +245,16 @@ def add_parser(subparsers, parent_parser):
         help="Show untracked files.",
     )
     data_status_parser.add_argument(
-        "--remote-refresh",
+        "--not-in-remote",
         action="store_true",
         default=False,
-        help="Refresh remote index.",
+        help="Show files not in remote.",
+    )
+    data_status_parser.add_argument(
+        "--no-remote-refresh",
+        dest="remote_refresh",
+        action="store_false",
+        help="Use cached remote index (don't check remote).",
     )
     data_status_parser.set_defaults(func=CmdDataStatus)
 

dvc/commands/experiments/__init__.py

@@ -82,6 +82,7 @@ def add_rev_selection_flags(
     experiments_subcmd_parser.add_argument(
         "--rev",
         type=str,
+        action="append",
         default=None,
         help=msg,
         metavar="<commit>",

dvc/commands/experiments/exec_run.py

@@ -20,6 +20,7 @@ class CmdExecutorRun(CmdBaseNoRepo):
             log_level=logger.getEffectiveLevel(),
             infofile=self.args.infofile,
             copy_paths=self.args.copy_paths,
+            message=self.args.message,
         )
         return 0
 
@@ -47,4 +48,11 @@ def add_parser(experiments_subparsers, parent_parser):
             " Only used if `--temp` or `--queue` is specified."
         ),
     )
+    exec_run_parser.add_argument(
+        "-M",
+        "--message",
+        type=str,
+        default=None,
+        help="Custom commit message to use when committing the experiment.",
+    )
     exec_run_parser.set_defaults(func=CmdExecutorRun)

dvc/commands/experiments/push.py

@@ -52,7 +52,9 @@ class CmdExperimentsPush(CmdBase):
             ui.write(humanize.get_summary(stats.items()))
 
         if project_url := result.get("url"):
-            ui.write("[yellow]View your experiments at", project_url, styled=True)
+            ui.rich_print(
+                "View your experiments at", project_url, style="yellow", soft_wrap=True
+            )
 
     def run(self):
         from dvc.repo.experiments.push import UploadError

dvc/commands/experiments/run.py

@@ -37,6 +37,7 @@ class CmdExperimentsRun(CmdRepro):
             tmp_dir=self.args.tmp_dir,
             machine=self.args.machine,
             copy_paths=self.args.copy_paths,
+            message=self.args.message,
             **self._common_kwargs,
         )
 
@@ -147,3 +148,10 @@ def _add_run_common(parser):
             " Only used if `--temp` or `--queue` is specified."
         ),
     )
+    parser.add_argument(
+        "-M",
+        "--message",
+        type=str,
+        default=None,
+        help="Custom commit message to use when committing the experiment.",
+    )

dvc/commands/experiments/save.py

@@ -16,6 +16,7 @@ class CmdExperimentsSave(CmdBase):
                 name=self.args.name,
                 force=self.args.force,
                 include_untracked=self.args.include_untracked,
+                message=self.args.message,
             )
         except DvcException:
             logger.exception("failed to save experiment")
@@ -75,4 +76,11 @@ def add_parser(experiments_subparsers, parent_parser):
         help="List of untracked paths to include in the experiment.",
         metavar="<path>",
     )
+    save_parser.add_argument(
+        "-M",
+        "--message",
+        type=str,
+        default=None,
+        help="Custom commit message to use when committing the experiment.",
+    )
     save_parser.set_defaults(func=CmdExperimentsSave)

dvc/commands/gc.py

@@ -10,7 +10,7 @@ logger = logging.getLogger(__name__)
 
 
 class CmdGC(CmdBase):
-    def run(self):  # noqa: C901
+    def run(self):  # noqa: C901, PLR0912
         from dvc.repo.gc import _validate_args
 
         _validate_args(
@@ -22,6 +22,8 @@ class CmdGC(CmdBase):
             workspace=self.args.workspace,
             rev=self.args.rev,
             num=self.args.num,
+            cloud=self.args.cloud,
+            not_in_remote=self.args.not_in_remote,
         )
 
         if self.args.rev:
@@ -46,6 +48,10 @@ class CmdGC(CmdBase):
 
         if self.args.all_experiments:
             msg += " and all experiments"
+
+        if self.args.not_in_remote:
+            msg += " that are present in the DVC remote"
+
         if self.args.repos:
             msg += " of the current and the following repos:"
 
@@ -74,6 +80,7 @@ class CmdGC(CmdBase):
             workspace=self.args.workspace,
             rev=self.args.rev,
             num=self.args.num,
+            not_in_remote=self.args.not_in_remote,
         )
         return 0
 
@@ -157,12 +164,18 @@ def add_parser(subparsers, parent_parser):
         default=False,
         help="Keep data files for all experiments.",
     )
+    gc_parser.add_argument(
+        "--not-in-remote",
+        action="store_true",
+        default=False,
+        help="Keep data files that are not present in the remote.",
+    )
     gc_parser.add_argument(
         "-c",
         "--cloud",
         action="store_true",
         default=False,
-        help="Collect garbage in remote repository.",
+        help="Collect garbage in remote storage in addition to local cache.",
     )
     gc_parser.add_argument(
         "-r",

dvc/commands/get.py

@@ -36,6 +36,7 @@ class CmdGet(CmdBaseNoRepo):
                 out=self.args.out,
                 rev=self.args.rev,
                 jobs=self.args.jobs,
+                force=self.args.force,
             )
             return 0
         except CloneError:
@@ -94,4 +95,11 @@ def add_parser(subparsers, parent_parser):
         ),
         metavar="<number>",
     )
+    get_parser.add_argument(
+        "-f",
+        "--force",
+        action="store_true",
+        default=False,
+        help="Override local file or folder if exists.",
+    )
     get_parser.set_defaults(func=CmdGet)

dvc/commands/get_url.py

@@ -14,7 +14,12 @@ class CmdGetUrl(CmdBaseNoRepo):
         from dvc.repo import Repo
 
         try:
-            Repo.get_url(self.args.url, out=self.args.out, jobs=self.args.jobs)
+            Repo.get_url(
+                self.args.url,
+                out=self.args.out,
+                jobs=self.args.jobs,
+                force=self.args.force,
+            )
             return 0
         except DvcException:
             logger.exception("failed to get '%s'", self.args.url)
@@ -46,4 +51,11 @@ def add_parser(subparsers, parent_parser):
         ),
         metavar="<number>",
     )
+    get_parser.add_argument(
+        "-f",
+        "--force",
+        action="store_true",
+        default=False,
+        help="Override local file or folder if exists.",
+    )
     get_parser.set_defaults(func=CmdGetUrl)

dvc/commands/imp_url.py

@@ -25,6 +25,7 @@ class CmdImportUrl(CmdBase):
                 labels=self.args.labels,
                 meta=self.args.meta,
                 jobs=self.args.jobs,
+                force=self.args.force,
                 version_aware=self.args.version_aware,
             )
         except DvcException:
@@ -112,6 +113,13 @@ def add_parser(subparsers, parent_parser):
         ),
         metavar="<number>",
     )
+    import_parser.add_argument(
+        "-f",
+        "--force",
+        action="store_true",
+        default=False,
+        help="Override local file or folder if exists.",
+    )
     import_parser.add_argument(
         "--version-aware",
         action="store_true",

dvc/commands/repro.py

@@ -38,6 +38,7 @@ class CmdRepro(CmdBase):
             "recursive": self.args.recursive,
             "force_downstream": self.args.force_downstream,
             "pull": self.args.pull,
+            "allow_missing": self.args.allow_missing,
         }
 
     @property
@@ -135,6 +136,12 @@ and then the stage name name.
             "from the run-cache."
         ),
     )
+    repro_parser.add_argument(
+        "--allow-missing",
+        action="store_true",
+        default=False,
+        help=("Skip stages with missing data but no other changes."),
+    )
     repro_parser.add_argument(
         "--dry",
         action="store_true",

dvc/compare.py

@@ -35,7 +35,7 @@ def with_value(value, default):
 
 
 class TabularData(MutableSequence[Sequence["CellT"]]):
-    def __init__(self, columns: Sequence[str], fill_value: str = ""):
+    def __init__(self, columns: Sequence[str], fill_value: Optional[str] = ""):
         self._columns: Dict[str, Column] = {name: Column() for name in columns}
         self._keys: List[str] = list(columns)
         self._fill_value = fill_value

dvc/config.py

@@ -79,9 +79,6 @@ class Config(dict):
         ConfigError: thrown if config has an invalid format.
     """
 
-    APPNAME = "dvc"
-    APPAUTHOR = "iterative"
-
     SYSTEM_LEVELS = ("system", "global")
     REPO_LEVELS = ("repo", "local")
     # In the order they shadow each other
@@ -121,14 +118,14 @@ class Config(dict):
 
     @classmethod
     def get_dir(cls, level):
-        from platformdirs import site_config_dir, user_config_dir
+        from dvc.dirs import global_config_dir, system_config_dir
 
         assert level in ("global", "system")
 
         if level == "global":
-            return user_config_dir(cls.APPNAME, cls.APPAUTHOR)
+            return global_config_dir()
         if level == "system":
-            return site_config_dir(cls.APPNAME, cls.APPAUTHOR)
+            return system_config_dir()
 
     @cached_property
     def files(self) -> Dict[str, str]:

dvc/config_schema.py

@@ -330,5 +330,7 @@ SCHEMA = {
     "studio": {
         "token": str,
         "url": str,
+        "repo_url": str,
+        Optional("offline", default=False): Bool,
     },
 }

dvc/dependency/repo.py

@@ -79,10 +79,14 @@ class RepoDependency(Dependency):
     ) -> "DVCFileSystem":
         from dvc.fs import DVCFileSystem
 
+        config = {"cache": self.repo.config["cache"]}
+        config["cache"]["dir"] = self.repo.cache.local.path
+
         return DVCFileSystem(
             url=self.def_repo[self.PARAM_URL],
             rev=rev or self._get_rev(locked=locked),
             subrepos=True,
+            config=config,
         )
 
     def _get_rev(self, locked: bool = True):

dvc/dirs.py

@@ -0,0 +1,26 @@
+import os
+
+import platformdirs
+
+from . import env
+
+APPNAME = "dvc"
+APPAUTHOR = "iterative"
+
+
+def system_config_dir():
+    return os.getenv(env.DVC_SYSTEM_CONFIG_DIR) or platformdirs.site_config_dir(
+        APPNAME, APPAUTHOR
+    )
+
+
+def global_config_dir():
+    return os.getenv(env.DVC_SYSTEM_CONFIG_DIR) or platformdirs.user_config_dir(
+        APPNAME, APPAUTHOR
+    )
+
+
+def site_cache_dir():
+    return os.getenv(env.DVC_SITE_CACHE_DIR) or platformdirs.site_cache_dir(
+        APPNAME, APPAUTHOR, opinion=True
+    )

dvc/env.py

@@ -1,14 +1,19 @@
+DVCLIVE_RESUME = "DVCLIVE_RESUME"
 DVC_CHECKPOINT = "DVC_CHECKPOINT"
 DVC_DAEMON = "DVC_DAEMON"
+DVC_EXP_AUTO_PUSH = "DVC_EXP_AUTO_PUSH"
+DVC_EXP_BASELINE_REV = "DVC_EXP_BASELINE_REV"
+DVC_EXP_GIT_REMOTE = "DVC_EXP_GIT_REMOTE"
+DVC_EXP_NAME = "DVC_EXP_NAME"
+DVC_GLOBAL_CONFIG_DIR = "DVC_GLOBAL_CONFIG_DIR"
+DVC_IGNORE_ISATTY = "DVC_IGNORE_ISATTY"
+DVC_NO_ANALYTICS = "DVC_NO_ANALYTICS"
 DVC_PAGER = "DVC_PAGER"
 DVC_ROOT = "DVC_ROOT"
-DVCLIVE_RESUME = "DVCLIVE_RESUME"
-DVC_IGNORE_ISATTY = "DVC_IGNORE_ISATTY"
-DVC_EXP_BASELINE_REV = "DVC_EXP_BASELINE_REV"
-DVC_EXP_NAME = "DVC_EXP_NAME"
-DVC_EXP_GIT_REMOTE = "DVC_EXP_GIT_REMOTE"
-DVC_EXP_AUTO_PUSH = "DVC_EXP_AUTO_PUSH"
-DVC_NO_ANALYTICS = "DVC_NO_ANALYTICS"
 DVC_SHOW_TRACEBACK = "DVC_SHOW_TRACEBACK"
+DVC_SITE_CACHE_DIR = "DVC_SITE_CACHE_DIR"
+DVC_STUDIO_OFFLINE = "DVC_STUDIO_OFFLINE"
+DVC_STUDIO_REPO_URL = "DVC_STUDIO_REPO_URL"
 DVC_STUDIO_TOKEN = "DVC_STUDIO_TOKEN"  # noqa: S105 # nosec B105
 DVC_STUDIO_URL = "DVC_STUDIO_URL"
+DVC_SYSTEM_CONFIG_DIR = "DVC_SYSTEM_CONFIG_DIR"

dvc/exceptions.py

@@ -1,4 +1,5 @@
 """Exceptions raised by the dvc."""
+import errno
 from typing import Dict, List
 
 from dvc.utils import format_link
@@ -206,6 +207,18 @@ class ETagMismatchError(DvcException):
         )
 
 
+class FileExistsLocallyError(FileExistsError, DvcException):
+    def __init__(self, path, hint=None):
+        import os.path
+
+        self.path = path
+        hint = "" if hint is None else f". {hint}"
+        path_typ = "directory" if os.path.isdir(path) else "file"
+        msg = f"The {path_typ} '{path}' already exists locally{hint}"
+        super().__init__(msg)
+        self.errno = errno.EEXIST
+
+
 class FileMissingError(DvcException):
     def __init__(self, path, hint=None):
         self.path = path

dvc/info.py

@@ -32,6 +32,7 @@ def get_dvc_info():
         f"Platform: Python {platform.python_version()} on {platform.platform()}",
         f"Subprojects:{_get_subprojects()}",
         f"Supports:{_get_supported_remotes()}",
+        f"Config:{_get_config_dirs()}",
     ]
 
     try:
@@ -42,7 +43,7 @@ def get_dvc_info():
             # `dvc config cache.shared group`.
             if os.path.exists(repo.cache.local.path):
                 info.append(f"Cache types: {_get_linktype_support_info(repo)}")
-                fs_type = get_fs_type(repo.cache.local.path)
+                fs_type = _get_fs_type(repo.cache.local.path)
                 info.append(f"Cache directory: {fs_type}")
             else:
                 info.append("Cache types: " + error_link("no-dvc-cache"))
@@ -51,7 +52,7 @@ def get_dvc_info():
             info.append(f"Remotes: {_get_remotes(repo.config)}")
 
             root_directory = repo.root_dir
-            fs_root = get_fs_type(os.path.abspath(root_directory))
+            fs_root = _get_fs_type(os.path.abspath(root_directory))
             info.append(f"Workspace directory: {fs_root}")
             info.append(f"Repo: {_get_dvc_repo_info(repo)}")
             info.append(f"Repo.site_cache_dir: {repo.site_cache_dir}")
@@ -136,7 +137,18 @@ def _get_supported_remotes():
     return "\n\t" + ",\n\t".join(supported_remotes)
 
 
-def get_fs_type(path):
+def _get_config_dirs():
+    from dvc.config import Config
+
+    dirs = [
+        f"Global: {Config.get_dir('global')}",
+        f"System: {Config.get_dir('system')}",
+    ]
+
+    return "\n\t" + "\n\t".join(dirs)
+
+
+def _get_fs_type(path):
     partition = {}
     for part in psutil.disk_partitions(all=True):
         if part.fstype:
@@ -155,11 +167,11 @@ def get_fs_type(path):
     return ("unknown", "none")
 
 
-def _get_dvc_repo_info(self):
-    if self.config.get("core", {}).get("no_scm", False):
+def _get_dvc_repo_info(repo):
+    if repo.config.get("core", {}).get("no_scm", False):
         return "dvc (no_scm)"
 
-    if self.root_dir != self.scm.root_dir:
+    if repo.root_dir != repo.scm.root_dir:
         return "dvc (subdir), git"
 
     return "dvc, git"

dvc/output.py

@@ -1,10 +1,11 @@
+import errno
 import logging
 import os
 import posixpath
 from collections import defaultdict
 from contextlib import suppress
 from operator import itemgetter
-from typing import TYPE_CHECKING, Dict, List, Optional, Set, Tuple, Type
+from typing import TYPE_CHECKING, Dict, List, Optional, Set, Tuple, Type, Union
 from urllib.parse import urlparse
 
 from funcy import collecting, first, project
@@ -25,26 +26,25 @@ from dvc_data.hashfile import check as ocheck
 from dvc_data.hashfile import load as oload
 from dvc_data.hashfile.build import build
 from dvc_data.hashfile.checkout import checkout
+from dvc_data.hashfile.db import HashFileDB, add_update_tree
 from dvc_data.hashfile.hash_info import HashInfo
 from dvc_data.hashfile.istextfile import istextfile
 from dvc_data.hashfile.meta import Meta
 from dvc_data.hashfile.transfer import transfer as otransfer
-from dvc_data.hashfile.tree import Tree
+from dvc_data.hashfile.tree import Tree, du
 from dvc_objects.errors import ObjectFormatError
 
 from .annotations import ANNOTATION_FIELDS, ANNOTATION_SCHEMA, Annotation
 from .fs import LocalFileSystem, RemoteMissingDepsError, Schemes, get_cloud_fs
-from .fs.callbacks import DEFAULT_CALLBACK
+from .fs.callbacks import DEFAULT_CALLBACK, Callback, TqdmCallback
 from .utils import relpath
 from .utils.fs import path_isin
 
 if TYPE_CHECKING:
-    from dvc_data.hashfile.db import HashFileDB
     from dvc_data.hashfile.obj import HashFile
     from dvc_data.index import DataIndexKey
     from dvc_objects.db import ObjectDB
 
-    from .fs.callbacks import Callback
     from .ignore import DvcIgnoreFilter
 
 logger = logging.getLogger(__name__)
@@ -276,6 +276,11 @@ class OutputIsIgnoredError(DvcException):
         super().__init__(f"Path '{match.file}' is ignored by\n{lines}")
 
 
+class CheckoutCallback(TqdmCallback):
+    # disable branching for checkouts
+    branch = Callback.branch  # type: ignore[assignment]
+
+
 class Output:
     IS_DEPENDENCY = False
 
@@ -417,8 +422,6 @@ class Output:
 
     def _compute_meta_hash_info_from_files(self) -> None:
         if self.files:
-            from dvc_data.hashfile.db import HashFileDB
-
             tree = Tree.from_list(self.files, hash_name=self.hash_name)
             tree.digest(with_meta=True)
             self.odb = HashFileDB(tree.fs, tree.path + ".odb")
@@ -741,15 +744,18 @@ class Output:
                     hardlink=hardlink,
                 )
             if relink:
-                self._checkout(
-                    filter_info or self.fs_path,
-                    self.fs,
-                    obj,
-                    self.cache,
-                    relink=True,
-                    state=self.repo.state,
-                    prompt=prompt.confirm,
-                )
+                rel = self.fs.path.relpath(filter_info or self.fs_path)
+                with CheckoutCallback(desc=f"Checking out {rel}", unit="files") as cb:
+                    self._checkout(
+                        filter_info or self.fs_path,
+                        self.fs,
+                        obj,
+                        self.cache,
+                        relink=True,
+                        state=self.repo.state,
+                        prompt=prompt.confirm,
+                        progress_callback=cb,
+                    )
                 self.set_exec()
 
     def _commit_granular_dir(self, filter_info, hardlink) -> Optional["HashFile"]:
@@ -891,9 +897,16 @@ class Output:
         checkpoint_reset: bool = False,
         **kwargs,
     ) -> Optional[Tuple[bool, Optional[bool]]]:
+        # callback passed act as a aggregate callback.
+        # do not let checkout to call set_size and change progressbar.
+        class CallbackProxy(Callback):
+            def relative_update(self, inc: int = 1) -> None:
+                progress_callback.relative_update(inc)
+                return super().relative_update(inc)
+
+        callback = CallbackProxy()
         if not self.use_cache:
-            if progress_callback != DEFAULT_CALLBACK:
-                progress_callback.relative_update(self.get_files_number(filter_info))
+            callback.relative_update(self.get_files_number(filter_info))
             return None
 
         obj = self.get_obj(filter_info=filter_info)
@@ -915,7 +928,7 @@ class Output:
                 obj,
                 self.cache,
                 force=force,
-                progress_callback=progress_callback,
+                progress_callback=callback,
                 relink=relink,
                 state=self.repo.state,
                 prompt=prompt.confirm,
@@ -1008,7 +1021,7 @@ class Output:
         if self.exists:
             self.cache.unprotect(self.fs_path)
 
-    def get_dir_cache(self, **kwargs):
+    def get_dir_cache(self, **kwargs) -> Optional["Tree"]:
         if not self.is_dir_checksum:
             raise DvcException("cannot get dir cache for file checksum")
 
@@ -1022,14 +1035,17 @@ class Output:
                 self.repo.cloud.pull([obj.hash_info], **kwargs)
 
         if self.obj:
+            assert isinstance(self.obj, Tree)
             return self.obj
 
         try:
-            self.obj = oload(self.cache, self.hash_info)
+            obj = oload(self.cache, self.hash_info)
+            assert isinstance(obj, Tree)
         except (FileNotFoundError, ObjectFormatError):
-            self.obj = None
+            obj = None
 
-        return self.obj
+        self.obj = obj
+        return obj
 
     def _collect_used_dir_cache(
         self, remote=None, force=False, jobs=None, filter_info=None
@@ -1165,7 +1181,7 @@ class Output:
 
     def merge(self, ancestor, other, allowed=None):
         from dvc_data.hashfile.tree import MergeError as TreeMergeError
-        from dvc_data.hashfile.tree import du, merge
+        from dvc_data.hashfile.tree import merge
 
         assert other
 
@@ -1198,6 +1214,167 @@ class Output:
             nfiles=len(merged),
         )
 
+    def unstage(self, path: str) -> Tuple["Meta", "Tree"]:
+        from pygtrie import Trie
+
+        from dvc_objects.fs.path import Path
+
+        assert isinstance(self.fs.path, Path)
+        rel_key = tuple(self.fs.path.parts(self.fs.path.relpath(path, self.fs_path)))
+
+        if self.hash_info:
+            tree = self.get_dir_cache()
+            if tree is None:
+                raise DvcException(f"could not read {self.hash_info.value!r}")
+        else:
+            tree = Tree()
+
+        trie = tree.as_trie()
+        assert isinstance(trie, Trie)
+
+        try:
+            del trie[rel_key:]  # type: ignore[misc]
+        except KeyError:
+            raise FileNotFoundError(  # noqa: B904
+                errno.ENOENT,
+                os.strerror(errno.ENOENT),
+                self.fs.path.relpath(path),
+            )
+
+        new = tree.from_trie(trie)
+        new.digest()
+        return Meta(nfiles=len(new), isdir=True), new
+
+    def apply(
+        self,
+        path: str,
+        obj: Union["Tree", "HashFile"],
+        meta: "Meta",
+    ) -> Tuple["Meta", "Tree"]:
+        from pygtrie import Trie
+
+        from dvc_objects.fs.path import Path
+
+        assert isinstance(self.fs.path, Path)
+        append_only = True
+        rel_key = tuple(self.fs.path.parts(self.fs.path.relpath(path, self.fs_path)))
+
+        if self.hash_info:
+            tree = self.get_dir_cache()
+            if tree is None:
+                raise DvcException(f"could not read {self.hash_info.value!r}")
+        else:
+            tree = Tree()
+
+        trie = tree.as_trie()
+        assert isinstance(trie, Trie)
+
+        try:
+            del trie[rel_key:]  # type: ignore[misc]
+        except KeyError:
+            pass
+        else:
+            append_only = False
+
+        items = {}
+        if isinstance(obj, Tree):
+            items = {(*rel_key, *key): (m, o) for key, m, o in obj}
+        else:
+            items = {rel_key: (meta, obj.hash_info)}
+        trie.update(items)
+
+        new = Tree.from_trie(trie)
+        new.digest()
+
+        size = self.meta.size if self.meta and self.meta.size else None
+        if append_only and size and meta.size is not None:
+            # if files were only appended, we can sum to the existing size
+            size += meta.size
+        elif self.hash_info and self.hash_info == new.hash_info:
+            # if hashes are same, sizes must have been the same
+            size = self.meta.size
+        else:
+            size = None
+
+        meta = Meta(nfiles=len(new), size=size, isdir=True)
+        return meta, new
+
+    def add(  # noqa: C901
+        self, path: Optional[str] = None, no_commit: bool = False, relink: bool = True
+    ) -> Optional["HashFile"]:
+        path = path or self.fs_path
+        if self.hash_info and not self.is_dir_checksum and self.fs_path != path:
+            raise DvcException(
+                f"Cannot modify '{self}' which is being tracked as a file"
+            )
+
+        assert self.repo
+        cache = self.cache if self.use_cache else self.repo.cache.local
+        assert isinstance(cache, HashFileDB)
+
+        new: "HashFile"
+        try:
+            assert self.hash_name
+            staging, meta, obj = build(
+                cache,
+                path,
+                self.fs,
+                self.hash_name,
+                ignore=self.dvcignore,
+                dry_run=not self.use_cache,
+            )
+        except FileNotFoundError as exc:
+            if self.fs_path == path:
+                raise self.DoesNotExistError(self) from exc
+            if not self.is_dir_checksum:
+                raise
+
+            meta, new = self.unstage(path)
+            staging, obj = None, None
+        else:
+            assert obj
+            assert staging
+            if self.fs_path != path:
+                meta, new = self.apply(path, obj, meta)
+                add_update_tree(staging, new)
+            else:
+                new = obj
+
+        self.obj = new
+        self.hash_info = self.obj.hash_info
+        self.meta = meta
+        self.files = None
+        self.ignore()
+
+        if no_commit or not self.use_cache:
+            return obj
+
+        if isinstance(new, Tree):
+            add_update_tree(cache, new)
+
+        if not obj:
+            return obj
+
+        assert staging
+        assert obj.hash_info
+        otransfer(staging, self.cache, {obj.hash_info}, hardlink=relink, shallow=False)
+
+        if relink:
+            rel = self.fs.path.relpath(path)
+            with CheckoutCallback(desc=f"Checking out {rel}", unit="files") as callback:
+                self._checkout(
+                    path,
+                    self.fs,
+                    obj,
+                    self.cache,
+                    relink=True,
+                    state=self.repo.state,
+                    prompt=prompt.confirm,
+                    progress_callback=callback,
+                )
+            self.set_exec()
+        return obj
+
     @property
     def fspath(self):
         return self.fs_path

dvc/repo/__init__.py

@@ -287,18 +287,9 @@ class Repo:
 
     @staticmethod
     def open(url, *args, **kwargs):  # noqa: A003
-        if url is None:
-            url = os.getcwd()
+        from .open_repo import open_repo
 
-        if os.path.exists(url):
-            try:
-                return Repo(url, *args, **kwargs)
-            except NotDvcRepoError:
-                pass  # fallthrough to external_repo
-
-        from dvc.external_repo import external_repo
-
-        return external_repo(url, *args, **kwargs)
+        return open_repo(url, *args, **kwargs)
 
     @cached_property
     def scm(self) -> Union["Git", "NoSCM"]:
@@ -563,12 +554,10 @@ class Repo:
         import getpass
         import hashlib
 
-        import platformdirs
-
+        from dvc.dirs import site_cache_dir
         from dvc.fs import GitFileSystem
 
-        default = platformdirs.site_cache_dir("dvc", "iterative", opinion=True)
-        cache_dir = self.config["core"].get("site_cache_dir") or default
+        cache_dir = self.config["core"].get("site_cache_dir") or site_cache_dir()
 
         if isinstance(self.fs, GitFileSystem):
             relparts = ()

dvc/repo/add.py

@@ -2,14 +2,16 @@ import logging
 import os
 from contextlib import contextmanager
 from itertools import tee
-from typing import TYPE_CHECKING, Any, Iterator, List, Optional
+from typing import TYPE_CHECKING, Any, Iterator, List, NamedTuple, Optional
 
 import colorama
 
 from dvc.exceptions import (
     CacheLinkError,
+    DvcException,
     InvalidArgumentError,
     OutputDuplicationError,
+    OutputNotFoundError,
     OverlappingOutputPathsError,
     RecursiveAddingWhileUsingFilename,
 )
@@ -29,13 +31,25 @@ Stages = List["Stage"]
 logger = logging.getLogger(__name__)
 
 
-OVERLAPPING_OUTPUT_FMT = (
+class StageInfo(NamedTuple):
+    stage: "Stage"
+    output_exists: bool
+
+
+OVERLAPPING_CHILD_FMT = (
     "Cannot add '{out}', because it is overlapping with other "
     "DVC tracked output: '{parent}'.\n"
     "To include '{out}' in '{parent}', run "
     "'dvc commit {parent_stage}'"
 )
 
+OVERLAPPING_PARENT_FMT = (
+    "Cannot add '{parent}', because it is overlapping with other "
+    "DVC tracked output: '{out}'.\n"
+    "To include '{out}' in '{parent}', run "
+    "'dvc remove {out_stage}' and then 'dvc add {parent}'"
+)
+
 
 def check_recursive_and_fname(args):
     if args.recursive and args.fname:
@@ -86,11 +100,18 @@ def translate_graph_error(stages: Stages) -> Iterator[None]:
     try:
         yield
     except OverlappingOutputPathsError as exc:
-        msg = OVERLAPPING_OUTPUT_FMT.format(
-            out=exc.overlapping_out.fs_path,
-            parent=exc.parent.fs_path,
-            parent_stage=exc.parent.stage.addressing,
-        )
+        if exc.parent in [o for s in stages for o in s.outs]:
+            msg = OVERLAPPING_PARENT_FMT.format(
+                out=exc.overlapping_out,
+                parent=exc.parent,
+                out_stage=exc.overlapping_out.stage.addressing,
+            )
+        else:
+            msg = OVERLAPPING_CHILD_FMT.format(
+                out=exc.overlapping_out,
+                parent=exc.parent,
+                parent_stage=exc.parent.stage.addressing,
+            )
         raise OverlappingOutputPathsError(  # noqa: B904
             exc.parent, exc.overlapping_out, msg
         )
@@ -100,7 +121,7 @@ def translate_graph_error(stages: Stages) -> Iterator[None]:
         )
 
 
-def progress_iter(stages: Stages) -> Iterator["Stage"]:
+def progress_iter(stages: List[StageInfo]) -> Iterator["StageInfo"]:
     total = len(stages)
     desc = "Adding..."
     with ui.progress(stages, total=total, desc=desc, unit="file", leave=True) as pbar:
@@ -108,10 +129,10 @@ def progress_iter(stages: Stages) -> Iterator["Stage"]:
             pbar.bar_format = desc
             pbar.refresh()
 
-        for stage in pbar:
+        for item in pbar:
             if total > 1:
-                pbar.set_msg(f"{stage.outs[0]}")
-            yield stage
+                pbar.set_msg(f"{item.stage.outs[0]}")
+            yield item
             if total == 1:  # restore bar format for stats
                 # pylint: disable=no-member
                 pbar.bar_format = pbar.BAR_FMT_DEFAULT
@@ -169,12 +190,14 @@ def add(
     desc = "Collecting targets"
     stages_it = create_stages(repo, add_targets, fname, transfer, **kwargs)
     stages = list(ui.progress(stages_it, desc=desc, unit="file"))
+
+    stages_list = [stage for stage, _ in stages]
     msg = "Collecting stages from the workspace"
-    with translate_graph_error(stages), ui.status(msg) as status:
+    with translate_graph_error(stages_list), ui.status(msg) as status:
         # remove existing stages that are to-be replaced with these
         # new stages for the graph checks.
         repo.check_graph(
-            stages=stages, callback=lambda: status.update("Checking graph")
+            stages=stages_list, callback=lambda: status.update("Checking graph")
         )
 
     odb = None
@@ -182,18 +205,18 @@ def add(
         odb = repo.cloud.get_remote_odb(kwargs.get("remote"), "add")
 
     with warn_link_failures() as link_failures:
-        for stage, source in zip(progress_iter(stages), sources):
+        for (stage, output_exists), source in zip(progress_iter(stages), sources):
+            out = stage.outs[0]
             if to_remote or to_cache:
                 stage.transfer(source, to_remote=to_remote, odb=odb, **kwargs)
             else:
                 try:
-                    stage.save()
-                    if not no_commit:
-                        stage.commit()
+                    path = out.fs.path.abspath(source) if output_exists else None
+                    stage.add_outs(path, no_commit=no_commit)
                 except CacheLinkError:
                     link_failures.append(str(stage.relpath))
             stage.dump()
-    return stages
+    return stages_list
 
 
 LARGE_DIR_RECURSIVE_ADD_WARNING = (
@@ -247,24 +270,34 @@ def create_stages(
     fname: Optional[str] = None,
     transfer: bool = False,
     external: bool = False,
+    force: bool = False,
     **kwargs: Any,
-) -> Iterator["Stage"]:
+) -> Iterator[StageInfo]:
     for target in targets:
         if kwargs.get("out"):
-            target = resolve_output(target, kwargs["out"])
+            target = resolve_output(target, kwargs["out"], force=force)
         path, wdir, out = resolve_paths(
             repo, target, always_local=transfer and not kwargs.get("out")
         )
 
-        stage = repo.stage.create(
-            single_stage=True,
-            validate=False,
-            fname=fname or path,
-            wdir=wdir,
-            outs=[out],
-            external=external,
-        )
+        try:
+            (out_obj,) = repo.find_outs_by_path(target, strict=False)
+            stage = out_obj.stage
+            if not stage.is_data_source:
+                raise DvcException(f"cannot update {out!r}: not a data source")
+            output_exists = True
+        except OutputNotFoundError:
+            stage = repo.stage.create(
+                single_stage=True,
+                validate=False,
+                fname=fname or path,
+                wdir=wdir,
+                outs=[out],
+                external=external,
+                force=force,
+            )
+            output_exists = False
 
         out_obj = stage.outs[0]
         out_obj.annot.update(**kwargs)
-        yield stage
+        yield StageInfo(stage, output_exists)

dvc/repo/artifacts.py

@@ -1,19 +1,26 @@
 import logging
-import os
 import re
-from typing import TYPE_CHECKING, Dict
+from pathlib import Path
+from typing import Dict, Optional
 
 from dvc.annotations import Artifact
-from dvc.dvcfile import FileMixin
+from dvc.dvcfile import PROJECT_FILE
+from dvc.exceptions import InvalidArgumentError
+from dvc.repo import Repo
 from dvc.utils import relpath
-
-if TYPE_CHECKING:
-    from dvc.repo import Repo
+from dvc.utils.serialize import modify_yaml
 
 logger = logging.getLogger(__name__)
 
 
-NAME_RE = re.compile(r"^[a-z]([a-z0-9-]*[a-z0-9])?$")
+# 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 name_is_compatible(name: str) -> bool:
@@ -22,22 +29,23 @@ def name_is_compatible(name: str) -> bool:
 
 def check_name_format(name: str) -> None:
     if not name_is_compatible(name):
-        logger.warning(
-            "Can't use '%s' as artifact name (ID)."
+        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). The first character must be a letter.",
-            name,
+            " (but not at the start or end)."
         )
 
 
-class ArtifactsFile(FileMixin):
-    from dvc.schema import SINGLE_ARTIFACT_SCHEMA as SCHEMA
-
-    def dump(self, stage, **kwargs):
-        raise NotImplementedError
-
-    def merge(self, ancestor, other, allowed=None):
-        raise NotImplementedError
+def check_for_nested_dvc_repo(dvcfile: Path):
+    if dvcfile.is_absolute():
+        raise InvalidArgumentError("Use relative path to dvc.yaml.")
+    path = dvcfile.parent
+    while path.name:
+        if (path / Repo.DVC_DIR).is_dir():
+            raise InvalidArgumentError(
+                f"Nested DVC repos like {path} are not supported."
+            )
+        path = path.parent
 
 
 class Artifacts:
@@ -50,18 +58,30 @@ class Artifacts:
             dvcfile,
             dvcfile_artifacts,
         ) in self.repo.index._artifacts.items():  # pylint: disable=protected-access
-            # read the artifacts.yaml file if needed
-            if isinstance(dvcfile_artifacts, str):
-                dvcfile_artifacts = ArtifactsFile(
-                    self.repo,
-                    os.path.join(os.path.dirname(dvcfile), dvcfile_artifacts),
-                    verify=False,
-                ).load()
-            if not dvcfile_artifacts:
-                continue
             dvcyaml = relpath(dvcfile, self.repo.root_dir)
             artifacts[dvcyaml] = {}
             for name, value in dvcfile_artifacts.items():
-                check_name_format(name)
+                try:
+                    check_name_format(name)
+                except InvalidArgumentError as e:
+                    logger.warning(e.msg)
                 artifacts[dvcyaml][name] = Artifact(**value)
         return artifacts
+
+    def add(self, name: str, artifact: Artifact, dvcfile: Optional[str] = None):
+        with self.repo.scm_context(quiet=True):
+            check_name_format(name)
+            dvcyaml = Path(dvcfile or PROJECT_FILE)
+            check_for_nested_dvc_repo(
+                dvcyaml.relative_to(self.repo.root_dir)
+                if dvcyaml.is_absolute()
+                else dvcyaml
+            )
+
+            with modify_yaml(dvcyaml) as data:
+                artifacts = data.setdefault("artifacts", {})
+                artifacts.update({name: artifact.to_dict()})
+
+            self.repo.scm_context.track_file(dvcfile)
+
+        return artifacts.get(name)

dvc/repo/data.py

@@ -172,7 +172,6 @@ def _diff_index_to_wtree(repo: "Repo", **kwargs: Any) -> Dict[str, List[str]]:
             repo.index.data["repo"],
             workspace,
             not_in_cache=True,
-            not_in_remote=True,
             **kwargs,
         )
 

dvc/repo/experiments/__init__.py

@@ -119,6 +119,7 @@ class Experiments:
         self,
         tmp_dir: bool = False,
         copy_paths: Optional[List[str]] = None,
+        message: Optional[str] = None,
         **kwargs,
     ):
         """Reproduce and checkout a single (standalone) experiment."""
@@ -126,7 +127,9 @@ class Experiments:
             self.tempdir_queue if tmp_dir else self.workspace_queue
         )
         self.queue_one(exp_queue, **kwargs)
-        results = self._reproduce_queue(exp_queue, copy_paths=copy_paths)
+        results = self._reproduce_queue(
+            exp_queue, copy_paths=copy_paths, message=message
+        )
         exp_rev = first(results)
         if exp_rev is not None:
             self._log_reproduced(results, tmp_dir=tmp_dir)
@@ -349,7 +352,11 @@ class Experiments:
 
     @unlocked_repo
     def _reproduce_queue(
-        self, queue: "BaseStashQueue", copy_paths: Optional[List[str]] = None, **kwargs
+        self,
+        queue: "BaseStashQueue",
+        copy_paths: Optional[List[str]] = None,
+        message: Optional[str] = None,
+        **kwargs,
     ) -> Dict[str, str]:
         """Reproduce queued experiments.
 
@@ -360,7 +367,7 @@ class Experiments:
             dict mapping successfully reproduced experiment revs to their
             results.
         """
-        exec_results = queue.reproduce(copy_paths=copy_paths)
+        exec_results = queue.reproduce(copy_paths=copy_paths, message=message)
 
         results: Dict[str, str] = {}
         for _, exp_result in exec_results.items():

dvc/repo/experiments/collect.py

@@ -304,7 +304,7 @@ def collect(
         experiments = list(
             itertools.chain.from_iterable(
                 _sorted_ranges(collected.get(baseline_rev, []))
-                for collected in (successful, active, queued, failed)
+                for collected in (active, successful, queued, failed)
             )
         )
         result.append(

dvc/repo/experiments/executor/base.py

@@ -7,6 +7,7 @@ from contextlib import contextmanager
 from dataclasses import asdict, dataclass
 from enum import IntEnum
 from functools import partial
+from itertools import chain
 from typing import (
     TYPE_CHECKING,
     Any,
@@ -42,6 +43,7 @@ from dvc.stage.serialize import to_lockfile
 from dvc.ui import ui
 from dvc.utils import dict_sha256, env2bool, relpath
 from dvc.utils.fs import remove
+from dvc.utils.studio import env_to_config
 
 if TYPE_CHECKING:
     from queue import Queue
@@ -261,10 +263,14 @@ class BaseExecutor(ABC):
         )
 
     @classmethod
-    def _get_top_level_paths(cls, repo: "Repo") -> Iterable["str"]:
-        yield from _collect_top_level_metrics(repo)
-        yield from _collect_top_level_params(repo)
-        yield from repo.index._plot_sources  # pylint: disable=protected-access
+    def _get_top_level_paths(cls, repo: "Repo") -> List["str"]:
+        return list(
+            chain(
+                _collect_top_level_metrics(repo),
+                _collect_top_level_params(repo),
+                repo.index._plot_sources,  # pylint: disable=protected-access
+            )
+        )
 
     @classmethod
     def save(
@@ -272,6 +278,7 @@ class BaseExecutor(ABC):
         info: "ExecutorInfo",
         force: bool = False,
         include_untracked: Optional[List[str]] = None,
+        message: Optional[str] = None,
     ) -> ExecutorResult:
         from dvc.dvcfile import LOCK_FILE
         from dvc.repo import Repo
@@ -303,6 +310,7 @@ class BaseExecutor(ABC):
                 exp_hash,
                 exp_name=info.name,
                 force=force,
+                message=message,
             )
             ref: Optional[str] = dvc.scm.get_ref(EXEC_BRANCH, follow=False)
             exp_ref = ExpRefInfo.from_ref(ref) if ref else None
@@ -456,6 +464,7 @@ class BaseExecutor(ABC):
         log_errors: bool = True,
         log_level: Optional[int] = None,
         copy_paths: Optional[List[str]] = None,
+        message: Optional[str] = None,
         **kwargs,
     ) -> "ExecutorResult":
         """Run dvc repro and return the result.
@@ -493,6 +502,7 @@ class BaseExecutor(ABC):
             infofile,
             log_errors=log_errors,
             copy_paths=copy_paths,
+            message=message,
             **kwargs,
         ) as dvc:
             if auto_push:
@@ -544,20 +554,16 @@ class BaseExecutor(ABC):
                 repro_force or checkpoint_reset,
             )
 
-            def after_repro():
-                paths = list(cls._get_top_level_paths(dvc))
-                if paths:
-                    logger.debug("Staging top-level files: %s", paths)
-                    dvc.scm_context.add(paths)
-
             stages = dvc_reproduce(
                 dvc,
                 *args,
                 on_unchanged=filter_pipeline,
                 checkpoint_func=checkpoint_func,
-                after_repro_callback=after_repro,
                 **kwargs,
             )
+            if paths := cls._get_top_level_paths(dvc):
+                logger.debug("Staging top-level files: %s", paths)
+                dvc.scm_context.add(paths)
 
             exp_hash = cls.hash_exp(stages)
             if not repro_dry:
@@ -569,6 +575,7 @@ class BaseExecutor(ABC):
                     auto_push,
                     git_remote,
                     repro_force,
+                    message=message,
                 )
                 info.result_hash = exp_hash
                 info.result_ref = ref
@@ -589,6 +596,7 @@ class BaseExecutor(ABC):
         auto_push,
         git_remote,
         repro_force,
+        message: Optional[str] = None,
     ) -> Tuple[Optional[str], Optional["ExpRefInfo"], bool]:
         is_checkpoint = any(stage.is_checkpoint for stage in stages)
         cls.commit(
@@ -597,6 +605,7 @@ class BaseExecutor(ABC):
             exp_name=info.name,
             force=repro_force,
             checkpoint=is_checkpoint,
+            message=message,
         )
         if auto_push:
             cls._auto_push(dvc, dvc.scm, git_remote)
@@ -624,6 +633,7 @@ class BaseExecutor(ABC):
         infofile: Optional[str] = None,
         log_errors: bool = True,
         copy_paths: Optional[List[str]] = None,
+        message: Optional[str] = None,
         **kwargs,
     ) -> Iterator["Repo"]:
         from dvc_studio_client.post_live_metrics import post_live_metrics
@@ -647,13 +657,23 @@ class BaseExecutor(ABC):
             else:
                 os.chdir(dvc.root_dir)
 
+            args_path = os.path.join(dvc.tmp_dir, cls.PACKED_ARGS_FILE)
+            if os.path.exists(args_path):
+                _, kwargs = cls.unpack_repro_args(args_path)
+            dvc_studio_config = dvc.config.get("studio")
+            # set missing config options using saved config
+            # inferring repo url will fail if not set here
+            run_env_config = env_to_config(kwargs.get("run_env", {}))
+            dvc_studio_config = {**run_env_config, **dvc_studio_config}
             try:
                 post_live_metrics(
                     "start",
                     info.baseline_rev,
-                    info.name,
+                    info.name,  # type: ignore[arg-type]
                     "dvc",
                     params=to_studio_params(dvc.params.show()),
+                    dvc_studio_config=dvc_studio_config,
+                    message=message,
                 )
                 logger.debug("Running repro in '%s'", os.getcwd())
                 yield dvc
@@ -675,10 +695,11 @@ class BaseExecutor(ABC):
                 post_live_metrics(
                     "done",
                     info.baseline_rev,
-                    info.name,
+                    info.name,  # type: ignore[arg-type]
                     "dvc",
                     experiment_rev=dvc.experiments.scm.get_ref(EXEC_BRANCH),
                     metrics=get_in(dvc.metrics.show(), ["", "data"]),
+                    dvc_studio_config=dvc_studio_config,
                 )
 
                 if infofile is not None:
@@ -750,6 +771,7 @@ class BaseExecutor(ABC):
         exp_name: Optional[str] = None,
         force: bool = False,
         checkpoint: bool = False,
+        message: Optional[str] = None,
     ):
         """Commit stages as an experiment and return the commit SHA."""
         rev = scm.get_rev()
@@ -779,7 +801,8 @@ class BaseExecutor(ABC):
                 logger.debug("Commit to new experiment branch '%s'", branch)
 
         scm.add([], update=True)
-        scm.commit(f"dvc: commit experiment {exp_hash}", no_verify=True)
+        message = message or f"dvc: commit experiment {exp_hash}"
+        scm.commit(message, no_verify=True)
         new_rev = scm.get_rev()
         if check_conflict:
             new_rev = cls._raise_ref_conflict(scm, branch, new_rev, checkpoint)

dvc/repo/experiments/executor/local.py

@@ -17,7 +17,7 @@ from dvc.repo.experiments.refs import (
     EXEC_HEAD,
     EXEC_MERGE,
     EXEC_NAMESPACE,
-    EXPS_TEMP,
+    TEMP_NAMESPACE,
 )
 from dvc.repo.experiments.utils import EXEC_TMP_DIR, get_exp_rwlock
 from dvc.scm import SCM, Git
@@ -90,9 +90,9 @@ class TempDirExecutor(BaseLocalExecutor):
         if infofile:
             self.info.dump_json(infofile)
 
-        temp_head = f"{EXPS_TEMP}/head-{uuid()}"
-        temp_merge = f"{EXPS_TEMP}/merge-{uuid()}"
-        temp_baseline = f"{EXPS_TEMP}/baseline-{uuid()}"
+        temp_head = f"{TEMP_NAMESPACE}/head-{uuid()}"
+        temp_merge = f"{TEMP_NAMESPACE}/merge-{uuid()}"
+        temp_baseline = f"{TEMP_NAMESPACE}/baseline-{uuid()}"
 
         temp_ref_dict = {
             temp_head: entry.head_rev,

dvc/repo/experiments/executor/ssh.py

@@ -243,6 +243,7 @@ class SSHExecutor(BaseExecutor):
         log_errors: bool = True,
         log_level: Optional[int] = None,
         copy_paths: Optional[List[str]] = None,  # noqa: ARG003
+        message: Optional[str] = None,  # noqa: ARG003
         **kwargs,
     ) -> "ExecutorResult":
         """Reproduce an experiment on a remote machine over SSH.

dvc/repo/experiments/push.py

@@ -20,6 +20,7 @@ from dvc.repo import locked
 from dvc.repo.scm_context import scm_context
 from dvc.scm import Git, TqdmGit, iter_revs
 from dvc.utils import env2bool
+from dvc.utils.collections import ensure_list
 
 from .exceptions import UnresolvedExpNamesError
 from .refs import ExpRefInfo
@@ -66,14 +67,39 @@ def notify_refs_to_studio(
     return d.get("url")
 
 
+def exp_refs_from_names(scm: "Git", exp_names: List[str]) -> Set["ExpRefInfo"]:
+    exp_ref_set = set()
+    exp_ref_dict = resolve_name(scm, exp_names)
+    unresolved_exp_names = []
+    for exp_name, exp_ref in exp_ref_dict.items():
+        if exp_ref is None:
+            unresolved_exp_names.append(exp_name)
+        else:
+            exp_ref_set.add(exp_ref)
+
+    if unresolved_exp_names:
+        raise UnresolvedExpNamesError(unresolved_exp_names)
+    return exp_ref_set
+
+
+def exp_refs_from_rev(scm: "Git", rev: List[str], num: int = 1) -> Set["ExpRefInfo"]:
+    exp_ref_set = set()
+    rev_dict = iter_revs(scm, rev, num)
+    rev_set = set(rev_dict.keys())
+    ref_info_dict = exp_refs_by_baseline(scm, rev_set)
+    for _, ref_info_list in ref_info_dict.items():
+        exp_ref_set.update(ref_info_list)
+    return exp_ref_set
+
+
 @locked
 @scm_context
-def push(  # noqa: C901
+def push(
     repo: "Repo",
     git_remote: str,
-    exp_names: Union[Iterable[str], str],
+    exp_names: Union[List[str], str],
     all_commits: bool = False,
-    rev: Optional[str] = None,
+    rev: Optional[Union[List[str], str]] = None,
     num: int = 1,
     force: bool = False,
     push_cache: bool = False,
@@ -83,29 +109,12 @@ def push(  # noqa: C901
     assert isinstance(repo.scm, Git)
     if all_commits:
         exp_ref_set.update(exp_refs(repo.scm))
-
-    else:
-        if exp_names:
-            if isinstance(exp_names, str):
-                exp_names = [exp_names]
-            exp_ref_dict = resolve_name(repo.scm, exp_names)
-
-            unresolved_exp_names = []
-            for exp_name, exp_ref in exp_ref_dict.items():
-                if exp_ref is None:
-                    unresolved_exp_names.append(exp_name)
-                else:
-                    exp_ref_set.add(exp_ref)
-
-            if unresolved_exp_names:
-                raise UnresolvedExpNamesError(unresolved_exp_names)
-
-        if rev:
-            rev_dict = iter_revs(repo.scm, [rev], num)
-            rev_set = set(rev_dict.keys())
-            ref_info_dict = exp_refs_by_baseline(repo.scm, rev_set)
-            for _, ref_info_list in ref_info_dict.items():
-                exp_ref_set.update(ref_info_list)
+    if exp_names:
+        exp_ref_set.update(exp_refs_from_names(repo.scm, ensure_list(exp_names)))
+    if rev:
+        if isinstance(rev, str):
+            rev = [rev]
+        exp_ref_set.update(exp_refs_from_rev(repo.scm, rev, num=num))
 
     push_result = _push(repo, git_remote, exp_ref_set, force)
 
@@ -118,15 +127,20 @@ def push(  # noqa: C901
     pushed_refs_info = (
         push_result[SyncStatus.UP_TO_DATE] + push_result[SyncStatus.SUCCESS]
     )
+
+    e = None
     if push_cache:
         try:
             result["uploaded"] = _push_cache(repo, pushed_refs_info, **kwargs)
         except Exception as exc:  # noqa: BLE001, pylint: disable=broad-except
-            raise UploadError("failed to push cache", result) from exc
+            e = exc
 
     pushed_refs = [str(r) for r in pushed_refs_info]
-    url = notify_refs_to_studio(repo, git_remote, pushed=pushed_refs)
-    return {**result, "url": url}
+    result["url"] = notify_refs_to_studio(repo, git_remote, pushed=pushed_refs)
+
+    if e:
+        raise UploadError("failed to push cache", result) from e
+    return result
 
 
 def _push(

dvc/repo/experiments/queue/base.py

@@ -17,8 +17,7 @@ from typing import (
     Union,
 )
 
-from dvc_studio_client.env import STUDIO_REPO_URL
-from dvc_studio_client.post_live_metrics import get_studio_repo_url
+from dvc_studio_client.post_live_metrics import get_studio_config
 from funcy import retry
 
 from dvc.dependency import ParamsDependency
@@ -39,6 +38,7 @@ from dvc.repo.experiments.utils import (
 )
 from dvc.ui import ui
 from dvc.utils.objects import cached_property
+from dvc.utils.studio import config_to_env
 
 from .utils import get_remote_executor_refs
 
@@ -254,7 +254,7 @@ class BaseStashQueue(ABC):
 
     @abstractmethod
     def reproduce(
-        self, copy_paths: Optional[List[str]] = None
+        self, copy_paths: Optional[List[str]] = None, message: Optional[str] = None
     ) -> Mapping[str, Mapping[str, str]]:
         """Reproduce queued experiments sequentially."""
 
@@ -420,12 +420,13 @@ class BaseStashQueue(ABC):
                     run_env[DVC_EXP_NAME] = name
                     if resume_rev:
                         run_env[DVCLIVE_RESUME] = "1"
-                    studio_repo_url = get_studio_repo_url()
-                    if studio_repo_url is not None:
-                        run_env[STUDIO_REPO_URL] = studio_repo_url
 
+                    # save studio config to read later by dvc and dvclive
+                    studio_config = get_studio_config(
+                        dvc_studio_config=self.repo.config.get("studio")
+                    )
+                    run_env = {**config_to_env(studio_config), **run_env}
                     self._pack_args(*args, run_env=run_env, **kwargs)
-
                     # save experiment as a stash commit
                     msg = self._stash_msg(
                         stash_head,

dvc/repo/experiments/queue/celery.py

@@ -125,6 +125,7 @@ class LocalCeleryQueue(BaseStashQueue):
             without_mingle=True,
             without_gossip=True,
             timeout=10,
+            loglevel="debug" if logger.getEffectiveLevel() <= logging.DEBUG else "info",
         )
 
     def _spawn_worker(self, num: int = 1):
@@ -143,6 +144,8 @@ class LocalCeleryQueue(BaseStashQueue):
         if num == 1:
             # automatically run celery cleanup when primary worker shuts down
             cmd.append("--clean")
+        if logger.getEffectiveLevel() <= logging.DEBUG:
+            cmd.append("-v")
         name = f"dvc-exp-worker-{num}"
 
         logger.debug("start a new worker: %s, node: %s", name, node_name)
@@ -177,12 +180,18 @@ class LocalCeleryQueue(BaseStashQueue):
         return started
 
     def put(
-        self, *args, copy_paths: Optional[List[str]] = None, **kwargs
+        self,
+        *args,
+        copy_paths: Optional[List[str]] = None,
+        message: Optional[str] = None,
+        **kwargs,
     ) -> QueueEntry:
         """Stash an experiment and add it to the queue."""
         with get_exp_rwlock(self.repo, writes=["workspace", CELERY_STASH]):
             entry = self._stash_exp(*args, **kwargs)
-        self.celery.signature(run_exp.s(entry.asdict(), copy_paths=copy_paths)).delay()
+        self.celery.signature(
+            run_exp.s(entry.asdict(), copy_paths=copy_paths, message=message)
+        ).delay()
         return entry
 
     # NOTE: Queue consumption should not be done directly. Celery worker(s)
@@ -261,7 +270,7 @@ class LocalCeleryQueue(BaseStashQueue):
                 yield QueueDoneResult(queue_entry, exp_result)
 
     def reproduce(
-        self, copy_paths: Optional[List[str]] = None
+        self, copy_paths: Optional[List[str]] = None, message: Optional[str] = None
     ) -> Mapping[str, Mapping[str, str]]:
         raise NotImplementedError
 

dvc/repo/experiments/queue/tasks.py

@@ -91,7 +91,11 @@ def cleanup_exp(executor: TempDirExecutor, infofile: str) -> None:
 
 
 @shared_task
-def run_exp(entry_dict: Dict[str, Any], copy_paths: Optional[List[str]] = None) -> None:
+def run_exp(
+    entry_dict: Dict[str, Any],
+    copy_paths: Optional[List[str]] = None,
+    message: Optional[str] = None,
+) -> None:
     """Run a full experiment.
 
     Experiment subtasks are executed inline as one atomic operation.
@@ -111,6 +115,8 @@ def run_exp(entry_dict: Dict[str, Any], copy_paths: Optional[List[str]] = None)
         if copy_paths:
             for path in copy_paths:
                 cmd.extend(["--copy-paths", path])
+        if message:
+            cmd.extend(["--message", message])
         proc_dict = queue.proc.run_signature(cmd, name=entry.stash_rev)()
         collect_exp.s(proc_dict, entry_dict)()
     finally:

dvc/repo/experiments/queue/tempdir.py

@@ -97,6 +97,7 @@ class TempDirQueue(WorkspaceQueue):
         entry: QueueEntry,
         executor: "BaseExecutor",
         copy_paths: Optional[List[str]] = None,
+        message: Optional[str] = None,
         **kwargs,
     ) -> Dict[str, Dict[str, str]]:
         from dvc.stage.monitor import CheckpointKilledError
@@ -113,6 +114,7 @@ class TempDirQueue(WorkspaceQueue):
                 log_level=logger.getEffectiveLevel(),
                 log_errors=True,
                 copy_paths=copy_paths,
+                message=message,
             )
             if not exec_result.exp_hash:
                 raise DvcException(f"Failed to reproduce experiment '{rev[:7]}'")

dvc/repo/experiments/queue/workspace.py

@@ -82,14 +82,16 @@ class WorkspaceQueue(BaseStashQueue):
         raise NotImplementedError
 
     def reproduce(
-        self, copy_paths: Optional[List[str]] = None
+        self, copy_paths: Optional[List[str]] = None, message: Optional[str] = None
     ) -> Dict[str, Dict[str, str]]:
         results: Dict[str, Dict[str, str]] = defaultdict(dict)
         try:
             while True:
                 entry, executor = self.get()
                 results.update(
-                    self._reproduce_entry(entry, executor, copy_paths=copy_paths)
+                    self._reproduce_entry(
+                        entry, executor, copy_paths=copy_paths, message=message
+                    )
                 )
         except ExpQueueEmptyError:
             pass
@@ -117,6 +119,7 @@ class WorkspaceQueue(BaseStashQueue):
                 infofile=infofile,
                 log_level=logger.getEffectiveLevel(),
                 log_errors=not isinstance(executor, WorkspaceExecutor),
+                message=kwargs.get("message"),
             )
             if not exec_result.exp_hash:
                 raise DvcException(f"Failed to reproduce experiment '{rev[:7]}'")

dvc/repo/experiments/refs.py

@@ -10,8 +10,9 @@ WORKSPACE_STASH = EXPS_STASH
 APPLY_NAMESPACE = f"{EXPS_NAMESPACE}/apply"
 APPLY_HEAD = f"{APPLY_NAMESPACE}/ORIG_HEAD"
 APPLY_STASH = f"{APPLY_NAMESPACE}/stash"
-CELERY_STASH = f"{EXPS_NAMESPACE}/celery/stash"
-CELERY_FAILED_STASH = f"{EXPS_NAMESPACE}/celery/failed"
+CELERY_NAMESPACE = f"{EXPS_NAMESPACE}/celery"
+CELERY_STASH = f"{CELERY_NAMESPACE}/stash"
+CELERY_FAILED_STASH = f"{CELERY_NAMESPACE}/failed"
 EXEC_NAMESPACE = f"{EXPS_NAMESPACE}/exec"
 EXEC_APPLY = f"{EXEC_NAMESPACE}/EXEC_APPLY"
 EXEC_CHECKPOINT = f"{EXEC_NAMESPACE}/EXEC_CHECKPOINT"
@@ -19,9 +20,14 @@ EXEC_BRANCH = f"{EXEC_NAMESPACE}/EXEC_BRANCH"
 EXEC_BASELINE = f"{EXEC_NAMESPACE}/EXEC_BASELINE"
 EXEC_HEAD = f"{EXEC_NAMESPACE}/EXEC_HEAD"
 EXEC_MERGE = f"{EXEC_NAMESPACE}/EXEC_MERGE"
-EXPS_TEMP = f"{EXPS_NAMESPACE}/temp"
+TEMP_NAMESPACE = f"{EXPS_NAMESPACE}/temp"
 STASHES = {WORKSPACE_STASH, CELERY_STASH}
-ITER_SKIP_NAMESPACES = {APPLY_NAMESPACE, EXEC_NAMESPACE}
+ITER_SKIP_NAMESPACES = {
+    APPLY_NAMESPACE,
+    CELERY_NAMESPACE,
+    EXEC_NAMESPACE,
+    TEMP_NAMESPACE,
+}
 
 
 class ExpRefInfo:

dvc/repo/experiments/run.py

@@ -20,6 +20,7 @@ def run(  # noqa: C901, PLR0912
     tmp_dir: bool = False,
     queue: bool = False,
     copy_paths: Optional[Iterable[str]] = None,
+    message: Optional[str] = None,
     **kwargs,
 ) -> Dict[str, str]:
     """Reproduce the specified targets as an experiment.
@@ -74,6 +75,7 @@ def run(  # noqa: C901, PLR0912
             params=path_overrides,
             tmp_dir=tmp_dir,
             copy_paths=copy_paths,
+            message=message,
             **kwargs,
         )
 
@@ -96,6 +98,7 @@ def run(  # noqa: C901, PLR0912
             targets=targets,
             params=sweep_overrides,
             copy_paths=copy_paths,
+            message=message,
             **kwargs,
         )
         if sweep_overrides:

dvc/repo/experiments/save.py

@@ -16,6 +16,7 @@ def save(
     name: Optional[str] = None,
     force: bool = False,
     include_untracked: Optional[List[str]] = None,
+    message: Optional[str] = None,
 ) -> Optional[str]:
     """Save the current workspace status as an experiment.
 
@@ -29,7 +30,10 @@ def save(
 
     try:
         save_result = executor.save(
-            executor.info, force=force, include_untracked=include_untracked
+            executor.info,
+            force=force,
+            include_untracked=include_untracked,
+            message=message,
         )
         result = queue.collect_executor(repo.experiments, executor, save_result)
     finally:

dvc/repo/experiments/show.py

@@ -62,7 +62,7 @@ def show(
 
 def tabulate(
     baseline_states: Iterable["ExpState"],
-    fill_value: str = "-",
+    fill_value: Optional[str] = "-",
     error_value: str = "!",
     **kwargs,
 ) -> Tuple["TabularData", Dict[str, Iterable[str]]]:
@@ -124,7 +124,7 @@ def _build_rows(
     baseline_states: Iterable["ExpState"],
     *,
     all_headers: Iterable[str],
-    fill_value: str,
+    fill_value: Optional[str],
     sort_by: Optional[str] = None,
     sort_order: Optional[Literal["asc", "desc"]] = None,
     **kwargs,
@@ -234,7 +234,7 @@ def _exp_range_rows(
     exp_range: "ExpRange",
     *,
     all_headers: Iterable[str],
-    fill_value: str,
+    fill_value: Optional[str],
     is_base: bool = False,
     **kwargs,
 ) -> Iterator[Tuple["CellT", ...]]:
@@ -276,7 +276,7 @@ def _data_cells(
     metrics_names: Mapping[str, Iterable[str]],
     params_names: Mapping[str, Iterable[str]],
     deps_names: Iterable[str],
-    fill_value: str = "-",
+    fill_value: Optional[str] = "-",
     error_value: str = "!",
     precision: Optional[int] = None,
     **kwargs,
@@ -317,10 +317,10 @@ def _data_cells(
 
 def format_time(
     timestamp: Optional[datetime],
-    fill_value: str = "-",
+    fill_value: Optional[str] = "-",
     iso: bool = False,
     **kwargs,
-) -> str:
+) -> Optional[str]:
     if not timestamp:
         return fill_value
     if iso:

dvc/repo/gc.py

@@ -1,5 +1,5 @@
 import logging
-from typing import TYPE_CHECKING, List, Optional
+from typing import TYPE_CHECKING, List, Optional, Set, Tuple
 
 from dvc.exceptions import InvalidArgumentError
 
@@ -7,11 +7,22 @@ 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
 
 logger = logging.getLogger(__name__)
 
 
 def _validate_args(**kwargs):
+    not_in_remote = kwargs.pop("not_in_remote", None)
+    cloud = kwargs.pop("cloud", None)
+    remote = kwargs.pop("remote", None)
+    if remote and not (cloud or not_in_remote):
+        raise InvalidArgumentError("`--remote` requires `--cloud` or `--not-in-remote`")
+    if not_in_remote and cloud:
+        raise InvalidArgumentError(
+            "`--not-in-remote` and `--cloud` are mutually exclusive"
+        )
     if not any(kwargs.values()):
         raise InvalidArgumentError(
             "Either of `-w|--workspace`, `-a|--all-branches`, `-T|--all-tags` "
@@ -22,8 +33,22 @@ 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):
+    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")
+
+        remote_oids.update(
+            remote_odb.list_oids_exists({x.value for x in obj_ids}, 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: PLR0913
+def gc(  # noqa: PLR0912, PLR0913, C901
     self: "Repo",
     all_branches: bool = False,
     cloud: bool = False,
@@ -39,6 +64,7 @@ def gc(  # noqa: PLR0913
     commit_date: Optional[str] = None,
     rev: Optional[str] = None,
     num: Optional[int] = None,
+    not_in_remote: bool = False,
 ):
     # require `workspace` to be true to come into effect.
     # assume `workspace` to be enabled if any of `all_tags`, `all_commits`,
@@ -52,6 +78,8 @@ def gc(  # noqa: PLR0913
         commit_date=commit_date,
         rev=rev,
         num=num,
+        cloud=cloud,
+        not_in_remote=not_in_remote,
     )
 
     from contextlib import ExitStack
@@ -64,13 +92,14 @@ def gc(  # noqa: PLR0913
         repos = []
     all_repos = [Repo(path) for path in repos]
 
-    used_obj_ids = set()
+    odb_to_obj_ids: List[Tuple[Optional["ObjectDB"], Set["HashInfo"]]] = []
+
     with ExitStack() as stack:
         for repo in all_repos:
             stack.enter_context(repo.lock)
 
         for repo in [*all_repos, self]:
-            for obj_ids in repo.used_objs(
+            for odb, obj_ids in repo.used_objs(
                 all_branches=all_branches,
                 with_deps=with_deps,
                 all_tags=all_tags,
@@ -82,13 +111,22 @@ def gc(  # noqa: PLR0913
                 jobs=jobs,
                 revs=[rev] if rev else None,
                 num=num or 1,
-            ).values():
-                used_obj_ids.update(obj_ids)
+            ).items():
+                odb_to_obj_ids.append((odb, obj_ids))
+
+    if not odb_to_obj_ids:
+        odb_to_obj_ids = [(None, set())]
+
+    if not_in_remote:
+        used_obj_ids = _used_obj_ids_not_in_remote(self, remote, jobs, odb_to_obj_ids)
+    else:
+        used_obj_ids = set()
+        for _, obj_ids in odb_to_obj_ids:
+            used_obj_ids.update(obj_ids)
 
     for scheme, odb in self.cache.by_scheme():
         if not odb:
             continue
-
         removed = ogc(odb, used_obj_ids, jobs=jobs)
         if not removed:
             logger.info("No unused '%s' cache to remove.", scheme)
@@ -96,9 +134,11 @@ def gc(  # noqa: PLR0913
     if not cloud:
         return
 
-    odb = self.cloud.get_remote_odb(remote, "gc -c")
-    removed = ogc(odb, used_obj_ids, jobs=jobs)
-    if removed:
-        get_index(odb).clear()
-    else:
-        logger.info("No unused cache to remove from remote.")
+    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:
+            get_index(remote_odb).clear()
+        else:
+            logger.info("No unused cache to remove from remote.")

dvc/repo/get.py

@@ -4,7 +4,6 @@ from typing import TYPE_CHECKING, Union
 
 from dvc.exceptions import DvcException
 from dvc.utils import resolve_output
-from dvc.utils.fs import remove
 
 if TYPE_CHECKING:
     from dvc.fs.dvc import DVCFileSystem
@@ -20,64 +19,39 @@ class GetDVCFileError(DvcException):
         )
 
 
-def get(url, path, out=None, rev=None, jobs=None):
-    import shortuuid
-
+def get(url, path, out=None, rev=None, jobs=None, force=False):
     from dvc.dvcfile import is_valid_filename
-    from dvc.external_repo import external_repo
     from dvc.fs.callbacks import Callback
+    from dvc.repo import Repo
 
-    out = resolve_output(path, out)
+    out = resolve_output(path, out, force=force)
 
     if is_valid_filename(out):
         raise GetDVCFileError()
 
-    # Creating a directory right beside the output to make sure that they
-    # are on the same filesystem, so we could take the advantage of
-    # reflink and/or hardlink. Not using tempfile.TemporaryDirectory
-    # because it will create a symlink to tmpfs, which defeats the purpose
-    # and won't work with reflink/hardlink.
-    dpath = os.path.dirname(os.path.abspath(out))
-    tmp_dir = os.path.join(dpath, "." + str(shortuuid.uuid()))
+    with Repo.open(
+        url=url,
+        rev=rev,
+        subrepos=True,
+        uninitialized=True,
+    ) as repo:
+        from dvc.fs.data import DataFileSystem
 
-    # Try any links possible to avoid data duplication.
-    #
-    # Not using symlink, because we need to remove cache after we
-    # are done, and to make that work we would have to copy data
-    # over anyway before removing the cache, so we might just copy
-    # it right away.
-    #
-    # Also, we can't use theoretical "move" link type here, because
-    # the same cache file might be used a few times in a directory.
-    cache_types = ["reflink", "hardlink", "copy"]
-    try:
-        with external_repo(
-            url=url,
-            rev=rev,
-            subrepos=True,
-            uninitialized=True,
-            cache_dir=tmp_dir,
-            cache_types=cache_types,
-        ) as repo:
-            from dvc.fs.data import DataFileSystem
+        fs: Union[DataFileSystem, "DVCFileSystem"]
+        if os.path.isabs(path):
+            fs = DataFileSystem(index=repo.index.data["local"])
+            fs_path = fs.from_os_path(path)
+        else:
+            fs = repo.dvcfs
+            fs_path = fs.from_os_path(path)
 
-            fs: Union[DataFileSystem, "DVCFileSystem"]
-            if os.path.isabs(path):
-                fs = DataFileSystem(index=repo.index.data["local"])
-                fs_path = fs.from_os_path(path)
-            else:
-                fs = repo.dvcfs
-                fs_path = fs.from_os_path(path)
-
-            with Callback.as_tqdm_callback(
-                desc=f"Downloading {fs.path.name(path)}",
-                unit="files",
-            ) as cb:
-                fs.get(
-                    fs_path,
-                    os.path.abspath(out),
-                    batch_size=jobs,
-                    callback=cb,
-                )
-    finally:
-        remove(tmp_dir)
+        with Callback.as_tqdm_callback(
+            desc=f"Downloading {fs.path.name(path)}",
+            unit="files",
+        ) as cb:
+            fs.get(
+                fs_path,
+                os.path.abspath(out),
+                batch_size=jobs,
+                callback=cb,
+            )

dvc/repo/get_url.py

@@ -6,8 +6,8 @@ from dvc.fs import download, parse_external_url
 from dvc.utils import resolve_output
 
 
-def get_url(url, out=None, *, config=None, jobs=None):
-    out = resolve_output(url, out)
+def get_url(url, out=None, *, config=None, jobs=None, force=False):
+    out = resolve_output(url, out, force=force)
     out = os.path.abspath(out)
     (out,) = output.loads_from(None, [out], use_cache=False)
 

dvc/repo/imp_url.py

@@ -30,10 +30,11 @@ def imp_url(  # noqa: C901, PLR0913
     labels=None,
     meta=None,
     jobs=None,
+    force=False,
     fs_config=None,
     version_aware: bool = False,
 ):
-    out = resolve_output(url, out)
+    out = resolve_output(url, out, force=force)
     path, wdir, out = resolve_paths(self, out, always_local=to_remote and not out)
 
     if to_remote and (no_exec or no_download or version_aware):

dvc/repo/metrics/show.py

@@ -47,7 +47,7 @@ def _collect_metrics(repo, targets, recursive):
 
 
 def _extract_metrics(metrics, path, rev):
-    if isinstance(metrics, (int, float)):
+    if isinstance(metrics, (int, float, str)):
         return metrics
 
     if not isinstance(metrics, dict):

dvc/repo/open_repo.py

@@ -3,7 +3,7 @@ import os
 import tempfile
 import threading
 from contextlib import contextmanager
-from typing import TYPE_CHECKING, Collection, Dict, Iterator, Optional, Tuple
+from typing import TYPE_CHECKING, Dict, Iterator, Optional, Tuple
 
 from funcy import retry, wrap_with
 
@@ -13,21 +13,17 @@ from dvc.scm import CloneError, map_scm_exception
 from dvc.utils import relpath
 
 if TYPE_CHECKING:
-    from dvc.types import StrPath
-
-    from .scm import Git
+    from dvc.scm import Git
 
 logger = logging.getLogger(__name__)
 
 
 @contextmanager
 @map_scm_exception()
-def external_repo(
+def _external_repo(
     url,
     rev: Optional[str] = None,
     for_write: bool = False,
-    cache_dir: Optional["StrPath"] = None,
-    cache_types: Optional[Collection[str]] = None,
     **kwargs,
 ) -> Iterator["Repo"]:
     logger.debug("Creating external repo %s@%s", url, rev)
@@ -37,12 +33,8 @@ def external_repo(
     # the tip of the default branch
     rev = rev or "refs/remotes/origin/HEAD"
 
-    cache_config = {
-        "cache": {"dir": cache_dir or _get_cache_dir(url), "type": cache_types}
-    }
-
     config = _get_remote_config(url) if os.path.isdir(url) else {}
-    config.update(cache_config)
+    config.update({"cache": {"dir": _get_cache_dir(url)}})
     config.update(kwargs.pop("config", None) or {})
 
     main_root = "/"
@@ -55,7 +47,7 @@ def external_repo(
         root_dir=path,
         url=url,
         config=config,
-        repo_factory=erepo_factory(url, main_root, cache_config),
+        repo_factory=erepo_factory(url, main_root, {"cache": config["cache"]}),
         rev=rev,
         **kwargs,
     )
@@ -70,6 +62,22 @@ def external_repo(
             _remove(path)
 
 
+def open_repo(url, *args, **kwargs):
+    if url is None:
+        url = os.getcwd()
+
+    if os.path.exists(url):
+        try:
+            config = _get_remote_config(url)
+            config.update(kwargs.get("config") or {})
+            kwargs["config"] = config
+            return Repo(url, *args, **kwargs)
+        except NotDvcRepoError:
+            pass  # fallthrough to _external_repo
+
+    return _external_repo(url, *args, **kwargs)
+
+
 def erepo_factory(url, root_dir, cache_config):
     from dvc.fs import localfs
 

dvc/repo/reproduce.py

@@ -82,7 +82,6 @@ def reproduce(  # noqa: C901, PLR0912
     recursive=False,
     pipeline=False,
     all_pipelines=False,
-    after_repro_callback=None,
     **kwargs,
 ):
     from .graph import get_pipeline, get_pipelines
@@ -126,10 +125,11 @@ def reproduce(  # noqa: C901, PLR0912
                 )
             )
 
-    result = _reproduce_stages(self.index.graph, list(stages), **kwargs)
-    if callable(after_repro_callback):
-        after_repro_callback()
-    return result
+    if kwargs.get("pull", False):
+        logger.debug("Pulling run cache")
+        self.stage_cache.pull(None)
+
+    return _reproduce_stages(self.index.graph, list(stages), **kwargs)
 
 
 def _reproduce_stages(  # noqa: C901

dvc/repo/worktree.py

@@ -173,7 +173,7 @@ def push_worktree(
         ) as cb:
             cb.set_size(total)
             try:
-                pushed += checkout(
+                stats = checkout(
                     new_index,
                     remote_obj.path,
                     remote_obj.fs,
@@ -184,6 +184,7 @@ def push_worktree(
                     jobs=jobs,
                     **diff_kwargs,
                 )
+                pushed += sum(len(changes) for changes in stats.values())
             except VersioningNotSupported:
                 logger.exception("")
                 raise DvcException(

dvc/schema.py

@@ -116,7 +116,6 @@ PLOT_DEFINITION = {
 SINGLE_PLOT_SCHEMA = {str: Any(PLOT_DEFINITION, None)}
 ARTIFACTS = "artifacts"
 SINGLE_ARTIFACT_SCHEMA = Schema({str: ARTIFACT_SCHEMA})
-ARTIFACTS_SCHEMA = Any(str, SINGLE_ARTIFACT_SCHEMA)
 FOREACH_IN = {
     Required(FOREACH_KWD): Any(dict, list, str),
     Required(DO_KWD): STAGE_DEFINITION,
@@ -130,7 +129,7 @@ MULTI_STAGE_SCHEMA = {
     VARS_KWD: VARS_SCHEMA,
     StageParams.PARAM_PARAMS: [str],
     StageParams.PARAM_METRICS: [str],
-    ARTIFACTS: ARTIFACTS_SCHEMA,
+    ARTIFACTS: SINGLE_ARTIFACT_SCHEMA,
 }
 
 COMPILED_SINGLE_STAGE_SCHEMA = Schema(SINGLE_STAGE_SCHEMA)

dvc/stage/__init__.py

@@ -319,20 +319,22 @@ class Stage(params.StageParams):
             env.update(current)
         return env
 
-    def changed_deps(self) -> bool:
+    def changed_deps(self, allow_missing: bool = False) -> bool:
         if self.frozen:
             return False
 
         if self.is_callback or self.always_changed or self.is_checkpoint:
             return True
 
-        return self._changed_deps()
+        return self._changed_deps(allow_missing=allow_missing)
 
     @rwlocked(read=["deps"])
-    def _changed_deps(self) -> bool:
+    def _changed_deps(self, allow_missing: bool = False) -> bool:
         for dep in self.deps:
             status = dep.status()
             if status:
+                if allow_missing and status[str(dep)] == "deleted":
+                    continue
                 logger.debug(
                     "Dependency '%s' of %s changed because it is '%s'.",
                     dep,
@@ -343,10 +345,12 @@ class Stage(params.StageParams):
         return False
 
     @rwlocked(read=["outs"])
-    def changed_outs(self) -> bool:
+    def changed_outs(self, allow_missing: bool = False) -> bool:
         for out in self.outs:
             status = out.status()
             if status:
+                if allow_missing and status[str(out)] == "not in cache":
+                    continue
                 logger.debug(
                     "Output '%s' of %s changed because it is '%s'.",
                     out,
@@ -364,13 +368,13 @@ class Stage(params.StageParams):
         return changed
 
     @rwlocked(read=["deps", "outs"])
-    def changed(self) -> bool:
+    def changed(self, allow_missing: bool = False) -> bool:
         is_changed = (
             # Short-circuit order: stage md5 is fast,
             # deps are expected to change
             self.changed_stage()
-            or self.changed_deps()
-            or self.changed_outs()
+            or self.changed_deps(allow_missing=allow_missing)
+            or self.changed_outs(allow_missing=allow_missing)
         )
         if is_changed:
             logger.debug("%s changed.", self)
@@ -416,12 +420,15 @@ class Stage(params.StageParams):
         (out,) = self.outs
         out.transfer(source, odb=odb, jobs=kwargs.get("jobs"))
         if not to_remote:
-            out.checkout()
+            out.checkout(force=kwargs.get("force"))
             out.ignore()
 
     @rwlocked(read=["deps"], write=["outs"])
     def reproduce(self, interactive=False, **kwargs) -> Optional["Stage"]:
-        if not (kwargs.get("force", False) or self.changed()):
+        if not (
+            kwargs.get("force", False)
+            or self.changed(kwargs.get("allow_missing", False))
+        ):
             if not isinstance(self, PipelineStage) and self.is_data_source:
                 logger.info("'%s' didn't change, skipping", self.addressing)
             else:
@@ -495,34 +502,34 @@ class Stage(params.StageParams):
                 if not allow_missing:
                     raise
 
-    def save_outs(self, allow_missing: bool = False):
-        from dvc.output import OutputDoesNotExistError
-
+    def get_versioned_outs(self) -> Dict[str, "Output"]:
         from .exceptions import StageFileDoesNotExistError, StageNotFound
 
         try:
             old = self.reload()
-            old_outs = {out.def_path: out for out in old.outs}
-            merge_versioned = any(
-                (
-                    out.files is not None
-                    or (out.meta is not None and out.meta.version_id is not None)
-                )
-                for out in old_outs.values()
-            )
         except (StageFileDoesNotExistError, StageNotFound):
-            merge_versioned = False
+            return {}
 
+        return {
+            out.def_path: out
+            for out in old.outs
+            if out.files is not None
+            or (out.meta is not None and out.meta.version_id is not None)
+        }
+
+    def save_outs(self, allow_missing: bool = False):
+        from dvc.output import OutputDoesNotExistError
+
+        old_versioned_outs = self.get_versioned_outs()
         for out in self.outs:
             try:
                 out.save()
             except OutputDoesNotExistError:
                 if not (allow_missing or out.checkpoint):
                     raise
-            if merge_versioned:
-                old_out = old_outs.get(out.def_path)
-                if old_out is not None:
-                    out.merge_version_meta(old_out)
+
+            if old_out := old_versioned_outs.get(out.def_path):
+                out.merge_version_meta(old_out)
 
     def ignore_outs(self) -> None:
         for out in self.outs:
@@ -560,8 +567,31 @@ class Stage(params.StageParams):
         if link_failures:
             raise CacheLinkError(link_failures)
 
+    @rwlocked(write=["outs"])
+    def add_outs(  # noqa: C901
+        self, filter_info=None, allow_missing: bool = False, **kwargs
+    ):
+        from dvc.output import OutputDoesNotExistError
+
+        link_failures = []
+        old_versioned_outs = self.get_versioned_outs()
+        for out in self.filter_outs(filter_info):
+            try:
+                out.add(filter_info, **kwargs)
+            except (FileNotFoundError, OutputDoesNotExistError):
+                if not (allow_missing or out.checkpoint):
+                    raise
+            except CacheLinkError:
+                link_failures.append(filter_info or out.fs_path)
+
+            if old_out := old_versioned_outs.get(out.def_path):
+                out.merge_version_meta(old_out)
+
+        if link_failures:
+            raise CacheLinkError(link_failures)
+
     @rwlocked(read=["deps", "outs"])
-    def run(
+    def run(  # noqa: C901
         self,
         dry=False,
         no_commit=False,
@@ -573,15 +603,20 @@ class Stage(params.StageParams):
         if (self.cmd or self.is_import) and not self.frozen and not dry:
             self.remove_outs(ignore_remove=False, force=False)
 
-        if (not self.frozen and self.is_import) or self.is_partial_import:
+        if (
+            self.is_import and (not self.frozen or kwargs.get("pull"))
+        ) or self.is_partial_import:
             self._sync_import(dry, force, kwargs.get("jobs", None), no_download)
         elif not self.frozen and self.cmd:
-            self._run_stage(dry, force, **kwargs)
-        else:
+            self._run_stage(dry, force, allow_missing=allow_missing, **kwargs)
+        elif kwargs.get("pull"):
+            logger.info("Pulling data for %s", self)
+            self.repo.pull(self.addressing, jobs=kwargs.get("jobs", None))
+            self.checkout()
+        elif not dry:
             args = ("outputs", "frozen ") if self.frozen else ("data sources", "")
             logger.info("Verifying %s in %s%s", *args, self)
-            if not dry:
-                self._check_missing_outputs()
+            self._check_missing_outputs()
 
         if not dry:
             if kwargs.get("checkpoint_func", None) or no_download:

dvc/stage/run.py

@@ -133,10 +133,25 @@ def cmd_run(stage, dry=False, checkpoint_func=None, run_env=None):
         _run(stage, executable, cmd, checkpoint_func=checkpoint_func, **kwargs)
 
 
+def _pull_missing_deps(stage):
+    for dep in stage.deps:
+        if not dep.exists:
+            stage.repo.pull(dep.def_path)
+
+
 def run_stage(
-    stage, dry=False, force=False, checkpoint_func=None, run_env=None, **kwargs
+    stage,
+    dry=False,
+    force=False,
+    checkpoint_func=None,
+    run_env=None,
+    allow_missing: bool = False,
+    **kwargs,
 ):
     if not (force or checkpoint_func):
+        if allow_missing and kwargs.get("pull") and not dry:
+            _pull_missing_deps(stage)
+
         from .cache import RunCacheNotFoundError
 
         try:

dvc/testing/README.rst

@@ -0,0 +1,19 @@
+DVC pytest plugin
+
+dvc.testing.benchmarks
+======================
+
+Benchmark test definitions are now part of ``dvc.testing``.
+For CLI usage and `bench.dvc.org <https://bench.dvc.org>`_ details see `dvc-bench <https://github.com/iterative/dvc-bench>`_.
+
+``dvc.testing.benchmarks`` structure:
+
+* cli: should be able to run these with any dvc (rpm, deb, pypi, snap, etc) (could be used in dvc-test repo too)
+
+  * commands: granular tests for individual commands. These should have a cached setup, so that we could use them during rapid development instead of our hand-written scripts. Every test could be run in a separate machine.
+  * stories: multistage start-to-end benchmarks, useful for testing workflows (e.g. in documentation, see test_sharing inspired by `Storing and sharing <https://dvc.org/doc/start/data-management/data-versioning#storing-and-sharing>`_. Every full story could be run in a separate machine.
+
+* api: for python api only.
+
+  * methods: granular tests for individual methods (e.g. ``api.open/read``). Same reasoning as in ``cli.commands``
+  * stories: same as ``cli.stories`` but for our api. E.g. imagine using our api with pandas or smth like that.

dvc/testing/benchmarks/cli/commands/test_add.py

@@ -0,0 +1,3 @@
+# pylint: disable=unused-argument
+def test_add(bench_dvc, tmp_dir, dvc, dataset):
+    bench_dvc("add", dataset)

dvc/testing/benchmarks/cli/commands/test_checkout.py

@@ -0,0 +1,39 @@
+# pylint: disable=unused-argument
+import os
+
+import pytest
+
+from dvc.fs import localfs
+from dvc_objects.fs import generic
+from dvc_objects.fs.utils import tmp_fname
+
+
+def _skip_unsupported_link(src, dest, link_type):
+    src_test_file = os.path.join(src, tmp_fname())
+    dest_test_file = os.path.join(dest, tmp_fname())
+    if not generic.test_links(
+        [link_type], localfs, src_test_file, localfs, dest_test_file
+    ):
+        pytest.skip(f"{link_type} not supported")
+
+
+def generate_test(*, link_type="copy"):
+    def _test_checkout_func(bench_dvc, tmp_dir, dvc, make_dataset):
+        dataset = make_dataset(dvcfile=True, cache=True, files=False)
+
+        _skip_unsupported_link((tmp_dir / ".dvc" / "cache"), tmp_dir, link_type)
+
+        with dvc.config.edit() as conf:
+            conf["cache"]["type"] = link_type
+
+        bench_dvc("checkout", dataset)
+        bench_dvc("checkout", name="noop")
+        (dataset / "new").write_text("new")
+        bench_dvc("checkout", "--force", name="update")
+
+    return _test_checkout_func
+
+
+test_checkout_copy = generate_test(link_type="copy")
+test_checkout_symlink = generate_test(link_type="symlink")
+test_checkout_hardlink = generate_test(link_type="hardlink")

dvc/testing/benchmarks/cli/commands/test_data_status.py

@@ -0,0 +1,49 @@
+# pylint: disable=unused-argument
+from shutil import rmtree
+
+import pytest
+
+pytestmark = pytest.mark.requires(minversion=(2, 15, 0), reason="new command")
+
+
+def test_data_status(
+    bench_dvc,
+    tmp_dir,
+    scm,
+    dvc,
+    make_dataset,
+):
+    args = ("data", "status")
+    dataset = make_dataset(cache=True, files=True, dvcfile=True, commit=False)
+    rmtree(dvc.tmp_dir)
+
+    bench_dvc(*args, name="new")
+    bench_dvc(*args, name="noop")
+
+    tmp_dir.scm_add(dataset.with_suffix(".dvc").name, commit="add dataset")
+
+    (dataset / "new").write_text("new")
+    bench_dvc(*args, name="changed")
+    bench_dvc(*args, name="changed-noop")
+
+
+def test_data_status_all_flags(bench_dvc, tmp_dir, scm, dvc, make_dataset):
+    args = (
+        "data",
+        "status",
+        "--granular",
+        "--unchanged",
+        "--untracked-files",
+        "--json",
+    )
+    dataset = make_dataset(cache=True, files=True, dvcfile=True, commit=False)
+    rmtree(dvc.tmp_dir)
+
+    bench_dvc(*args, name="new")
+    bench_dvc(*args, name="noop")
+
+    tmp_dir.scm_add(dataset.with_suffix(".dvc").name, commit="add dataset")
+
+    (dataset / "new").write_text("new")
+    bench_dvc(*args, name="changed")
+    bench_dvc(*args, name="changed-noop")

dvc/testing/benchmarks/cli/commands/test_diff.py

@@ -0,0 +1,9 @@
+# pylint: disable=unused-argument
+def test_diff(bench_dvc, tmp_dir, scm, dvc, make_dataset):
+    dataset = make_dataset(cache=True, files=True, dvcfile=True, commit=True)
+    bench_dvc("diff")
+    bench_dvc("diff", name="noop")
+
+    (dataset / "new").write_text("new")
+    bench_dvc("diff", name="changed")
+    bench_dvc("diff", name="changed-noop")

dvc/testing/benchmarks/cli/commands/test_exp_show.py

@@ -0,0 +1,14 @@
+# pylint: disable=unused-argument
+import pytest
+
+
+@pytest.mark.requires(minversion=(2, 10, 0))
+def test_exp_show(make_project, monkeypatch, bench_dvc, dvc_bin):
+    url = "https://github.com/efiop/lstm_seq2seq"
+    rev = "dvc-bench"
+    path = make_project(url, rev=rev)
+    monkeypatch.chdir(path)
+
+    dvc_bin("exp", "pull", "-A", "--no-cache", "origin")
+
+    bench_dvc("exp", "show", "-A", "--no-pager")

dvc/testing/benchmarks/cli/commands/test_gc.py

@@ -0,0 +1,4 @@
+# pylint: disable=unused-argument
+def test_gc(bench_dvc, tmp_dir, dvc, make_dataset):
+    make_dataset(files=False, cache=True, dvcfile=False)
+    bench_dvc("gc", "-f", "-w")

dvc/testing/benchmarks/cli/commands/test_get.py

@@ -0,0 +1,6 @@
+# pylint: disable=unused-argument
+def test_get(bench_dvc, tmp_dir, scm, dvc, make_dataset, remote):
+    dataset = make_dataset(
+        cache=False, files=False, dvcfile=True, commit=True, remote=True
+    )
+    bench_dvc("get", tmp_dir, dataset.name, "-o", "new")

dvc/testing/benchmarks/cli/commands/test_get_url.py

@@ -0,0 +1,6 @@
+# pylint: disable=unused-argument
+def test_get_url(bench_dvc, tmp_dir, scm, dvc, make_dataset):
+    dataset = make_dataset(
+        cache=False, files=True, dvcfile=False, commit=False, remote=False
+    )
+    bench_dvc("get-url", str(dataset), "new")

dvc/testing/benchmarks/cli/commands/test_help.py

@@ -0,0 +1,2 @@
+def test_help(bench_dvc):
+    bench_dvc("--help", rounds=100)

dvc/testing/benchmarks/cli/commands/test_import.py

@@ -0,0 +1,6 @@
+# pylint: disable=unused-argument
+def test_import(bench_dvc, tmp_dir, scm, dvc, make_dataset, remote):
+    dataset = make_dataset(
+        cache=False, files=False, dvcfile=True, commit=True, remote=True
+    )
+    bench_dvc("import", tmp_dir, dataset.name, "-o", "new")

dvc/testing/benchmarks/cli/commands/test_import_url.py

@@ -0,0 +1,6 @@
+# pylint: disable=unused-argument
+def test_import_url(bench_dvc, tmp_dir, scm, dvc, make_dataset):
+    dataset = make_dataset(
+        cache=False, files=True, dvcfile=False, commit=False, remote=False
+    )
+    bench_dvc("import-url", str(dataset), "new")

dvc/testing/benchmarks/cli/commands/test_init.py

@@ -0,0 +1,14 @@
+# pylint: disable=unused-argument
+import shutil
+
+
+def test_init(bench_dvc, tmp_dir, scm):
+    def _cleanup_dir():
+        for item in tmp_dir.iterdir():
+            if item.is_dir():
+                if item.name != ".git":
+                    shutil.rmtree(item)
+            else:
+                item.unlink()
+
+    bench_dvc("init", setup=_cleanup_dir, rounds=100, warmup_rounds=1)

dvc/testing/benchmarks/cli/commands/test_ls.py

@@ -0,0 +1,7 @@
+# pylint: disable=unused-argument
+def test_list(bench_dvc, tmp_dir, scm, dvc, make_dataset, remote):
+    make_dataset(cache=False, files=False, dvcfile=True, commit=True, remote=True)
+    bench_dvc("list", tmp_dir)
+    bench_dvc("list", tmp_dir, "--dvc-only", name="dvc-only")
+    bench_dvc("list", tmp_dir, "--recursive", name="recursive")
+    bench_dvc("list", tmp_dir, "dataset", name="shallow")

dvc/testing/benchmarks/cli/commands/test_plots.py

@@ -0,0 +1,20 @@
+import pytest
+
+from dvc.repo import Repo
+from dvc.testing.benchmarks.fixtures import _pull
+
+# pylint: disable=redefined-outer-name,unused-argument
+
+# pylint: disable=unused-argument
+
+
+@pytest.mark.requires(minversion=(2, 34, 0), reason="top-level plots not supported")
+def test_plots(project, bench_dvc):
+    with Repo() as dvc:
+        _pull(dvc)
+
+    kwargs = {"rounds": 5, "iterations": 3, "warmup_rounds": 2}
+    bench_dvc("plots", "show", name="show", **kwargs)
+    bench_dvc("plots", "show", "--json", name="show-json", **kwargs)
+    bench_dvc("plots", "diff", "HEAD", name="diff", **kwargs)
+    bench_dvc("plots", "diff", "HEAD", "--json", name="diff-json", **kwargs)

dvc/testing/benchmarks/cli/commands/test_pull.py

@@ -0,0 +1,4 @@
+# pylint: disable=unused-argument
+def test_pull(bench_dvc, tmp_dir, dvc, make_dataset, remote):
+    make_dataset(cache=False, dvcfile=True, files=False, remote=True)
+    bench_dvc("pull")

dvc/testing/benchmarks/cli/commands/test_push.py

@@ -0,0 +1,4 @@
+# pylint: disable=unused-argument
+def test_push(bench_dvc, tmp_dir, dvc, make_dataset, remote):
+    make_dataset(cache=True, dvcfile=True, files=False)
+    bench_dvc("push")

dvc/testing/benchmarks/cli/commands/test_status.py

@@ -0,0 +1,9 @@
+# pylint: disable=unused-argument
+def test_status(bench_dvc, tmp_dir, dvc, make_dataset):
+    dataset = make_dataset(files=True, dvcfile=True, cache=True)
+    bench_dvc("status")
+    bench_dvc("status", name="noop")
+
+    (dataset / "new").write_text("new")
+    bench_dvc("status", name="changed")
+    bench_dvc("status", name="changed-noop")

dvc/testing/benchmarks/cli/commands/test_update.py

@@ -0,0 +1,9 @@
+# pylint: disable=unused-argument
+def test_update(bench_dvc, tmp_dir, dvc, make_dataset):
+    dataset = make_dataset(
+        cache=False, files=True, dvcfile=False, commit=False, remote=False
+    )
+    dvc.imp_url(str(dataset), "new")
+    (dataset / "new").write_text("new")
+    bench_dvc("update", "new.dvc")
+    bench_dvc("update", "new.dvc", name="noop")

dvc/testing/benchmarks/cli/stories/test_modify_data.py

@@ -0,0 +1,57 @@
+"""
+Tests for modifications to an existing dataset.
+"""
+import glob
+import os
+import random
+import shutil
+import sys
+
+import pytest
+
+# pylint: disable=unused-argument,unexpected-keyword-arg
+
+
+@pytest.mark.skipif(sys.version_info < (3, 10), reason="requires 3.10 glob.glob")
+def test_partial_add(bench_dvc, tmp_dir, dvc, dataset, remote):
+    random.seed(4231)
+    # Move some files to create a partial dataset
+    os.makedirs("partial-copy")
+    for f in glob.glob("*", root_dir=dataset, recursive=True):  # type: ignore[call-arg]
+        if random.random() > 0.5:  # noqa: S311 # nosec
+            shutil.move(dataset / f, tmp_dir / "partial-copy" / f)
+
+    # Add/push partial dataset
+    dvc.add(str(dataset))
+    dvc.push()
+
+    # Add more files to the dataset
+    shutil.copytree("partial-copy", dataset, dirs_exist_ok=True)
+
+    # Benchmark operations for adding files to a dataset
+    bench_dvc("add", dataset, name="partial-add")
+    bench_dvc("push", name="partial-add")
+    bench_dvc("gc", "-f", "-w", name="noop")
+    bench_dvc("gc", "-f", "-w", "-c", name="cloud-noop")
+
+
+@pytest.mark.skipif(sys.version_info < (3, 10), reason="requires 3.10 glob.glob")
+def test_partial_remove(bench_dvc, tmp_dir, dvc, dataset, remote):
+    random.seed(5232)
+    # Add/push full dataset
+    dvc.add(str(dataset))
+    dvc.push()
+
+    # Remove some files
+    for f in glob.glob("*", root_dir=dataset, recursive=True):  # type: ignore[call-arg]
+        if random.random() > 0.5:  # noqa: S311 # nosec
+            if os.path.isfile(dataset / f):
+                os.remove(dataset / f)
+            elif os.path.isdir(dataset / f):
+                shutil.rmtree(dataset / f)
+
+    # Benchmark operations for removing files from dataset
+    bench_dvc("add", dataset)
+    bench_dvc("push")
+    bench_dvc("gc", "-f", "-w")
+    bench_dvc("gc", "-f", "-w", "-c", name="cloud")

dvc/testing/benchmarks/cli/stories/use_cases/test_sharing.py

@@ -0,0 +1,19 @@
+import shutil
+
+# pylint: disable=unused-argument
+
+
+def test_sharing(bench_dvc, tmp_dir, dvc, dataset, remote):
+    bench_dvc("add", dataset)
+    bench_dvc("add", dataset, name="noop")
+
+    bench_dvc("push")
+    bench_dvc("push", name="noop")
+
+    shutil.rmtree(dataset)
+    shutil.rmtree(tmp_dir / ".dvc" / "cache")
+
+    bench_dvc("pull")
+    bench_dvc("pull", name="noop")
+
+    bench_dvc("checkout", name="noop")

dvc/testing/benchmarks/conftest.py

@@ -0,0 +1 @@
+from .fixtures import *  # noqa, pylint: disable=wildcard-import

dvc/testing/benchmarks/fixtures.py

@@ -0,0 +1,242 @@
+import os
+import shutil
+from pathlib import Path
+from subprocess import check_output  # nosec B404
+
+import pytest
+from dulwich.porcelain import clone
+from funcy import first
+from packaging import version
+
+# pylint: disable=redefined-outer-name,unused-argument
+
+
+@pytest.fixture(scope="session")
+def bench_config(request):
+    return request.config.bench_config
+
+
+@pytest.fixture(scope="session")
+def make_dvc_venv(tmp_path_factory):
+    def _make_dvc_venv(name):
+        from pytest_virtualenv import VirtualEnv
+
+        name = _sanitize_venv_name(name)
+        venv_dir = tmp_path_factory.mktemp(f"dvc-venv-{name}")
+        return VirtualEnv(workspace=venv_dir)
+
+    return _make_dvc_venv
+
+
+def _sanitize_venv_name(name):
+    return name.replace("/", "-").replace("\\", "-")
+
+
+@pytest.fixture(scope="session")
+def dvc_venvs():
+    return {}
+
+
+@pytest.fixture(scope="session")
+def dvc_git_repo(tmp_path_factory, bench_config):
+    url = bench_config.dvc_git_repo
+
+    if os.path.isdir(url):
+        return url
+
+    tmp_path = tmp_path_factory.mktemp("dvc-git-repo")
+    clone(url, os.fspath(tmp_path))
+
+    return tmp_path
+
+
+@pytest.fixture(scope="session")
+def dvc_bench_git_repo(tmp_path_factory, bench_config):
+    url = bench_config.dvc_bench_git_repo
+
+    if os.path.isdir(url):
+        return Path(url)
+
+    tmp_path = tmp_path_factory.mktemp("dvc-bench-git-repo")
+    clone(url, os.fspath(tmp_path))
+
+    return tmp_path
+
+
+@pytest.fixture(scope="session")
+def make_dvc_bin(
+    dvc_rev,
+    dvc_venvs,
+    make_dvc_venv,
+    dvc_git_repo,
+    bench_config,
+    request,
+):
+    if dvc_rev:
+        venv = dvc_venvs.get(dvc_rev)
+        if not venv:
+            venv = make_dvc_venv(dvc_rev)
+            venv.run("pip install -U pip")
+            if bench_config.dvc_install_deps:
+                egg = f"dvc[{bench_config.dvc_install_deps}]"
+            else:
+                egg = "dvc"
+            venv.run(f"pip install git+file://{dvc_git_repo}@{dvc_rev}#egg={egg}")
+            if dvc_rev in ["2.18.1", "2.11.0", "2.6.3"]:
+                venv.run("pip install fsspec==2022.11.0")
+            dvc_venvs[dvc_rev] = venv
+        dvc_bin = venv.virtualenv / "bin" / "dvc"
+    else:
+        dvc_bin = bench_config.dvc_bin
+
+    def _dvc_bin(*args):
+        return check_output([dvc_bin, *args], text=True)  # nosec B603
+
+    _dvc_bin.version = parse_tuple(_dvc_bin("--version"))  # type: ignore[attr-defined]
+    return _dvc_bin
+
+
+def parse_tuple(version_string):
+    from packaging.version import Version
+
+    parsed = version.parse(version_string)
+    assert isinstance(parsed, Version)
+    return (parsed.major, parsed.minor, parsed.micro)
+
+
+@pytest.fixture
+def dvc_bin(request, make_dvc_bin):
+    if marker := request.node.get_closest_marker("requires"):
+        minversion = marker.kwargs.get("minversion") or first(marker.args)
+        assert minversion, (
+            "'minversion' needs to be specified as"
+            " a positional or a keyword argument"
+        )
+        reason = marker.kwargs.get("reason", "")
+        if isinstance(minversion, str):
+            minversion = parse_tuple(minversion)
+        if make_dvc_bin.version < minversion:
+            version_repr = ".".join(map(str, minversion))
+            pytest.skip(f"requires dvc>={version_repr}: {reason}")
+    return make_dvc_bin
+
+
+@pytest.fixture
+def make_bench(request):
+    def _make_bench(name):
+        import pytest_benchmark.plugin
+
+        # hack from https://github.com/ionelmc/pytest-benchmark/issues/166
+        bench = pytest_benchmark.plugin.benchmark.__pytest_wrapped__.obj(request)
+
+        suffix = f"-{name}"
+
+        def add_suffix(_name):
+            start, sep, end = _name.partition("[")
+            return start + suffix + sep + end
+
+        bench.name = add_suffix(bench.name)
+        bench.fullname = add_suffix(bench.fullname)
+
+        return bench
+
+    return _make_bench
+
+
+@pytest.fixture
+def bench_dvc(dvc_bin, make_bench):
+    def _bench_dvc(*args, **kwargs):
+        name = kwargs.pop("name", None)
+        name = f"-{name}" if name else ""
+        bench = make_bench(args[0] + name)
+        return bench.pedantic(dvc_bin, args=args, **kwargs)
+
+    return _bench_dvc
+
+
+def _pull(repo, *args):
+    from dvc.exceptions import CheckoutError, DownloadError
+
+    while True:
+        try:
+            return repo.pull(*args)
+        except (CheckoutError, DownloadError):
+            pass
+
+
+@pytest.fixture
+def make_dataset(request, bench_config, tmp_dir, dvc_bench_git_repo):
+    def _make_dataset(
+        dvcfile=False, files=True, cache=False, commit=False, remote=False
+    ):
+        from dvc.repo import Repo
+
+        path = tmp_dir / "dataset"
+        root = dvc_bench_git_repo
+        src = root / "data" / bench_config.size / "dataset"
+        src_dvc = src.with_suffix(".dvc")
+
+        dvc = Repo(root)
+
+        _pull(dvc, [str(src_dvc)])
+        if files:
+            shutil.copytree(src, path)
+        if dvcfile:
+            shutil.copy(src.with_suffix(".dvc"), path.with_suffix(".dvc"))
+        if cache:
+            shutil.copytree(root / ".dvc" / "cache", tmp_dir / ".dvc" / "cache")
+        if remote:
+            assert dvcfile
+            assert not cache
+            assert tmp_dir.dvc
+            # FIXME temporary hack, we should try to push from home repo
+            # directly to this remote instead
+            shutil.copytree(root / ".dvc" / "cache", tmp_dir / ".dvc" / "cache")
+            tmp_dir.dvc.push([str(path.with_suffix(".dvc").relative_to(tmp_dir))])
+            shutil.rmtree(tmp_dir / ".dvc" / "cache")
+        if commit:
+            assert dvcfile
+            assert tmp_dir.scm
+            tmp_dir.scm.add([str(path.with_suffix(".dvc").relative_to(tmp_dir))])
+            tmp_dir.scm.commit("add dataset")
+        return path
+
+    return _make_dataset
+
+
+@pytest.fixture
+def dataset(make_dataset):
+    return make_dataset(dvcfile=False, files=True, cache=False)
+
+
+@pytest.fixture
+def remote_dataset():
+    pytest.skip("fixme")
+
+
+@pytest.fixture
+def make_project(tmp_path_factory):
+    def _make_project(url, rev=None):
+        path = os.fspath(tmp_path_factory.mktemp("dvc-project"))
+
+        if rev:
+            rev = rev.encode("ascii")
+
+        clone(url, path, branch=rev)
+        return path
+
+    return _make_project
+
+
+@pytest.fixture
+def project(bench_config, monkeypatch, make_project):
+    rev = bench_config.project_rev
+    url = bench_config.project_git_repo
+
+    if os.path.isdir(url):
+        path = url
+        assert not rev
+    else:
+        path = make_project(url, rev=rev)
+
+    monkeypatch.chdir(path)

dvc/testing/benchmarks/plugin.py

@@ -0,0 +1,95 @@
+DEFAULT_DVC_BIN = "dvc"
+DEFAULT_DVC_GIT_REPO = "https://github.com/iterative/dvc"
+DEFAULT_DVC_BENCH_GIT_REPO = "https://github.com/iterative/dvc-bench"
+DEFAULT_PROJECT_GIT_REPO = "https://github.com/iterative/example-get-started"
+
+
+def pytest_generate_tests(metafunc):
+    str_revs = metafunc.config.getoption("--dvc-revs")
+    revs = str_revs.split(",") if str_revs else [None]
+    if "dvc_rev" in metafunc.fixturenames:
+        metafunc.parametrize("dvc_rev", revs, scope="session")
+
+
+class DVCBenchConfig:
+    def __init__(self):
+        self.size = "small"
+        self.dvc_bin = DEFAULT_DVC_BIN
+        self.dvc_revs = None
+        self.dvc_git_repo = DEFAULT_DVC_GIT_REPO
+        self.dvc_install_deps = None
+        self.dvc_bench_git_repo = DEFAULT_DVC_BENCH_GIT_REPO
+        self.project_rev = None
+        self.project_git_repo = DEFAULT_PROJECT_GIT_REPO
+
+
+def pytest_configure(config):
+    config.addinivalue_line(
+        "markers", "requires(minversion): mark a test as requiring minimum DVC version"
+    )
+
+    config.bench_config = DVCBenchConfig()
+    config.bench_config.size = config.getoption("--size")
+    config.bench_config.dvc_bin = config.getoption("--dvc-bin")
+    config.bench_config.dvc_revs = config.getoption("--dvc-revs")
+    config.bench_config.dvc_git_repo = config.getoption("--dvc-git-repo")
+    config.bench_config.dvc_install_deps = config.getoption("--dvc-install-deps")
+    config.bench_config.dvc_bench_git_repo = config.getoption("--dvc-bench-git-repo")
+    config.bench_config.project_rev = config.getoption("--project-rev")
+    config.bench_config.project_git_repo = config.getoption("--project-git-repo")
+
+
+def pytest_addoption(parser):
+    parser.addoption(
+        "--size",
+        choices=["tiny", "small", "large", "mnist"],
+        default="small",
+        help="Size of the dataset/datafile to use in tests",
+    )
+
+    parser.addoption(
+        "--dvc-bin",
+        type=str,
+        default=DEFAULT_DVC_BIN,
+        help="Path to dvc binary",
+    )
+
+    parser.addoption(
+        "--dvc-revs",
+        type=str,
+        help=("Comma-separated list of DVC revisions to test (overrides `--dvc-bin`)"),
+    )
+
+    parser.addoption(
+        "--dvc-git-repo",
+        type=str,
+        default=DEFAULT_DVC_GIT_REPO,
+        help="Path or url to dvc git repo",
+    )
+
+    parser.addoption(
+        "--dvc-install-deps",
+        type=str,
+        default="",
+        help="Comma-separated list of DVC installation packages",
+    )
+
+    parser.addoption(
+        "--dvc-bench-git-repo",
+        type=str,
+        default=DEFAULT_DVC_BENCH_GIT_REPO,
+        help="Path or url to dvc-bench git repo (for loading benchmark datasets)",
+    )
+
+    parser.addoption(
+        "--project-rev",
+        type=str,
+        help="Project revision to test",
+    )
+
+    parser.addoption(
+        "--project-git-repo",
+        type=str,
+        default=DEFAULT_PROJECT_GIT_REPO,
+        help="Path or url to dvc project",
+    )

dvc/testing/plugin.py

@@ -0,0 +1,20 @@
+from .benchmarks.fixtures import *  # noqa, pylint: disable=wildcard-import
+from .fixtures import *  # noqa, pylint: disable=wildcard-import
+
+
+def pytest_generate_tests(metafunc):
+    from .benchmarks.plugin import pytest_generate_tests as bench_generate_tests
+
+    bench_generate_tests(metafunc)
+
+
+def pytest_addoption(parser):
+    from .benchmarks.plugin import pytest_addoption as bench_addoption
+
+    bench_addoption(parser)
+
+
+def pytest_configure(config):
+    from .benchmarks.plugin import pytest_configure as bench_configure
+
+    bench_configure(config)

dvc/ui/table.py

@@ -14,7 +14,7 @@ if TYPE_CHECKING:
 SHOW_MAX_WIDTH = 1024
 
 
-CellT = Union[str, "RichText"]  # RichText is mostly compatible with str
+CellT = Union[str, "RichText", None]  # RichText is mostly compatible with str
 Row = Sequence[CellT]
 TableData = Sequence[Row]
 Headers = Sequence[str]

dvc/utils/__init__.py

@@ -255,10 +255,11 @@ def env2bool(var, undefined=False):
     return bool(re.search("1|y|yes|true", var, flags=re.I))
 
 
-def resolve_output(inp, out):
-    import errno
+def resolve_output(inp, out, force=False):
     from urllib.parse import urlparse
 
+    from dvc.exceptions import FileExistsLocallyError
+
     name = os.path.basename(os.path.normpath(urlparse(inp).path))
     if not out:
         ret = name
@@ -267,8 +268,9 @@ def resolve_output(inp, out):
     else:
         ret = out
 
-    if os.path.exists(ret):
-        raise FileExistsError(errno.EEXIST, os.strerror(errno.EEXIST), ret)
+    if os.path.exists(ret) and not force:
+        hint = "\nTo override it, re-run with '--force'."
+        raise FileExistsLocallyError(ret, hint=hint)
 
     return ret
 

dvc/utils/diff.py

@@ -26,6 +26,7 @@ def _diff_vals(old, new, with_unchanged):
     res = {"old": old, "new": new}
     if isinstance(new, (int, float)) and isinstance(old, (int, float)):
         res["diff"] = new - old
+
     return res
 
 

dvc/utils/hydra.py

@@ -1,3 +1,4 @@
+import logging
 from pathlib import Path
 from typing import TYPE_CHECKING, List
 
@@ -9,6 +10,9 @@ if TYPE_CHECKING:
     from dvc.types import StrPath
 
 
+logger = logging.getLogger(__name__)
+
+
 def compose_and_dump(
     output_file: "StrPath",
     config_dir: str,
@@ -44,6 +48,9 @@ def compose_and_dump(
         dumper(output_file, OmegaConf.to_object(cfg))
     else:
         Path(output_file).write_text(OmegaConf.to_yaml(cfg), encoding="utf-8")
+    logger.trace(  # type: ignore[attr-defined]
+        "Hydra composition enabled. Contents dumped to %s:\n %s", output_file, cfg
+    )
 
 
 def apply_overrides(path: "StrPath", overrides: List[str]) -> None:

dvc/utils/studio.py

@@ -6,6 +6,13 @@ import requests
 from funcy import compact, ignore
 from requests.adapters import HTTPAdapter
 
+from dvc.env import (
+    DVC_STUDIO_OFFLINE,
+    DVC_STUDIO_REPO_URL,
+    DVC_STUDIO_TOKEN,
+    DVC_STUDIO_URL,
+)
+
 if TYPE_CHECKING:
     from requests import Response
 
@@ -80,3 +87,29 @@ def notify_refs(
             "received response: %s (status=%r)", d, r.status_code
         )
     return d
+
+
+def config_to_env(config: Dict[str, Any]) -> Dict[str, Any]:
+    env = {}
+    if "offline" in config:
+        env[DVC_STUDIO_OFFLINE] = config["offline"]
+    if "repo_url" in config:
+        env[DVC_STUDIO_REPO_URL] = config["repo_url"]
+    if "token" in config:
+        env[DVC_STUDIO_TOKEN] = config["token"]
+    if "url" in config:
+        env[DVC_STUDIO_URL] = config["url"]
+    return env
+
+
+def env_to_config(env: Dict[str, Any]) -> Dict[str, Any]:
+    config = {}
+    if DVC_STUDIO_OFFLINE in env:
+        config["offline"] = env[DVC_STUDIO_OFFLINE]
+    if DVC_STUDIO_REPO_URL in env:
+        config["repo_url"] = env[DVC_STUDIO_REPO_URL]
+    if DVC_STUDIO_TOKEN in env:
+        config["token"] = env[DVC_STUDIO_TOKEN]
+    if DVC_STUDIO_URL in env:
+        config["url"] = env[DVC_STUDIO_URL]
+    return config

pyproject.toml

@@ -34,11 +34,11 @@ dependencies = [
     "configobj>=5.0.6",
     "distro>=1.3",
     "dpath<3,>=2.1.0",
-    "dvc-data>=0.47.1,<0.48",
+    "dvc-data>=0.51.0,<0.52",
     "dvc-http>=2.29.0",
-    "dvc-render>=0.3.1,<0.4.0",
-    "dvc-studio-client>=0.6.1,<1",
-    "dvc-task>=0.2.0,<1",
+    "dvc-render>=0.3.1,<1",
+    "dvc-studio-client>=0.9.2,<1",
+    "dvc-task>=0.2.1,<1",
     "flatten_dict<1,>=0.4.1",
     "flufl.lock>=5",
     "funcy>=1.14",
@@ -73,8 +73,8 @@ gdrive = ["dvc-gdrive==2.19.2"]
 gs = ["dvc-gs==2.22.0"]
 hdfs = ["dvc-hdfs==2.19"]
 lint = [
-    "mypy==1.2.0",
-    "pylint==2.17.2",
+    "mypy==1.3.0",
+    "pylint==2.17.4",
     "types-colorama",
     "types-psutil",
     "types-requests",
@@ -87,8 +87,13 @@ s3 = ["dvc-s3==2.22.0"]
 ssh = ["dvc-ssh>=2.22.1,<3"]
 ssh_gssapi = ["dvc-ssh[gssapi]>=2.22.1,<3"]
 terraform = ["tpi[ssh]>=2.1"]
-testing = ["pytest-test-utils"]
+testing = [
+  "pytest-test-utils",
+  "pytest-benchmark[histogram]",
+  "pytest-virtualenv",
+]
 tests = [
+    "dvc[testing]",
     "beautifulsoup4>=4.4",
     "dvc-ssh",
     "filelock",
@@ -114,6 +119,8 @@ Source = "https://github.com/iterative/dvc"
 dvc = "dvc.cli:main"
 [project.entry-points."fsspec.specs"]
 dvc = "dvc.api:DVCFileSystem"
+[project.entry-points."pytest11"]
+dvc-testing = "dvc.testing.plugin"
 
 [tool.setuptools]
 license-files = ["LICENSE"]
@@ -149,6 +156,8 @@ filterwarnings = [
     # and https://github.com/celery/celery/issues/7528
     # drop when celery==5.3 && kombu==5.3 releases
     "ignore:SelectableGroups dict interface is deprecated:DeprecationWarning",
+    # tpi imports pkg_resources
+    "ignore:pkg_resources is deprecated as an API:DeprecationWarning",
     # see https://github.com/networkx/networkx/issues/5723.
     "ignore:nx.nx_pydot.* depends on the pydot package, which has.*known issues and is not actively maintained:DeprecationWarning",
     # remove when new version of pytest-cov gets released
@@ -221,7 +230,9 @@ module = [
     "pygtrie.*",
     "pyinstrument",
     "pyparsing",
+    "pytest_benchmark.*",
     "pytest_docker.plugin",
+    "pytest_virtualenv.*",
     "ruamel.*",
     "ruamel.yaml.*",
     "shortuuid",
@@ -285,6 +296,7 @@ ignore-names = ["M", "SCM"]
 [tool.ruff.per-file-ignores]
 "dvc/commands/**" = ["N806"]
 "dvc/testing/**" = ["ARG002"]
+"dvc/testing/benchmarks/**" = ["ARG001"]
 "scripts/**" = ["T201", "INP001", "TRY002"]
 
 [tool.ruff.pylint]

tests/conftest.py

@@ -5,6 +5,7 @@ from contextlib import suppress
 
 import pytest
 
+from dvc import env
 from dvc.testing.fixtures import *  # noqa, pylint: disable=wildcard-import
 
 from .dir_helpers import *  # noqa, pylint: disable=wildcard-import
@@ -60,7 +61,7 @@ def enable_ui():
 
 @pytest.fixture(autouse=True)
 def clean_repos():
-    from dvc.external_repo import clean_repos
+    from dvc.repo.open_repo import clean_repos
 
     clean_repos()
 
@@ -206,7 +207,13 @@ defaultBranch=master
     import pygit2
 
     pygit2.settings.search_path[pygit2.GIT_CONFIG_LEVEL_GLOBAL] = str(home_dir)
+
+    monkeypatch.setenv(env.DVC_SYSTEM_CONFIG_DIR, os.fspath(path / "system"))
+    monkeypatch.setenv(env.DVC_GLOBAL_CONFIG_DIR, os.fspath(path / "global"))
+    monkeypatch.setenv(env.DVC_SITE_CACHE_DIR, os.fspath(path / "site_cache_dir"))
+
     yield
+
     monkeypatch.undo()
 
 

tests/func/api/test_experiments.py

@@ -2,9 +2,10 @@ import pytest
 
 from dvc import api
 from dvc.repo.experiments.exceptions import ExperimentExistsError
+from tests.unit.repo.experiments.conftest import exp_stage  # noqa: F401
 
 
-def test_exp_save(tmp_dir, dvc, scm, mocker):
+def test_exp_save(tmp_dir, dvc, scm):
     tmp_dir.scm_gen({"foo": "foo"}, commit="initial")
 
     api.exp_save()
@@ -16,3 +17,18 @@ def test_exp_save(tmp_dir, dvc, scm, mocker):
     ):
         api.exp_save("foo")
     api.exp_save("foo", force=True)
+
+
+def test_exp_show(tmp_dir, dvc, scm, exp_stage):  # noqa: F811
+    exps = api.exp_show()
+
+    assert len(exps) == 2
+    assert isinstance(exps, list)
+    assert isinstance(exps[0], dict)
+    assert isinstance(exps[1], dict)
+    # Postprocessing casting to float
+    assert exps[0]["metrics.yaml:foo"] == 1.0
+    # Postprocessing using `None` as fill value
+    assert exps[0]["State"] is None
+    # Postprocessing empty string as `None`
+    assert exps[0]["Experiment"] is None

tests/func/api/test_scm.py

@@ -0,0 +1,30 @@
+from dvc.api.scm import all_branches, all_commits, all_tags
+
+
+def test_all_branches(tmp_dir, scm, dvc):
+    assert all_branches() == ["master"]
+
+    with tmp_dir.branch("branch", new=True):
+        tmp_dir.scm_gen("branch", "branch", "commit")
+
+    assert all_branches() == ["branch", "master"]
+
+
+def test_all_commits(tmp_dir, scm, dvc):
+    first = scm.get_rev()
+    assert all_commits() == [first]
+
+    tmp_dir.scm_gen("foo", "foo", "commit")
+    second = scm.get_rev()
+
+    assert set(all_commits()) == {first, second}
+
+
+def test_all_tags(tmp_dir, scm, dvc):
+    scm.tag("v1")
+    assert all_tags() == ["v1"]
+
+    tmp_dir.scm_gen("foo", "foo", "commit")
+    scm.tag("v2")
+
+    assert set(all_tags()) == {"v1", "v2"}