pyinstaller: drop google-api-client hooks

Bumps [iterative-telemetry](https://github.com/iterative/telemetry-python) from 0.0.5 to 0.0.6.
- [Release notes](https://github.com/iterative/telemetry-python/releases)
- [Commits](https://github.com/iterative/telemetry-python/compare/0.0.5...0.0.6)

---
updated-dependencies:
- dependency-name: iterative-telemetry
  dependency-type: direct:production
  update-type: version-update:semver-patch
...

Signed-off-by: dependabot[bot] <[email protected]>

.github/workflows/codeql.yml

@@ -0,0 +1,39 @@
+name: "CodeQL"
+
+on:
+  push:
+    branches: ["main"]
+  pull_request:
+    branches: ["main"]
+  schedule:
+    - cron: '32 19 * * 2'
+  workflow_dispatch:
+
+permissions:
+  contents: read
+
+jobs:
+  analyze:
+    name: Analyze
+    runs-on: ubuntu-latest
+    permissions:
+      actions: read
+      contents: read
+      security-events: write
+
+    strategy:
+      fail-fast: false
+      matrix:
+        language: ['python']
+
+    steps:
+    - name: Checkout repository
+      uses: actions/checkout@v3
+
+    - name: Initialize CodeQL
+      uses: github/codeql-action/init@v2
+      with:
+        languages: ${{ matrix.language }}
+
+    - name: Perform CodeQL Analysis
+      uses: github/codeql-action/analyze@v2

.github/workflows/packages.yaml

@@ -104,7 +104,7 @@ jobs:
         password: ${{ secrets.PYPI_TOKEN }}
 
     - name: Publish to Test PyPI
-      if: ${{ github.event_name == 'push' && github.ref == 'refs/heads/main' }}
+      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 }}

.github/workflows/tests.yaml

@@ -46,13 +46,15 @@ jobs:
       fail-fast: false
       matrix:
         os: [ubuntu-20.04, windows-latest]
-        pyv: ["3.8", "3.9", "3.10"]
+        pyv: ["3.8", "3.9", "3.10", "3.11"]
         pytest-filter:
         - "import or plot or live or experiment"
         - "not (import or plot or live or experiment)"
         include:
-        - {os: ubuntu-latest, pyv: "3.11"}
         - {os: macos-latest, pyv: "3.8"}
+        - {os: macos-latest, pyv: "3.9"}
+        - {os: macos-latest, pyv: "3.10"}
+        - {os: macos-latest, pyv: "3.11"}
 
     steps:
     - uses: actions/checkout@v3

dvc/exceptions.py

@@ -132,10 +132,17 @@ class NotDvcRepoError(DvcException):
 class CyclicGraphError(DvcException):
     def __init__(self, stages):
         assert isinstance(stages, list)
-        msg = "Pipeline has a cycle involving: {}.".format(
-            ", ".join(s.addressing for s in stages)
+        stage_part = "stage" if len(stages) == 1 else "stages"
+        msg = (
+            "Same item(s) are defined as both a dependency and an output "
+            "in {stage_part}: {stage}."
+        )
+        super().__init__(
+            msg.format(
+                stage_part=stage_part,
+                stage=", ".join(s.addressing for s in stages),
+            )
         )
-        super().__init__(msg)
 
 
 class ConfirmRemoveError(DvcException):

dvc/output.py

@@ -526,7 +526,7 @@ class Output:
             entry.path = dep.fs_path
             entry.meta = dep.meta
             if not entry.obj:
-                if not dep.obj:
+                if not dep.obj and dep.files:
                     dep.obj = dep.get_obj()
                 entry.obj = dep.obj
         return entry

dvc/repo/data.py

@@ -12,6 +12,7 @@ from typing import (
     Optional,
     Tuple,
     TypedDict,
+    Union,
     cast,
 )
 
@@ -19,10 +20,9 @@ from dvc.fs.git import GitFileSystem
 from dvc.ui import ui
 
 if TYPE_CHECKING:
-    from scmrepo.base import Base
-
     from dvc.output import Output
     from dvc.repo import Repo
+    from dvc.scm import Git, NoSCM
     from dvc_data.hashfile.db import HashFileDB
     from dvc_data.hashfile.hash_info import HashInfo
     from dvc_data.hashfile.obj import HashFile
@@ -140,7 +140,9 @@ class GitInfo(TypedDict, total=False):
     is_dirty: bool
 
 
-def _git_info(scm: "Base", untracked_files: str = "all") -> GitInfo:
+def _git_info(
+    scm: Union["Git", "NoSCM"], untracked_files: str = "all"
+) -> GitInfo:
     from scmrepo.exceptions import SCMError
 
     from dvc.scm import NoSCM

dvc/repo/experiments/__init__.py

@@ -4,7 +4,7 @@ import re
 import time
 from typing import Dict, Iterable, Optional
 
-from funcy import cached_property, first
+from funcy import cached_property, chain, first
 
 from dvc.exceptions import DvcException
 from dvc.ui import ui
@@ -171,7 +171,18 @@ class Experiments:
     ) -> Dict[str, str]:
         results: Dict[str, str] = {}
         if entries is None:
-            entries = list(self.celery_queue.iter_queued())
+            entries = list(
+                chain(
+                    self.celery_queue.iter_active(),
+                    self.celery_queue.iter_queued(),
+                )
+            )
+
+        logger.debug(
+            "reproduce all these entries '%s'",
+            entries,
+        )
+
         if not entries:
             return results
 
@@ -189,7 +200,10 @@ class Experiments:
                     time.sleep(1)
                 self.celery_queue.follow(entry)
                 # wait for task collection to complete
-                result = self.celery_queue.get_result(entry)
+                try:
+                    result = self.celery_queue.get_result(entry)
+                except FileNotFoundError:
+                    result = None
                 if result is None or result.exp_hash is None:
                     name = entry.name or entry.stash_rev[:7]
                     failed.append(name)

dvc/repo/experiments/diff.py

@@ -7,7 +7,7 @@ logger = logging.getLogger(__name__)
 
 
 def diff(repo, *args, a_rev=None, b_rev=None, param_deps=False, **kwargs):
-    from dvc.repo.experiments.show import _collect_experiment_commit
+    from dvc.repo.experiments.show import collect_experiment_commit
     from dvc.scm import resolve_rev
 
     if repo.scm.no_commits:
@@ -15,15 +15,15 @@ def diff(repo, *args, a_rev=None, b_rev=None, param_deps=False, **kwargs):
 
     if a_rev:
         rev = resolve_rev(repo.scm, a_rev)
-        old = _collect_experiment_commit(repo, rev, param_deps=param_deps)
+        old = collect_experiment_commit(repo, rev, param_deps=param_deps)
     else:
-        old = _collect_experiment_commit(repo, "HEAD", param_deps=param_deps)
+        old = collect_experiment_commit(repo, "HEAD", param_deps=param_deps)
 
     if b_rev:
         rev = resolve_rev(repo.scm, b_rev)
-        new = _collect_experiment_commit(repo, rev, param_deps=param_deps)
+        new = collect_experiment_commit(repo, rev, param_deps=param_deps)
     else:
-        new = _collect_experiment_commit(
+        new = collect_experiment_commit(
             repo, "workspace", param_deps=param_deps
         )
 

dvc/repo/experiments/executor/base.py

@@ -602,29 +602,22 @@ class BaseExecutor(ABC):
             logger.debug("Running repro in '%s'", os.getcwd())
             yield dvc
             info.status = TaskStatus.SUCCESS
-            if infofile is not None:
-                info.dump_json(infofile)
-
         except CheckpointKilledError:
             info.status = TaskStatus.FAILED
-            if infofile is not None:
-                info.dump_json(infofile)
             raise
         except DvcException:
             if log_errors:
                 logger.exception("")
             info.status = TaskStatus.FAILED
-            if infofile is not None:
-                info.dump_json(infofile)
             raise
         except Exception:
             if log_errors:
                 logger.exception("unexpected error")
             info.status = TaskStatus.FAILED
-            if infofile is not None:
-                info.dump_json(infofile)
             raise
         finally:
+            if infofile is not None:
+                info.dump_json(infofile)
             dvc.close()
             os.chdir(old_cwd)
 

dvc/repo/experiments/executor/local.py

@@ -47,9 +47,9 @@ class BaseLocalExecutor(BaseExecutor):
         return SCM(self.root_dir)
 
     def cleanup(self, infofile: str):
-        super().cleanup(infofile)
         self.scm.close()
         del self.scm
+        super().cleanup(infofile)
 
     def collect_cache(
         self, repo: "Repo", exp_ref: "ExpRefInfo", run_cache: bool = True

dvc/repo/experiments/queue/celery.py

@@ -23,12 +23,7 @@ from dvc.exceptions import DvcException
 from dvc.ui import ui
 
 from ..exceptions import UnresolvedQueueExpNamesError
-from ..executor.base import (
-    EXEC_TMP_DIR,
-    ExecutorInfo,
-    ExecutorResult,
-    TaskStatus,
-)
+from ..executor.base import EXEC_TMP_DIR, ExecutorInfo, ExecutorResult
 from .base import BaseStashQueue, QueueDoneResult, QueueEntry, QueueGetResult
 from .tasks import run_exp
 from .utils import fetch_running_exp_from_temp_dir
@@ -187,6 +182,7 @@ class LocalCeleryQueue(BaseStashQueue):
                 continue
             args, kwargs, _embed = msg.decode()
             entry_dict = kwargs.get("entry_dict", args[0])
+            logger.debug("Found queued task %s", entry_dict["stash_rev"])
             yield _MessageEntry(msg, QueueEntry.from_dict(entry_dict))
 
     def _iter_processed(self) -> Generator[_MessageEntry, None, None]:
@@ -203,6 +199,7 @@ class LocalCeleryQueue(BaseStashQueue):
             task_id = msg.headers["id"]
             result: AsyncResult = AsyncResult(task_id)
             if not result.ready():
+                logger.debug("Found active task %s", entry.stash_rev)
                 yield _TaskEntry(result, entry)
 
     def _iter_done_tasks(self) -> Generator[_TaskEntry, None, None]:
@@ -211,6 +208,7 @@ class LocalCeleryQueue(BaseStashQueue):
             task_id = msg.headers["id"]
             result: AsyncResult = AsyncResult(task_id)
             if result.ready():
+                logger.debug("Found done task %s", entry.stash_rev)
                 yield _TaskEntry(result, entry)
 
     def iter_active(self) -> Generator[QueueEntry, None, None]:
@@ -243,48 +241,50 @@ class LocalCeleryQueue(BaseStashQueue):
     def reproduce(self) -> Mapping[str, Mapping[str, str]]:
         raise NotImplementedError
 
-    def get_result(
+    def _load_info(self, rev: str) -> ExecutorInfo:
+        infofile = self.get_infofile_path(rev)
+        return ExecutorInfo.load_json(infofile)
+
+    def _get_done_result(
         self, entry: QueueEntry, timeout: Optional[float] = None
-    ) -> Optional[ExecutorResult]:
+    ) -> Optional["ExecutorResult"]:
         from celery.exceptions import TimeoutError as _CeleryTimeout
 
-        def _load_info(rev: str) -> ExecutorInfo:
-            infofile = self.get_infofile_path(rev)
-            return ExecutorInfo.load_json(infofile)
-
-        def _load_collected(rev: str) -> Optional[ExecutorResult]:
-            executor_info = _load_info(rev)
-            if executor_info.status > TaskStatus.SUCCESS:
+        for msg, processed_entry in self._iter_processed():
+            if entry.stash_rev == processed_entry.stash_rev:
+                task_id = msg.headers["id"]
+                result: AsyncResult = AsyncResult(task_id)
+                if not result.ready():
+                    logger.debug(
+                        "Waiting for exp task '%s' to complete", result.id
+                    )
+                    try:
+                        result.get(timeout=timeout)
+                    except _CeleryTimeout as exc:
+                        raise DvcException(
+                            "Timed out waiting for exp to finish."
+                        ) from exc
+                executor_info = self._load_info(entry.stash_rev)
                 return executor_info.result
-            raise FileNotFoundError
+        raise FileNotFoundError
+
+    def get_result(
+        self, entry: QueueEntry, timeout: Optional[float] = None
+    ) -> Optional["ExecutorResult"]:
 
         try:
-            return _load_collected(entry.stash_rev)
+            return self._get_done_result(entry, timeout)
         except FileNotFoundError:
-            # Infofile will not be created until execution begins
             pass
 
         for queue_entry in self.iter_queued():
             if entry.stash_rev == queue_entry.stash_rev:
                 raise DvcException("Experiment has not been started.")
-        for result, active_entry in self._iter_active_tasks():
-            if entry.stash_rev == active_entry.stash_rev:
-                logger.debug(
-                    "Waiting for exp task '%s' to complete", result.id
-                )
-                try:
-                    result.get(timeout=timeout)
-                except _CeleryTimeout as exc:
-                    raise DvcException(
-                        "Timed out waiting for exp to finish."
-                    ) from exc
-                executor_info = _load_info(entry.stash_rev)
-                return executor_info.result
 
         # NOTE: It's possible for an exp to complete while iterating through
         # other queued and active tasks, in which case the exp will get moved
         # out of the active task list, and needs to be loaded here.
-        return _load_collected(entry.stash_rev)
+        return self._get_done_result(entry, timeout)
 
     def kill(self, revs: Collection[str]) -> None:
         to_kill: Set[QueueEntry] = set()

dvc/repo/experiments/queue/utils.py

@@ -1,6 +1,8 @@
 import logging
 from typing import TYPE_CHECKING, Dict
 
+from scmrepo.exceptions import SCMError
+
 from ..executor.base import ExecutorInfo, TaskStatus
 
 logger = logging.getLogger(__name__)
@@ -52,7 +54,7 @@ def fetch_running_exp_from_temp_dir(
                     result[rev]["last"] = last_rev
                     if last_rev:
                         result[last_rev] = info.asdict()
-            except InvalidRemoteSCMRepo:
+            except (InvalidRemoteSCMRepo, SCMError):
                 # ignore stale info files
                 del result[rev]
     return result

dvc/repo/experiments/run.py

@@ -29,8 +29,7 @@ def run(
     of `repro` for that experiment.
     """
     if run_all:
-        entries = list(repo.experiments.celery_queue.iter_queued())
-        return repo.experiments.reproduce_celery(entries, jobs=jobs)
+        return repo.experiments.reproduce_celery(jobs=jobs)
 
     hydra_sweep = None
     if params:
@@ -62,6 +61,10 @@ def run(
         )
 
     if hydra_sweep:
+        if kwargs.get("name"):
+            raise InvalidArgumentError(
+                "Sweep overrides can't be used alongside `--name`"
+            )
         from dvc.utils.hydra import get_hydra_sweeps
 
         sweeps = get_hydra_sweeps(path_overrides)

dvc/repo/experiments/show.py

@@ -3,16 +3,29 @@ from collections import OrderedDict, defaultdict
 from datetime import datetime
 from enum import Enum
 from itertools import chain
-from typing import TYPE_CHECKING, Any, Callable, Dict, List, Optional, Union
+from typing import (
+    TYPE_CHECKING,
+    Any,
+    Callable,
+    Dict,
+    List,
+    Optional,
+    Tuple,
+    Union,
+)
+
+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 iter_revs
+from dvc.scm import SCMError, iter_revs, resolve_rev
 from dvc.utils import error_handler, onerror_collect, relpath
 
 from .refs import ExpRefInfo
 
 if TYPE_CHECKING:
+    from scmrepo.git.objects import GitCommit
+
     from dvc.repo import Repo
 
 logger = logging.getLogger(__name__)
@@ -25,34 +38,49 @@ class ExpStatus(Enum):
     Failed = 3
 
 
+def _is_scm_error(collected_exp: Dict[str, Any]) -> bool:
+    if "error" in collected_exp and (
+        isinstance(collected_exp["error"], SCMError)
+        or isinstance(collected_exp["error"], InnerScmError)
+    ):
+        return True
+    return False
+
+
+def _show_onerror_collect(result: Dict, exception: Exception, *args, **kwargs):
+    onerror_collect(result, exception, *args, **kwargs)
+    result["data"] = {}
+
+
 @error_handler
-def _collect_experiment_commit(
+def collect_experiment_commit(
     repo: "Repo",
     exp_rev: str,
     status: ExpStatus = ExpStatus.Success,
     param_deps=False,
-    running=None,
+    running: Optional[Dict[str, Any]] = None,
     onerror: Optional[Callable] = None,
-):
+) -> Dict[str, Any]:
     from dvc.dependency import ParamsDependency, RepoDependency
 
-    res: Dict[str, Optional[Any]] = defaultdict(dict)
+    result: Dict[str, Any] = defaultdict(dict)
+    running = running or {}
     for rev in repo.brancher(revs=[exp_rev]):
         if rev == "workspace":
             if exp_rev != "workspace":
                 continue
-            res["timestamp"] = None
+            result["timestamp"] = None
         else:
             commit = repo.scm.resolve_commit(rev)
-            res["timestamp"] = datetime.fromtimestamp(commit.commit_time)
+            result["timestamp"] = datetime.fromtimestamp(commit.commit_time)
 
         params = _gather_params(
             repo, rev=rev, targets=None, deps=param_deps, onerror=onerror
         )
         if params:
-            res["params"] = params
+            result["params"] = params
 
-        res["deps"] = {
+        result["deps"] = {
             relpath(dep.fs_path, repo.root_dir): {
                 "hash": dep.hash_info.value,
                 "size": dep.meta.size,
@@ -61,7 +89,7 @@ def _collect_experiment_commit(
             for dep in repo.index.deps
             if not isinstance(dep, (ParamsDependency, RepoDependency))
         }
-        res["outs"] = {
+        result["outs"] = {
             relpath(out.fs_path, repo.root_dir): {
                 "hash": out.hash_info.value,
                 "size": out.meta.size,
@@ -73,14 +101,14 @@ def _collect_experiment_commit(
             if not (out.is_metric or out.is_plot)
         }
 
-        res["status"] = status.name
+        result["status"] = status.name
         if status == ExpStatus.Running:
-            res["executor"] = running.get(exp_rev, {}).get("location", None)
+            result["executor"] = running.get(exp_rev, {}).get("location", None)
         else:
-            res["executor"] = None
+            result["executor"] = None
 
         if status == ExpStatus.Failed:
-            res["error"] = {
+            result["error"] = {
                 "msg": "Experiment run failed.",
                 "type": "",
             }
@@ -89,54 +117,102 @@ def _collect_experiment_commit(
             vals = _gather_metrics(
                 repo, targets=None, rev=rev, recursive=False, onerror=onerror
             )
-            res["metrics"] = vals
+            result["metrics"] = vals
 
-    return res
+    return result
 
 
-def _collect_experiment_branch(
-    res,
-    repo,
-    branch,
-    baseline,
-    onerror: Optional[Callable] = None,
-    running=None,
+def _collect_complete_experiment(
+    repo: "Repo",
+    baseline: str,
+    exp_rev: str,
+    running: Dict[str, Any],
+    revs: List[str],
     **kwargs,
-):
-    from dvc.scm import resolve_rev
+) -> Dict[str, Dict[str, Any]]:
+    results: "OrderedDict[str, Dict[str, Any]]" = OrderedDict()
 
-    exp_rev = resolve_rev(repo.scm, branch)
-    prev = None
-    revs = list(repo.scm.branch_revs(exp_rev, baseline))
+    checkpoint: bool = True if len(revs) > 1 else False
+    prev = ""
     for rev in revs:
         status = ExpStatus.Running if rev in running else ExpStatus.Success
-        collected_exp = _collect_experiment_commit(
+        collected_exp = collect_experiment_commit(
             repo,
             rev,
-            onerror=onerror,
             status=status,
             running=running,
             **kwargs,
         )
-        if len(revs) > 1:
+        if _is_scm_error(collected_exp):
+            return {}
+        if checkpoint:
             exp = {"checkpoint_tip": exp_rev}
             if prev:
-                res[prev]["data"][  # type: ignore[unreachable]
+                results[prev]["data"][  # type: ignore[unreachable]
                     "checkpoint_parent"
                 ] = rev
-            if rev in res:
-                res[rev]["data"].update(exp)
-                res.move_to_end(rev)
+            if rev in results:
+                results[rev]["data"].update(exp)
+                results.move_to_end(rev)
             else:
                 exp.update(collected_exp["data"])
         else:
             exp = collected_exp["data"]
-        if rev not in res:
-            res[rev] = {"data": exp}
+        if rev not in results:
+            results[rev] = {"data": exp}
         prev = rev
-    if len(revs) > 1:
-        res[prev]["data"]["checkpoint_parent"] = baseline
-    return res
+    if checkpoint and prev:
+        results[prev]["data"]["checkpoint_parent"] = baseline
+    return results
+
+
+def _collect_branch(
+    repo: "Repo",
+    baseline: str,
+    running: Dict[str, Any],
+    **kwargs,
+) -> Dict[str, Dict[str, Any]]:
+    results: Dict[str, Dict[str, Any]] = OrderedDict()
+    status = ExpStatus.Running if baseline in running else ExpStatus.Success
+    results["baseline"] = collect_experiment_commit(
+        repo,
+        baseline,
+        status=status,
+        running=running,
+        **kwargs,
+    )
+    if baseline == "workspace" or _is_scm_error(results["baseline"]):
+        return results
+
+    ref_info = ExpRefInfo(baseline_sha=baseline)
+    commits: List[Tuple[str, "GitCommit", str, List[str]]] = []
+
+    for ref in repo.scm.iter_refs(base=str(ref_info)):
+        try:
+            commit = repo.scm.resolve_commit(ref)
+            exp_rev = resolve_rev(repo.scm, ref)
+            revs = list(repo.scm.branch_revs(exp_rev, baseline))
+        except (SCMError, InnerScmError):
+            continue
+        commits.append((ref, commit, exp_rev, revs))
+
+    for exp_ref, _, exp_rev, revs in sorted(
+        commits, key=lambda x: x[1].commit_time, reverse=True
+    ):
+        ref_info = ExpRefInfo.from_ref(exp_ref)
+        assert ref_info.baseline_sha == baseline
+        collected_exp = _collect_complete_experiment(
+            repo,
+            baseline=baseline,
+            exp_rev=exp_rev,
+            running=running,
+            revs=revs,
+            **kwargs,
+        )
+        if _is_scm_error(collected_exp):
+            continue
+        results.update(collected_exp)
+    return results
 
 
 def get_names(repo: "Repo", result: Dict[str, Dict[str, Any]]):
@@ -172,13 +248,11 @@ def get_names(repo: "Repo", result: Dict[str, Dict[str, Any]]):
                 rev = baseline
                 if rev == "workspace":
                     continue
-                name = names.get(rev, None)
-            name = name or exact_name[rev]
+            name = names.get(rev, None) or exact_name[rev]
             if name:
                 rev_result["data"]["name"] = name
 
 
-# flake8: noqa: C901
 def _collect_active_experiment(
     repo: "Repo",
     found_revs: Dict[str, List[str]],
@@ -194,13 +268,16 @@ def _collect_active_experiment(
         if entry.baseline_rev in found_revs and (
             stash_rev not in running or not running[stash_rev].get("last")
         ):
-            result[entry.baseline_rev][stash_rev] = _collect_experiment_commit(
+            collected_exp = collect_experiment_commit(
                 repo,
                 stash_rev,
                 status=ExpStatus.Running,
                 running=running,
                 **kwargs,
             )
+            if _is_scm_error(collected_exp):
+                continue
+            result[entry.baseline_rev][stash_rev] = collected_exp
     return result
 
 
@@ -214,13 +291,16 @@ def _collect_queued_experiment(
     for entry in repo.experiments.celery_queue.iter_queued():
         stash_rev = entry.stash_rev
         if entry.baseline_rev in found_revs:
-            result[entry.baseline_rev][stash_rev] = _collect_experiment_commit(
+            collected_exp = collect_experiment_commit(
                 repo,
                 stash_rev,
                 status=ExpStatus.Queued,
                 running=running,
                 **kwargs,
             )
+            if _is_scm_error(collected_exp):
+                continue
+            result[entry.baseline_rev][stash_rev] = collected_exp
     return result
 
 
@@ -235,14 +315,16 @@ def _collect_failed_experiment(
         entry = queue_done_result.entry
         stash_rev = entry.stash_rev
         if entry.baseline_rev in found_revs:
-            experiment = _collect_experiment_commit(
+            collected_exp = collect_experiment_commit(
                 repo,
                 stash_rev,
                 status=ExpStatus.Failed,
                 running=running,
                 **kwargs,
             )
-            result[entry.baseline_rev][stash_rev] = experiment
+            if _is_scm_error(collected_exp):
+                continue
+            result[entry.baseline_rev][stash_rev] = collected_exp
     return result
 
 
@@ -273,7 +355,7 @@ def show(
         return {}
 
     if onerror is None:
-        onerror = onerror_collect
+        onerror = _show_onerror_collect
 
     res: Dict[str, Dict] = defaultdict(OrderedDict)
 
@@ -316,7 +398,6 @@ def show(
             repo,
             found_revs,
             running,
-            sha_only=sha_only,
             param_deps=param_deps,
             onerror=onerror,
         )
@@ -324,47 +405,20 @@ def show(
         else {}
     )
 
-    for rev in found_revs:
-        status = ExpStatus.Running if rev in running else ExpStatus.Success
-        res[rev]["baseline"] = _collect_experiment_commit(
+    for baseline in found_revs:
+        res[baseline] = _collect_branch(
             repo,
-            rev,
-            status=status,
-            param_deps=param_deps,
+            baseline,
             running=running,
+            param_deps=param_deps,
             onerror=onerror,
         )
 
-        if rev == "workspace":
-            continue
-
-        ref_info = ExpRefInfo(baseline_sha=rev)
-        commits = [
-            (ref, repo.scm.resolve_commit(ref))
-            for ref in repo.scm.iter_refs(base=str(ref_info))
-        ]
-        for exp_ref, _ in sorted(
-            commits, key=lambda x: x[1].commit_time, reverse=True
-        ):
-            ref_info = ExpRefInfo.from_ref(exp_ref)
-            assert ref_info.baseline_sha == rev
-            _collect_experiment_branch(
-                res[rev],
-                repo,
-                exp_ref,
-                rev,
-                param_deps=param_deps,
-                running=running,
-                onerror=onerror,
-            )
-
-    if not hide_failed:
-        update_new(res, failed_experiments)
+    update_new(res, failed_experiments)
 
     update_new(res, active_experiment)
 
-    if not hide_queued:
-        update_new(res, queued_experiment)
+    update_new(res, queued_experiment)
 
     if not sha_only:
         get_names(repo, res)

dvc/repo/imports.py

@@ -44,7 +44,7 @@ def unfetched_view(
 
         return True
 
-    unfetched = index.targets_view(targets, filter_fn=need_fetch, **kwargs)
+    unfetched = index.targets_view(targets, stage_filter=need_fetch, **kwargs)
     return unfetched, changed_deps
 
 
@@ -53,7 +53,7 @@ def partial_view(
 ) -> "IndexView":
     return index.targets_view(
         targets,
-        filter_fn=lambda s: s.is_partial_import,
+        stage_filter=lambda s: s.is_partial_import,
         **kwargs,
     )
 

dvc/repo/index.py

@@ -7,8 +7,10 @@ from typing import (
     Iterable,
     Iterator,
     List,
+    NamedTuple,
     Optional,
     Set,
+    Tuple,
 )
 
 from funcy import cached_property, nullcontext
@@ -151,24 +153,34 @@ class Index:
     def targets_view(
         self,
         targets: "TargetType",
-        filter_fn: Optional[Callable[["Stage"], bool]] = None,
+        stage_filter: Optional[Callable[["Stage"], bool]] = None,
+        outs_filter: Optional[Callable[["Output"], bool]] = None,
         **kwargs: Any,
     ) -> "IndexView":
         """Return read-only view of index for the specified targets.
 
         Args:
             targets: Targets to collect
-            filter_fn: Optional stage filter to be applied after collecting
+            stage_filter: Optional stage filter to be applied after collecting
+                targets.
+            outs_filter: Optional output filter to be applied after collecting
                 targets.
 
         Additional kwargs will be passed into the stage collector.
+
+        Note:
+            If both stage_filter and outs_filter are provided, stage_filter
+            will be applied first, and the resulting view will only contain
+            outputs from stages that matched stage_filter. Outputs from stages
+            that did not match will be excluded from the view (whether or not
+            the output would have matched outs_filter).
         """
         stage_infos = [
             stage_info
             for stage_info in self._collect_targets(targets, **kwargs)
-            if not filter_fn or filter_fn(stage_info.stage)
+            if not stage_filter or stage_filter(stage_info.stage)
         ]
-        return IndexView(self, stage_infos)
+        return IndexView(self, stage_infos, outs_filter=outs_filter)
 
     @property
     def outs(self) -> Iterator["Output"]:
@@ -338,17 +350,26 @@ class Index:
         return dict_md5(self.dumpd())
 
 
+class _DataPrefixes(NamedTuple):
+    explicit: Set["DataIndexKey"]
+    recursive: Set["DataIndexKey"]
+
+
 class IndexView:
     """Read-only view of Index.data using filtered stages."""
 
     def __init__(  # pylint: disable=redefined-outer-name
-        self, index: Index, stage_infos: Iterable["StageInfo"]
+        self,
+        index: Index,
+        stage_infos: Iterable["StageInfo"],
+        outs_filter: Optional[Callable[["Output"], bool]],
     ):
         self._index = index
         self._stage_infos = stage_infos
         # NOTE: stage_infos might have the same stage multiple times but with
         # different filter_info
         self._stages = list({stage for stage, _ in stage_infos})
+        self._outs_filter = outs_filter
 
     def __len__(self) -> int:
         return len(self._stages)
@@ -377,30 +398,32 @@ class IndexView:
             yield from stage.deps
 
     @property
-    def outs(self) -> Iterator["Output"]:
-        outs = set()
+    def _filtered_outs(self) -> Iterator[Tuple["Output", Optional[str]]]:
         for stage, filter_info in self._stage_infos:
             for out in stage.filter_outs(filter_info):
-                outs.add(out)
-        yield from outs
+                if not self._outs_filter or self._outs_filter(out):
+                    yield out, filter_info
+
+    @property
+    def outs(self) -> Iterator["Output"]:
+        yield from {out for (out, _) in self._filtered_outs}
 
     @cached_property
-    def _data_prefixes(self) -> Dict[str, Set["DataIndexKey"]]:
+    def _data_prefixes(self) -> Dict[str, "_DataPrefixes"]:
         from collections import defaultdict
 
-        prefixes: Dict[str, Set["DataIndexKey"]] = defaultdict(set)
-        for stage, filter_info in self._stage_infos:
-            for out in stage.filter_outs(filter_info):
-                workspace, key = out.index_key
-                if filter_info and out.fs.path.isin(filter_info, out.fs_path):
-                    key = (
-                        *key,
-                        out.fs.path.relparts(filter_info, out.fs_path),
-                    )
-                prefixes[workspace].add(key)
-                prefixes[workspace].update(
-                    key[:i] for i in range(len(key), 0, -1)
-                )
+        prefixes: Dict[str, "_DataPrefixes"] = defaultdict(
+            lambda: _DataPrefixes(set(), set())
+        )
+        for out, filter_info in self._filtered_outs:
+            workspace, key = out.index_key
+            if filter_info and out.fs.path.isin(filter_info, out.fs_path):
+                key = key + out.fs.path.relparts(filter_info, out.fs_path)
+            if out.meta.isdir:
+                prefixes[workspace].recursive.add(key)
+            prefixes[workspace].explicit.update(
+                key[:i] for i in range(len(key), 0, -1)
+            )
         return prefixes
 
     @cached_property
@@ -411,9 +434,10 @@ class IndexView:
 
         def key_filter(workspace: str, key: "DataIndexKey"):
             try:
-                return key in self._data_prefixes[workspace] or any(
+                prefixes = self._data_prefixes[workspace]
+                return key in prefixes.explicit or any(
                     key[: len(prefix)] == prefix
-                    for prefix in self._data_prefixes[workspace]
+                    for prefix in prefixes.recursive
                 )
             except KeyError:
                 return False

dvc/repo/plots/__init__.py

@@ -553,7 +553,7 @@ def parse(fs, path, props=None, **kwargs):
         return _load_sv(path=path, fs=fs, delimiter="\t", header=header)
     if extension in LOADERS or extension in (".yml", ".yaml"):
         return LOADERS[extension](path=path, fs=fs)
-    if extension in (".jpeg", ".jpg", ".gif", ".png"):
+    if extension in (".jpeg", ".jpg", ".gif", ".png", ".svg"):
         with fs.open(path, "rb") as fd:
             return fd.read()
     raise PlotMetricTypeError(path)

pyproject.toml

@@ -52,12 +52,12 @@ dependencies = [
     "typing-extensions>=3.7.4",
     "scmrepo==0.1.3",
     "dvc-render==0.0.12",
-    "dvc-task==0.1.4",
+    "dvc-task==0.1.5",
     "dvclive>=1.0",
     "dvc-data==0.25.3",
     "dvc-http==2.27.2",
     "hydra-core>=1.1.0",
-    "iterative-telemetry==0.0.5",
+    "iterative-telemetry==0.0.6",
 ]
 dynamic = ["version"]
 classifiers = [
@@ -71,8 +71,8 @@ classifiers = [
 
 [project.optional-dependencies]
 azure = ["dvc-azure==2.20.4"]
-gdrive = ["dvc-gdrive==2.19.0"]
-gs = ["dvc-gs==2.19.1"]
+gdrive = ["dvc-gdrive==2.19.1"]
+gs = ["dvc-gs==2.20.0"]
 hdfs = ["dvc-hdfs==2.19.0; python_version < '3.11'"]
 oss = ["dvc-oss==2.19.0"]
 s3 = ["dvc-s3==2.21.0"]
@@ -110,7 +110,7 @@ tests = [
     # and, protected method calls in our tests
     "pylint-plugin-utils==0.7",
     # type-checking
-    "mypy==0.982",
+    "mypy==0.990",
     "types-requests>=2.27.15",
     "types-tabulate>=0.8.6",
     "types-toml>=0.10.4",
@@ -201,6 +201,7 @@ show_traceback = true
 pretty = true
 # See https://mypy.readthedocs.io/en/latest/running_mypy.html#missing-imports.
 ignore_missing_imports = true
+implicit_optional = true
 check_untyped_defs = false
 # Warnings
 warn_no_return = true

scripts/pyinstaller/hooks/hook-pydrive2.py

@@ -3,4 +3,3 @@ from PyInstaller.utils.hooks import (  # pylint:disable=import-error
 )
 
 datas = copy_metadata("pydrive2")
-datas += copy_metadata("google-api-python-client")

tests/func/experiments/test_experiments.py

@@ -536,7 +536,7 @@ def test_run_celery(tmp_dir, scm, dvc, exp_stage, mocker):
     repro_spy = mocker.spy(dvc.experiments, "reproduce_celery")
     results = dvc.experiments.run(run_all=True)
     assert len(results) == 2
-    repro_spy.assert_called_once_with(entries=mocker.ANY, jobs=1)
+    repro_spy.assert_called_once_with(jobs=1)
 
     expected = {"foo: 2", "foo: 3"}
     metrics = set()

tests/func/experiments/test_queue.py

@@ -11,64 +11,54 @@ def to_dict(tasks):
     return status_dict
 
 
-@pytest.fixture
-def success_tasks(tmp_dir, dvc, scm, test_queue, exp_stage):
-    queue_length = 3
-    name_list = []
-    for i in range(queue_length):
-        name = f"success{i}"
-        name_list.append(name)
-        dvc.experiments.run(
-            exp_stage.addressing, params=[f"foo={i}"], queue=True, name=name
-        )
-    dvc.experiments.run(run_all=True)
-    return name_list
-
-
-@pytest.fixture
-def failed_tasks(tmp_dir, dvc, scm, test_queue, failed_exp_stage):
-    queue_length = 3
-    name_list = []
-    for i in range(queue_length):
-        name = f"failed{i}"
-        name_list.append(name)
-        dvc.experiments.run(
-            failed_exp_stage.addressing,
-            params=[f"foo={i+queue_length}"],
-            queue=True,
-            name=name,
-        )
-    dvc.experiments.run(run_all=True)
-    return name_list
-
-
-@pytest.mark.xfail(strict=False, reason="pytest-celery flaky")
 @pytest.mark.parametrize("follow", [True, False])
 def test_celery_logs(
     tmp_dir,
     scm,
     dvc,
     failed_exp_stage,
-    test_queue,
     follow,
     capsys,
 ):
+    celery_queue = dvc.experiments.celery_queue
     dvc.experiments.run(failed_exp_stage.addressing, queue=True)
     dvc.experiments.run(run_all=True)
 
-    queue = dvc.experiments.celery_queue
-    done_result = first(queue.iter_done())
+    done_result = first(celery_queue.iter_done())
+
     name = done_result.entry.stash_rev
     captured = capsys.readouterr()
-    queue.logs(name, follow=follow)
+    celery_queue.logs(name, follow=follow)
     captured = capsys.readouterr()
     assert "failed to reproduce 'failed-copy-file'" in captured.out
 
 
-@pytest.mark.xfail(strict=False, reason="pytest-celery flaky")
-def test_queue_remove_done(dvc, failed_tasks, success_tasks):
-    assert len(dvc.experiments.celery_queue.failed_stash) == 3
-    status = to_dict(dvc.experiments.celery_queue.status())
+def test_queue_remove_done(
+    dvc,
+    exp_stage,
+    failed_exp_stage,
+):
+    queue_length = 3
+    success_tasks = []
+    failed_tasks = []
+    celery_queue = dvc.experiments.celery_queue
+    for i in range(queue_length):
+        name = f"success{i}"
+        success_tasks.append(name)
+        dvc.experiments.run(
+            exp_stage.addressing, params=[f"foo={i}"], queue=True, name=name
+        )
+        name_fail = f"failed{i}"
+        failed_tasks.append(name_fail)
+        dvc.experiments.run(
+            failed_exp_stage.addressing,
+            params=[f"foo={i+queue_length}"],
+            queue=True,
+            name=name_fail,
+        )
+    dvc.experiments.run(run_all=True)
+    assert len(celery_queue.failed_stash) == 3
+    status = to_dict(celery_queue.status())
     assert len(status) == 6
     for name in failed_tasks:
         assert status[name] == "Failed"
@@ -76,27 +66,21 @@ def test_queue_remove_done(dvc, failed_tasks, success_tasks):
         assert status[name] == "Success"
 
     with pytest.raises(InvalidArgumentError):
-        dvc.experiments.celery_queue.remove(failed_tasks[:2] + ["non-exist"])
-    assert len(dvc.experiments.celery_queue.status()) == 6
+        celery_queue.remove(failed_tasks[:2] + ["non-exist"])
+    assert len(celery_queue.status()) == 6
 
     to_remove = [failed_tasks[0], success_tasks[2]]
-    assert set(dvc.experiments.celery_queue.remove(to_remove)) == set(
-        to_remove
-    )
+    assert set(celery_queue.remove(to_remove)) == set(to_remove)
 
-    assert len(dvc.experiments.celery_queue.failed_stash) == 2
-    status = to_dict(dvc.experiments.celery_queue.status())
+    assert len(celery_queue.failed_stash) == 2
+    status = to_dict(celery_queue.status())
     assert set(status) == set(failed_tasks[1:] + success_tasks[:2])
 
-    assert dvc.experiments.celery_queue.clear(failed=True) == failed_tasks[1:]
+    assert set(celery_queue.clear(failed=True)) == set(failed_tasks[1:])
 
-    assert len(dvc.experiments.celery_queue.failed_stash) == 0
-    assert set(to_dict(dvc.experiments.celery_queue.status())) == set(
-        success_tasks[:2]
-    )
+    assert len(celery_queue.failed_stash) == 0
+    assert set(to_dict(celery_queue.status())) == set(success_tasks[:2])
 
-    assert (
-        dvc.experiments.celery_queue.clear(success=True) == success_tasks[:2]
-    )
+    assert set(celery_queue.clear(success=True)) == set(success_tasks[:2])
 
-    assert dvc.experiments.celery_queue.status() == []
+    assert celery_queue.status() == []

tests/func/experiments/test_set_params.py

@@ -144,5 +144,18 @@ def test_hydra_sweep(
 
 
 def test_hydra_sweep_requires_queue(params_repo, dvc):
-    with pytest.raises(InvalidArgumentError):
+    with pytest.raises(
+        InvalidArgumentError,
+        match="Sweep overrides can't be used without `--queue`",
+    ):
         dvc.experiments.run(params=["db=mysql,postgresql"])
+
+
+def test_hydra_sweep_cant_use_name(tmp_dir, params_repo, dvc):
+    with pytest.raises(
+        InvalidArgumentError,
+        match="Sweep overrides can't be used alongside `--name`",
+    ):
+        dvc.experiments.run(
+            params=["db=mysql,postgresql"], queue=True, name="foo"
+        )

tests/func/experiments/test_show.py

@@ -1,10 +1,12 @@
 import logging
 import os
+import random
 from datetime import datetime
 from unittest.mock import ANY
 
 import pytest
 from funcy import first, get_in
+from scmrepo.exceptions import SCMError
 
 from dvc.cli import main
 from dvc.repo.experiments.executor.base import (
@@ -172,7 +174,6 @@ def test_show_queued(tmp_dir, scm, dvc, exp_stage):
 
 
 @pytest.mark.vscode
-@pytest.mark.xfail(strict=False, reason="pytest-celery flaky")
 def test_show_failed_experiment(tmp_dir, scm, dvc, failed_exp_stage):
     baseline_rev = scm.get_rev()
     timestamp = datetime.fromtimestamp(
@@ -645,6 +646,9 @@ def test_show_csv(tmp_dir, scm, dvc, exp_stage, capsys):
     result1 = dvc.experiments.run(exp_stage.addressing, params=["foo=2"])
     rev1 = first(result1)
     ref_info1 = first(exp_refs_by_rev(scm, rev1))
+
+    # at least 1 second gap between these experiments to make sure
+    # the previous experiment to be regarded as branch_base
     time.sleep(1)
     result2 = dvc.experiments.run(exp_stage.addressing, params=["foo=3"])
     rev2 = first(result2)
@@ -972,3 +976,122 @@ def test_show_sorted_deps(tmp_dir, dvc, scm, capsys):
     assert main(["exp", "show", "--csv"]) == 0
     cap = capsys.readouterr()
     assert "a,b,c,z" in cap.out
+
+
+@pytest.mark.vscode
+def test_show_queued_error(tmp_dir, scm, dvc, exp_stage, mocker):
+    baseline_rev = scm.get_rev()
+
+    dvc.experiments.run(
+        exp_stage.addressing, params=["foo=2"], queue=True, name="test_name"
+    )
+    exp_rev_2 = dvc.experiments.scm.resolve_rev(f"{CELERY_STASH}@{{0}}")
+    commit_2 = scm.resolve_commit(exp_rev_2)
+
+    dvc.experiments.run(exp_stage.addressing, params=["foo=3"], queue=True)
+    exp_rev_3 = dvc.experiments.scm.resolve_rev(f"{CELERY_STASH}@{{0}}")
+
+    def resolve_commit(rev):
+        if rev == exp_rev_3:
+            raise SCMError
+        else:
+            return commit_2
+
+    mocker.patch.object(
+        scm,
+        "resolve_commit",
+        side_effect=mocker.MagicMock(side_effect=resolve_commit),
+    )
+
+    results = dvc.experiments.show()[baseline_rev]
+    assert len(results) == 2
+    exp_2 = results[exp_rev_2]["data"]
+    assert exp_2["status"] == "Queued"
+    assert exp_2["params"]["params.yaml"] == {"data": {"foo": 2}}
+
+
+@pytest.mark.vscode
+def test_show_completed_error(tmp_dir, scm, dvc, exp_stage, mocker):
+    baseline_rev = scm.get_rev()
+
+    result_2 = dvc.experiments.run(exp_stage.addressing, params=["foo=2"])
+    exp_rev_2 = first(result_2)
+    commit_2 = scm.resolve_commit(exp_rev_2)
+    result_3 = dvc.experiments.run(exp_stage.addressing, params=["foo=3"])
+    exp_rev_3 = first(result_3)
+
+    def resolve_commit(rev):
+        if rev == exp_rev_3:
+            raise SCMError
+        else:
+            return commit_2
+
+    mocker.patch.object(
+        scm,
+        "resolve_commit",
+        side_effect=mocker.MagicMock(side_effect=resolve_commit),
+    )
+    experiments = dvc.experiments.show()[baseline_rev]
+    assert len(experiments) == 2
+    assert exp_rev_2 in experiments
+
+
+@pytest.mark.vscode
+def test_show_checkpoint_error(tmp_dir, scm, dvc, checkpoint_stage, mocker):
+    baseline_rev = scm.get_rev()
+    results = dvc.experiments.run(
+        checkpoint_stage.addressing, params=["foo=2"]
+    )
+    exp_rev = first(results)
+    exp_ref = str(first(exp_refs_by_rev(scm, exp_rev)))
+
+    results = dvc.experiments.show()[baseline_rev]
+    assert len(results) == checkpoint_stage.iterations + 1
+
+    checkpoints = {}
+    for rev in results:
+        if rev != "baseline":
+            checkpoints[rev] = scm.resolve_commit(rev)
+    checkpoints[exp_ref] = scm.resolve_commit(exp_ref)
+    checkpoints[baseline_rev] = scm.resolve_commit(baseline_rev)
+
+    failed_rev = random.choice(list(checkpoints.keys()))
+
+    def resolve_commit(rev):
+        if rev == failed_rev:
+            raise SCMError
+        return checkpoints[rev]
+
+    mocker.patch.object(
+        scm,
+        "resolve_commit",
+        side_effect=mocker.MagicMock(side_effect=resolve_commit),
+    )
+    results = dvc.experiments.show()[baseline_rev]
+    assert len(results) == 1
+
+
+@pytest.mark.vscode
+def test_show_baseline_error(tmp_dir, scm, dvc, exp_stage, mocker):
+    baseline_rev = scm.get_rev()
+    branch = scm.active_branch()
+
+    result_2 = dvc.experiments.run(exp_stage.addressing, params=["foo=2"])
+    exp_rev_2 = first(result_2)
+    commit_2 = scm.resolve_commit(exp_rev_2)
+
+    def resolve_commit(rev):
+        if rev == baseline_rev:
+            raise SCMError
+        else:
+            return commit_2
+
+    mocker.patch.object(
+        scm,
+        "resolve_commit",
+        side_effect=mocker.MagicMock(side_effect=resolve_commit),
+    )
+    experiments = dvc.experiments.show()[baseline_rev]
+    assert len(experiments) == 1
+    assert experiments["baseline"]["data"] == {"name": branch}
+    assert isinstance(experiments["baseline"]["error"], SCMError)

tests/func/plots/test_show.py

@@ -257,16 +257,21 @@ def test_log_errors(
     )
 
 
-def test_plots_binary(tmp_dir, scm, dvc, run_copy_metrics, custom_template):
-    with open("image.jpg", "wb") as fd:
+@pytest.mark.parametrize("ext", ["jpg", "svg"])
+def test_plots_binary(
+    tmp_dir, scm, dvc, run_copy_metrics, custom_template, ext
+):
+    file1 = f"image.{ext}"
+    file2 = f"plot.{ext}"
+    with open(file1, "wb") as fd:
         fd.write(b"content")
 
-    dvc.add(["image.jpg"])
+    dvc.add([file1])
     run_copy_metrics(
-        "image.jpg",
-        "plot.jpg",
+        file1,
+        file2,
         commit="run training",
-        plots=["plot.jpg"],
+        plots=[file2],
         name="s2",
         single_stage=False,
     )
@@ -276,12 +281,12 @@ def test_plots_binary(tmp_dir, scm, dvc, run_copy_metrics, custom_template):
 
     scm.tag("v1")
 
-    with open("plot.jpg", "wb") as fd:
+    with open(file2, "wb") as fd:
         fd.write(b"content2")
 
     result = dvc.plots.show(revs=["v1", "workspace"])
-    assert get_plot(result, "v1", file="plot.jpg") == b"content"
-    assert get_plot(result, "workspace", file="plot.jpg") == b"content2"
+    assert get_plot(result, "v1", file=file2) == b"content"
+    assert get_plot(result, "workspace", file=file2) == b"content2"
 
 
 def test_collect_non_existing_dir(tmp_dir, dvc, run_copy_metrics):

tests/func/test_repo_index.py

@@ -303,3 +303,85 @@ def test_getitem(tmp_dir, dvc, run_copy):
 
     with pytest.raises(KeyError):
         _ = index["no-valid-stage-name"]
+
+
+def test_view_granular_dir(tmp_dir, scm, dvc, run_copy):
+    tmp_dir.dvc_gen(
+        {"dir": {"subdir": {"in-subdir": "in-subdir"}, "in-dir": "in-dir"}},
+        commit="init",
+    )
+    index = Index(dvc)
+
+    # view should include the specific target, parent dirs, and children
+    # view should exclude any siblings of the target
+    view = index.targets_view("dir/subdir")
+    data_index = view.data["repo"]
+    assert ("dir",) in data_index
+    assert (
+        "dir",
+        "subdir",
+    ) in data_index
+    assert ("dir", "subdir", "in-subdir") in data_index
+    assert (
+        "dir",
+        "in-dir",
+    ) not in data_index
+
+
+def test_view_stage_filter(tmp_dir, scm, dvc, run_copy):
+    (stage1,) = tmp_dir.dvc_gen("foo", "foo")
+    stage2 = run_copy("foo", "bar", name="copy-foo-bar")
+    tmp_dir.commit([s.outs[0].fspath for s in (stage1, stage2)], msg="add")
+    index = Index(dvc)
+
+    view = index.targets_view(None)
+    assert set(view.stages) == {stage1, stage2}
+    assert {out.fs_path for out in view.outs} == {
+        out.fs_path for out in (stage1.outs + stage2.outs)
+    }
+
+    view = index.targets_view(
+        None, stage_filter=lambda s: getattr(s, "name", "").startswith("copy")
+    )
+    assert set(view.stages) == {stage2}
+    assert {out.fs_path for out in view.outs} == {
+        out.fs_path for out in stage2.outs
+    }
+
+
+def test_view_outs_filter(tmp_dir, scm, dvc, run_copy):
+    (stage1,) = tmp_dir.dvc_gen("foo", "foo")
+    stage2 = run_copy("foo", "bar", name="copy-foo-bar")
+    tmp_dir.commit([s.outs[0].fspath for s in (stage1, stage2)], msg="add")
+    index = Index(dvc)
+
+    view = index.targets_view(None, outs_filter=lambda o: o.def_path == "foo")
+    assert set(view.stages) == {stage1, stage2}
+    assert {out.fs_path for out in view.outs} == {
+        out.fs_path for out in stage1.outs
+    }
+
+
+def test_view_combined_filter(tmp_dir, scm, dvc, run_copy):
+    (stage1,) = tmp_dir.dvc_gen("foo", "foo")
+    stage2 = run_copy("foo", "bar", name="copy-foo-bar")
+    tmp_dir.commit([s.outs[0].fspath for s in (stage1, stage2)], msg="add")
+    index = Index(dvc)
+
+    view = index.targets_view(
+        None,
+        stage_filter=lambda s: getattr(s, "name", "").startswith("copy"),
+        outs_filter=lambda o: o.def_path == "foo",
+    )
+    assert set(view.stages) == {stage2}
+    assert set(view.outs) == set()
+
+    view = index.targets_view(
+        None,
+        stage_filter=lambda s: getattr(s, "name", "").startswith("copy"),
+        outs_filter=lambda o: o.def_path == "bar",
+    )
+    assert set(view.stages) == {stage2}
+    assert {out.fs_path for out in view.outs} == {
+        out.fs_path for out in stage2.outs
+    }

tests/unit/repo/experiments/test_executor_status.py

@@ -82,21 +82,34 @@ def test_workspace_executor_success_status(dvc, scm, exp_stage, queue_type):
         assert not os.path.exists(infofile)
 
 
-@pytest.mark.parametrize("queue_type", ["workspace_queue", "tempdir_queue"])
+@pytest.mark.parametrize(
+    "queue_type",
+    ["workspace_queue", "tempdir_queue"],
+)
 def test_workspace_executor_failed_status(
     dvc, scm, failed_exp_stage, queue_type
 ):
-    workspace_queue = getattr(dvc.experiments, queue_type)
-    queue_entry = workspace_queue.put(
+    queue = getattr(dvc.experiments, queue_type)
+    queue.put(
         params={"params.yaml": ["foo=1"]},
         targets=failed_exp_stage.addressing,
         name="failed",
     )
-    name = workspace_queue._EXEC_NAME or queue_entry.stash_rev
+    entry, executor = queue.get()
+    name = queue._EXEC_NAME or entry.stash_rev
+    infofile = queue.get_infofile_path(name)
+    rev = entry.stash_rev
 
-    infofile = workspace_queue.get_infofile_path(name)
     with pytest.raises(DvcException):
-        workspace_queue.reproduce()
+        executor.reproduce(
+            info=executor.info,
+            rev=rev,
+            infofile=infofile,
+        )
+    executor_info = ExecutorInfo.load_json(infofile)
+    assert executor_info.status == TaskStatus.FAILED
+
+    cleanup_exp.s(executor, infofile)()
     if queue_type == "workspace_queue":
         assert not os.path.exists(infofile)
     else:

tests/unit/repo/experiments/test_show.py

@@ -0,0 +1,27 @@
+from funcy import first
+from scmrepo.git import Git
+
+from dvc.repo.experiments.show import get_names
+
+
+def test_get_show_branch(tmp_dir, scm: "Git", dvc, exp_stage):
+    new_branch = "new"
+
+    baseline = scm.get_rev()
+    base_branch = scm.active_branch()
+    results = dvc.experiments.run(exp_stage.addressing, params=["foo=2"])
+    exp_a = first(results)
+    dvc.experiments.branch(exp_a, new_branch)
+
+    scm.checkout(new_branch, force=True)
+
+    result = {
+        "workspace": {"baseline": {"data": {}}},
+        exp_a: {"baseline": {"data": {}}},
+        baseline: {"baseline": {"data": {}}, exp_a: {"data": {}}},
+    }
+
+    get_names(dvc, result)
+    assert result[exp_a]["baseline"]["data"] == {"name": new_branch}
+    assert result[baseline]["baseline"]["data"] == {"name": base_branch}
+    assert result[baseline][exp_a]["data"] == {"name": new_branch}