exp show: cache collected git revs

dvc/commands/experiments/show.py

@@ -453,8 +453,12 @@ def _normalize_headers(names, count):
 
 
 def _format_json(item):
+    from dvc.repo.experiments.show import _CachedError
+
     if isinstance(item, (date, datetime)):
         return item.isoformat()
+    if isinstance(item, _CachedError):
+        return {"type": getattr(item, "typ", "_CachedError"), "msg": str(item)}
     return encode_exception(item)
 
 
@@ -472,6 +476,7 @@ class CmdExperimentsShow(CmdBase):
                 sha_only=self.args.sha,
                 param_deps=self.args.param_deps,
                 fetch_running=self.args.fetch_running,
+                force=self.args.force,
             )
         except DvcException:
             logger.exception("failed to show experiments")
@@ -650,4 +655,10 @@ def add_parser(experiments_subparsers, parent_parser):
         action="store_false",
         help=argparse.SUPPRESS,
     )
+    experiments_show_parser.add_argument(
+        "-f",
+        "--force",
+        action="store_true",
+        help="Force re-collection of experiments instead of loading from exp cache.",
+    )
     experiments_show_parser.set_defaults(func=CmdExperimentsShow)

dvc/repo/experiments/__init__.py

@@ -11,6 +11,7 @@ from dvc.ui import ui
 from dvc.utils import relpath
 from dvc.utils.objects import cached_property
 
+from .cache import ExpCache
 from .exceptions import (
     BaselineMismatchError,
     ExperimentExistsError,
@@ -92,6 +93,10 @@ class Experiments:
     def apply_stash(self) -> ApplyStash:
         return ApplyStash(self.scm, APPLY_STASH)
 
+    @cached_property
+    def cache(self) -> ExpCache:
+        return ExpCache(self.repo)
+
     @property
     def stash_revs(self) -> Dict[str, "ExpStashEntry"]:
         revs = {}

dvc/repo/experiments/show.py

@@ -18,12 +18,11 @@ from typing import (
 
 from scmrepo.exceptions import SCMError as InnerScmError
 
-from dvc.repo.metrics.show import _gather_metrics
-from dvc.repo.params.show import _gather_params
-from dvc.scm import Git, SCMError, iter_revs, resolve_rev
-from dvc.utils import error_handler, onerror_collect, relpath
+from dvc.exceptions import DvcException
+from dvc.scm import Git, RevError, SCMError, iter_revs, resolve_rev
 
 from .refs import ExpRefInfo
+from .serialize import SerializableError, SerializableExp
 
 if TYPE_CHECKING:
     from scmrepo.git.objects import GitCommit
@@ -40,85 +39,113 @@ class ExpStatus(Enum):
     Failed = 3
 
 
+class _CachedError(DvcException):
+    def __init__(self, msg, typ, *args):
+        super().__init__(msg, *args)
+        self.typ = typ
+
+
 def _is_scm_error(collected_exp: Dict[str, Any]) -> bool:
     if "error" in collected_exp and (
-        isinstance(collected_exp["error"], (SCMError, InnerScmError))
+        isinstance(collected_exp["error"], (_CachedError, SCMError, InnerScmError))
     ):
         return True
     return False
 
 
-def _show_onerror_collect(result: Dict, exception: Exception, *args, **kwargs):
-    onerror_collect(result, exception, *args, **kwargs)
-    result["data"] = {}
+def _format_exp(exp: SerializableExp) -> Dict[str, Any]:
+    # SerializableExp always includes error but we need to strip it from show
+    # output when it is false-y to maintain compatibility w/tools that consume
+    # json output and assume that "error" key presence means there was an error
+    exp_dict = exp.dumpd()
+    if "error" in exp_dict and not exp_dict["error"]:
+        del exp_dict["error"]
+    return {"data": exp_dict}
+
+
+def _format_error(error: SerializableError):
+    msg = error.msg or "None"
+    return {"data": {}, "error": _CachedError(msg, error.type)}
 
 
-@error_handler
 def collect_experiment_commit(
     repo: "Repo",
     exp_rev: str,
     status: ExpStatus = ExpStatus.Success,
-    param_deps=False,
-    running: Optional[Dict[str, Any]] = None,
-    onerror: Optional[Callable] = None,
+    param_deps: bool = False,
+    force: bool = False,
+    **kwargs,
 ) -> Dict[str, Any]:
-    from dvc.dependency import ParamsDependency, RepoDependency
+    cache = repo.experiments.cache
+    # TODO: support filtering serialized exp when param_deps is set
+    if exp_rev != "workspace" and not (force or param_deps):
+        cached_exp = cache.get(exp_rev)
+        if cached_exp:
+            if status == ExpStatus.Running:
+                # expire cached queued exp entry once we start running it
+                cache.delete(exp_rev)
+            elif isinstance(cached_exp, SerializableError):
+                return _format_error(cached_exp)
+            else:
+                return _format_exp(cached_exp)
+    try:
+        exp = _collect_from_repo(
+            repo,
+            exp_rev,
+            status=status,
+            param_deps=param_deps,
+            force=force,
+            **kwargs,
+        )
+        if exp_rev != "workspace" and not param_deps:
+            cache.put(exp, force=True)
+        return _format_exp(exp)
+    except Exception as exc:  # noqa: BLE001, pylint: disable=broad-except
+        logger.debug("", exc_info=True)
+        error = SerializableError(str(exc), type(exc).__name__)
+        if not (exp_rev == "workspace" or param_deps or status == ExpStatus.Running):
+            cache.put(error, rev=exp_rev, force=True)
+        return _format_error(error)
 
-    result: Dict[str, Any] = defaultdict(dict)
+
+def _collect_from_repo(
+    repo: "Repo",
+    exp_rev: str,
+    status: ExpStatus = ExpStatus.Success,
+    running: Optional[Dict[str, Any]] = None,
+    **kwargs,
+) -> "SerializableExp":
     running = running or {}
     for rev in repo.brancher(revs=[exp_rev]):
         if rev == "workspace":
             if exp_rev != "workspace":
                 continue
-            result["timestamp"] = None
+            timestamp: Optional[datetime] = None
         else:
             commit = repo.scm.resolve_commit(rev)
-            result["timestamp"] = datetime.fromtimestamp(commit.commit_time)
+            timestamp = datetime.fromtimestamp(commit.commit_time)
 
-        params = _gather_params(repo, targets=None, deps=param_deps, onerror=onerror)
-        if params:
-            result["params"] = params
-
-        result["deps"] = {
-            relpath(dep.fs_path, repo.root_dir): {
-                "hash": dep.hash_info.value,
-                "size": dep.meta.size,
-                "nfiles": dep.meta.nfiles,
-            }
-            for dep in repo.index.deps
-            if not isinstance(dep, (ParamsDependency, RepoDependency))
-        }
-        result["outs"] = {
-            relpath(out.fs_path, repo.root_dir): {
-                "hash": out.hash_info.value,
-                "size": out.meta.size,
-                "nfiles": out.meta.nfiles,
-                "use_cache": out.use_cache,
-                "is_data_source": out.stage.is_data_source,
-            }
-            for out in repo.index.outs
-            if not (out.is_metric or out.is_plot)
-        }
-
-        result["status"] = status.name
         if status == ExpStatus.Running:
-            result["executor"] = running.get(exp_rev, {}).get("location", None)
+            executor: Optional[str] = running.get(exp_rev, {}).get("location", None)
         else:
-            result["executor"] = None
+            executor = None
 
         if status == ExpStatus.Failed:
-            result["error"] = {
-                "msg": "Experiment run failed.",
-                "type": "",
-            }
-
-        if status not in {ExpStatus.Queued, ExpStatus.Failed}:
-            vals = _gather_metrics(
-                repo, targets=None, rev=rev, recursive=False, onerror=onerror
+            error: Optional["SerializableError"] = SerializableError(
+                "Experiment run failed."
             )
-            result["metrics"] = vals
+        else:
+            error = None
 
-    return result
+        return SerializableExp.from_repo(
+            repo,
+            rev=exp_rev,
+            timestamp=timestamp,
+            status=status.name,
+            executor=executor,
+            error=error,
+        )
+    raise RevError(f"nonexistent exp rev: '{exp_rev}'")
 
 
 def _collect_complete_experiment(
@@ -396,12 +423,11 @@ def show(  # noqa: PLR0913
     param_deps=False,
     onerror: Optional[Callable] = None,
     fetch_running: bool = True,
+    force: bool = False,
 ):
     if repo.scm.no_commits:
         return {}
 
-    onerror = onerror or _show_onerror_collect
-
     res: Dict[str, Dict] = defaultdict(OrderedDict)
 
     if not any([revs, all_branches, all_tags, all_commits]):
@@ -427,7 +453,7 @@ def show(  # noqa: PLR0913
             found_revs,
             running,
             param_deps=param_deps,
-            onerror=onerror,
+            force=force,
         )
         if not hide_queued
         else {}
@@ -439,6 +465,7 @@ def show(  # noqa: PLR0913
         running,
         param_deps=param_deps,
         onerror=onerror,
+        force=force,
     )
 
     failed_experiments = (
@@ -448,6 +475,7 @@ def show(  # noqa: PLR0913
             running,
             param_deps=param_deps,
             onerror=onerror,
+            force=force,
         )
         if not hide_failed
         else {}
@@ -460,6 +488,7 @@ def show(  # noqa: PLR0913
             running=running,
             param_deps=param_deps,
             onerror=onerror,
+            force=force,
         )
 
     update_new(res, failed_experiments)

tests/func/experiments/test_show.py

@@ -12,6 +12,7 @@ from dvc.cli import main
 from dvc.repo.experiments.executor.base import BaseExecutor, ExecutorInfo, TaskStatus
 from dvc.repo.experiments.queue.base import QueueEntry
 from dvc.repo.experiments.refs import CELERY_STASH, ExpRefInfo
+from dvc.repo.experiments.show import _CachedError
 from dvc.repo.experiments.utils import EXEC_PID_DIR, EXEC_TMP_DIR, exp_refs_by_rev
 from dvc.utils import relpath
 from dvc.utils.serialize import YAMLFileCorruptedError
@@ -71,6 +72,7 @@ def test_show_simple(tmp_dir, scm, dvc, exp_stage):
     assert dvc.experiments.show()["workspace"] == {
         "baseline": {
             "data": {
+                "rev": "workspace",
                 "deps": {
                     "copy.py": {
                         "hash": ANY,
@@ -102,6 +104,7 @@ def test_show_experiment(tmp_dir, scm, dvc, exp_stage, workspace):
 
     expected_baseline = {
         "data": {
+            "rev": ANY,
             "deps": {
                 "copy.py": {
                     "hash": ANY,
@@ -177,6 +180,7 @@ def test_show_failed_experiment(tmp_dir, scm, dvc, failed_exp_stage):
 
     expected_baseline = {
         "data": {
+            "rev": ANY,
             "deps": {
                 "copy.py": {
                     "hash": ANY,
@@ -196,9 +200,11 @@ def test_show_failed_experiment(tmp_dir, scm, dvc, failed_exp_stage):
 
     expected_failed = {
         "data": {
+            "rev": ANY,
             "timestamp": ANY,
             "params": {"params.yaml": {"data": {"foo": 2}}},
             "deps": {"copy.py": {"hash": None, "size": None, "nfiles": None}},
+            "metrics": {},
             "outs": {},
             "status": "Failed",
             "executor": None,
@@ -432,6 +438,7 @@ def test_show_running_workspace(
     assert dvc.experiments.show().get("workspace") == {
         "baseline": {
             "data": {
+                "rev": "workspace",
                 "deps": {
                     "copy.py": {
                         "hash": ANY,
@@ -608,7 +615,8 @@ def test_show_with_broken_repo(tmp_dir, scm, dvc, exp_stage, caplog):
     assert get_in(baseline[rev2], paths) == {"data": {"foo": 3}}
 
     paths = ["workspace", "baseline", "error"]
-    assert isinstance(get_in(result, paths), YAMLFileCorruptedError)
+    assert isinstance(get_in(result, paths), (_CachedError, YAMLFileCorruptedError))
+    assert "YAML file structure is corrupted" in str(get_in(result, paths))
 
 
 def test_show_csv(tmp_dir, scm, dvc, exp_stage, capsys):
@@ -1040,7 +1048,7 @@ def test_show_checkpoint_error(tmp_dir, scm, dvc, checkpoint_stage, mocker):
         "resolve_commit",
         side_effect=mocker.MagicMock(side_effect=resolve_commit),
     )
-    results = dvc.experiments.show()[baseline_rev]
+    results = dvc.experiments.show(force=True)[baseline_rev]
     assert len(results) == 1
 
 
@@ -1066,4 +1074,4 @@ def test_show_baseline_error(tmp_dir, scm, dvc, exp_stage, mocker):
     experiments = dvc.experiments.show()[baseline_rev]
     assert len(experiments) == 1
     assert experiments["baseline"]["data"] == {"name": branch}
-    assert isinstance(experiments["baseline"]["error"], SCMError)
+    assert isinstance(experiments["baseline"]["error"], (SCMError, _CachedError))

tests/unit/command/test_experiments.py

@@ -100,6 +100,7 @@ def test_experiments_show(dvc, scm, mocker):
             "1",
             "--rev",
             "foo",
+            "--force",
         ]
     )
     assert cli_args.func == CmdExperimentsShow
@@ -121,6 +122,7 @@ def test_experiments_show(dvc, scm, mocker):
         sha_only=True,
         param_deps=True,
         fetch_running=True,
+        force=True,
     )