yappi: add --yappi-separate-threads flag

fix: #8450

1. Add a new function check_rwlock to check and autocorrect rwlock info.
2. Add related tests for it.

.github/release-drafter.yml

@@ -1,49 +0,0 @@
-# Config for https://github.com/apps/release-drafter
-name-template: "$RESOLVED_VERSION 🦉 "
-tag-template: "$RESOLVED_VERSION"
-categories:
-  - title: "🚀 New Features and Enhancements "
-    labels:
-      - "feature"
-      - "enhancement"
-      - "ui"
-  - title: "💣 Breaking Changes "
-    labels:
-      - "breaking-change"
-  - title: "🏇 Optimizations "
-    labels:
-      - "optimize"
-      - "performance"
-  - title: "🐛 Bug Fixes "
-    labels:
-      - "bugfix"
-      - "bug"
-  - title: "🔨 Maintenance "
-    labels:
-      - "maintenance"
-      - "refactoring"
-      - "chore"
-      - "build"
-      - "ci"
-change-template: "- $TITLE (#$NUMBER) @$AUTHOR"
-version-resolver:
-  major:
-    labels:
-      - "major"
-  minor:
-    labels:
-      - "minor"
-  patch:
-    labels:
-      - "patch"
-  default: patch
-exclude-labels:
-  - "skip-changelog"
-template: |
-  > Refer to https://dvc.org/doc/install for installation instructions.
-
-  ## Changes
-
-  $CHANGES
-
-  Thanks again to $CONTRIBUTORS for the contributions! 🎉

.pre-commit-config.yaml

@@ -6,7 +6,7 @@ repos:
       - id: black
         language_version: python3
     repo: https://github.com/ambv/black
-    rev: 22.8.0
+    rev: 22.10.0
   - repo: https://github.com/pre-commit/pre-commit-hooks
     rev: v4.3.0
     hooks:
@@ -18,7 +18,7 @@ repos:
           - --ignore-words-list
           - ba,datas,fo,uptodate,cachable,falsy
     repo: https://github.com/codespell-project/codespell
-    rev: v2.2.1
+    rev: v2.2.2
   - hooks:
       - id: isort
         language_version: python3

dvc/_debug.py

@@ -1,16 +1,14 @@
 from contextlib import ExitStack, contextmanager
 from datetime import datetime
-from typing import TYPE_CHECKING, Callable, Union
+from typing import TYPE_CHECKING, Callable, Optional, Union
 
 if TYPE_CHECKING:
     from argparse import Namespace
+    from types import FrameType
 
 
 @contextmanager
-def viztracer_profile(
-    path: Union[Callable[[], str], str],
-    max_stack_depth: int = -1,
-):
+def viztracer_profile(path: Union[Callable[[], str], str], depth: int = -1):
     try:
         import viztracer  # pylint: disable=import-error
     except ImportError:
@@ -18,7 +16,7 @@ def viztracer_profile(
         yield
         return
 
-    tracer = viztracer.VizTracer(max_stack_depth=max_stack_depth)
+    tracer = viztracer.VizTracer(max_stack_depth=depth)
 
     tracer.start()
     yield
@@ -29,7 +27,9 @@ def viztracer_profile(
 
 @contextmanager
 def yappi_profile(
-    path: Union[Callable[[], str], str] = None, wall_clock: bool = True
+    path: Union[Callable[[], str], str] = None,
+    wall_clock: Optional[bool] = True,
+    separate_threads: Optional[bool] = False,
 ):
     try:
         import yappi  # pylint: disable=import-error
@@ -44,14 +44,26 @@ def yappi_profile(
     yield
     yappi.stop()
 
-    # pylint:disable=no-member
-    if path:
-        stats = yappi.get_func_stats()
-        fpath = path() if callable(path) else path
-        stats.save(fpath, "callgrind")
+    threads = yappi.get_thread_stats()
+    stats = {}
+    if separate_threads:
+        for thread in threads:
+            ctx_id = thread.id
+            stats[ctx_id] = yappi.get_func_stats(ctx_id=ctx_id)
     else:
-        yappi.get_func_stats().print_all()
-        yappi.get_thread_stats().print_all()
+        stats[None] = yappi.get_func_stats()
+
+    fpath = path() if callable(path) else path
+    for ctx_id, st in stats.items():
+        if fpath:
+            out = f"{fpath}-{ctx_id}" if ctx_id is not None else fpath
+            st.save(out, type="callgrind")
+        else:
+            if ctx_id is not None:
+                print(f"\nThread {ctx_id}")
+            st.print_all()  # pylint:disable=no-member
+            if ctx_id is None:
+                threads.print_all()  # pylint:disable=no-member
 
     yappi.clear_stats()
 
@@ -109,6 +121,38 @@ def debug():
         raise  # prevent from jumping ahead
 
 
+def _sigshow(_, frame: Optional["FrameType"]) -> None:
+    from shutil import get_terminal_size
+    from traceback import format_stack
+
+    from dvc.ui import ui
+
+    lines = "\u2015" * get_terminal_size().columns
+    ui.error_write(lines, "\n", *format_stack(frame), lines, sep="")
+
+
+@contextmanager
+def show_stack():
+    r"""Show stack trace on SIGQUIT (Ctrl-\) or SIGINFO (Ctrl-T on macOS)."""
+    import signal
+
+    signal.signal(signal.SIGQUIT, _sigshow)
+    try:
+        signal.signal(signal.SIGINFO, _sigshow)  # only available on macOS
+    except AttributeError:
+        pass
+    yield
+
+
+def _get_path_func(tool: str, ext: str):
+    fmt = f"{tool}.dvc-{{now:%Y%m%d}}_{{now:%H%M%S}}.{ext}"
+
+    def func(now: Optional["datetime"] = None) -> str:
+        return fmt.format(now=now or datetime.now())
+
+    return func
+
+
 @contextmanager
 def debugtools(args: "Namespace" = None, **kwargs):
     kw = vars(args) if args else {}
@@ -121,19 +165,21 @@ def debugtools(args: "Namespace" = None, **kwargs):
             stack.enter_context(profile(kw.get("cprofile_dump")))
         if kw.get("instrument") or kw.get("instrument_open"):
             stack.enter_context(instrument(kw.get("instrument_open", False)))
+        if kw.get("show_stack", False):
+            stack.enter_context(show_stack())
         if kw.get("yappi"):
-            output = "callgrind.dvc-{0:%Y%m%d}_{0:%H%M%S}.out"
+            path_func = _get_path_func("callgrind", "out")
             stack.enter_context(
-                yappi_profile(path=lambda: output.format(datetime.now()))
-            )
-        if kw.get("viztracer") or kw.get("viztracer_depth"):
-            output = "viztracer.dvc-{0:%Y%m%d}_{0:%H%M%S}.json"
-            stack.enter_context(
-                viztracer_profile(
-                    path=lambda: output.format(datetime.now()),
-                    max_stack_depth=kw.get("viztracer_depth") or -1,
+                yappi_profile(
+                    path=path_func,
+                    separate_threads=kw.get("yappi_separate_threads"),
                 )
             )
+        if kw.get("viztracer") or kw.get("viztracer_depth"):
+            path_func = _get_path_func("viztracer", "json")
+            depth = kw.get("viztracer_depth") or -1
+            prof = viztracer_profile(path=path_func, depth=depth)
+            stack.enter_context(prof)
         yield
 
 
@@ -146,6 +192,12 @@ def add_debugging_flags(parser):
     parser.add_argument(
         "--yappi", action="store_true", default=False, help=SUPPRESS
     )
+    parser.add_argument(
+        "--yappi-separate-threads",
+        action="store_true",
+        default=False,
+        help=SUPPRESS,
+    )
     parser.add_argument(
         "--viztracer", action="store_true", default=False, help=SUPPRESS
     )
@@ -160,3 +212,10 @@ def add_debugging_flags(parser):
     parser.add_argument(
         "--instrument-open", action="store_true", default=False, help=SUPPRESS
     )
+    parser.add_argument(
+        "--show-stack",
+        "--ss",
+        action="store_true",
+        default=False,
+        help=SUPPRESS,
+    )

dvc/commands/experiments/ls.py

@@ -16,13 +16,17 @@ class CmdExperimentsList(CmdBase):
             num=self.args.num,
             git_remote=self.args.git_remote,
         )
+        tags = self.repo.scm.describe(exps)
+        remained = {baseline for baseline, tag in tags.items() if tag is None}
+        base = "refs/heads/"
+        ref_heads = self.repo.scm.describe(remained, base=base)
+
         for baseline in exps:
-            tag = self.repo.scm.describe(baseline)
-            if not tag:
-                branch = self.repo.scm.describe(baseline, base="refs/heads")
-                if branch:
-                    tag = branch.split("/")[-1]
-            name = tag if tag else baseline[:7]
+            name = (
+                tags[baseline]
+                or ref_heads[baseline][len(base) :]
+                or baseline[:7]
+            )
             if not name_only:
                 print(f"{name}:")
             for exp_name in exps[baseline]:

dvc/commands/experiments/show.py

@@ -95,11 +95,8 @@ def _collect_rows(
 
         exp = results.get("data", {})
 
-        if exp.get("running"):
-            state = "Running"
-        elif exp.get("queued"):
-            state = "Queued"
-        else:
+        state = exp.get("status")
+        if state == "Success":
             state = fill_value
 
         is_baseline = rev == "baseline"
@@ -476,6 +473,8 @@ class CmdExperimentsShow(CmdBase):
                 all_branches=self.args.all_branches,
                 all_tags=self.args.all_tags,
                 all_commits=self.args.all_commits,
+                hide_queued=self.args.hide_queued,
+                hide_failed=self.args.hide_failed,
                 revs=self.args.rev,
                 num=self.args.num,
                 sha_only=self.args.sha,
@@ -594,6 +593,18 @@ def add_parser(experiments_subparsers, parent_parser):
         default=False,
         help="Always show git commit SHAs instead of branch/tag names.",
     )
+    experiments_show_parser.add_argument(
+        "--hide-failed",
+        action="store_true",
+        default=False,
+        help="Hide failed experiments in the table.",
+    )
+    experiments_show_parser.add_argument(
+        "--hide-queued",
+        action="store_true",
+        default=False,
+        help="Hide queued experiments in the table.",
+    )
     experiments_show_parser.add_argument(
         "--json",
         "--show-json",

dvc/commands/imp_url.py

@@ -25,6 +25,7 @@ class CmdImportUrl(CmdBase):
                 labels=self.args.labels,
                 meta=self.args.meta,
                 jobs=self.args.jobs,
+                version_aware=self.args.version_aware,
             )
         except DvcException:
             logger.exception(
@@ -108,5 +109,14 @@ def add_parser(subparsers, parent_parser):
         ),
         metavar="<number>",
     )
+    import_parser.add_argument(
+        "--version-aware",
+        action="store_true",
+        default=False,
+        help=(
+            "Import using cloud versioning. Implied if the URL contains a "
+            "version ID."
+        ),
+    )
     _add_annotating_args(import_parser)
     import_parser.set_defaults(func=CmdImportUrl)

dvc/config_schema.py

@@ -180,6 +180,8 @@ SCHEMA = {
                     "user": str,
                     "password": str,
                     "ask_password": Bool,
+                    "passphrase": str,
+                    "ask_passphrase": Bool,
                     "keyfile": str,
                     "timeout": Coerce(int),
                     "gss_auth": Bool,

dvc/data_cloud.py

@@ -11,6 +11,7 @@ if TYPE_CHECKING:
     from dvc_data.hashfile.db import HashFileDB
     from dvc_data.hashfile.hash_info import HashInfo
     from dvc_data.hashfile.status import CompareStatusResult
+    from dvc_data.hashfile.transfer import TransferResult
 
 logger = logging.getLogger(__name__)
 
@@ -106,14 +107,10 @@ class DataCloud:
         dest_odb: "HashFileDB",
         objs: Iterable["HashInfo"],
         **kwargs,
-    ):
-        from dvc.exceptions import FileTransferError
-        from dvc_data.hashfile.transfer import TransferError, transfer
+    ) -> "TransferResult":
+        from dvc_data.hashfile.transfer import transfer
 
-        try:
-            return transfer(src_odb, dest_odb, objs, **kwargs)
-        except TransferError as exc:
-            raise FileTransferError(exc.fails) from exc
+        return transfer(src_odb, dest_odb, objs, **kwargs)
 
     def push(
         self,
@@ -121,7 +118,7 @@ class DataCloud:
         jobs: Optional[int] = None,
         remote: Optional[str] = None,
         odb: Optional["HashFileDB"] = None,
-    ):
+    ) -> "TransferResult":
         """Push data items in a cloud-agnostic way.
 
         Args:
@@ -148,7 +145,7 @@ class DataCloud:
         jobs: Optional[int] = None,
         remote: Optional[str] = None,
         odb: Optional["HashFileDB"] = None,
-    ):
+    ) -> "TransferResult":
         """Pull data items in a cloud-agnostic way.
 
         Args:

dvc/dependency/base.py

@@ -1,6 +1,7 @@
-from typing import Type
+from typing import Dict, Type
 
 from dvc.exceptions import DvcException
+from dvc.fs import download as fs_download
 from dvc.output import Output
 
 
@@ -30,5 +31,34 @@ class Dependency(Output):
     )  # type: Type[DvcException]
     IsStageFileError = DependencyIsStageFileError  # type: Type[DvcException]
 
+    def workspace_status(self) -> Dict[str, str]:
+        if self.fs.version_aware:
+            old_fs_path = self.fs_path
+            try:
+                self.fs_path = self.fs.path.version_path(self.fs_path, None)
+                if self.changed_meta():
+                    return {str(self): "update available"}
+            finally:
+                self.fs_path = old_fs_path
+        return super().workspace_status()
+
     def update(self, rev=None):
-        pass
+        if self.fs.version_aware:
+            self.fs_path = self.fs.path.version_path(self.fs_path, rev)
+            self.meta = self.get_meta()
+            self.def_path = self.fs.path.version_path(
+                self.def_path, self.meta.version_id
+            )
+            self.fs_path = self.fs.path.version_path(
+                self.fs_path, self.meta.version_id
+            )
+
+    def download(self, to, jobs=None):
+        fs_download(self.fs, self.fs_path, to, jobs=jobs)
+
+    def save(self):
+        super().save()
+        if self.fs.version_aware:
+            self.fs_path = self.fs.path.version_path(
+                self.fs_path, self.meta.version_id
+            )

dvc/fs/__init__.py

@@ -26,6 +26,7 @@ from dvc_objects.fs.errors import (  # noqa: F401
 )
 from dvc_objects.fs.path import Path  # noqa: F401
 
+from .callbacks import Callback
 from .data import DataFileSystem  # noqa: F401
 from .dvc import DVCFileSystem  # noqa: F401
 from .git import GitFileSystem  # noqa: F401
@@ -47,6 +48,22 @@ known_implementations.update(
 # pylint: enable=unused-import
 
 
+def download(fs, fs_path, to, jobs=None):
+    with Callback.as_tqdm_callback(
+        desc=f"Downloading {fs.path.name(fs_path)}",
+        unit="files",
+    ) as cb:
+        fs.get(fs_path, to.fs_path, batch_size=jobs, callback=cb)
+
+
+def parse_external_url(url, config=None):
+    remote_config = dict(config) if config else {}
+    remote_config["url"] = url
+    fs_cls, fs_config, fs_path = get_cloud_fs(None, **remote_config)
+    fs = fs_cls(**fs_config)
+    return fs, fs_path
+
+
 def get_fs_config(config, **kwargs):
     name = kwargs.get("name")
     if name:

dvc/fs/callbacks.py

@@ -1,6 +1,6 @@
 # pylint: disable=unused-import
 from contextlib import ExitStack
-from typing import TYPE_CHECKING, Optional
+from typing import TYPE_CHECKING, Any, BinaryIO, Dict, Optional, Union
 
 from funcy import cached_property
 
@@ -75,10 +75,16 @@ class RichCallback(Callback):
             visible=not self.disable,
         )
 
-    def branch(self, path_1, path_2, kwargs, child: Optional[Callback] = None):
+    def branch(
+        self,
+        path_1: Union[str, BinaryIO],
+        path_2: str,
+        kwargs: Dict[str, Any],
+        child: "Callback" = None,
+    ):
         child = child or RichCallback(
             progress=self.progress,
-            desc=path_1,
+            desc=path_1 if isinstance(path_1, str) else path_2,
             bytes=True,
             transient=self._transient,
         )

dvc/ignore.py

@@ -300,12 +300,25 @@ class DvcIgnoreFilter:
         return [fs_dict[name] for name in chain(dirs, nondirs)]
 
     def walk(self, fs: FileSystem, path: AnyFSPath, **kwargs):
+        detail = kwargs.get("detail", False)
         ignore_subrepos = kwargs.pop("ignore_subrepos", True)
         if fs.protocol == Schemes.LOCAL:
             for root, dirs, files in fs.walk(path, **kwargs):
-                dirs[:], files[:] = self(
-                    root, dirs, files, ignore_subrepos=ignore_subrepos
-                )
+                if detail:
+                    all_dnames = set(dirs.keys())
+                    all_fnames = set(files.keys())
+                    dnames, fnames = self(
+                        root,
+                        all_dnames,
+                        all_fnames,
+                        ignore_subrepos=ignore_subrepos,
+                    )
+                    list(map(dirs.pop, all_dnames - set(dnames)))
+                    list(map(files.pop, all_fnames - set(fnames)))
+                else:
+                    dirs[:], files[:] = self(
+                        root, dirs, files, ignore_subrepos=ignore_subrepos
+                    )
                 yield root, dirs, files
         else:
             yield from fs.walk(path, **kwargs)

dvc/output.py

@@ -26,10 +26,12 @@ 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.index import DataIndexEntry
 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 .utils import relpath
 from .utils.fs import path_isin
 
@@ -38,6 +40,8 @@ if TYPE_CHECKING:
     from dvc_data.index import DataIndexKey
     from dvc_objects.db import ObjectDB
 
+    from .fs.callbacks import Callback
+
 logger = logging.getLogger(__name__)
 
 
@@ -511,6 +515,23 @@ class Output:
             key = self.fs.path.parts(no_drive)[1:]
         return workspace, key
 
+    def get_entry(self) -> "DataIndexEntry":
+        from dvc.config import NoRemoteError
+
+        try:
+            remote = self.repo.cloud.get_remote_odb(self.remote)
+        except NoRemoteError:
+            remote = None
+
+        return DataIndexEntry(
+            meta=self.meta,
+            obj=self.obj,
+            hash_info=self.hash_info,
+            odb=self.odb,
+            cache=self.odb,
+            remote=remote,
+        )
+
     def changed_checksum(self):
         return self.hash_info != self.get_hash()
 
@@ -530,7 +551,7 @@ class Output:
 
     def changed_meta(self) -> bool:
         if self.fs.version_aware and self.meta.version_id:
-            return self.meta.version_id == self.get_meta().version_id
+            return self.meta.version_id != self.get_meta().version_id
         return False
 
     def workspace_status(self):
@@ -797,15 +818,6 @@ class Output:
                 f"binary file '{self.fs_path}' cannot be used as metrics."
             )
 
-    def download(self, to, jobs=None):
-        from dvc.fs.callbacks import Callback
-
-        with Callback.as_tqdm_callback(
-            desc=f"Downloading {self.fs.path.name(self.fs_path)}",
-            unit="files",
-        ) as cb:
-            self.fs.get(self.fs_path, to.fs_path, batch_size=jobs, callback=cb)
-
     def get_obj(
         self, filter_info: Optional[str] = None, **kwargs
     ) -> Optional["HashFile"]:
@@ -825,6 +837,7 @@ class Output:
         else:
             return None
 
+        assert obj
         fs_path = self.fs.path
         if filter_info and filter_info != self.fs_path:
             prefix = fs_path.relparts(filter_info, self.fs_path)
@@ -835,7 +848,7 @@ class Output:
     def checkout(
         self,
         force=False,
-        progress_callback=None,
+        progress_callback: "Callback" = DEFAULT_CALLBACK,
         relink=False,
         filter_info=None,
         allow_missing=False,
@@ -843,9 +856,9 @@ class Output:
         **kwargs,
     ):
         if not self.use_cache:
-            if progress_callback:
-                progress_callback(
-                    self.fs_path, self.get_files_number(filter_info)
+            if progress_callback != DEFAULT_CALLBACK:
+                progress_callback.relative_update(
+                    self.get_files_number(filter_info)
                 )
             return None
 

dvc/repo/__init__.py

@@ -3,7 +3,7 @@ import os
 from collections import defaultdict
 from contextlib import contextmanager
 from functools import wraps
-from typing import TYPE_CHECKING, Callable, List, Optional
+from typing import TYPE_CHECKING, Callable, Optional
 
 from funcy import cached_property
 
@@ -18,7 +18,6 @@ if TYPE_CHECKING:
     from dvc.fs import FileSystem
     from dvc.repo.scm_context import SCMContext
     from dvc.scm import Base
-    from dvc.stage import Stage
 
 logger = logging.getLogger(__name__)
 
@@ -447,44 +446,6 @@ class Repo:
 
         return used
 
-    def partial_imports(
-        self,
-        targets=None,
-        all_branches=False,
-        all_tags=False,
-        all_commits=False,
-        all_experiments=False,
-        commit_date: Optional[str] = None,
-        recursive=False,
-        revs=None,
-        num=1,
-    ) -> List["Stage"]:
-        """Get the stages related to the given target and collect dependencies
-        which are missing outputs.
-
-        This is useful to retrieve files which have been imported to the repo
-        using --no-download.
-
-        Returns:
-            A list of partially imported stages
-        """
-        from itertools import chain
-
-        partial_imports = chain.from_iterable(
-            self.index.partial_imports(targets, recursive=recursive)
-            for _ in self.brancher(
-                revs=revs,
-                all_branches=all_branches,
-                all_tags=all_tags,
-                all_commits=all_commits,
-                all_experiments=all_experiments,
-                commit_date=commit_date,
-                num=num,
-            )
-        )
-
-        return list(partial_imports)
-
     @property
     def stages(self):  # obsolete, only for backward-compatibility
         return self.index.stages

dvc/repo/brancher.py

@@ -77,6 +77,7 @@ def brancher(  # noqa: E302
         from dvc.fs import GitFileSystem
 
         for sha, names in found_revs.items():
+            self.__dict__.pop("index", None)
             self.fs = GitFileSystem(scm=scm, rev=sha)
             self.root_dir = self.fs.path.join("/", *repo_root_parts)
 

dvc/repo/checkout.py

@@ -7,7 +7,6 @@ from dvc.exceptions import (
     CheckoutErrorSuggestGit,
     NoOutputOrStageError,
 )
-from dvc.progress import Tqdm
 from dvc.utils import relpath
 
 from . import locked
@@ -80,6 +79,7 @@ def checkout(
     allow_missing=False,
     **kwargs,
 ):
+    from dvc.fs.callbacks import Callback
 
     stats = {"added": [], "deleted": [], "modified": [], "failed": []}
     if not targets:
@@ -91,13 +91,16 @@ def checkout(
 
     pairs = _collect_pairs(self, targets, with_deps, recursive)
     total = get_all_files_numbers(pairs)
-    with Tqdm(
-        total=total, unit="file", desc="Checkout", disable=total == 0
-    ) as pbar:
+    with Callback.as_tqdm_callback(
+        unit="file",
+        desc="Checkout",
+        disable=total == 0,
+    ) as cb:
+        cb.set_size(total)
         for stage, filter_info in pairs:
             result = stage.checkout(
                 force=force,
-                progress_callback=pbar.update_msg,
+                progress_callback=cb,
                 relink=relink,
                 filter_info=filter_info,
                 allow_missing=allow_missing,

dvc/repo/experiments/__init__.py

@@ -2,7 +2,7 @@ import logging
 import os
 import re
 import time
-from typing import Any, Dict, Iterable, Optional
+from typing import Dict, Iterable, Optional
 
 from funcy import cached_property, first
 
@@ -16,7 +16,7 @@ from .exceptions import (
     InvalidExpRefError,
     MultipleBranchError,
 )
-from .executor.base import BaseExecutor, ExecutorInfo, TaskStatus
+from .executor.base import BaseExecutor
 from .queue.base import BaseStashQueue, QueueEntry
 from .queue.celery import LocalCeleryQueue
 from .queue.tempdir import TempDirQueue
@@ -25,7 +25,6 @@ from .refs import (
     CELERY_FAILED_STASH,
     CELERY_STASH,
     EXEC_APPLY,
-    EXEC_BRANCH,
     EXEC_CHECKPOINT,
     EXEC_NAMESPACE,
     EXPS_NAMESPACE,
@@ -211,8 +210,9 @@ class Experiments:
 
     def _log_reproduced(self, revs: Iterable[str], tmp_dir: bool = False):
         names = []
+        rev_names = self.get_exact_name(revs)
         for rev in revs:
-            name = self.get_exact_name(rev)
+            name = rev_names[rev]
             names.append(name if name else rev[:7])
         ui.write("\nRan experiment(s): {}".format(", ".join(names)))
         if tmp_dir:
@@ -411,88 +411,41 @@ class Experiments:
             raise MultipleBranchError(rev, ref_infos)
         return str(ref_infos[0])
 
-    def get_exact_name(self, rev: str):
+    def get_exact_name(self, revs: Iterable[str]) -> Dict[str, Optional[str]]:
         """Returns preferred name for the specified revision.
 
         Prefers tags, branches (heads), experiments in that orer.
         """
+        result: Dict[str, Optional[str]] = {}
         exclude = f"{EXEC_NAMESPACE}/*"
-        ref = self.scm.describe(rev, base=EXPS_NAMESPACE, exclude=exclude)
-        if ref:
-            try:
-                name = ExpRefInfo.from_ref(ref).name
-                if name:
-                    return name
-            except InvalidExpRefError:
-                pass
-        if rev in self.stash_revs:
-            return self.stash_revs[rev].name
-        return None
-
-    def get_running_exps(self, fetch_refs: bool = True) -> Dict[str, Any]:
-        """Return info for running experiments."""
-        result = {}
-        infofile = self.workspace_queue.get_infofile_path("workspace")
-        result.update(
-            self._fetch_running_exp("workspace", infofile, fetch_refs)
+        ref_dict = self.scm.describe(
+            revs, base=EXPS_NAMESPACE, exclude=exclude
         )
-        for queue in (self.tempdir_queue, self.celery_queue):
-            for entry in queue.iter_active():
-                infofile = queue.get_infofile_path(entry.stash_rev)
-                result.update(
-                    self._fetch_running_exp(
-                        entry.stash_rev, infofile, fetch_refs
-                    )
-                )
+        for rev in revs:
+            name: Optional[str] = None
+            ref = ref_dict[rev]
+            if ref:
+                try:
+                    name = ExpRefInfo.from_ref(ref).name
+                except InvalidExpRefError:
+                    pass
+            if not name:
+                if rev in self.stash_revs:
+                    name = self.stash_revs[rev].name
+                elif rev in self.celery_queue.failed_stash.stash_revs:
+                    name = self.celery_queue.failed_stash.stash_revs[rev].name
+            result[rev] = name
         return result
 
-    def _fetch_running_exp(
-        self, rev: str, infofile: str, fetch_refs: bool
-    ) -> Dict[str, Any]:
-        from dvc.scm import InvalidRemoteSCMRepo
-        from dvc.utils.serialize import load_json
-
-        from .executor.local import TempDirExecutor
-
-        result: Dict[str, Any] = {}
-        try:
-            info = ExecutorInfo.from_dict(load_json(infofile))
-        except OSError:
-            return result
-        if info.result is None:
-            if rev == "workspace":
-                # If we are appending to a checkpoint branch in a workspace
-                # run, show the latest checkpoint as running.
-                if info.status > TaskStatus.RUNNING:
-                    return result
-                last_rev = self.scm.get_ref(EXEC_BRANCH)
-                if last_rev:
-                    result[last_rev] = info.asdict()
-                else:
-                    result[rev] = info.asdict()
-            else:
-                result[rev] = info.asdict()
-                if info.git_url and fetch_refs:
-
-                    def on_diverged(_ref: str, _checkpoint: bool):
-                        return False
-
-                    executor = TempDirExecutor.from_info(info)
-                    try:
-                        for ref in executor.fetch_exps(
-                            self.scm,
-                            on_diverged=on_diverged,
-                        ):
-                            logger.debug(
-                                "Updated running experiment '%s'.", ref
-                            )
-                            last_rev = self.scm.get_ref(ref)
-                            result[rev]["last"] = last_rev
-                            if last_rev:
-                                result[last_rev] = info.asdict()
-                    except InvalidRemoteSCMRepo:
-                        # ignore stale info files
-                        del result[rev]
+    def get_running_exps(self, fetch_refs: bool = True) -> Dict[str, Dict]:
+        """Return info for running experiments."""
+        result = {}
+        for queue in (
+            self.workspace_queue,
+            self.tempdir_queue,
+            self.celery_queue,
+        ):
+            result.update(queue.get_running_exps(fetch_refs))
         return result
 
     def apply(self, *args, **kwargs):

dvc/repo/experiments/executor/base.py

@@ -20,6 +20,8 @@ from typing import (
     Union,
 )
 
+from scmrepo.exceptions import SCMError
+
 from dvc.env import DVC_EXP_AUTO_PUSH, DVC_EXP_GIT_REMOTE
 from dvc.exceptions import DvcException
 from dvc.stage.serialize import to_lockfile
@@ -361,21 +363,25 @@ class BaseExecutor(ABC):
             return False
 
         # fetch experiments
-        dest_scm.fetch_refspecs(
-            self.git_url,
-            [f"{ref}:{ref}" for ref in refs],
-            on_diverged=on_diverged_ref,
-            force=force,
-            **kwargs,
-        )
-        # update last run checkpoint (if it exists)
-        if has_checkpoint:
+        try:
             dest_scm.fetch_refspecs(
                 self.git_url,
-                [f"{EXEC_CHECKPOINT}:{EXEC_CHECKPOINT}"],
-                force=True,
+                [f"{ref}:{ref}" for ref in refs],
+                on_diverged=on_diverged_ref,
+                force=force,
                 **kwargs,
             )
+            # update last run checkpoint (if it exists)
+            if has_checkpoint:
+                dest_scm.fetch_refspecs(
+                    self.git_url,
+                    [f"{EXEC_CHECKPOINT}:{EXEC_CHECKPOINT}"],
+                    force=True,
+                    **kwargs,
+                )
+        except SCMError:
+            pass
+
         return refs
 
     @classmethod

dvc/repo/experiments/queue/base.py

@@ -18,11 +18,12 @@ from typing import (
     Union,
 )
 
-from funcy import cached_property
+from funcy import cached_property, retry
 
 from dvc.dependency import ParamsDependency
 from dvc.env import DVCLIVE_RESUME
 from dvc.exceptions import DvcException
+from dvc.lock import LockError
 from dvc.ui import ui
 
 from ..exceptions import CheckpointExistsError, ExperimentExistsError
@@ -560,6 +561,7 @@ class BaseStashQueue(ABC):
         self.scm.add(list(params.keys()))
 
     @staticmethod
+    @retry(180, errors=LockError, timeout=1)
     @scm_locked
     def init_executor(
         exp: "Experiments",
@@ -601,8 +603,9 @@ class BaseStashQueue(ABC):
         )
 
     @staticmethod
+    @retry(180, errors=LockError, timeout=1)
     @scm_locked
-    def collect_executor(
+    def collect_git(
         exp: "Experiments",
         executor: BaseExecutor,
         exec_result: ExecutorResult,
@@ -626,6 +629,17 @@ class BaseStashQueue(ABC):
                 logger.debug("Collected experiment '%s'.", exp_rev[:7])
                 results[exp_rev] = exec_result.exp_hash
 
+        return results
+
+    @classmethod
+    def collect_executor(
+        cls,
+        exp: "Experiments",
+        executor: BaseExecutor,
+        exec_result: ExecutorResult,
+    ) -> Dict[str, str]:
+        results = cls.collect_git(exp, executor, exec_result)
+
         if exec_result.ref_info is not None:
             executor.collect_cache(exp.repo, exec_result.ref_info)
 
@@ -692,3 +706,11 @@ class BaseStashQueue(ABC):
                 entry.stash_rev,
                 message=f"commit: {msg}",
             )
+
+    @abstractmethod
+    def get_running_exps(self, fetch_refs: bool = True) -> Dict[str, Dict]:
+        """Get the execution info of the currently running experiments
+
+        Args:
+            fetch_ref (bool): fetch completed checkpoints or not.
+        """

dvc/repo/experiments/queue/celery.py

@@ -31,6 +31,7 @@ from ..executor.base import (
 )
 from .base import BaseStashQueue, QueueDoneResult, QueueEntry, QueueGetResult
 from .tasks import run_exp
+from .utils import fetch_running_exp_from_temp_dir
 
 if TYPE_CHECKING:
     from dvc_task.app import FSApp
@@ -373,3 +374,18 @@ class LocalCeleryQueue(BaseStashQueue):
         from .remove import celery_remove
 
         return celery_remove(self, *args, **kwargs)
+
+    def get_running_exps(self, fetch_refs: bool = True) -> Dict[str, Dict]:
+        """Get the execution info of the currently running experiments
+
+        Args:
+            fetch_ref (bool): fetch completed checkpoints or not.
+        """
+        result: Dict[str, Dict] = {}
+        for entry in self.iter_active():
+            result.update(
+                fetch_running_exp_from_temp_dir(
+                    self, entry.stash_rev, fetch_refs
+                )
+            )
+        return result

dvc/repo/experiments/queue/tempdir.py

@@ -15,6 +15,7 @@ from ..executor.base import (
 )
 from ..executor.local import TempDirExecutor
 from .base import BaseStashQueue, QueueEntry, QueueGetResult
+from .utils import fetch_running_exp_from_temp_dir
 from .workspace import WorkspaceQueue
 
 if TYPE_CHECKING:
@@ -98,3 +99,13 @@ class TempDirQueue(WorkspaceQueue):
         exec_result: ExecutorResult,
     ) -> Dict[str, str]:
         return BaseStashQueue.collect_executor(exp, executor, exec_result)
+
+    def get_running_exps(self, fetch_refs: bool = True) -> Dict[str, Dict]:
+        result: Dict[str, Dict] = {}
+        for entry in self.iter_active():
+            result.update(
+                fetch_running_exp_from_temp_dir(
+                    self, entry.stash_rev, fetch_refs
+                )
+            )
+        return result

dvc/repo/experiments/queue/utils.py

@@ -0,0 +1,58 @@
+import logging
+from typing import TYPE_CHECKING, Dict
+
+from ..executor.base import ExecutorInfo, TaskStatus
+
+logger = logging.getLogger(__name__)
+
+
+if TYPE_CHECKING:
+    from .base import BaseStashQueue
+
+
+def fetch_running_exp_from_temp_dir(
+    queue: "BaseStashQueue", rev: str, fetch_refs: bool
+) -> Dict[str, Dict]:
+    """Fetch status of running exps out of current working directory
+
+    Args:
+        queue (BaseStashQueue):
+        rev (str): stash revision of the experiment
+        fetch_refs (bool): fetch running checkpoint results to local or not.
+
+    Returns:
+        Dict[str, Dict]: _description_
+    """
+    from dvc.scm import InvalidRemoteSCMRepo
+    from dvc.utils.serialize import load_json
+
+    from ..executor.local import TempDirExecutor
+
+    result: Dict[str, Dict] = {}
+    infofile = queue.get_infofile_path(rev)
+    try:
+        info = ExecutorInfo.from_dict(load_json(infofile))
+    except OSError:
+        return result
+    if info.status < TaskStatus.FAILED:
+        result[rev] = info.asdict()
+        if info.git_url and fetch_refs and info.status > TaskStatus.PREPARING:
+
+            def on_diverged(_ref: str, _checkpoint: bool):
+                return False
+
+            executor = TempDirExecutor.from_info(info)
+            try:
+                for ref in executor.fetch_exps(
+                    queue.scm,
+                    on_diverged=on_diverged,
+                ):
+                    logger.debug("Updated running experiment '%s'.", ref)
+                    last_rev = queue.scm.get_ref(ref)
+                    result[rev]["last"] = last_rev
+                    if last_rev:
+                        result[last_rev] = info.asdict()
+            except InvalidRemoteSCMRepo:
+                # ignore stale info files
+                del result[rev]
+    return result

dvc/repo/experiments/queue/workspace.py

@@ -1,13 +1,25 @@
+import json
 import logging
 from collections import defaultdict
-from typing import TYPE_CHECKING, Collection, Dict, Generator, Optional
+from typing import TYPE_CHECKING, Collection, Dict, Generator, List, Optional
 
+import psutil
 from funcy import first
+from voluptuous import Invalid
 
 from dvc.exceptions import DvcException
+from dvc.lock import make_lock
+from dvc.rwlock import RWLOCK_FILE, RWLOCK_LOCK, SCHEMA
+from dvc.utils import relpath
+from dvc.utils.fs import remove
 
 from ..exceptions import ExpQueueEmptyError
-from ..executor.base import BaseExecutor, ExecutorResult
+from ..executor.base import (
+    BaseExecutor,
+    ExecutorInfo,
+    ExecutorResult,
+    TaskStatus,
+)
 from ..executor.local import WorkspaceExecutor
 from ..refs import EXEC_BRANCH
 from .base import BaseStashQueue, QueueDoneResult, QueueEntry, QueueGetResult
@@ -150,3 +162,89 @@ class WorkspaceQueue(BaseStashQueue):
         follow: bool = False,
     ):
         raise NotImplementedError
+
+    def check_rwlock(
+        self,
+        hardlink: bool = False,
+        autocorrect: bool = False,
+    ) -> bool:
+        """Check and autocorrect the RWLock status for file paths.
+
+        Args:
+            hardlink (bool): use hardlink lock to guard rwlock file when on
+                            edit.
+            autocorrect (bool): autocorrect corrupted rwlock file.
+
+        Return:
+            (bool): if the pid alive.
+        """
+        path = self.repo.fs.path.join(self.repo.tmp_dir, RWLOCK_FILE)
+
+        rwlock_guard = make_lock(
+            self.repo.fs.path.join(self.repo.tmp_dir, RWLOCK_LOCK),
+            tmp_dir=self.repo.tmp_dir,
+            hardlink_lock=hardlink,
+        )
+        with rwlock_guard:
+            try:
+                with self.repo.fs.open(path, encoding="utf-8") as fobj:
+                    lock: Dict[str, List[Dict]] = SCHEMA(json.load(fobj))
+                file_path = first(lock["read"])
+                if not file_path:
+                    return False
+                lock_info = first(lock["read"][file_path])
+                pid = int(lock_info["pid"])
+                if psutil.pid_exists(pid):
+                    return True
+                cmd = lock_info["cmd"]
+                logger.warning(
+                    "Process '%s' with (Pid %s), in RWLock-file '%s'"
+                    " had been killed.",
+                    cmd,
+                    pid,
+                    relpath(path),
+                )
+            except FileNotFoundError:
+                return False
+            except json.JSONDecodeError:
+                logger.warning(
+                    "Unable to read RWLock-file '%s'. JSON structure is"
+                    " corrupted",
+                    relpath(path),
+                )
+            except Invalid:
+                logger.warning("RWLock-file '%s' format error.", relpath(path))
+            if autocorrect:
+                logger.warning(
+                    "Delete corrupted RWLock-file '%s'", relpath(path)
+                )
+                remove(path)
+            return False
+
+    def get_running_exps(self, fetch_refs: bool = True) -> Dict[str, Dict]:
+        from dvc.utils.serialize import load_json
+
+        assert self._EXEC_NAME
+        result: Dict[str, Dict] = {}
+
+        if not self.check_rwlock(autocorrect=True):
+            return result
+
+        infofile = self.get_infofile_path(self._EXEC_NAME)
+
+        try:
+            info = ExecutorInfo.from_dict(load_json(infofile))
+        except OSError:
+            return result
+
+        if info.status < TaskStatus.FAILED:
+            # If we are appending to a checkpoint branch in a workspace
+            # run, show the latest checkpoint as running.
+            if info.status == TaskStatus.SUCCESS:
+                return result
+            last_rev = self.scm.get_ref(EXEC_BRANCH)
+            if last_rev:
+                result[last_rev] = info.asdict()
+            else:
+                result[self._EXEC_NAME] = info.asdict()
+        return result

dvc/repo/experiments/show.py

@@ -1,6 +1,7 @@
 import logging
 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
 
@@ -17,16 +18,21 @@ if TYPE_CHECKING:
 logger = logging.getLogger(__name__)
 
 
+class ExpStatus(Enum):
+    Success = 0
+    Queued = 1
+    Running = 2
+    Failed = 3
+
+
 @error_handler
 def _collect_experiment_commit(
-    repo,
-    exp_rev,
-    stash=False,
-    sha_only=True,
+    repo: "Repo",
+    exp_rev: str,
+    status: ExpStatus = ExpStatus.Success,
     param_deps=False,
     running=None,
     onerror: Optional[Callable] = None,
-    is_baseline: bool = False,
 ):
     from dvc.dependency import ParamsDependency, RepoDependency
 
@@ -67,37 +73,35 @@ def _collect_experiment_commit(
             if not (out.is_metric or out.is_plot)
         }
 
-        res["queued"] = stash
-        if running is not None and exp_rev in running:
-            res["running"] = True
-            res["executor"] = running[exp_rev].get("location")
+        res["status"] = status.name
+        if status == ExpStatus.Running:
+            res["executor"] = running.get(exp_rev, {}).get("location", None)
         else:
-            res["running"] = False
             res["executor"] = None
-        if not stash:
+
+        if status == ExpStatus.Failed:
+            res["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
             )
             res["metrics"] = vals
 
-        if not sha_only and rev != "workspace":
-            name: Optional[str] = None
-            if is_baseline:
-                for refspec in ["refs/tags", "refs/heads"]:
-                    name = repo.scm.describe(rev, base=refspec)
-                    if name:
-                        break
-            if not name:
-                name = repo.experiments.get_exact_name(rev)
-            if name:
-                name = name.rsplit("/")[-1]
-                res["name"] = name
-
     return res
 
 
 def _collect_experiment_branch(
-    res, repo, branch, baseline, onerror: Optional[Callable] = None, **kwargs
+    res,
+    repo,
+    branch,
+    baseline,
+    onerror: Optional[Callable] = None,
+    running=None,
+    **kwargs,
 ):
     from dvc.scm import resolve_rev
 
@@ -105,8 +109,14 @@ def _collect_experiment_branch(
     prev = None
     revs = list(repo.scm.branch_revs(exp_rev, baseline))
     for rev in revs:
+        status = ExpStatus.Running if rev in running else ExpStatus.Success
         collected_exp = _collect_experiment_commit(
-            repo, rev, onerror=onerror, **kwargs
+            repo,
+            rev,
+            onerror=onerror,
+            status=status,
+            running=running,
+            **kwargs,
         )
         if len(revs) > 1:
             exp = {"checkpoint_tip": exp_rev}
@@ -129,12 +139,129 @@ def _collect_experiment_branch(
     return res
 
 
+def get_names(repo: "Repo", result: Dict[str, Dict[str, Any]]):
+
+    rev_set = set()
+    baseline_set = set()
+    for baseline in result:
+        for rev in result[baseline]:
+            if rev == "baseline":
+                rev = baseline
+                baseline_set.add(rev)
+            if rev != "workspace":
+                rev_set.add(rev)
+
+    names: Dict[str, Optional[str]] = {}
+    for base in ("refs/tags/", "refs/heads/"):
+        if rev_set:
+            names.update(
+                (rev, ref[len(base) :])
+                for rev, ref in repo.scm.describe(
+                    baseline_set, base=base
+                ).items()
+                if ref is not None
+            )
+            rev_set.difference_update(names.keys())
+
+    exact_name = repo.experiments.get_exact_name(rev_set)
+
+    for baseline, baseline_results in result.items():
+        for rev, rev_result in baseline_results.items():
+            name: Optional[str] = None
+            if rev == "baseline":
+                rev = baseline
+                if rev == "workspace":
+                    continue
+                name = names.get(rev, None)
+            name = name 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]],
+    running: Dict[str, Any],
+    **kwargs,
+) -> Dict[str, Dict[str, Any]]:
+    result: Dict[str, Dict] = defaultdict(OrderedDict)
+    for entry in chain(
+        repo.experiments.tempdir_queue.iter_active(),
+        repo.experiments.celery_queue.iter_active(),
+    ):
+        stash_rev = entry.stash_rev
+        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(
+                repo,
+                stash_rev,
+                status=ExpStatus.Running,
+                running=running,
+                **kwargs,
+            )
+    return result
+
+
+def _collect_queued_experiment(
+    repo: "Repo",
+    found_revs: Dict[str, List[str]],
+    running: Dict[str, Any],
+    **kwargs,
+) -> Dict[str, Dict[str, Any]]:
+    result: Dict[str, Dict] = defaultdict(OrderedDict)
+    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(
+                repo,
+                stash_rev,
+                status=ExpStatus.Queued,
+                running=running,
+                **kwargs,
+            )
+    return result
+
+
+def _collect_failed_experiment(
+    repo: "Repo",
+    found_revs: Dict[str, List[str]],
+    running: Dict[str, Any],
+    **kwargs,
+) -> Dict[str, Dict[str, Any]]:
+    result: Dict[str, Dict] = defaultdict(OrderedDict)
+    for queue_done_result in repo.experiments.celery_queue.iter_failed():
+        entry = queue_done_result.entry
+        stash_rev = entry.stash_rev
+        if entry.baseline_rev in found_revs:
+            experiment = _collect_experiment_commit(
+                repo,
+                stash_rev,
+                status=ExpStatus.Failed,
+                running=running,
+                **kwargs,
+            )
+            result[entry.baseline_rev][stash_rev] = experiment
+    return result
+
+
+def update_new(
+    to_dict: Dict[str, Dict[str, Any]], from_dict: Dict[str, Dict[str, Any]]
+):
+    for baseline, experiments in from_dict.items():
+        for rev, experiment in experiments.items():
+            to_dict[baseline][rev] = to_dict[baseline].get(rev, experiment)
+
+
 def show(
     repo: "Repo",
     all_branches=False,
     all_tags=False,
     revs: Union[List[str], str, None] = None,
     all_commits=False,
+    hide_queued=False,
+    hide_failed=False,
     sha_only=False,
     num=1,
     param_deps=False,
@@ -160,17 +287,52 @@ def show(
         iter_revs(repo.scm, revs, num, all_branches, all_tags, all_commits)
     )
 
-    running = repo.experiments.get_running_exps(fetch_refs=fetch_running)
+    running: Dict[str, Dict] = repo.experiments.get_running_exps(
+        fetch_refs=fetch_running
+    )
+
+    queued_experiment = (
+        _collect_queued_experiment(
+            repo,
+            found_revs,
+            running,
+            param_deps=param_deps,
+            onerror=onerror,
+        )
+        if not hide_queued
+        else {}
+    )
+
+    active_experiment = _collect_active_experiment(
+        repo,
+        found_revs,
+        running,
+        param_deps=param_deps,
+        onerror=onerror,
+    )
+
+    failed_experiments = (
+        _collect_failed_experiment(
+            repo,
+            found_revs,
+            running,
+            sha_only=sha_only,
+            param_deps=param_deps,
+            onerror=onerror,
+        )
+        if not hide_failed
+        else {}
+    )
 
     for rev in found_revs:
+        status = ExpStatus.Running if rev in running else ExpStatus.Success
         res[rev]["baseline"] = _collect_experiment_commit(
             repo,
             rev,
-            sha_only=sha_only,
+            status=status,
             param_deps=param_deps,
             running=running,
             onerror=onerror,
-            is_baseline=True,
         )
 
         if rev == "workspace":
@@ -191,31 +353,20 @@ def show(
                 repo,
                 exp_ref,
                 rev,
-                sha_only=sha_only,
                 param_deps=param_deps,
                 running=running,
                 onerror=onerror,
             )
 
-        # collect standalone & celery experiments
-        for entry in chain(
-            repo.experiments.tempdir_queue.iter_active(),
-            repo.experiments.celery_queue.iter_active(),
-            repo.experiments.celery_queue.iter_queued(),
-        ):
-            stash_rev = entry.stash_rev
-            if entry.baseline_rev in found_revs:
-                if stash_rev not in running or not running[stash_rev].get(
-                    "last"
-                ):
-                    experiment = _collect_experiment_commit(
-                        repo,
-                        stash_rev,
-                        sha_only=sha_only,
-                        stash=stash_rev not in running,
-                        param_deps=param_deps,
-                        running=running,
-                        onerror=onerror,
-                    )
-                    res[entry.baseline_rev][stash_rev] = experiment
+    if not hide_failed:
+        update_new(res, failed_experiments)
+
+    update_new(res, active_experiment)
+
+    if not hide_queued:
+        update_new(res, queued_experiment)
+
+    if not sha_only:
+        get_names(repo, res)
+
     return res

dvc/repo/fetch.py

@@ -3,12 +3,15 @@ from contextlib import suppress
 from typing import TYPE_CHECKING, Optional
 
 from dvc.config import NoRemoteError
-from dvc.exceptions import DownloadError, FileTransferError
+from dvc.exceptions import DownloadError
 from dvc.fs import Schemes
 
 from . import locked
 
 if TYPE_CHECKING:
+    from dvc.repo import Repo
+    from dvc.types import TargetType
+    from dvc_data.hashfile.transfer import TransferResult
     from dvc_objects.db.base import ObjectDB
 
 logger = logging.getLogger(__name__)
@@ -42,7 +45,7 @@ def fetch(
     run_cache=False,
     revs=None,
     odb: Optional["ObjectDB"] = None,
-):
+) -> int:
     """Download data items from a cloud and imported repositories
 
     Returns:
@@ -55,6 +58,9 @@ def fetch(
         config.NoRemoteError: thrown when downloading only local files and no
             remote is configured
     """
+    from dvc.repo.imports import save_imports
+    from dvc_data.hashfile.transfer import TransferResult
+
     if isinstance(targets, str):
         targets = [targets]
 
@@ -63,116 +69,102 @@ def fetch(
         if _remote.worktree:
             return _fetch_worktree(self, _remote)
 
-    used = self.used_objs(
-        targets,
-        all_branches=all_branches,
-        all_tags=all_tags,
-        all_commits=all_commits,
-        with_deps=with_deps,
-        force=True,
-        remote=remote,
-        jobs=jobs,
-        recursive=recursive,
-        revs=revs,
-    )
-
-    downloaded = 0
-    failed = 0
+    failed_count = 0
 
     try:
         if run_cache:
             self.stage_cache.pull(remote)
     except DownloadError as exc:
-        failed += exc.amount
+        failed_count += exc.amount
 
+    no_remote_msg: Optional[str] = None
+    result = TransferResult(set(), set())
+    try:
+        d, f = _fetch(
+            self,
+            targets,
+            all_branches=all_branches,
+            all_tags=all_tags,
+            all_commits=all_commits,
+            with_deps=with_deps,
+            force=True,
+            remote=remote,
+            jobs=jobs,
+            recursive=recursive,
+            revs=revs,
+            odb=odb,
+        )
+        result.transferred.update(d)
+        result.failed.update(f)
+    except NoRemoteError as exc:
+        no_remote_msg = str(exc)
+
+    for rev in self.brancher(
+        revs=revs,
+        all_branches=all_branches,
+        all_tags=all_tags,
+        all_commits=all_commits,
+    ):
+        imported = save_imports(
+            self,
+            targets,
+            unpartial=not rev or rev == "workspace",
+            recursive=recursive,
+        )
+        result.transferred.update(imported)
+        result.failed.difference_update(imported)
+
+    failed_count += len(result.failed)
+
+    if failed_count:
+        if no_remote_msg:
+            logger.error(no_remote_msg)
+        raise DownloadError(failed_count)
+
+    return len(result.transferred)
+
+
+def _fetch(
+    repo: "Repo",
+    targets: "TargetType",
+    remote: Optional[str] = None,
+    jobs: Optional[int] = None,
+    odb: Optional["ObjectDB"] = None,
+    **kwargs,
+) -> "TransferResult":
+    from dvc_data.hashfile.transfer import TransferResult
+
+    result = TransferResult(set(), set())
+    used = repo.used_objs(
+        targets,
+        remote=remote,
+        jobs=jobs,
+        **kwargs,
+    )
     if odb:
         all_ids = set()
         for _odb, obj_ids in used.items():
             all_ids.update(obj_ids)
-        d, f = _fetch(
-            self,
+        d, f = repo.cloud.pull(
             all_ids,
             jobs=jobs,
             remote=remote,
             odb=odb,
         )
-        downloaded += d
-        failed += f
+        result.transferred.update(d)
+        result.failed.update(f)
     else:
         for src_odb, obj_ids in sorted(
             used.items(),
             key=lambda item: item[0] is not None
             and item[0].fs.protocol == Schemes.MEMORY,
         ):
-            d, f = _fetch(
-                self,
+            d, f = repo.cloud.pull(
                 obj_ids,
                 jobs=jobs,
                 remote=remote,
                 odb=src_odb,
             )
-            downloaded += d
-            failed += f
-
-    d, f = _fetch_partial_imports(
-        self,
-        targets,
-        all_branches=all_branches,
-        all_tags=all_tags,
-        all_commits=all_commits,
-        recursive=recursive,
-        revs=revs,
-    )
-    downloaded += d
-    failed += f
-
-    if failed:
-        raise DownloadError(failed)
-
-    return downloaded
-
-
-def _fetch(repo, obj_ids, **kwargs):
-    downloaded = 0
-    failed = 0
-    try:
-        downloaded += repo.cloud.pull(obj_ids, **kwargs)
-    except FileTransferError as exc:
-        failed += exc.amount
-    return downloaded, failed
-
-
-def _fetch_partial_imports(repo, targets, **kwargs):
-    from dvc.stage.exceptions import DataSourceChanged
-
-    downloaded = 0
-    failed = 0
-    for stage in repo.partial_imports(targets, **kwargs):
-        dep = stage.deps[0]
-        out = stage.outs[0]
-        try:
-            if dep.changed_checksum():
-                raise DataSourceChanged(f"{stage} ({dep})")
-
-            logger.info("Importing '%s' -> '%s'", dep, out)
-            if stage.is_repo_import:
-                obj = dep.get_obj()
-                meta = dep.get_meta()
-                out.hash_info = obj.hash_info
-                out.meta = meta
-            else:
-                out.transfer(
-                    dep.def_path, jobs=kwargs.get("jobs"), odb=repo.odb.local
-                )
-        except DataSourceChanged as exc:
-            logger.warning(f"{exc}")
-            failed += 1
-            continue
-        if not any(
-            kwargs.get(kw, None)
-            for kw in ("all_branches", "all_tags", "all_commits", "revs")
-        ):
-            stage.dump()
-
-        downloaded += 1
-    return downloaded, failed
+            result.transferred.update(d)
+            result.failed.update(f)
+    return result

dvc/repo/get_url.py

@@ -1,19 +1,17 @@
 import os
 
+from dvc import output
+from dvc.exceptions import URLMissingError
+from dvc.fs import download, parse_external_url
+from dvc.utils import resolve_output
 
-def get_url(url, out=None, jobs=None):
-    from dvc import dependency, output
-    from dvc.utils import resolve_output
 
+def get_url(url, out=None, *, config=None, jobs=None):
     out = resolve_output(url, out)
-
-    if os.path.exists(url):
-        url = os.path.abspath(url)
-
     out = os.path.abspath(out)
-
-    (dep,) = dependency.loads_from(None, [url])
     (out,) = output.loads_from(None, [out], use_cache=False)
-    if not dep.exists:
-        raise dep.DoesNotExistError(dep)
-    dep.download(out, jobs=jobs)
+
+    src_fs, src_path = parse_external_url(url, config)
+    if not src_fs.exists(src_path):
+        raise URLMissingError(url)
+    download(src_fs, src_path, out, jobs=jobs)

dvc/repo/imp_url.py

@@ -1,4 +1,5 @@
 import os
+from typing import TYPE_CHECKING
 
 from dvc.repo.scm_context import scm_context
 from dvc.utils import relpath, resolve_output, resolve_paths
@@ -7,6 +8,9 @@ from dvc.utils.fs import path_isin
 from ..exceptions import InvalidArgumentError, OutputDuplicationError
 from . import locked
 
+if TYPE_CHECKING:
+    from dvc.dvcfile import DVCFile
+
 
 @locked
 @scm_context
@@ -27,6 +31,7 @@ def imp_url(
     meta=None,
     jobs=None,
     fs_config=None,
+    version_aware: bool = False,
 ):
     from dvc.dvcfile import Dvcfile
     from dvc.stage import Stage, create_stage, restore_fields
@@ -54,6 +59,11 @@ def imp_url(
     ):
         url = relpath(url, wdir)
 
+    if version_aware:
+        if fs_config is None:
+            fs_config = {}
+        fs_config["version_aware"] = True
+
     stage = create_stage(
         Stage,
         self,
@@ -68,7 +78,7 @@ def imp_url(
 
     out_obj = stage.outs[0]
     out_obj.annot.update(desc=desc, type=type, labels=labels, meta=meta)
-    dvcfile = Dvcfile(self, stage.path)
+    dvcfile: "DVCFile" = Dvcfile(self, stage.path)  # type: ignore
     dvcfile.remove()
 
     try:

dvc/repo/imports.py

@@ -0,0 +1,132 @@
+import logging
+from tempfile import TemporaryDirectory
+from typing import TYPE_CHECKING, List, Set, Tuple, Union
+
+if TYPE_CHECKING:
+    from dvc.dependency.base import Dependency
+    from dvc.repo import Repo
+    from dvc.repo.index import Index, IndexView
+    from dvc.stage import Stage
+    from dvc.types import TargetType
+    from dvc_data.hashfile import HashInfo
+
+logger = logging.getLogger(__name__)
+
+
+def unfetched_view(
+    index: "Index", targets: "TargetType", unpartial: bool = False, **kwargs
+) -> Tuple["IndexView", List["Dependency"]]:
+    """Return index view of imports which have not been fetched.
+
+    Returns:
+        Tuple in the form (view, changed_deps) where changed_imports is a list
+        of import dependencies that cannot be fetched due to changed data
+        source.
+    """
+    changed_deps: List["Dependency"] = []
+
+    def need_fetch(stage: "Stage") -> bool:
+        if (
+            not stage.is_import
+            or stage.is_repo_import
+            or (stage.is_partial_import and not unpartial)
+        ):
+            return False
+
+        out = stage.outs[0]
+        if not out.changed_cache():
+            return False
+
+        dep = stage.deps[0]
+        if dep.changed_checksum():
+            changed_deps.append(dep)
+            return False
+
+        return True
+
+    unfetched = index.targets_view(targets, filter_fn=need_fetch, **kwargs)
+    return unfetched, changed_deps
+
+
+def partial_view(
+    index: "Index", targets: "TargetType", **kwargs
+) -> "IndexView":
+    return index.targets_view(
+        targets,
+        filter_fn=lambda s: s.is_partial_import,
+        **kwargs,
+    )
+
+
+def unpartial_imports(index: Union["Index", "IndexView"]) -> int:
+    """Update any outs in the index which are no longer partial imports.
+
+    Returns:
+        Total number of files which were unpartialed.
+    """
+    updated = 0
+    for out in index.outs:
+        # we need to use view[key] here and since the out fields have not been
+        # updated yet (out.get_entry() would return the partial-import state)
+        workspace, key = out.index_key
+        entry = index.data[workspace][key]
+        if out.stage.is_partial_import:
+            if out.stage.is_repo_import:
+                dep = out.stage.deps[0]
+                out.hash_info = dep.get_obj().hash_info
+                out.meta = dep.get_meta()
+            else:
+                out.hash_info = entry.hash_info
+                out.meta = entry.meta
+            out.stage.dump()
+            updated += out.meta.nfiles if out.meta.nfiles is not None else 1
+    return updated
+
+
+def save_imports(
+    repo: "Repo", targets: "TargetType", unpartial: bool = False, **kwargs
+) -> Set["HashInfo"]:
+    """Save (download) imports from their original source location.
+
+    Imports which are already cached will not be downloaded.
+
+    Returns:
+        Objects which were downloaded from source location.
+    """
+    from dvc.stage.exceptions import DataSourceChanged
+    from dvc.utils.fs import makedirs
+    from dvc_data.index import checkout, md5, save
+    from dvc_objects.fs.callbacks import Callback
+
+    downloaded: Set["HashInfo"] = set()
+
+    unfetched, changed = unfetched_view(
+        repo.index, targets, unpartial=unpartial, **kwargs
+    )
+    for dep in changed:
+        logger.warning(str(DataSourceChanged(f"{dep.stage} ({dep})")))
+
+    data_view = unfetched.data["repo"]
+    if len(data_view):
+        cache = repo.odb.local
+        if not cache.fs.exists(cache.path):
+            makedirs(cache.path)
+        with TemporaryDirectory(dir=cache.path) as tmpdir:
+            with Callback.as_tqdm_callback(
+                desc="Downloading imports from source",
+                unit="files",
+            ) as cb:
+                checkout(data_view, tmpdir, cache.fs, callback=cb)
+            md5(data_view)
+            save(data_view, odb=cache, hardlink=True)
+
+        downloaded.update(
+            entry.hash_info
+            for _, entry in data_view.iteritems()
+            if not entry.meta.isdir and entry.hash_info is not None
+        )
+
+    if unpartial:
+        unpartial_imports(partial_view(repo.index, targets, **kwargs))
+
+    return downloaded

dvc/repo/index.py

@@ -1,13 +1,13 @@
 from contextlib import suppress
 from typing import (
     TYPE_CHECKING,
+    Any,
     Callable,
     Dict,
     Iterable,
     Iterator,
     List,
     Optional,
-    Sequence,
     Set,
 )
 
@@ -22,11 +22,11 @@ if TYPE_CHECKING:
     from dvc.dependency import Dependency, ParamsDependency
     from dvc.fs import FileSystem
     from dvc.output import Output
-    from dvc.repo.stage import StageLoad
+    from dvc.repo.stage import StageInfo, StageLoad
     from dvc.stage import Stage
     from dvc.types import StrPath, TargetType
     from dvc_data.hashfile.hash_info import HashInfo
-    from dvc_data.index import DataIndex
+    from dvc_data.index import DataIndex, DataIndexKey, DataIndexView
     from dvc_objects.db import ObjectDB
 
 
@@ -64,6 +64,7 @@ class Index:
         self.stage_collector: "StageLoad" = repo.stage
         if stages is not None:
             self.stages: List["Stage"] = stages
+        self._collected_targets: Dict[int, List["StageInfo"]] = {}
 
     @cached_property
     def stages(self) -> List["Stage"]:  # pylint: disable=method-hidden
@@ -113,6 +114,62 @@ class Index:
 
         return self.filter(is_stage_inside_path)
 
+    @staticmethod
+    def _hash_targets(
+        targets: Iterable[Optional[str]],
+        **kwargs: Any,
+    ) -> int:
+        return hash(
+            (
+                frozenset(targets),
+                kwargs.get("with_deps", False),
+                kwargs.get("recursive", False),
+            )
+        )
+
+    def _collect_targets(
+        self, targets: Optional["TargetType"], **kwargs: Any
+    ) -> List["StageInfo"]:
+        from itertools import chain
+
+        from dvc.repo.stage import StageInfo
+        from dvc.utils.collections import ensure_list
+
+        targets = ensure_list(targets)
+        if not targets:
+            return [StageInfo(stage) for stage in self.stages]
+        targets_hash = self._hash_targets(targets, **kwargs)
+        if targets_hash not in self._collected_targets:
+            self._collected_targets[targets_hash] = list(
+                chain.from_iterable(
+                    self.stage_collector.collect_granular(target, **kwargs)
+                    for target in targets
+                )
+            )
+        return self._collected_targets[targets_hash]
+
+    def targets_view(
+        self,
+        targets: "TargetType",
+        filter_fn: Optional[Callable[["Stage"], 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
+                targets.
+
+        Additional kwargs will be passed into the stage collector.
+        """
+        stage_infos = [
+            stage_info
+            for stage_info in self._collect_targets(targets, **kwargs)
+            if not filter_fn or filter_fn(stage_info.stage)
+        ]
+        return IndexView(self, stage_infos)
+
     @property
     def outs(self) -> Iterator["Output"]:
         for stage in self:
@@ -171,8 +228,7 @@ class Index:
     def data(self) -> "Dict[str, DataIndex]":
         from collections import defaultdict
 
-        from dvc.config import NoRemoteError
-        from dvc_data.index import DataIndex, DataIndexEntry
+        from dvc_data.index import DataIndex
 
         by_workspace: dict = defaultdict(DataIndex)
 
@@ -184,25 +240,18 @@ class Index:
                 continue
 
             workspace, key = out.index_key
-
-            try:
-                remote = self.repo.cloud.get_remote_odb(out.remote)
-            except NoRemoteError:
-                remote = None
-
             data_index = by_workspace[workspace]
 
             if out.files:
                 out.obj = out.get_obj()
 
-            data_index[key] = DataIndexEntry(
-                meta=out.meta,
-                obj=out.obj,
-                hash_info=out.hash_info,
-                odb=out.odb,
-                cache=out.odb,
-                remote=remote,
-            )
+            entry = out.get_entry()
+            entry.key = key
+            if out.stage.is_import and not out.stage.is_repo_import:
+                entry.fs = out.stage.deps[0].fs
+                entry.path = out.stage.deps[0].fs_path
+                entry.meta = out.stage.deps[0].meta
+            data_index.add(entry)
 
         return dict(by_workspace)
 
@@ -216,22 +265,11 @@ class Index:
         jobs: int = None,
     ) -> "ObjectContainer":
         from collections import defaultdict
-        from itertools import chain
-
-        from dvc.utils.collections import ensure_list
 
         used: "ObjectContainer" = defaultdict(set)
-        collect_targets: Sequence[Optional[str]] = (None,)
-        if targets:
-            collect_targets = ensure_list(targets)
-
-        pairs = chain.from_iterable(
-            self.stage_collector.collect_granular(
-                target, recursive=recursive, with_deps=with_deps
-            )
-            for target in collect_targets
+        pairs = self._collect_targets(
+            targets, recursive=recursive, with_deps=with_deps
         )
-
         for stage, filter_info in pairs:
             for odb, objs in stage.get_used_objs(
                 remote=remote,
@@ -242,26 +280,6 @@ class Index:
                 used[odb].update(objs)
         return used
 
-    def partial_imports(
-        self,
-        targets: "TargetType" = None,
-        recursive: bool = False,
-    ) -> List["Stage"]:
-        from itertools import chain
-
-        from dvc.utils.collections import ensure_list
-
-        collect_targets: Sequence[Optional[str]] = (None,)
-        if targets:
-            collect_targets = ensure_list(targets)
-
-        pairs = chain.from_iterable(
-            self.stage_collector.collect_granular(target, recursive=recursive)
-            for target in collect_targets
-        )
-
-        return [stage for stage, _ in pairs if stage.is_partial_import]
-
     # Following methods help us treat the collection as a set-like structure
     # and provides faux-immutability.
     # These methods do not preserve stages order.
@@ -327,6 +345,89 @@ class Index:
         return dict_md5(self.dumpd())
 
 
+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
+        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})
+
+    def __len__(self) -> int:
+        return len(self._stages)
+
+    def __contains__(self, stage: "Stage") -> bool:
+        # as we are keeping stages inside a list, it might be slower.
+        return stage in self._stages
+
+    def __iter__(self) -> Iterator["Stage"]:
+        yield from self._stages
+
+    def __getitem__(self, item: str) -> "Stage":
+        """Get a stage by its addressing attribute."""
+        for stage in self:
+            if stage.addressing == item:
+                return stage
+        raise KeyError(f"{item} - available stages are {self.stages}")
+
+    @property
+    def stages(self):
+        return self._stages
+
+    @property
+    def deps(self) -> Iterator["Dependency"]:
+        for stage in self:
+            yield from stage.deps
+
+    @property
+    def outs(self) -> Iterator["Output"]:
+        outs = set()
+        for stage, filter_info in self._stage_infos:
+            for out in stage.filter_outs(filter_info):
+                outs.add(out)
+        yield from outs
+
+    @cached_property
+    def _data_prefixes(self) -> Dict[str, Set["DataIndexKey"]]:
+        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)
+                )
+        return prefixes
+
+    @cached_property
+    def data(self) -> "Dict[str, DataIndexView]":
+        from functools import partial
+
+        from dvc_data.index import view
+
+        def key_filter(workspace: str, key: "DataIndexKey"):
+            try:
+                return key in self._data_prefixes[workspace]
+            except KeyError:
+                return False
+
+        return {
+            workspace: view(index, partial(key_filter, workspace))
+            for workspace, index in self._index.data.items()
+        }
+
+
 if __name__ == "__main__":
     import logging
 

dvc/repo/ls_url.py

@@ -1,13 +1,5 @@
 from dvc.exceptions import URLMissingError
-from dvc.fs import get_cloud_fs
-
-
-def parse_external_url(url, config=None):
-    remote_config = dict(config) if config else {}
-    remote_config["url"] = url
-    fs_cls, fs_config, fs_path = get_cloud_fs(None, **remote_config)
-    fs = fs_cls(**fs_config)
-    return fs, fs_path
+from dvc.fs import parse_external_url
 
 
 def ls_url(url, *, config=None, recursive=False):

dvc/repo/plots/__init__.py

@@ -444,6 +444,10 @@ def _collect_pipeline_files(repo, targets: List[str], props, onerror=None):
         if isinstance(dvcfile, PipelineFile):
             dvcfile_path = _relpath(repo.dvcfs, dvcfile.path)
             dvcfile_defs = dvcfile.load().get("plots", {})
+            if isinstance(dvcfile_defs, list):
+                dvcfile_defs = {
+                    k: v for elem in dvcfile_defs for k, v in elem.items()
+                }
             resolved = _resolve_definitions(
                 repo.dvcfs,
                 targets,

dvc/repo/push.py

@@ -1,6 +1,6 @@
 from typing import TYPE_CHECKING, Optional
 
-from dvc.exceptions import FileTransferError, UploadError
+from dvc.exceptions import UploadError
 
 from ..utils import glob_targets
 from . import locked
@@ -10,8 +10,8 @@ if TYPE_CHECKING:
 
 
 def _push_worktree(repo, remote):
-    from dvc_data.hashfile.tree import tree_from_index
     from dvc_data.index import checkout
+    from dvc_data.index.save import build_tree
 
     index = repo.index.data["repo"]
     checkout(index, remote.path, remote.fs)
@@ -39,8 +39,7 @@ def _push_worktree(repo, remote):
                         repo.fs.path.relparts(fs_path, out.fs_path)
                     )
                     entry.hash_info = hash_info
-                tree_meta, new_tree = tree_from_index(index, key)
-                new_tree.digest()
+                tree_meta, new_tree = build_tree(index, key)
                 out.obj = new_tree
                 out.hash_info = new_tree.hash_info
                 out.meta = tree_meta
@@ -103,18 +102,18 @@ def push(
             if not include_imports and dest_odb and dest_odb.read_only:
                 continue
             all_ids.update(obj_ids)
-        try:
-            pushed += self.cloud.push(all_ids, jobs, remote=remote, odb=odb)
-        except FileTransferError as exc:
-            raise UploadError(exc.amount)
+        result = self.cloud.push(all_ids, jobs, remote=remote, odb=odb)
+        if result.failed:
+            raise UploadError(len(result.failed))
+        pushed += len(result.transferred)
     else:
         for dest_odb, obj_ids in used.items():
             if dest_odb and dest_odb.read_only:
                 continue
-            try:
-                pushed += self.cloud.push(
-                    obj_ids, jobs, remote=remote, odb=odb or dest_odb
-                )
-            except FileTransferError as exc:
-                raise UploadError(exc.amount)
+            result = self.cloud.push(
+                obj_ids, jobs, remote=remote, odb=odb or dest_odb
+            )
+            if result.failed:
+                raise UploadError(len(result.failed))
+            pushed += len(result.transferred)
     return pushed

dvc/repo/status.py

@@ -14,7 +14,9 @@ def _joint_status(pairs):
     status_info = {}
 
     for stage, filter_info in pairs:
-        if stage.frozen and not stage.is_repo_import:
+        if stage.frozen and not (
+            stage.is_repo_import or stage.is_versioned_import
+        ):
             logger.warning(
                 "%s is frozen. Its dependencies are"
                 " not going to be shown in the status output.",

dvc/rwlock.py

@@ -1,4 +1,5 @@
 import json
+import logging
 import os
 from collections import defaultdict
 from contextlib import contextmanager
@@ -10,6 +11,9 @@ from .fs import localfs
 from .lock import make_lock
 from .utils import relpath
 
+logger = logging.getLogger(__name__)
+
+
 INFO_SCHEMA = {Required("pid"): int, Required("cmd"): str}
 
 SCHEMA = Schema(
@@ -19,6 +23,9 @@ SCHEMA = Schema(
     }
 )
 
+RWLOCK_FILE = "rwlock"
+RWLOCK_LOCK = "rwlock.lock"
+
 
 class RWLockFileCorruptedError(DvcException):
     def __init__(self, path):
@@ -35,10 +42,10 @@ class RWLockFileFormatError(DvcException):
 
 @contextmanager
 def _edit_rwlock(lock_dir, fs, hardlink):
-    path = fs.path.join(lock_dir, "rwlock")
+    path = fs.path.join(lock_dir, RWLOCK_FILE)
 
     rwlock_guard = make_lock(
-        fs.path.join(lock_dir, "rwlock.lock"),
+        fs.path.join(lock_dir, RWLOCK_LOCK),
         tmp_dir=lock_dir,
         hardlink_lock=hardlink,
     )

dvc/schema.py

@@ -118,7 +118,7 @@ PLOT_DEFINITION = {
     Output.PARAM_PLOT_TITLE: str,
     Output.PARAM_PLOT_TEMPLATE: str,
 }
-PLOTS_SCHEMA = {str: Any(PLOT_DEFINITION, None)}
+SINGLE_PLOT_SCHEMA = {str: Any(PLOT_DEFINITION, None)}
 FOREACH_IN = {
     Required(FOREACH_KWD): Any(dict, list, str),
     Required(DO_KWD): STAGE_DEFINITION,
@@ -127,9 +127,9 @@ SINGLE_PIPELINE_STAGE_SCHEMA = {
     str: either_or(STAGE_DEFINITION, FOREACH_IN, [FOREACH_KWD, DO_KWD])
 }
 MULTI_STAGE_SCHEMA = {
+    PLOTS: Any(SINGLE_PLOT_SCHEMA, [SINGLE_PLOT_SCHEMA]),
     STAGES: SINGLE_PIPELINE_STAGE_SCHEMA,
     VARS_KWD: VARS_SCHEMA,
-    PLOTS: PLOTS_SCHEMA,
 }
 
 COMPILED_SINGLE_STAGE_SCHEMA = Schema(SINGLE_STAGE_SCHEMA)

dvc/stage/__init__.py

@@ -257,6 +257,10 @@ class Stage(params.StageParams):
 
         return isinstance(self.deps[0], RepoDependency)
 
+    @property
+    def is_versioned_import(self):
+        return self.is_import and self.deps[0].fs.version_aware
+
     @property
     def is_checkpoint(self):
         """
@@ -620,7 +624,9 @@ class Stage(params.StageParams):
     @rwlocked(read=["deps", "outs"])
     def status(self, check_updates=False, filter_info=None):
         ret = []
-        show_import = self.is_repo_import and check_updates
+        show_import = (
+            self.is_repo_import or self.is_versioned_import
+        ) and check_updates
 
         if not self.frozen or show_import:
             self._status_deps(ret)

dvc/stage/cache.py

@@ -9,7 +9,6 @@ from funcy import cached_property, first
 from dvc import fs
 from dvc.exceptions import DvcException
 from dvc.utils import dict_sha256, relpath
-from dvc_data.hashfile.transfer import _log_exceptions
 
 if TYPE_CHECKING:
     from dvc_objects.db import ObjectDB
@@ -222,7 +221,7 @@ class StageCache:
 
         from_fs = from_odb.fs
         to_fs = to_odb.fs
-        func = _log_exceptions(fs.generic.copy)
+        func = fs.generic.log_exceptions(fs.generic.copy)
         runs = from_fs.path.join(from_odb.path, "runs")
 
         http_odb = next(

dvc/testing/api_tests.py

@@ -38,7 +38,7 @@ class TestAPI:
     )
     def test_filesystem(
         self,
-        M,  # from `pytest-test-utils`
+        M,
         tmp_dir,
         make_tmp_dir,
         scm,

dvc/testing/workspace_tests.py

@@ -3,6 +3,7 @@ import os
 import pytest
 
 from dvc.exceptions import URLMissingError
+from dvc.repo import Repo
 from dvc.repo.ls_url import ls_url, parse_external_url
 
 
@@ -165,6 +166,8 @@ class TestLsUrl:
         cloud.gen(
             {"dir/foo": "foo contents", "dir/subdir/bar": "bar contents"}
         )
+        if not (cloud / "dir").is_dir():
+            pytest.skip("Cannot create directories on this cloud")
         fs, _ = parse_external_url(cloud.url, cloud.config)
         result = ls_url(str(cloud / "dir"), config=cloud.config)
         match_files(
@@ -180,6 +183,8 @@ class TestLsUrl:
         cloud.gen(
             {"dir/foo": "foo contents", "dir/subdir/bar": "bar contents"}
         )
+        if not (cloud / "dir").is_dir():
+            pytest.skip("Cannot create directories on this cloud")
         fs, _ = parse_external_url(cloud.url, cloud.config)
         result = ls_url(
             str(cloud / "dir"), config=cloud.config, recursive=True
@@ -196,3 +201,40 @@ class TestLsUrl:
     def test_nonexistent(self, cloud):
         with pytest.raises(URLMissingError):
             ls_url(str(cloud / "dir"), config=cloud.config)
+
+
+class TestGetUrl:
+    def test_get_file(self, cloud, tmp_dir):
+        cloud.gen({"foo": "foo contents"})
+
+        Repo.get_url(str(cloud / "foo"), "foo_imported", config=cloud.config)
+
+        assert (tmp_dir / "foo_imported").is_file()
+        assert (tmp_dir / "foo_imported").read_text() == "foo contents"
+
+    def test_get_dir(self, cloud, tmp_dir):
+        cloud.gen({"foo": {"foo": "foo contents"}})
+        if not (cloud / "foo").is_dir():
+            pytest.skip("Cannot create directories on this cloud")
+
+        Repo.get_url(str(cloud / "foo"), "foo_imported", config=cloud.config)
+
+        assert (tmp_dir / "foo_imported").is_dir()
+        assert (tmp_dir / "foo_imported" / "foo").is_file()
+        assert (tmp_dir / "foo_imported" / "foo").read_text() == "foo contents"
+
+    @pytest.mark.parametrize("dname", [".", "dir", "dir/subdir"])
+    def test_get_url_to_dir(self, cloud, tmp_dir, dname):
+        cloud.gen({"src": {"foo": "foo contents"}})
+        if not (cloud / "src").is_dir():
+            pytest.skip("Cannot create directories on this cloud")
+        tmp_dir.gen({"dir": {"subdir": {}}})
+
+        Repo.get_url(str(cloud / "src" / "foo"), dname, config=cloud.config)
+
+        assert (tmp_dir / dname).is_dir()
+        assert (tmp_dir / dname / "foo").read_text() == "foo contents"
+
+    def test_get_url_nonexistent(self, cloud):
+        with pytest.raises(URLMissingError):
+            Repo.get_url(str(cloud / "nonexistent"), config=cloud.config)

setup.cfg

@@ -63,11 +63,11 @@ install_requires =
     rich>=10.13.0
     pyparsing>=2.4.7
     typing-extensions>=3.7.4
-    scmrepo==0.1.1
-    dvc-render==0.0.11
-    dvc-task==0.1.3
+    scmrepo==0.1.2
+    dvc-render==0.0.12
+    dvc-task==0.1.4
     dvclive>=0.10.0
-    dvc-data==0.14.0
+    dvc-data==0.22.0
     dvc-http==2.27.2
     hydra-core>=1.1.0
     iterative-telemetry==0.0.5
@@ -95,11 +95,11 @@ hdfs =
     dvc-hdfs==2.19.0; python_version < '3.11'
 oss = dvc-oss==2.19.0
 s3 = dvc-s3==2.20.1
-ssh = dvc-ssh==2.19.0
+ssh = dvc-ssh==2.20.0
 # gssapi should not be included in all_remotes, because it doesn't have wheels
 # for Linux, so it will fail to compile if user doesn't have all the
 # requirements, including kerberos itself.
-ssh_gssapi = dvc-ssh[gssapi]==2.19.0
+ssh_gssapi = dvc-ssh[gssapi]==2.20.0
 webdav = dvc-webdav==2.19.0
 webhdfs = dvc-webhdfs==2.19.0
 # requests-kerberos requires krb5 & gssapi, which does not provide wheels Linux
@@ -109,12 +109,12 @@ testing =
     pytest-test-utils==0.0.8
 tests =
     %(terraform)s
-    dvc-ssh==2.19.0
+    dvc-ssh==2.20.0
     # Test requirements
     pytest==7.1.3
     pytest-cov==4.0.0
     pytest-xdist==2.5.0
-    pytest-mock==3.9.0
+    pytest-mock==3.10.0
     pytest-lazy-fixture==0.6.3
     %(testing)s
     # https://github.com/docker/docker-py/issues/2902
@@ -123,11 +123,9 @@ tests =
     pytest-timeout==2.1.0
     filelock==3.8.0
     # pylint requirements
-    pylint==2.15.2
+    pylint==2.15.4
     # we use this to suppress pytest-related false positives in our tests.
     pylint-pytest==1.1.2
-    # see https://github.com/PyCQA/pylint/issues/7488
-    astroid==2.12.9
     # we use this to suppress some messages in tests, eg: foo/bar naming,
     # and, protected method calls in our tests
     pylint-plugin-utils==0.7

tests/func/experiments/test_experiments.py

@@ -38,7 +38,7 @@ def test_new_simple(tmp_dir, scm, dvc, exp_stage, mocker, name, workspace):
         assert (tmp_dir / "metrics.yaml").read_text().strip() == "foo: 2"
 
     exp_name = name if name else ref_info.name
-    assert dvc.experiments.get_exact_name(exp) == exp_name
+    assert dvc.experiments.get_exact_name([exp])[exp] == exp_name
     assert resolve_rev(scm, exp_name) == exp
 
 
@@ -481,7 +481,7 @@ def test_subdir(tmp_dir, scm, dvc, workspace):
     with fs.open("dir/metrics.yaml", mode="r", encoding="utf-8") as fobj:
         assert fobj.read().strip() == "foo: 2"
 
-    assert dvc.experiments.get_exact_name(exp) == ref_info.name
+    assert dvc.experiments.get_exact_name([exp])[exp] == ref_info.name
     assert resolve_rev(scm, ref_info.name) == exp
 
 
@@ -525,7 +525,7 @@ def test_subrepo(tmp_dir, scm, workspace):
     with fs.open("dir/repo/metrics.yaml", mode="r", encoding="utf-8") as fobj:
         assert fobj.read().strip() == "foo: 2"
 
-    assert subrepo.dvc.experiments.get_exact_name(exp) == ref_info.name
+    assert subrepo.dvc.experiments.get_exact_name([exp])[exp] == ref_info.name
     assert resolve_rev(scm, ref_info.name) == exp
 
 

tests/func/experiments/test_show.py

@@ -17,10 +17,20 @@ from dvc.repo.experiments.executor.base import (
 from dvc.repo.experiments.queue.base import QueueEntry
 from dvc.repo.experiments.refs import CELERY_STASH, ExpRefInfo
 from dvc.repo.experiments.utils import exp_refs_by_rev
+from dvc.utils import relpath
 from dvc.utils.fs import makedirs
 from dvc.utils.serialize import YAMLFileCorruptedError
 from tests.func.test_repro_multistage import COPY_SCRIPT
 
+LOCK_CONTENTS = {
+    "read": {
+        "data/MNIST": [{"pid": 54062, "cmd": "dvc exp run"}],
+    },
+    "write": {
+        "data/MNIST": {"pid": 54062, "cmd": "dvc exp run"},
+    },
+}
+
 
 def make_executor_info(**kwargs):
     # set default values for required info fields
@@ -48,6 +58,21 @@ def test_show_no_commits(tmp_dir):
     assert Repo.init().experiments.show() == {}
 
 
+@pytest.mark.vscode
+def test_show_branch_and_tag_name(tmp_dir, scm, dvc, exp_stage):
+    with tmp_dir.branch("new/branch", new=True):
+        tmp_dir.scm_gen("branch", "branch", "commit")
+        branch_rev = scm.get_rev()
+
+    result = dvc.experiments.show(all_branches=True)
+    assert result[branch_rev]["baseline"]["data"]["name"] == "new/branch"
+
+    scm.tag("new/tag")
+    tag_rev = scm.get_rev()
+    result = dvc.experiments.show(all_tags=True)
+    assert result[tag_rev]["baseline"]["data"]["name"] == "new/tag"
+
+
 @pytest.mark.vscode
 def test_show_simple(tmp_dir, scm, dvc, exp_stage):
     assert dvc.experiments.show()["workspace"] == {
@@ -63,8 +88,7 @@ def test_show_simple(tmp_dir, scm, dvc, exp_stage):
                 "metrics": {"metrics.yaml": {"data": {"foo": 1}}},
                 "outs": {},
                 "params": {"params.yaml": {"data": {"foo": 1}}},
-                "queued": False,
-                "running": False,
+                "status": "Success",
                 "executor": None,
                 "timestamp": None,
             }
@@ -97,8 +121,7 @@ def test_show_experiment(tmp_dir, scm, dvc, exp_stage, workspace):
             "metrics": {"metrics.yaml": {"data": {"foo": 1}}},
             "outs": {},
             "params": {"params.yaml": {"data": {"foo": 1}}},
-            "queued": False,
-            "running": False,
+            "status": "Success",
             "executor": None,
             "timestamp": timestamp,
             "name": "master",
@@ -130,7 +153,7 @@ def test_show_queued(tmp_dir, scm, dvc, exp_stage):
     assert len(results) == 2
     exp = results[exp_rev]["data"]
     assert exp["name"] == "test_name"
-    assert exp["queued"]
+    assert exp["status"] == "Queued"
     assert exp["params"]["params.yaml"] == {"data": {"foo": 2}}
 
     # test that only queued experiments for the current baseline are returned
@@ -145,10 +168,68 @@ def test_show_queued(tmp_dir, scm, dvc, exp_stage):
     results = dvc.experiments.show()[new_rev]
     assert len(results) == 2
     exp = results[exp_rev]["data"]
-    assert exp["queued"]
+    assert exp["status"] == "Queued"
     assert exp["params"]["params.yaml"] == {"data": {"foo": 3}}
 
 
+@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(
+        scm.gitpython.repo.rev_parse(baseline_rev).committed_date
+    )
+
+    dvc.experiments.run(
+        failed_exp_stage.addressing, params=["foo=2"], queue=True
+    )
+    exp_rev = dvc.experiments.scm.resolve_rev(f"{CELERY_STASH}@{{0}}")
+    dvc.experiments.run(run_all=True)
+    experiments = dvc.experiments.show()[baseline_rev]
+
+    expected_baseline = {
+        "data": {
+            "deps": {
+                "copy.py": {
+                    "hash": ANY,
+                    "size": ANY,
+                    "nfiles": None,
+                }
+            },
+            "metrics": {},
+            "outs": {},
+            "params": {"params.yaml": {"data": {"foo": 1}}},
+            "status": "Success",
+            "executor": None,
+            "timestamp": timestamp,
+            "name": "master",
+        }
+    }
+
+    expected_failed = {
+        "data": {
+            "timestamp": ANY,
+            "params": {"params.yaml": {"data": {"foo": 2}}},
+            "deps": {"copy.py": {"hash": None, "size": None, "nfiles": None}},
+            "outs": {},
+            "status": "Failed",
+            "executor": None,
+            "error": {
+                "msg": "Experiment run failed.",
+                "type": "",
+            },
+        }
+    }
+
+    assert len(experiments) == 2
+    for rev, exp in experiments.items():
+        if rev == "baseline":
+            assert exp == expected_baseline
+        else:
+            assert rev == exp_rev
+            assert exp == expected_failed
+
+
 @pytest.mark.vscode
 @pytest.mark.parametrize("workspace", [True, False])
 def test_show_checkpoint(
@@ -175,7 +256,7 @@ def test_show_checkpoint(
 
     for i, rev in enumerate(checkpoints):
         if i == 0:
-            name = dvc.experiments.get_exact_name(rev)
+            name = dvc.experiments.get_exact_name([rev])[rev]
             name = f"{rev[:7]} [{name}]"
             fs = "╓"
         elif i == len(checkpoints) - 1:
@@ -340,12 +421,20 @@ def test_show_sort(tmp_dir, scm, dvc, exp_stage, caplog):
 
 @pytest.mark.vscode
 @pytest.mark.parametrize(
-    "status, running", [(TaskStatus.RUNNING, True), (TaskStatus.FAILED, False)]
+    "status, pid_exists",
+    [
+        (TaskStatus.RUNNING, True),
+        (TaskStatus.RUNNING, False),
+        (TaskStatus.FAILED, False),
+    ],
 )
 def test_show_running_workspace(
-    tmp_dir, scm, dvc, exp_stage, capsys, status, running
+    tmp_dir, scm, dvc, exp_stage, capsys, caplog, status, pid_exists, mocker
 ):
+    from dvc.rwlock import RWLOCK_FILE
+
     pid_dir = os.path.join(dvc.tmp_dir, EXEC_TMP_DIR, EXEC_PID_DIR)
+    lock_file = relpath(os.path.join(dvc.tmp_dir, RWLOCK_FILE), str(tmp_dir))
     info = make_executor_info(
         location=BaseExecutor.DEFAULT_LOCATION, status=status
     )
@@ -356,8 +445,10 @@ def test_show_running_workspace(
     )
     makedirs(os.path.dirname(pidfile), True)
     (tmp_dir / pidfile).dump_json(info.asdict())
+    (tmp_dir / lock_file).dump_json(LOCK_CONTENTS)
+
+    mocker.patch("psutil.pid_exists", return_value=pid_exists)
 
-    print(dvc.experiments.show())
     assert dvc.experiments.show().get("workspace") == {
         "baseline": {
             "data": {
@@ -371,9 +462,12 @@ def test_show_running_workspace(
                 "metrics": {"metrics.yaml": {"data": {"foo": 1}}},
                 "params": {"params.yaml": {"data": {"foo": 1}}},
                 "outs": {},
-                "queued": False,
-                "running": True if running else False,
-                "executor": info.location if running else None,
+                "status": "Running"
+                if status == TaskStatus.RUNNING and pid_exists
+                else "Success",
+                "executor": info.location
+                if status == TaskStatus.RUNNING and pid_exists
+                else None,
                 "timestamp": None,
             }
         }
@@ -381,9 +475,20 @@ def test_show_running_workspace(
     capsys.readouterr()
     assert main(["exp", "show", "--csv"]) == 0
     cap = capsys.readouterr()
-    if running:
-        assert "Running" in cap.out
-        assert info.location in cap.out
+    if status == TaskStatus.RUNNING:
+        if pid_exists:
+            assert "Running" in cap.out
+            assert info.location in cap.out
+        else:
+            cmd = LOCK_CONTENTS["read"]["data/MNIST"][0]["cmd"]
+            pid = LOCK_CONTENTS["read"]["data/MNIST"][0]["pid"]
+            assert (
+                f"Process '{cmd}' with (Pid {pid}), in RWLock-file "
+                f"'{lock_file}' had been killed."
+            ) in caplog.text
+            assert (
+                f"Delete corrupted RWLock-file '{lock_file}'"
+            ) in caplog.text
 
 
 def test_show_running_tempdir(tmp_dir, scm, dvc, exp_stage, mocker):
@@ -418,20 +523,20 @@ def test_show_running_tempdir(tmp_dir, scm, dvc, exp_stage, mocker):
     makedirs(os.path.dirname(pidfile), True)
     (tmp_dir / pidfile).dump_json(info.asdict())
     mock_fetch = mocker.patch.object(
-        dvc.experiments,
-        "_fetch_running_exp",
+        dvc.experiments.tempdir_queue,
+        "get_running_exps",
         return_value={exp_rev: info.asdict()},
     )
 
     results = dvc.experiments.show()
     mock_fetch.assert_has_calls(
-        [mocker.call(stash_rev, pidfile, True)],
+        [mocker.call(True)],
     )
     exp_data = get_in(results, [baseline_rev, exp_rev, "data"])
-    assert exp_data["running"]
+    assert exp_data["status"] == "Running"
     assert exp_data["executor"] == info.location
 
-    assert not results["workspace"]["baseline"]["data"]["running"]
+    assert results["workspace"]["baseline"]["data"]["status"] == "Success"
 
 
 def test_show_running_celery(tmp_dir, scm, dvc, exp_stage, mocker):
@@ -453,11 +558,10 @@ def test_show_running_celery(tmp_dir, scm, dvc, exp_stage, mocker):
 
     results = dvc.experiments.show()
     exp_data = get_in(results, [baseline_rev, exp_rev, "data"])
-    assert not exp_data["queued"]
-    assert exp_data["running"]
+    assert exp_data["status"] == "Running"
     assert exp_data["executor"] == info.location
 
-    assert not results["workspace"]["baseline"]["data"]["running"]
+    assert results["workspace"]["baseline"]["data"]["status"] == "Success"
 
 
 def test_show_running_checkpoint(tmp_dir, scm, dvc, checkpoint_stage, mocker):
@@ -479,11 +583,17 @@ def test_show_running_checkpoint(tmp_dir, scm, dvc, checkpoint_stage, mocker):
         "iter_active",
         return_value=entries,
     )
+    mocker.patch.object(
+        dvc.experiments.celery_queue,
+        "iter_failed",
+        return_value=[],
+    )
     pidfile = queue.get_infofile_path(entries[0].stash_rev)
     info = make_executor_info(
         git_url="foo.git",
         baseline_rev=baseline_rev,
         location=TempDirExecutor.DEFAULT_LOCATION,
+        status=TaskStatus.RUNNING,
     )
     makedirs(os.path.dirname(pidfile), True)
     (tmp_dir / pidfile).dump_json(info.asdict())
@@ -495,10 +605,10 @@ def test_show_running_checkpoint(tmp_dir, scm, dvc, checkpoint_stage, mocker):
     results = dvc.experiments.show()
 
     checkpoint_res = get_in(results, [baseline_rev, checkpoint_rev, "data"])
-    assert checkpoint_res["running"]
+    assert checkpoint_res["status"] == "Running"
     assert checkpoint_res["executor"] == info.location
 
-    assert not results["workspace"]["baseline"]["data"]["running"]
+    assert results["workspace"]["baseline"]["data"]["status"] == "Success"
 
 
 def test_show_with_broken_repo(tmp_dir, scm, dvc, exp_stage, caplog):

tests/func/test_dvcfile.py

@@ -15,6 +15,16 @@ from dvc.stage.exceptions import StageFileDoesNotExistError
 from dvc.stage.loader import StageNotFound
 from dvc.utils.strictyaml import YAMLValidationError
 
+STAGE_EXAMPLE = {
+    "stage1": {
+        "cmd": "cp foo bar",
+        "desc": "stage desc",
+        "meta": {"key1": "value1", "key2": "value2"},
+        "deps": ["foo"],
+        "outs": [{"bar": {"desc": "bar desc", "meta": {"key": "value"}}}],
+    }
+}
+
 
 def test_run_load_one_for_multistage(tmp_dir, dvc):
     tmp_dir.gen("foo", "foo")
@@ -386,19 +396,7 @@ def test_dvcfile_try_dumping_parametrized_stage(tmp_dir, dvc, data, name):
 
 
 def test_dvcfile_load_dump_stage_with_desc_meta(tmp_dir, dvc):
-    data = {
-        "stages": {
-            "stage1": {
-                "cmd": "cp foo bar",
-                "desc": "stage desc",
-                "meta": {"key1": "value1", "key2": "value2"},
-                "deps": ["foo"],
-                "outs": [
-                    {"bar": {"desc": "bar desc", "meta": {"key": "value"}}}
-                ],
-            }
-        }
-    }
+    data = {"stages": STAGE_EXAMPLE}
     (tmp_dir / "dvc.yaml").dump(data)
 
     stage = dvc.stage.load_one(name="stage1")
@@ -411,3 +409,34 @@ def test_dvcfile_load_dump_stage_with_desc_meta(tmp_dir, dvc):
     # sanity check
     stage.dump()
     assert (tmp_dir / "dvc.yaml").parse() == data
+
+
+@pytest.mark.parametrize(
+    "data",
+    (
+        {
+            "plots": {
+                "path/to/plot": {"x": "value", "y": "value"},
+                "path/to/another/plot": {"x": "value", "y": "value"},
+            },
+            "stages": STAGE_EXAMPLE,
+        },
+        {
+            "plots": [
+                {"path/to/plot": {"x": "value", "y": "value"}},
+                {"path/to/another/plot": {"x": "value", "y": "value"}},
+            ],
+            "stages": STAGE_EXAMPLE,
+        },
+    ),
+)
+def test_dvcfile_load_with_plots(tmp_dir, dvc, data):
+    (tmp_dir / "dvc.yaml").dump(data)
+    plots = list(dvc.plots.collect())
+    top_level_plots = plots[0]["workspace"]["definitions"]["data"]["dvc.yaml"][
+        "data"
+    ]
+    assert all(
+        name in top_level_plots
+        for name in ("path/to/plot", "path/to/another/plot")
+    )

tests/func/test_get_url.py

@@ -2,8 +2,9 @@ import os
 
 import pytest
 
-from dvc.dependency.base import DependencyDoesNotExistError
+from dvc.exceptions import URLMissingError
 from dvc.repo import Repo
+from dvc.testing.workspace_tests import TestGetUrl as _TestGetUrl
 
 
 def test_get_file(tmp_dir):
@@ -26,7 +27,7 @@ def test_get_dir(tmp_dir):
 
 
 @pytest.mark.parametrize("dname", [".", "dir", "dir/subdir"])
-def test_get_url_to_dir(tmp_dir, tmp_path_factory, dname):
+def test_get_url_to_dir(tmp_dir, dname):
     tmp_dir.gen({"src": {"foo": "foo contents"}, "dir": {"subdir": {}}})
 
     Repo.get_url(os.path.join("src", "foo"), dname)
@@ -36,5 +37,9 @@ def test_get_url_to_dir(tmp_dir, tmp_path_factory, dname):
 
 
 def test_get_url_nonexistent(tmp_dir):
-    with pytest.raises(DependencyDoesNotExistError):
+    with pytest.raises(URLMissingError):
         Repo.get_url("nonexistent")
+
+
+class TestGetUrl(_TestGetUrl):
+    pass

tests/func/test_ignore.py

@@ -18,10 +18,6 @@ def _to_pattern_info_list(str_list: List):
     return [PatternInfo(a, "") for a in str_list]
 
 
-def walk_files(dvc, *args):
-    yield from dvc.dvcignore.find(*args)
-
-
 @pytest.mark.parametrize("filename", ["ignored", "тест"])
 def test_ignore(tmp_dir, dvc, filename):
     tmp_dir.gen({"dir": {filename: filename, "other": "text2"}})
@@ -29,13 +25,68 @@ def test_ignore(tmp_dir, dvc, filename):
 
     dvc._reset()
 
-    result = walk_files(dvc, dvc.fs, tmp_dir)
+    result = dvc.dvcignore.find(dvc.fs, tmp_dir)
     assert set(result) == {
         (tmp_dir / DvcIgnore.DVCIGNORE_FILE).fs_path,
         (tmp_dir / "dir" / "other").fs_path,
     }
 
 
+def test_walk(tmp_dir, dvc):
+    tmp_dir.gen(
+        {
+            "foo": "foo",
+            "bar": "bar",
+            "dir": {
+                "foo": "foo",
+                "bar": "bar",
+                "baz": "baz",
+                "subdir": {"foo": "foo", "qux": "qux"},
+            },
+        }
+    )
+    tmp_dir.gen(DvcIgnore.DVCIGNORE_FILE, "dir/bar\nfoo")
+
+    dvc._reset()
+
+    result = list(dvc.dvcignore.walk(dvc.fs, tmp_dir))
+    assert result[0][0] == str(tmp_dir)
+    assert result[0][1] == ["dir"]
+    assert set(result[0][2]) == {"bar", ".dvcignore"}
+    assert result[1][0] == str(tmp_dir / "dir")
+    assert result[1][1] == ["subdir"]
+    assert result[1][2] == ["baz"]
+    assert result[2][0] == str(tmp_dir / "dir" / "subdir")
+    assert result[2][1] == []
+    assert result[2][2] == ["qux"]
+
+    result = list(dvc.dvcignore.walk(dvc.fs, tmp_dir, detail=True))
+    assert result == [
+        (
+            str(tmp_dir),
+            {"dir": dvc.fs.info(tmp_dir / "dir")},
+            {
+                "bar": dvc.fs.info(tmp_dir / "bar"),
+                ".dvcignore": dvc.fs.info(tmp_dir / ".dvcignore"),
+            },
+        ),
+        (
+            str(tmp_dir / "dir"),
+            {
+                "subdir": dvc.fs.info(tmp_dir / "dir" / "subdir"),
+            },
+            {
+                "baz": dvc.fs.info(tmp_dir / "dir" / "baz"),
+            },
+        ),
+        (
+            str(tmp_dir / "dir" / "subdir"),
+            {},
+            {"qux": dvc.fs.info(tmp_dir / "dir" / "subdir" / "qux")},
+        ),
+    ]
+
+
 def test_rename_ignored_file(tmp_dir, dvc):
     tmp_dir.gen({"dir": {"ignored": "...", "other": "text"}})
 
@@ -133,7 +184,7 @@ def test_ignore_on_branch(tmp_dir, scm, dvc):
 
     dvc._reset()
 
-    result = walk_files(dvc, dvc.fs, tmp_dir)
+    result = dvc.dvcignore.find(dvc.fs, tmp_dir)
     assert set(result) == {
         (tmp_dir / "foo").fs_path,
         (tmp_dir / "bar").fs_path,
@@ -155,7 +206,7 @@ def test_match_nested(tmp_dir, dvc):
         }
     )
     dvc._reset()
-    result = walk_files(dvc, dvc.fs, tmp_dir)
+    result = dvc.dvcignore.find(dvc.fs, tmp_dir)
     assert set(result) == {
         (tmp_dir / DvcIgnore.DVCIGNORE_FILE).fs_path,
         (tmp_dir / "foo").fs_path,
@@ -167,7 +218,7 @@ def test_ignore_external(tmp_dir, scm, dvc, tmp_path_factory):
     ext_dir = TmpDir(os.fspath(tmp_path_factory.mktemp("external_dir")))
     ext_dir.gen({"y.backup": "y", "tmp": {"file": "ext tmp"}})
 
-    result = walk_files(dvc, dvc.fs, ext_dir)
+    result = dvc.dvcignore.find(dvc.fs, ext_dir)
     assert set(result) == {
         (ext_dir / "y.backup").fs_path,
         (ext_dir / "tmp" / "file").fs_path,
@@ -207,7 +258,7 @@ def test_ignore_blank_line(tmp_dir, dvc):
     tmp_dir.gen({"dir": {"ignored": "text", "other": "text2"}})
     tmp_dir.gen(DvcIgnore.DVCIGNORE_FILE, "foo\n\ndir/ignored")
     dvc._reset()
-    result = walk_files(dvc, dvc.fs, tmp_dir / "dir")
+    result = dvc.dvcignore.find(dvc.fs, tmp_dir / "dir")
     assert set(result) == {(tmp_dir / "dir" / "other").fs_path}
 
 
@@ -242,7 +293,7 @@ def test_ignore_file_in_parent_path(
     tmp_dir.gen(data_struct)
     tmp_dir.gen(DvcIgnore.DVCIGNORE_FILE, "\n".join(pattern_list))
     dvc._reset()
-    result = walk_files(dvc, dvc.fs, tmp_dir / "dir")
+    result = dvc.dvcignore.find(dvc.fs, tmp_dir / "dir")
     assert set(result) == {
         (tmp_dir / relpath).fs_path for relpath in result_set
     }
@@ -265,7 +316,7 @@ def test_ignore_sub_directory(tmp_dir, dvc):
     tmp_dir.gen({"dir": {DvcIgnore.DVCIGNORE_FILE: "doc/fortz"}})
 
     dvc._reset()
-    result = walk_files(dvc, dvc.fs, tmp_dir / "dir")
+    result = dvc.dvcignore.find(dvc.fs, tmp_dir / "dir")
     assert set(result) == {
         (tmp_dir / "dir" / "a" / "doc" / "fortz" / "a").fs_path,
         (tmp_dir / "dir" / DvcIgnore.DVCIGNORE_FILE).fs_path,
@@ -277,7 +328,7 @@ def test_ignore_directory(tmp_dir, dvc):
     tmp_dir.gen({"dir": {"fortz": {}, "a": {"fortz": {}}}})
     tmp_dir.gen({"dir": {DvcIgnore.DVCIGNORE_FILE: "fortz"}})
     dvc._reset()
-    result = walk_files(dvc, dvc.fs, tmp_dir / "dir")
+    result = dvc.dvcignore.find(dvc.fs, tmp_dir / "dir")
     assert set(result) == {
         (tmp_dir / "dir" / DvcIgnore.DVCIGNORE_FILE).fs_path
     }
@@ -288,7 +339,7 @@ def test_multi_ignore_file(tmp_dir, dvc, monkeypatch):
     tmp_dir.gen(DvcIgnore.DVCIGNORE_FILE, "dir/subdir/*_ignore")
     tmp_dir.gen({"dir": {DvcIgnore.DVCIGNORE_FILE: "!subdir/not_ignore"}})
     dvc._reset()
-    result = walk_files(dvc, dvc.fs, tmp_dir / "dir")
+    result = dvc.dvcignore.find(dvc.fs, tmp_dir / "dir")
     assert set(result) == {
         (tmp_dir / "dir" / "subdir" / "not_ignore").fs_path,
         (tmp_dir / "dir" / DvcIgnore.DVCIGNORE_FILE).fs_path,
@@ -386,7 +437,7 @@ def test_ignore_in_added_dir(tmp_dir, dvc):
     dvc._reset()
 
     ignored_path = tmp_dir / "dir" / "sub" / "ignored"
-    result = walk_files(dvc, dvc.fs, ignored_path)
+    result = dvc.dvcignore.find(dvc.fs, ignored_path)
     assert set(result) == set()
     assert ignored_path.exists()
 

tests/unit/command/test_debug.py

@@ -1,19 +0,0 @@
-from dvc.cli import main
-
-
-def test_viztracer(tmp_dir, dvc, mocker):
-    viztracer_profile = mocker.patch("dvc._debug.viztracer_profile")
-
-    assert main(["status", "--viztracer"]) == 0
-
-    args = viztracer_profile.call_args[1]
-    assert callable(args["path"])
-    assert args["max_stack_depth"] == -1
-
-    assert main(["status", "--viztracer", "--viztracer-depth", "5"]) == 0
-    args = viztracer_profile.call_args[1]
-    assert args["max_stack_depth"] == 5
-
-    assert main(["status", "--viztracer-depth", "2"]) == 0
-    args = viztracer_profile.call_args[1]
-    assert args["max_stack_depth"] == 2

tests/unit/command/test_experiments.py

@@ -95,6 +95,8 @@ def test_experiments_show(dvc, scm, mocker):
             "--all-tags",
             "--all-branches",
             "--all-commits",
+            "--hide-queued",
+            "--hide-failed",
             "--sha",
             "--param-deps",
             "-n",
@@ -115,6 +117,8 @@ def test_experiments_show(dvc, scm, mocker):
         all_tags=True,
         all_branches=True,
         all_commits=True,
+        hide_queued=True,
+        hide_failed=True,
         num=1,
         revs="foo",
         sha_only=True,
@@ -433,8 +437,7 @@ def test_show_experiments_csv(capsys):
                             }
                         }
                     },
-                    "queued": False,
-                    "running": False,
+                    "status": "Success",
                     "executor": None,
                     "metrics": {
                         "scores.json": {
@@ -470,8 +473,7 @@ def test_show_experiments_csv(capsys):
                             }
                         }
                     },
-                    "queued": False,
-                    "running": False,
+                    "status": "Success",
                     "executor": None,
                     "metrics": {
                         "scores.json": {
@@ -506,8 +508,7 @@ def test_show_experiments_csv(capsys):
                             }
                         }
                     },
-                    "queued": True,
-                    "running": True,
+                    "status": "Running",
                     "executor": None,
                     "metrics": {
                         "scores.json": {
@@ -561,8 +562,7 @@ def test_show_experiments_md(capsys):
                 "data": {
                     "timestamp": None,
                     "params": {"params.yaml": {"data": {"foo": 1}}},
-                    "queued": False,
-                    "running": False,
+                    "status": "Success",
                     "executor": None,
                     "metrics": {
                         "scores.json": {"data": {"bar": 0.9544670443829399}}
@@ -599,8 +599,7 @@ def test_show_experiments_sort_by(capsys, sort_order):
                             }
                         }
                     },
-                    "queued": False,
-                    "running": False,
+                    "status": "Success",
                     "executor": None,
                     "metrics": {},
                 }
@@ -617,8 +616,7 @@ def test_show_experiments_sort_by(capsys, sort_order):
                             }
                         }
                     },
-                    "queued": False,
-                    "running": False,
+                    "status": "Success",
                     "executor": None,
                     "metrics": {},
                     "name": "master",
@@ -634,8 +632,7 @@ def test_show_experiments_sort_by(capsys, sort_order):
                             }
                         }
                     },
-                    "queued": False,
-                    "running": False,
+                    "status": "Success",
                     "executor": None,
                     "metrics": {},
                     "name": "exp-89140",
@@ -651,8 +648,7 @@ def test_show_experiments_sort_by(capsys, sort_order):
                             }
                         }
                     },
-                    "queued": False,
-                    "running": False,
+                    "status": "Success",
                     "executor": None,
                     "metrics": {},
                     "name": "exp-43537",
@@ -668,8 +664,7 @@ def test_show_experiments_sort_by(capsys, sort_order):
                             }
                         }
                     },
-                    "queued": False,
-                    "running": False,
+                    "status": "Success",
                     "executor": None,
                     "metrics": {},
                     "name": "exp-4f89e",
@@ -915,8 +910,7 @@ def test_show_experiments_pcp(tmp_dir, mocker):
                 "data": {
                     "timestamp": None,
                     "params": {"params.yaml": {"data": {"foo": 1}}},
-                    "queued": False,
-                    "running": False,
+                    "status": "Success",
                     "executor": None,
                     "metrics": {
                         "scores.json": {"data": {"bar": 0.9544670443829399}}

tests/unit/command/test_imp_url.py

@@ -41,6 +41,7 @@ def test_import_url(mocker):
         labels=None,
         meta=None,
         jobs=4,
+        version_aware=False,
     )
 
 
@@ -98,6 +99,7 @@ def test_import_url_no_exec_download_flags(mocker, flag, expected):
         type=None,
         labels=None,
         meta=None,
+        version_aware=False,
         **expected
     )
 
@@ -135,6 +137,7 @@ def test_import_url_to_remote(mocker):
         labels=None,
         meta=None,
         jobs=None,
+        version_aware=False,
     )
 
 

tests/unit/fs/test_ssh.py

@@ -1,184 +0,0 @@
-# pylint: disable=W0613
-
-import getpass
-import os
-from unittest.mock import mock_open, patch
-
-import pytest
-from dvc_ssh import DEFAULT_PORT, SSHFileSystem
-
-
-def test_get_kwargs_from_urls():
-    user = "test"
-    host = "123.45.67.89"
-    port = 1234
-    path = "/path/to/dir"
-
-    # URL ssh://[user@]host.xz[:port]/path
-    url = f"ssh://{user}@{host}:{port}{path}"
-
-    assert SSHFileSystem._get_kwargs_from_urls(url) == {
-        "username": user,
-        "host": host,
-        "port": port,
-    }
-
-    # SCP-like URL ssh://[user@]host.xz:/absolute/path
-    url = f"ssh://{user}@{host}:{path}"
-
-    assert SSHFileSystem._get_kwargs_from_urls(url) == {
-        "username": user,
-        "host": host,
-    }
-
-
-def test_init():
-    fs = SSHFileSystem(
-        user="test", host="12.34.56.78", port="1234", password="xxx"
-    )
-    assert fs.fs_args["username"] == "test"
-    assert fs.fs_args["host"] == "12.34.56.78"
-    assert fs.fs_args["port"] == "1234"
-    assert fs.fs_args["password"] == fs.fs_args["passphrase"] == "xxx"
-
-
-def test_ssh_ask_password(mocker):
-    mocker.patch("dvc_ssh.ask_password", return_value="fish")
-    fs = SSHFileSystem(user="test", host="2.2.2.2", ask_password=True)
-    assert fs.fs_args["password"] == fs.fs_args["passphrase"] == "fish"
-
-
-mock_ssh_config = """
-Host example.com
-   User ubuntu
-   HostName 1.2.3.4
-   Port 1234
-   IdentityFile ~/.ssh/not_default.key
-"""
-
-
-@pytest.mark.parametrize(
-    "config,expected_host",
-    [
-        ({"host": "example.com"}, "1.2.3.4"),
-        ({"host": "not_in_ssh_config.com"}, "not_in_ssh_config.com"),
-    ],
-)
-@patch(
-    "builtins.open",
-    new_callable=mock_open,
-    read_data=mock_ssh_config,
-)
-def test_ssh_host_override_from_config(mock_file, config, expected_host):
-    fs = SSHFileSystem(**config)
-    assert fs.fs_args["host"] == expected_host
-
-
-@pytest.mark.parametrize(
-    "config,expected_user",
-    [
-        ({"host": "example.com", "user": "test1"}, "test1"),
-        ({"host": "example.com"}, "ubuntu"),
-        ({"host": "not_in_ssh_config.com", "user": "test1"}, "test1"),
-        ({"host": "not_in_ssh_config.com"}, getpass.getuser()),
-    ],
-)
-@patch(
-    "builtins.open",
-    new_callable=mock_open,
-    read_data=mock_ssh_config,
-)
-def test_ssh_user(mock_file, config, expected_user):
-    fs = SSHFileSystem(**config)
-    assert fs.fs_args["username"] == expected_user
-
-
-@pytest.mark.parametrize(
-    "config,expected_port",
-    [
-        ({"host": "example.com"}, 1234),
-        ({"host": "example.com", "port": 4321}, 4321),
-        ({"host": "not_in_ssh_config.com"}, DEFAULT_PORT),
-        ({"host": "not_in_ssh_config.com", "port": 2222}, 2222),
-    ],
-)
-@patch(
-    "builtins.open",
-    new_callable=mock_open,
-    read_data=mock_ssh_config,
-)
-def test_ssh_port(mock_file, config, expected_port):
-    fs = SSHFileSystem(**config)
-    assert fs.fs_args["port"] == expected_port
-
-
-@pytest.mark.parametrize(
-    "config,expected_keyfile",
-    [
-        (
-            {"host": "example.com", "keyfile": "dvc_config.key"},
-            ["dvc_config.key"],
-        ),
-        (
-            {"host": "example.com"},
-            ["~/.ssh/not_default.key"],
-        ),
-        (
-            {
-                "host": "not_in_ssh_config.com",
-                "keyfile": "dvc_config.key",
-            },
-            ["dvc_config.key"],
-        ),
-        ({"host": "not_in_ssh_config.com"}, None),
-    ],
-)
-@patch(
-    "builtins.open",
-    new_callable=mock_open,
-    read_data=mock_ssh_config,
-)
-def test_ssh_keyfile(mock_file, config, expected_keyfile):
-    fs = SSHFileSystem(**config)
-    expected_keyfiles = (
-        [os.path.expanduser(path) for path in expected_keyfile]
-        if expected_keyfile
-        else expected_keyfile
-    )
-    assert fs.fs_args.get("client_keys") == expected_keyfiles
-
-
-mock_ssh_multi_key_config = """
-IdentityFile file_1
-
-Host example.com
-    IdentityFile file_2
-"""
-
-
-@patch(
-    "builtins.open",
-    new_callable=mock_open,
-    read_data=mock_ssh_multi_key_config,
-)
-def test_ssh_multi_identity_files(mock_file):
-    fs = SSHFileSystem(host="example.com")
-    assert fs.fs_args.get("client_keys") == ["file_1", "file_2"]
-
-
-@pytest.mark.parametrize(
-    "config,expected_gss_auth",
-    [
-        ({"host": "example.com", "gss_auth": True}, True),
-        ({"host": "example.com", "gss_auth": False}, False),
-        ({"host": "not_in_ssh_config.com"}, False),
-    ],
-)
-@patch(
-    "builtins.open",
-    new_callable=mock_open,
-    read_data=mock_ssh_config,
-)
-def test_ssh_gss_auth(mock_file, config, expected_gss_auth):
-    fs = SSHFileSystem(**config)
-    assert fs.fs_args["gss_auth"] == expected_gss_auth