build(deps-dev): Bump types-requests from 2.27.9 to 2.27.10

Bumps [types-requests](https://github.com/python/typeshed) from 2.27.9 to 2.27.10.
- [Release notes](https://github.com/python/typeshed/releases)
- [Commits](https://github.com/python/typeshed/commits)

---
updated-dependencies:
- dependency-name: types-requests
  dependency-type: direct:development
  update-type: version-update:semver-patch
...

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

.git_archival.txt

@@ -0,0 +1 @@
+ref-names: $Format:%D$

.gitattributes

@@ -0,0 +1 @@
+.git_archival.txt  export-subst

.github/mergify.yml

@@ -1,7 +1,7 @@
 pull_request_rules:
   - name: backport patches to 1.11
     conditions:
-      - base=master
+      - base=main
       - label=backport-1.11
     actions:
       backport:

.github/workflows/benchmarks.yaml

@@ -17,7 +17,7 @@ jobs:
       - uses: actions/setup-python@v2
         with:
             python-version: 3.7
-      - uses: iterative/dvc-bench@master
+      - uses: iterative/dvc-bench@main
         with:
             pytest_options: "-k 'test_init or test_help'"
 

.github/workflows/packages.yaml

@@ -2,7 +2,7 @@ name: Build packages
 on:
   push:
     branches:
-    - master
+    - main
   release:
     types: [released, prereleased]
   workflow_dispatch:

.github/workflows/tests.yaml

@@ -67,10 +67,6 @@ jobs:
       run: |
         pip install --upgrade pip setuptools wheel
         pip install -e ".[dev]"
-    - name: setup git
-      run: |
-        git config --global user.email "[email protected]"
-        git config --global user.name "DVC Tester"
     - name: run tests
       timeout-minutes: 40
       run: >-
@@ -83,7 +79,7 @@ jobs:
         file: ./coverage.xml
         fail_ci_if_error: false
   notify:
-    if: github.ref == 'refs/heads/master' && failure()
+    if: github.ref == 'refs/heads/main' && failure()
     needs:
       - lint
       - tests
@@ -94,6 +90,6 @@ jobs:
       env:
         SLACK_WEBHOOK: ${{ secrets.SLACK_WEBHOOK }}
         SLACK_COLOR: ${{ job.status }}
-        SLACK_MESSAGE: 'CI Failed on master :boom:'
+        SLACK_MESSAGE: 'CI Failed on main :boom:'
         SLACK_TITLE: CI Status
         SLACK_USERNAME: DVC-CI

.pre-commit-config.yaml

@@ -6,9 +6,9 @@ repos:
       - id: black
         language_version: python3
     repo: https://github.com/ambv/black
-    rev: 21.12b0
+    rev: 22.1.0
   - repo: https://github.com/pre-commit/pre-commit-hooks
-    rev: v4.0.1
+    rev: v4.1.0
     hooks:
       - id: check-toml
       - id: check-yaml

README.rst

@@ -103,7 +103,7 @@ Snap (Snapcraft/Linux)
 
 This corresponds to the latest tagged release.
 Add ``--beta`` for the latest tagged release candidate,
-or ``--edge`` for the latest ``master`` version.
+or ``--edge`` for the latest ``main`` version.
 
 Choco (Chocolatey/Windows)
 --------------------------
@@ -183,8 +183,8 @@ Fedora / CentOS (rpm)
 Comparison to related technologies
 ==================================
 
-#. Data Engineering tools such as `AirFlow <https://airflow.apache.org/>`,
-   `Luigi <https://github.com/spotify/luigi>`, and others - in DVC data,
+#. Data Engineering tools such as `AirFlow <https://airflow.apache.org/>`_,
+   `Luigi <https://github.com/spotify/luigi>`_, and others - in DVC data,
    model and ML pipelines represent a single ML project focused on data
    scientists' experience.  Data engineering tools orchestrate multiple data
    projects and focus on efficient execution. A DVC project can be used from
@@ -212,7 +212,7 @@ Comparison to related technologies
 #. `DVC Studio <https://studio.iterative.ai/>`_ - official online
    platform for DVC projects.  It can be used to manage data and models, run
    and track experiments, and visualize and share results.  Also, it
-   integrates with `CML (CI/CD for ML) <https://cml.dev/>` for training
+   integrates with `CML (CI/CD for ML) <https://cml.dev/>`__ for training
    models in the cloud or Kubernetes.
 
 
@@ -254,7 +254,7 @@ Barrak, A., Eghan, E.E. and Adams, B. `On the Co-evolution of ML Pipelines and S
    :target: https://dvc.org
    :alt: DVC logo
 
-.. |CI| image:: https://github.com/iterative/dvc/workflows/Tests/badge.svg?branch=master
+.. |CI| image:: https://github.com/iterative/dvc/workflows/Tests/badge.svg?branch=main
    :target: https://github.com/iterative/dvc/actions
    :alt: GHA Tests
 
@@ -262,7 +262,7 @@ Barrak, A., Eghan, E.E. and Adams, B. `On the Co-evolution of ML Pipelines and S
    :target: https://codeclimate.com/github/iterative/dvc
    :alt: Code Climate
 
-.. |Coverage| image:: https://codecov.io/gh/iterative/dvc/branch/master/graph/badge.svg
+.. |Coverage| image:: https://codecov.io/gh/iterative/dvc/branch/main/graph/badge.svg
    :target: https://codecov.io/gh/iterative/dvc
    :alt: Codecov
 

dvc/__main__.py

@@ -1,7 +1,7 @@
 """Main entry point for DVC command line tool."""
 import sys
 
-from dvc.main import main
+from dvc.cli import main
 
 if __name__ == "__main__":
     sys.exit(main(sys.argv[1:]))

dvc/_debug.py

@@ -1,10 +1,32 @@
 from contextlib import ExitStack, contextmanager
+from datetime import datetime
 from typing import TYPE_CHECKING, Callable, Union
 
 if TYPE_CHECKING:
     from argparse import Namespace
 
 
+@contextmanager
+def viztracer_profile(
+    path: Union[Callable[[], str], str],
+    max_stack_depth: int = -1,
+):
+    try:
+        import viztracer  # pylint: disable=import-error
+    except ImportError:
+        print("Failed to run profiler, viztracer is not installed")
+        yield
+        return
+
+    tracer = viztracer.VizTracer(max_stack_depth=max_stack_depth)
+
+    tracer.start()
+    yield
+    tracer.stop()
+
+    tracer.save(path() if callable(path) else path)
+
+
 @contextmanager
 def yappi_profile(
     path: Union[Callable[[], str], str] = None, wall_clock: bool = True
@@ -100,12 +122,18 @@ def debugtools(args: "Namespace" = None, **kwargs):
         if kw.get("instrument") or kw.get("instrument_open"):
             stack.enter_context(instrument(kw.get("instrument_open", False)))
         if kw.get("yappi"):
-            from datetime import datetime
-
             output = "callgrind.dvc-{0:%Y%m%d}_{0:%H%M%S}.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,
+                )
+            )
         yield
 
 
@@ -118,6 +146,10 @@ def add_debugging_flags(parser):
     parser.add_argument(
         "--yappi", action="store_true", default=False, help=SUPPRESS
     )
+    parser.add_argument(
+        "--viztracer", action="store_true", default=False, help=SUPPRESS
+    )
+    parser.add_argument("--viztracer-depth", type=int, help=SUPPRESS)
     parser.add_argument("--cprofile-dump", help=SUPPRESS)
     parser.add_argument(
         "--pdb", action="store_true", default=False, help=SUPPRESS

dvc/api.py

@@ -21,13 +21,14 @@ def get_url(path, repo=None, rev=None, remote=None):
     with Repo.open(repo, rev=rev, subrepos=True, uninitialized=True) as _repo:
         fs_path = _repo.fs.path.join(_repo.root_dir, path)
         with reraise(FileNotFoundError, PathMissingError(path, repo)):
-            metadata = _repo.repo_fs.metadata(fs_path)
+            info = _repo.repo_fs.info(fs_path)
 
-        if not metadata.is_dvc:
+        if not info["isdvc"]:
             raise OutputNotFoundError(path, repo)
 
-        cloud = metadata.repo.cloud
-        md5 = metadata.repo.dvcfs.info(fs_path)["md5"]
+        cloud = info["repo"].cloud
+        dvc_path = _repo.fs.path.relpath(fs_path, info["repo"].root_dir)
+        md5 = info["repo"].dvcfs.info(dvc_path)["md5"]
         return cloud.get_url_for(remote, checksum=md5)
 
 

dvc/cli/__init__.py

@@ -0,0 +1,138 @@
+"""This module provides an entrypoint to the dvc cli and parsing utils."""
+
+import logging
+
+# Workaround for CPython bug. See [1] and [2] for more info.
+# [1] https://github.com/aws/aws-cli/blob/1.16.277/awscli/clidriver.py#L55
+# [2] https://bugs.python.org/issue29288
+
+"".encode("idna")
+
+logger = logging.getLogger("dvc")
+
+
+class DvcParserError(Exception):
+    """Base class for CLI parser errors."""
+
+    def __init__(self):
+        super().__init__("parser error")
+
+
+def parse_args(argv=None):
+    """Parses CLI arguments.
+
+    Args:
+        argv: optional list of arguments to parse. sys.argv is used by default.
+
+    Raises:
+        DvcParserError: raised for argument parsing errors.
+    """
+    from .parser import get_main_parser
+
+    parser = get_main_parser()
+    args = parser.parse_args(argv)
+    return args
+
+
+def main(argv=None):  # noqa: C901
+    """Main entry point for dvc CLI.
+
+    Args:
+        argv: optional list of arguments to parse. sys.argv is used by default.
+
+    Returns:
+        int: command's return code.
+    """
+    from dvc._debug import debugtools
+    from dvc.config import ConfigError
+    from dvc.exceptions import DvcException, NotDvcRepoError
+    from dvc.logger import FOOTER, disable_other_loggers
+
+    args = None
+    disable_other_loggers()
+
+    outerLogLevel = logger.level
+    try:
+        args = parse_args(argv)
+
+        level = None
+        if args.quiet:
+            level = logging.CRITICAL
+        elif args.verbose == 1:
+            level = logging.DEBUG
+        elif args.verbose > 1:
+            level = logging.TRACE
+
+        if level is not None:
+            logger.setLevel(level)
+
+        logger.trace(args)
+
+        if not args.quiet:
+            from dvc.ui import ui
+
+            ui.enable()
+
+        with debugtools(args):
+            cmd = args.func(args)
+            ret = cmd.do_run()
+    except ConfigError:
+        logger.exception("configuration error")
+        ret = 251
+    except KeyboardInterrupt:
+        logger.exception("interrupted by the user")
+        ret = 252
+    except NotDvcRepoError:
+        logger.exception("")
+        ret = 253
+    except DvcException:
+        ret = 255
+        logger.exception("")
+    except DvcParserError:
+        ret = 254
+    except Exception as exc:  # noqa, pylint: disable=broad-except
+        # pylint: disable=no-member
+        import errno
+
+        if isinstance(exc, OSError) and exc.errno == errno.EMFILE:
+            from dvc.utils import error_link
+
+            logger.exception(
+                "too many open files, please visit "
+                "{} to see how to handle this "
+                "problem".format(error_link("many-files")),
+                extra={"tb_only": True},
+            )
+        else:
+            from dvc.info import get_dvc_info
+
+            logger.exception("unexpected error")
+
+            dvc_info = get_dvc_info()
+            logger.debug("Version info for developers:\n%s", dvc_info)
+
+            logger.info(FOOTER)
+        ret = 255
+
+    try:
+        from dvc import analytics
+
+        if analytics.is_enabled():
+            analytics.collect_and_send_report(args, ret)
+
+        return ret
+    finally:
+        logger.setLevel(outerLogLevel)
+
+        from dvc.fs.pool import close_pools
+
+        # Closing pools by-hand to prevent weird messages when closing SSH
+        # connections. See https://github.com/iterative/dvc/issues/3248 for
+        # more info.
+        close_pools()
+
+        from dvc.external_repo import clean_repos
+
+        # Remove cached repos in the end of the call, these are anonymous
+        # so won't be reused by any other subsequent run anyway.
+        clean_repos()

dvc/cli/command.py

@@ -5,29 +5,6 @@ from abc import ABC, abstractmethod
 logger = logging.getLogger(__name__)
 
 
-def fix_subparsers(subparsers):
-    """Workaround for bug in Python 3. See more info at:
-    https://bugs.python.org/issue16308
-    https://github.com/iterative/dvc/issues/769
-
-    Args:
-        subparsers: subparsers to fix.
-    """
-    subparsers.required = True
-    subparsers.dest = "cmd"
-
-
-def append_doc_link(help_message, path):
-    from dvc.utils import format_link
-
-    if not path:
-        return help_message
-    doc_base = "https://man.dvc.org/"
-    return "{message}\nDocumentation: {link}".format(
-        message=help_message, link=format_link(doc_base + path)
-    )
-
-
 class CmdBase(ABC):
     UNINITIALIZED = False
 

dvc/cli/parser.py

@@ -1,11 +1,10 @@
-"""DVC command line interface"""
+"""Main parser for the dvc cli."""
 import argparse
 import logging
 import os
 import sys
 
-from ._debug import add_debugging_flags
-from .command import (
+from dvc.commands import (
     add,
     cache,
     check_ignore,
@@ -45,8 +44,8 @@ from .command import (
     update,
     version,
 )
-from .command.base import fix_subparsers
-from .exceptions import DvcParserError
+
+from . import DvcParserError
 
 logger = logging.getLogger(__name__)
 
@@ -96,7 +95,7 @@ def _find_parser(parser, cmd_cls):
     defaults = parser._defaults  # pylint: disable=protected-access
     if not cmd_cls or cmd_cls == defaults.get("func"):
         parser.print_help()
-        raise DvcParserError()
+        raise DvcParserError
 
     actions = parser._actions  # pylint: disable=protected-access
     for action in actions:
@@ -143,6 +142,8 @@ def get_parent_parser():
     When overwriting `-q` or `-v`, you need to instantiate a new object
     in order to prevent some weird behavior.
     """
+    from dvc._debug import add_debugging_flags
+
     parent_parser = argparse.ArgumentParser(add_help=False)
     add_debugging_flags(parent_parser)
     log_level_group = parent_parser.add_mutually_exclusive_group()
@@ -207,23 +208,11 @@ def get_main_parser():
         help="Use `dvc COMMAND --help` for command-specific help.",
     )
 
+    from .utils import fix_subparsers
+
     fix_subparsers(subparsers)
 
     for cmd in COMMANDS:
         cmd.add_parser(subparsers, parent_parser)
 
     return parser
-
-
-def parse_args(argv=None):
-    """Parses CLI arguments.
-
-    Args:
-        argv: optional list of arguments to parse. sys.argv is used by default.
-
-    Raises:
-        dvc.exceptions.DvcParserError: raised for argument parsing errors.
-    """
-    parser = get_main_parser()
-    args = parser.parse_args(argv)
-    return args

dvc/cli/utils.py

@@ -0,0 +1,31 @@
+def fix_subparsers(subparsers):
+    """Workaround for bug in Python 3. See more info at:
+    https://bugs.python.org/issue16308
+    https://github.com/iterative/dvc/issues/769
+
+    Args:
+        subparsers: subparsers to fix.
+    """
+    subparsers.required = True
+    subparsers.dest = "cmd"
+
+
+def append_doc_link(help_message, path):
+    from dvc.utils import format_link
+
+    if not path:
+        return help_message
+    doc_base = "https://man.dvc.org/"
+    return "{message}\nDocumentation: {link}".format(
+        message=help_message, link=format_link(doc_base + path)
+    )
+
+
+def fix_plumbing_subparsers(subparsers):
+    # metavar needs to be explicitly set in order to hide plumbing subcommands
+    # from the 'positional arguments' choices list
+    # see: https://bugs.python.org/issue22848
+    cmds = [
+        cmd for cmd, parser in subparsers.choices.items() if parser.add_help
+    ]
+    subparsers.metavar = "{{{}}}".format(",".join(cmds))

dvc/commands/add.py

@@ -1,8 +1,9 @@
 import argparse
 import logging
 
-from dvc.command import completion
-from dvc.command.base import CmdBase, append_doc_link
+from dvc.cli.command import CmdBase
+from dvc.cli.utils import append_doc_link
+from dvc.commands import completion
 
 logger = logging.getLogger(__name__)
 
@@ -101,7 +102,9 @@ def add_parser(subparsers, parent_parser):
         "--jobs",
         type=int,
         help=(
-            "Number of jobs to run simultaneously. "
+            "Only used along with '--to-remote'. "
+            "Number of jobs to run simultaneously "
+            "when pushing data to remote."
             "The default value is 4 * cpu_count(). "
             "For SSH remotes, the default is 4. "
         ),

dvc/commands/cache.py

@@ -1,8 +1,8 @@
 import argparse
 
-from dvc.command import completion
-from dvc.command.base import append_doc_link, fix_subparsers
-from dvc.command.config import CmdConfig
+from dvc.cli.utils import append_doc_link, fix_subparsers
+from dvc.commands import completion
+from dvc.commands.config import CmdConfig
 from dvc.ui import ui
 
 
@@ -28,7 +28,7 @@ class CmdCacheDir(CmdConfig):
 
 
 def add_parser(subparsers, parent_parser):
-    from dvc.command.config import parent_config_parser
+    from dvc.commands.config import parent_config_parser
 
     CACHE_HELP = "Manage cache settings."
 

dvc/commands/check_ignore.py

@@ -1,7 +1,8 @@
 import argparse
 
-from dvc.command import completion
-from dvc.command.base import CmdBase, append_doc_link
+from dvc.cli.command import CmdBase
+from dvc.cli.utils import append_doc_link
+from dvc.commands import completion
 from dvc.ui import ui
 
 

dvc/commands/checkout.py

@@ -1,8 +1,9 @@
 import argparse
 import operator
 
-from dvc.command import completion
-from dvc.command.base import CmdBase, append_doc_link
+from dvc.cli.command import CmdBase
+from dvc.cli.utils import append_doc_link
+from dvc.commands import completion
 from dvc.exceptions import CheckoutError
 from dvc.ui import ui
 

dvc/commands/commit.py

@@ -1,8 +1,9 @@
 import argparse
 import logging
 
-from dvc.command import completion
-from dvc.command.base import CmdBase, append_doc_link
+from dvc.cli.command import CmdBase
+from dvc.cli.utils import append_doc_link
+from dvc.commands import completion
 
 logger = logging.getLogger(__name__)
 

dvc/commands/completion.py

@@ -3,7 +3,8 @@ import logging
 
 import shtab
 
-from dvc.command.base import CmdBaseNoRepo, append_doc_link
+from dvc.cli.command import CmdBaseNoRepo
+from dvc.cli.utils import append_doc_link
 from dvc.ui import ui
 
 logger = logging.getLogger(__name__)
@@ -82,7 +83,7 @@ EXPERIMENT = {"bash": "_dvc_compgen_exps", "zsh": "_dvc_compadd_exps"}
 
 class CmdCompletion(CmdBaseNoRepo):
     def run(self):
-        from dvc.cli import get_main_parser
+        from dvc.cli.parser import get_main_parser
 
         parser = get_main_parser()
         shell = self.args.shell

dvc/commands/config.py

@@ -2,7 +2,8 @@ import argparse
 import logging
 import os
 
-from dvc.command.base import CmdBaseNoRepo, append_doc_link
+from dvc.cli.command import CmdBaseNoRepo
+from dvc.cli.utils import append_doc_link
 from dvc.ui import ui
 
 logger = logging.getLogger(__name__)

dvc/commands/daemon.py

@@ -1,5 +1,6 @@
-from dvc.command import completion
-from dvc.command.base import CmdBaseNoRepo, fix_subparsers
+from dvc.cli.command import CmdBaseNoRepo
+from dvc.cli.utils import fix_subparsers
+from dvc.commands import completion
 
 
 class CmdDaemonBase(CmdBaseNoRepo):

dvc/commands/dag.py

@@ -1,6 +1,7 @@
 import argparse
 
-from dvc.command.base import CmdBase, append_doc_link
+from dvc.cli.command import CmdBase
+from dvc.cli.utils import append_doc_link
 from dvc.ui import ui
 
 

dvc/commands/data_sync.py

@@ -1,8 +1,9 @@
 import argparse
 import logging
 
-from dvc.command import completion
-from dvc.command.base import CmdBase, append_doc_link
+from dvc.cli.command import CmdBase
+from dvc.cli.utils import append_doc_link
+from dvc.commands import completion
 
 logger = logging.getLogger(__name__)
 
@@ -18,7 +19,7 @@ class CmdDataBase(CmdBase):
 
 class CmdDataPull(CmdDataBase):
     def log_summary(self, stats):
-        from dvc.command.checkout import log_changes
+        from dvc.commands.checkout import log_changes
 
         log_changes(stats)
         super().log_summary(stats)
@@ -97,7 +98,7 @@ class CmdDataFetch(CmdDataBase):
 
 
 def shared_parent_parser():
-    from dvc.cli import get_parent_parser
+    from dvc.cli.parser import get_parent_parser
 
     # Parent parser used in pull/push/status
     parent_parser = argparse.ArgumentParser(
@@ -127,7 +128,7 @@ def shared_parent_parser():
 
 
 def add_parser(subparsers, _parent_parser):
-    from dvc.command.status import CmdDataStatus
+    from dvc.commands.status import CmdDataStatus
 
     # Pull
     PULL_HELP = "Download tracked files or directories from remote storage."

dvc/commands/destroy.py

@@ -1,7 +1,8 @@
 import argparse
 import logging
 
-from dvc.command.base import CmdBase, append_doc_link
+from dvc.cli.command import CmdBase
+from dvc.cli.utils import append_doc_link
 
 logger = logging.getLogger(__name__)
 

dvc/commands/diff.py

@@ -2,8 +2,9 @@ import argparse
 import logging
 import os
 
-from dvc.command import completion
-from dvc.command.base import CmdBase, append_doc_link
+from dvc.cli.command import CmdBase
+from dvc.cli.utils import append_doc_link
+from dvc.commands import completion
 from dvc.ui import ui
 
 logger = logging.getLogger(__name__)

dvc/commands/experiments/__init__.py

@@ -0,0 +1,60 @@
+import argparse
+
+from dvc.cli.utils import (
+    append_doc_link,
+    fix_plumbing_subparsers,
+    fix_subparsers,
+)
+from dvc.commands.experiments import (
+    apply,
+    branch,
+    diff,
+    exec_run,
+    gc,
+    init,
+    ls,
+    pull,
+    push,
+    remove,
+    run,
+    show,
+)
+
+SUB_COMMANDS = [
+    apply,
+    branch,
+    diff,
+    exec_run,
+    gc,
+    init,
+    ls,
+    pull,
+    push,
+    remove,
+    run,
+    show,
+]
+
+
+def add_parser(subparsers, parent_parser):
+    EXPERIMENTS_HELP = "Commands to run and compare experiments."
+
+    experiments_parser = subparsers.add_parser(
+        "experiments",
+        parents=[parent_parser],
+        aliases=["exp"],
+        description=append_doc_link(EXPERIMENTS_HELP, "exp"),
+        formatter_class=argparse.RawDescriptionHelpFormatter,
+        help=EXPERIMENTS_HELP,
+    )
+
+    experiments_subparsers = experiments_parser.add_subparsers(
+        dest="cmd",
+        help="Use `dvc experiments CMD --help` to display "
+        "command-specific help.",
+    )
+
+    fix_subparsers(experiments_subparsers)
+    for cmd in SUB_COMMANDS:
+        cmd.add_parser(experiments_subparsers, parent_parser)
+    fix_plumbing_subparsers(experiments_subparsers)

dvc/commands/experiments/apply.py

@@ -0,0 +1,42 @@
+import argparse
+import logging
+
+from dvc.cli.command import CmdBase
+from dvc.cli.utils import append_doc_link
+from dvc.commands import completion
+
+logger = logging.getLogger(__name__)
+
+
+class CmdExperimentsApply(CmdBase):
+    def run(self):
+
+        self.repo.experiments.apply(
+            self.args.experiment, force=self.args.force
+        )
+
+        return 0
+
+
+def add_parser(experiments_subparsers, parent_parser):
+
+    EXPERIMENTS_APPLY_HELP = (
+        "Apply the changes from an experiment to your workspace."
+    )
+    experiments_apply_parser = experiments_subparsers.add_parser(
+        "apply",
+        parents=[parent_parser],
+        description=append_doc_link(EXPERIMENTS_APPLY_HELP, "exp/apply"),
+        help=EXPERIMENTS_APPLY_HELP,
+        formatter_class=argparse.RawDescriptionHelpFormatter,
+    )
+    experiments_apply_parser.add_argument(
+        "--no-force",
+        action="store_false",
+        dest="force",
+        help="Fail if this command would overwrite conflicting changes.",
+    )
+    experiments_apply_parser.add_argument(
+        "experiment", help="Experiment to be applied."
+    ).complete = completion.EXPERIMENT
+    experiments_apply_parser.set_defaults(func=CmdExperimentsApply)

dvc/commands/experiments/branch.py

@@ -0,0 +1,34 @@
+import argparse
+import logging
+
+from dvc.cli.command import CmdBase
+from dvc.cli.utils import append_doc_link
+
+logger = logging.getLogger(__name__)
+
+
+class CmdExperimentsBranch(CmdBase):
+    def run(self):
+
+        self.repo.experiments.branch(self.args.experiment, self.args.branch)
+
+        return 0
+
+
+def add_parser(experiments_subparsers, parent_parser):
+
+    EXPERIMENTS_BRANCH_HELP = "Promote an experiment to a Git branch."
+    experiments_branch_parser = experiments_subparsers.add_parser(
+        "branch",
+        parents=[parent_parser],
+        description=append_doc_link(EXPERIMENTS_BRANCH_HELP, "exp/branch"),
+        help=EXPERIMENTS_BRANCH_HELP,
+        formatter_class=argparse.RawDescriptionHelpFormatter,
+    )
+    experiments_branch_parser.add_argument(
+        "experiment", help="Experiment to be promoted."
+    )
+    experiments_branch_parser.add_argument(
+        "branch", help="Git branch name to use."
+    )
+    experiments_branch_parser.set_defaults(func=CmdExperimentsBranch)

dvc/commands/experiments/diff.py

@@ -0,0 +1,118 @@
+import argparse
+import logging
+
+from dvc.cli.command import CmdBase
+from dvc.cli.utils import append_doc_link
+from dvc.commands import completion
+from dvc.commands.metrics import DEFAULT_PRECISION
+from dvc.exceptions import DvcException
+from dvc.ui import ui
+
+logger = logging.getLogger(__name__)
+
+
+class CmdExperimentsDiff(CmdBase):
+    def run(self):
+
+        try:
+            diff = self.repo.experiments.diff(
+                a_rev=self.args.a_rev,
+                b_rev=self.args.b_rev,
+                all=self.args.all,
+                param_deps=self.args.param_deps,
+            )
+        except DvcException:
+            logger.exception("failed to show experiments diff")
+            return 1
+
+        if self.args.json:
+            ui.write_json(diff)
+        else:
+            from dvc.compare import show_diff
+
+            precision = self.args.precision or DEFAULT_PRECISION
+            diffs = [("metrics", "Metric"), ("params", "Param")]
+            for idx, (key, title) in enumerate(diffs):
+                if idx:
+                    # we are printing tables even in `--quiet` mode
+                    # so we should also be printing the "table" separator
+                    ui.write(force=True)
+
+                show_diff(
+                    diff[key],
+                    title=title,
+                    markdown=self.args.markdown,
+                    no_path=self.args.no_path,
+                    on_empty_diff="diff not supported",
+                    precision=precision if key == "metrics" else None,
+                    a_rev=self.args.a_rev,
+                    b_rev=self.args.b_rev,
+                )
+
+        return 0
+
+
+def add_parser(experiments_subparsers, parent_parser):
+
+    EXPERIMENTS_DIFF_HELP = (
+        "Show changes between experiments in the DVC repository."
+    )
+
+    experiments_diff_parser = experiments_subparsers.add_parser(
+        "diff",
+        parents=[parent_parser],
+        description=append_doc_link(EXPERIMENTS_DIFF_HELP, "exp/diff"),
+        help=EXPERIMENTS_DIFF_HELP,
+        formatter_class=argparse.RawDescriptionHelpFormatter,
+    )
+    experiments_diff_parser.add_argument(
+        "a_rev", nargs="?", help="Old experiment to compare (defaults to HEAD)"
+    ).complete = completion.EXPERIMENT
+    experiments_diff_parser.add_argument(
+        "b_rev",
+        nargs="?",
+        help="New experiment to compare (defaults to the current workspace)",
+    ).complete = completion.EXPERIMENT
+    experiments_diff_parser.add_argument(
+        "--all",
+        action="store_true",
+        default=False,
+        help="Show unchanged metrics/params as well.",
+    )
+    experiments_diff_parser.add_argument(
+        "--param-deps",
+        action="store_true",
+        default=False,
+        help="Show only params that are stage dependencies.",
+    )
+    experiments_diff_parser.add_argument(
+        "--json",
+        "--show-json",
+        action="store_true",
+        default=False,
+        help="Show output in JSON format.",
+    )
+    experiments_diff_parser.add_argument(
+        "--md",
+        "--show-md",
+        action="store_true",
+        default=False,
+        dest="markdown",
+        help="Show tabulated output in the Markdown format (GFM).",
+    )
+    experiments_diff_parser.add_argument(
+        "--no-path",
+        action="store_true",
+        default=False,
+        help="Don't show metric/param path.",
+    )
+    experiments_diff_parser.add_argument(
+        "--precision",
+        type=int,
+        help=(
+            "Round metrics/params to `n` digits precision after the decimal "
+            f"point. Rounds to {DEFAULT_PRECISION} digits by default."
+        ),
+        metavar="<n>",
+    )
+    experiments_diff_parser.set_defaults(func=CmdExperimentsDiff)

dvc/commands/experiments/exec_run.py

@@ -0,0 +1,42 @@
+import logging
+
+from dvc.cli.command import CmdBaseNoRepo
+
+logger = logging.getLogger(__name__)
+
+
+class CmdExecutorRun(CmdBaseNoRepo):
+    """Run an experiment executor."""
+
+    def run(self):
+        from dvc.repo.experiments.executor.base import (
+            BaseExecutor,
+            ExecutorInfo,
+        )
+        from dvc.utils.serialize import load_json
+
+        info = ExecutorInfo.from_dict(load_json(self.args.infofile))
+        BaseExecutor.reproduce(
+            info=info,
+            rev="",
+            queue=None,
+            log_level=logger.getEffectiveLevel(),
+            infofile=self.args.infofile,
+        )
+        return 0
+
+
+def add_parser(experiments_subparsers, parent_parser):
+    EXEC_RUN_HELP = "Run an experiment executor."
+    exec_run_parser = experiments_subparsers.add_parser(
+        "exec-run",
+        parents=[parent_parser],
+        description=EXEC_RUN_HELP,
+        add_help=False,
+    )
+    exec_run_parser.add_argument(
+        "--infofile",
+        help="Path to executor info file",
+        default=None,
+    )
+    exec_run_parser.set_defaults(func=CmdExecutorRun)

dvc/commands/experiments/gc.py

@@ -0,0 +1,129 @@
+import argparse
+import logging
+
+from dvc.cli.command import CmdBase
+from dvc.cli.utils import append_doc_link
+from dvc.exceptions import InvalidArgumentError
+from dvc.ui import ui
+
+logger = logging.getLogger(__name__)
+
+
+def _raise_error_if_all_disabled(**kwargs):
+    if not any(kwargs.values()):
+        raise InvalidArgumentError(
+            "Either of `-w|--workspace`, `-a|--all-branches`, `-T|--all-tags` "
+            "or `--all-commits` needs to be set."
+        )
+
+
+class CmdExperimentsGC(CmdBase):
+    def run(self):
+        _raise_error_if_all_disabled(
+            all_branches=self.args.all_branches,
+            all_tags=self.args.all_tags,
+            all_commits=self.args.all_commits,
+            workspace=self.args.workspace,
+        )
+
+        msg = "This will remove all experiments except those derived from "
+
+        msg += "the workspace"
+        if self.args.all_commits:
+            msg += " and all git commits"
+        elif self.args.all_branches and self.args.all_tags:
+            msg += " and all git branches and tags"
+        elif self.args.all_branches:
+            msg += " and all git branches"
+        elif self.args.all_tags:
+            msg += " and all git tags"
+        msg += " of the current repo."
+        if self.args.queued:
+            msg += " Run queued experiments will be preserved."
+        else:
+            msg += " Run queued experiments will be removed."
+
+        logger.warning(msg)
+
+        msg = "Are you sure you want to proceed?"
+        if not self.args.force and not ui.confirm(msg):
+            return 1
+
+        removed = self.repo.experiments.gc(
+            all_branches=self.args.all_branches,
+            all_tags=self.args.all_tags,
+            all_commits=self.args.all_commits,
+            workspace=self.args.workspace,
+            queued=self.args.queued,
+        )
+
+        if removed:
+            ui.write(
+                f"Removed {removed} experiments.",
+                "To remove unused cache files",
+                "use 'dvc gc'.",
+            )
+        else:
+            ui.write("No experiments to remove.")
+        return 0
+
+
+def add_parser(experiments_subparsers, parent_parser):
+
+    EXPERIMENTS_GC_HELP = "Garbage collect unneeded experiments."
+    EXPERIMENTS_GC_DESCRIPTION = (
+        "Removes all experiments which are not derived from the specified"
+        "Git revisions."
+    )
+    experiments_gc_parser = experiments_subparsers.add_parser(
+        "gc",
+        parents=[parent_parser],
+        description=append_doc_link(EXPERIMENTS_GC_DESCRIPTION, "exp/gc"),
+        help=EXPERIMENTS_GC_HELP,
+        formatter_class=argparse.RawDescriptionHelpFormatter,
+    )
+    experiments_gc_parser.add_argument(
+        "-w",
+        "--workspace",
+        action="store_true",
+        default=False,
+        help="Keep experiments derived from the current workspace.",
+    )
+    experiments_gc_parser.add_argument(
+        "-a",
+        "--all-branches",
+        action="store_true",
+        default=False,
+        help="Keep experiments derived from the tips of all Git branches.",
+    )
+    experiments_gc_parser.add_argument(
+        "-T",
+        "--all-tags",
+        action="store_true",
+        default=False,
+        help="Keep experiments derived from all Git tags.",
+    )
+    experiments_gc_parser.add_argument(
+        "-A",
+        "--all-commits",
+        action="store_true",
+        default=False,
+        help="Keep experiments derived from all Git commits.",
+    )
+    experiments_gc_parser.add_argument(
+        "--queued",
+        action="store_true",
+        default=False,
+        help=(
+            "Keep queued experiments (experiments run queue will be cleared "
+            "by default)."
+        ),
+    )
+    experiments_gc_parser.add_argument(
+        "-f",
+        "--force",
+        action="store_true",
+        default=False,
+        help="Force garbage collection - automatically agree to all prompts.",
+    )
+    experiments_gc_parser.set_defaults(func=CmdExperimentsGC)

dvc/commands/experiments/init.py

@@ -0,0 +1,188 @@
+import argparse
+import logging
+
+from funcy import compact
+
+from dvc.cli.command import CmdBase
+from dvc.cli.utils import append_doc_link
+from dvc.exceptions import InvalidArgumentError
+from dvc.ui import ui
+
+logger = logging.getLogger(__name__)
+
+
+class CmdExperimentsInit(CmdBase):
+    DEFAULT_NAME = "train"
+    CODE = "src"
+    DATA = "data"
+    MODELS = "models"
+    DEFAULT_METRICS = "metrics.json"
+    DEFAULT_PARAMS = "params.yaml"
+    PLOTS = "plots"
+    DVCLIVE = "dvclive"
+    DEFAULTS = {
+        "code": CODE,
+        "data": DATA,
+        "models": MODELS,
+        "metrics": DEFAULT_METRICS,
+        "params": DEFAULT_PARAMS,
+        "plots": PLOTS,
+        "live": DVCLIVE,
+    }
+    EXP_LINK = "https://s.dvc.org/g/exp/run"
+
+    def run(self):
+        from dvc.commands.stage import parse_cmd
+
+        cmd = parse_cmd(self.args.command)
+        if not self.args.interactive and not cmd:
+            raise InvalidArgumentError("command is not specified")
+
+        from dvc.repo.experiments.init import init
+
+        defaults = {}
+        if not self.args.explicit:
+            config = self.repo.config["exp"]
+            defaults.update({**self.DEFAULTS, **config})
+
+        cli_args = compact(
+            {
+                "cmd": cmd,
+                "code": self.args.code,
+                "data": self.args.data,
+                "models": self.args.models,
+                "metrics": self.args.metrics,
+                "params": self.args.params,
+                "plots": self.args.plots,
+                "live": self.args.live,
+            }
+        )
+
+        initialized_stage = init(
+            self.repo,
+            name=self.args.name,
+            type=self.args.type,
+            defaults=defaults,
+            overrides=cli_args,
+            interactive=self.args.interactive,
+            force=self.args.force,
+        )
+
+        text = ui.rich_text.assemble(
+            "\n" if self.args.interactive else "",
+            "Created ",
+            (self.args.name, "bright_blue"),
+            " stage in ",
+            ("dvc.yaml", "green"),
+            ".",
+        )
+        if not self.args.run:
+            text.append_text(
+                ui.rich_text.assemble(
+                    " To run, use ",
+                    ('"dvc exp run"', "green"),
+                    ".\nSee ",
+                    (self.EXP_LINK, "repr.url"),
+                    ".",
+                )
+            )
+
+        ui.write(text, styled=True)
+        if self.args.run:
+            return self.repo.experiments.run(
+                targets=[initialized_stage.addressing]
+            )
+
+        return 0
+
+
+def add_parser(experiments_subparsers, parent_parser):
+
+    EXPERIMENTS_INIT_HELP = "Quickly setup any project to use experiments."
+    experiments_init_parser = experiments_subparsers.add_parser(
+        "init",
+        parents=[parent_parser],
+        description=append_doc_link(EXPERIMENTS_INIT_HELP, "exp/init"),
+        formatter_class=argparse.RawDescriptionHelpFormatter,
+        help=EXPERIMENTS_INIT_HELP,
+    )
+    experiments_init_parser.add_argument(
+        "command",
+        nargs=argparse.REMAINDER,
+        help="Command to execute.",
+        metavar="command",
+    )
+    experiments_init_parser.add_argument(
+        "--run",
+        action="store_true",
+        help="Run the experiment after initializing it",
+    )
+    experiments_init_parser.add_argument(
+        "--interactive",
+        "-i",
+        action="store_true",
+        help="Prompt for values that are not provided",
+    )
+    experiments_init_parser.add_argument(
+        "-f",
+        "--force",
+        action="store_true",
+        default=False,
+        help="Overwrite existing stage",
+    )
+    experiments_init_parser.add_argument(
+        "--explicit",
+        action="store_true",
+        default=False,
+        help="Only use the path values explicitly provided",
+    )
+    experiments_init_parser.add_argument(
+        "--name",
+        "-n",
+        help="Name of the stage to create (default: %(default)s)",
+        default=CmdExperimentsInit.DEFAULT_NAME,
+    )
+    experiments_init_parser.add_argument(
+        "--code",
+        help="Path to the source file or directory "
+        "which your experiments depend"
+        f" (default: {CmdExperimentsInit.CODE})",
+    )
+    experiments_init_parser.add_argument(
+        "--data",
+        help="Path to the data file or directory "
+        "which your experiments depend"
+        f" (default: {CmdExperimentsInit.DATA})",
+    )
+    experiments_init_parser.add_argument(
+        "--models",
+        help="Path to the model file or directory for your experiments"
+        f" (default: {CmdExperimentsInit.MODELS})",
+    )
+    experiments_init_parser.add_argument(
+        "--params",
+        help="Path to the parameters file for your experiments"
+        f" (default: {CmdExperimentsInit.DEFAULT_PARAMS})",
+    )
+    experiments_init_parser.add_argument(
+        "--metrics",
+        help="Path to the metrics file for your experiments"
+        f" (default: {CmdExperimentsInit.DEFAULT_METRICS})",
+    )
+    experiments_init_parser.add_argument(
+        "--plots",
+        help="Path to the plots file or directory for your experiments"
+        f" (default: {CmdExperimentsInit.PLOTS})",
+    )
+    experiments_init_parser.add_argument(
+        "--live",
+        help="Path to log dvclive outputs for your experiments"
+        f" (default: {CmdExperimentsInit.DVCLIVE})",
+    )
+    experiments_init_parser.add_argument(
+        "--type",
+        choices=["default", "dl"],
+        default="default",
+        help="Select type of stage to create (default: %(default)s)",
+    )
+    experiments_init_parser.set_defaults(func=CmdExperimentsInit)

dvc/commands/experiments/ls.py

@@ -0,0 +1,73 @@
+import argparse
+import logging
+
+from dvc.cli.command import CmdBase
+from dvc.cli.utils import append_doc_link
+
+logger = logging.getLogger(__name__)
+
+
+class CmdExperimentsList(CmdBase):
+    def run(self):
+        names_only = self.args.names_only
+        exps = self.repo.experiments.ls(
+            rev=self.args.rev,
+            git_remote=self.args.git_remote,
+            all_=self.args.all,
+        )
+        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]
+            if not names_only:
+                print(f"{name}:")
+            for exp_name in exps[baseline]:
+                indent = "" if names_only else "\t"
+                print(f"{indent}{exp_name}")
+
+        return 0
+
+
+def add_parser(experiments_subparsers, parent_parser):
+
+    EXPERIMENTS_LIST_HELP = "List local and remote experiments."
+    experiments_list_parser = experiments_subparsers.add_parser(
+        "list",
+        parents=[parent_parser],
+        description=append_doc_link(EXPERIMENTS_LIST_HELP, "exp/list"),
+        help=EXPERIMENTS_LIST_HELP,
+        formatter_class=argparse.RawDescriptionHelpFormatter,
+    )
+    experiments_list_parser.add_argument(
+        "--rev",
+        type=str,
+        default=None,
+        help=(
+            "List experiments derived from the specified revision. "
+            "Defaults to HEAD if neither `--rev` nor `--all` are specified."
+        ),
+        metavar="<rev>",
+    )
+    experiments_list_parser.add_argument(
+        "--all", action="store_true", help="List all experiments."
+    )
+    experiments_list_parser.add_argument(
+        "--names-only",
+        action="store_true",
+        help="Only output experiment names (without parent commits).",
+    )
+    experiments_list_parser.add_argument(
+        "git_remote",
+        nargs="?",
+        default=None,
+        help=(
+            "Optional Git remote name or Git URL. "
+            "If provided, experiments from the specified Git repository "
+            " will be listed instead of local ones."
+        ),
+        metavar="[<git_remote>]",
+    )
+    experiments_list_parser.set_defaults(func=CmdExperimentsList)

dvc/commands/experiments/pull.py

@@ -0,0 +1,96 @@
+import argparse
+import logging
+
+from dvc.cli.command import CmdBase
+from dvc.cli.utils import append_doc_link
+from dvc.ui import ui
+
+logger = logging.getLogger(__name__)
+
+
+class CmdExperimentsPull(CmdBase):
+    def run(self):
+
+        self.repo.experiments.pull(
+            self.args.git_remote,
+            self.args.experiment,
+            force=self.args.force,
+            pull_cache=self.args.pull_cache,
+            dvc_remote=self.args.dvc_remote,
+            jobs=self.args.jobs,
+            run_cache=self.args.run_cache,
+        )
+
+        ui.write(
+            f"Pulled experiment '{self.args.experiment}'",
+            f"from Git remote '{self.args.git_remote}'.",
+        )
+        if not self.args.pull_cache:
+            ui.write(
+                "To pull cached outputs for this experiment"
+                "from DVC remote storage,"
+                "re-run this command without '--no-cache'."
+            )
+
+        return 0
+
+
+def add_parser(experiments_subparsers, parent_parser):
+    EXPERIMENTS_PULL_HELP = "Pull an experiment from a Git remote."
+    experiments_pull_parser = experiments_subparsers.add_parser(
+        "pull",
+        parents=[parent_parser],
+        description=append_doc_link(EXPERIMENTS_PULL_HELP, "exp/pull"),
+        help=EXPERIMENTS_PULL_HELP,
+        formatter_class=argparse.RawDescriptionHelpFormatter,
+    )
+    experiments_pull_parser.add_argument(
+        "-f",
+        "--force",
+        action="store_true",
+        help="Replace local experiment already exists.",
+    )
+    experiments_pull_parser.add_argument(
+        "--no-cache",
+        action="store_false",
+        dest="pull_cache",
+        help=(
+            "Do not pull cached outputs for this experiment from DVC remote "
+            "storage."
+        ),
+    )
+    experiments_pull_parser.add_argument(
+        "-r",
+        "--remote",
+        dest="dvc_remote",
+        metavar="<name>",
+        help="Name of the DVC remote to use when pulling cached outputs.",
+    )
+    experiments_pull_parser.add_argument(
+        "-j",
+        "--jobs",
+        type=int,
+        metavar="<number>",
+        help=(
+            "Number of jobs to run simultaneously when pulling from DVC "
+            "remote storage."
+        ),
+    )
+    experiments_pull_parser.add_argument(
+        "--run-cache",
+        action="store_true",
+        default=False,
+        help="Pull run history for all stages.",
+    )
+    experiments_pull_parser.add_argument(
+        "git_remote",
+        help="Git remote name or Git URL.",
+        metavar="<git_remote>",
+    )
+    experiments_pull_parser.add_argument(
+        "experiment",
+        nargs="+",
+        help="Experiments to pull.",
+        metavar="<experiment>",
+    )
+    experiments_pull_parser.set_defaults(func=CmdExperimentsPull)

dvc/commands/experiments/push.py

@@ -0,0 +1,97 @@
+import argparse
+import logging
+
+from dvc.cli.command import CmdBase
+from dvc.cli.utils import append_doc_link
+from dvc.commands import completion
+from dvc.ui import ui
+
+logger = logging.getLogger(__name__)
+
+
+class CmdExperimentsPush(CmdBase):
+    def run(self):
+
+        self.repo.experiments.push(
+            self.args.git_remote,
+            self.args.experiment,
+            force=self.args.force,
+            push_cache=self.args.push_cache,
+            dvc_remote=self.args.dvc_remote,
+            jobs=self.args.jobs,
+            run_cache=self.args.run_cache,
+        )
+
+        ui.write(
+            f"Pushed experiment '{self.args.experiment}'"
+            f"to Git remote '{self.args.git_remote}'."
+        )
+        if not self.args.push_cache:
+            ui.write(
+                "To push cached outputs",
+                "for this experiment to DVC remote storage,"
+                "re-run this command without '--no-cache'.",
+            )
+
+        return 0
+
+
+def add_parser(experiments_subparsers, parent_parser):
+    EXPERIMENTS_PUSH_HELP = "Push a local experiment to a Git remote."
+    experiments_push_parser = experiments_subparsers.add_parser(
+        "push",
+        parents=[parent_parser],
+        description=append_doc_link(EXPERIMENTS_PUSH_HELP, "exp/push"),
+        help=EXPERIMENTS_PUSH_HELP,
+        formatter_class=argparse.RawDescriptionHelpFormatter,
+    )
+    experiments_push_parser.add_argument(
+        "-f",
+        "--force",
+        action="store_true",
+        help="Replace experiment in the Git remote if it already exists.",
+    )
+    experiments_push_parser.add_argument(
+        "--no-cache",
+        action="store_false",
+        dest="push_cache",
+        help=(
+            "Do not push cached outputs for this experiment to DVC remote "
+            "storage."
+        ),
+    )
+    experiments_push_parser.add_argument(
+        "-r",
+        "--remote",
+        dest="dvc_remote",
+        metavar="<name>",
+        help="Name of the DVC remote to use when pushing cached outputs.",
+    )
+    experiments_push_parser.add_argument(
+        "-j",
+        "--jobs",
+        type=int,
+        metavar="<number>",
+        help=(
+            "Number of jobs to run simultaneously when pushing to DVC remote "
+            "storage."
+        ),
+    )
+    experiments_push_parser.add_argument(
+        "--run-cache",
+        action="store_true",
+        default=False,
+        help="Push run history for all stages.",
+    )
+    experiments_push_parser.add_argument(
+        "git_remote",
+        help="Git remote name or Git URL.",
+        metavar="<git_remote>",
+    )
+    experiments_push_parser.add_argument(
+        "experiment",
+        nargs="+",
+        help="Experiments to push.",
+        metavar="<experiment>",
+    ).complete = completion.EXPERIMENT
+    experiments_push_parser.set_defaults(func=CmdExperimentsPush)

dvc/commands/experiments/remove.py

@@ -0,0 +1,55 @@
+import argparse
+import logging
+
+from dvc.cli.command import CmdBase
+from dvc.cli.utils import append_doc_link
+
+logger = logging.getLogger(__name__)
+
+
+class CmdExperimentsRemove(CmdBase):
+    def run(self):
+
+        self.repo.experiments.remove(
+            exp_names=self.args.experiment,
+            queue=self.args.queue,
+            clear_all=self.args.all,
+            remote=self.args.git_remote,
+        )
+
+        return 0
+
+
+def add_parser(experiments_subparsers, parent_parser):
+
+    EXPERIMENTS_REMOVE_HELP = "Remove experiments."
+    experiments_remove_parser = experiments_subparsers.add_parser(
+        "remove",
+        parents=[parent_parser],
+        description=append_doc_link(EXPERIMENTS_REMOVE_HELP, "exp/remove"),
+        help=EXPERIMENTS_REMOVE_HELP,
+        formatter_class=argparse.RawDescriptionHelpFormatter,
+    )
+    remove_group = experiments_remove_parser.add_mutually_exclusive_group()
+    remove_group.add_argument(
+        "--queue", action="store_true", help="Remove all queued experiments."
+    )
+    remove_group.add_argument(
+        "-A",
+        "--all",
+        action="store_true",
+        help="Remove all committed experiments.",
+    )
+    remove_group.add_argument(
+        "-g",
+        "--git-remote",
+        metavar="<git_remote>",
+        help="Name or URL of the Git remote to remove the experiment from",
+    )
+    experiments_remove_parser.add_argument(
+        "experiment",
+        nargs="*",
+        help="Experiments to remove.",
+        metavar="<experiment>",
+    )
+    experiments_remove_parser.set_defaults(func=CmdExperimentsRemove)

dvc/commands/experiments/run.py

@@ -0,0 +1,142 @@
+import argparse
+import logging
+
+from dvc.cli.utils import append_doc_link
+from dvc.commands import completion
+from dvc.commands.repro import CmdRepro
+from dvc.commands.repro import add_arguments as add_repro_arguments
+from dvc.exceptions import InvalidArgumentError
+from dvc.ui import ui
+
+logger = logging.getLogger(__name__)
+
+
+class CmdExperimentsRun(CmdRepro):
+    def run(self):
+        from dvc.compare import show_metrics
+
+        if self.args.checkpoint_resume:
+            if self.args.reset:
+                raise InvalidArgumentError(
+                    "--reset and --rev are mutually exclusive."
+                )
+            if not (self.args.queue or self.args.tmp_dir):
+                raise InvalidArgumentError(
+                    "--rev can only be used in conjunction with "
+                    "--queue or --temp."
+                )
+
+        if self.args.reset:
+            ui.write("Any existing checkpoints will be reset and re-run.")
+
+        results = self.repo.experiments.run(
+            name=self.args.name,
+            queue=self.args.queue,
+            run_all=self.args.run_all,
+            jobs=self.args.jobs,
+            params=self.args.set_param,
+            checkpoint_resume=self.args.checkpoint_resume,
+            reset=self.args.reset,
+            tmp_dir=self.args.tmp_dir,
+            machine=self.args.machine,
+            **self._repro_kwargs,
+        )
+
+        if self.args.metrics and results:
+            metrics = self.repo.metrics.show(revs=list(results))
+            metrics.pop("workspace", None)
+            show_metrics(metrics)
+
+        return 0
+
+
+def add_parser(experiments_subparsers, parent_parser):
+
+    EXPERIMENTS_RUN_HELP = "Run or resume an experiment."
+    experiments_run_parser = experiments_subparsers.add_parser(
+        "run",
+        parents=[parent_parser],
+        description=append_doc_link(EXPERIMENTS_RUN_HELP, "exp/run"),
+        help=EXPERIMENTS_RUN_HELP,
+        formatter_class=argparse.RawDescriptionHelpFormatter,
+    )
+    _add_run_common(experiments_run_parser)
+    experiments_run_parser.add_argument(
+        "-r",
+        "--rev",
+        type=str,
+        dest="checkpoint_resume",
+        help=(
+            "Continue the specified checkpoint experiment. Can only be used "
+            "in conjunction with --queue or --temp."
+        ),
+        metavar="<experiment_rev>",
+    ).complete = completion.EXPERIMENT
+    experiments_run_parser.add_argument(
+        "--reset",
+        action="store_true",
+        help="Reset existing checkpoints and restart the experiment.",
+    )
+    experiments_run_parser.set_defaults(func=CmdExperimentsRun)
+
+
+def _add_run_common(parser):
+    """Add common args for 'exp run' and 'exp resume'."""
+    # inherit arguments from `dvc repro`
+    add_repro_arguments(parser)
+    parser.add_argument(
+        "-n",
+        "--name",
+        default=None,
+        help=(
+            "Human-readable experiment name. If not specified, a name will "
+            "be auto-generated."
+        ),
+        metavar="<name>",
+    )
+    parser.add_argument(
+        "-S",
+        "--set-param",
+        action="append",
+        default=[],
+        help="Use the specified param value when reproducing pipelines.",
+        metavar="[<filename>:]<param_name>=<param_value>",
+    )
+    parser.add_argument(
+        "--queue",
+        action="store_true",
+        default=False,
+        help="Stage this experiment in the run queue for future execution.",
+    )
+    parser.add_argument(
+        "--run-all",
+        action="store_true",
+        default=False,
+        help="Execute all experiments in the run queue. Implies --temp.",
+    )
+    parser.add_argument(
+        "-j",
+        "--jobs",
+        type=int,
+        default=1,
+        help="Run the specified number of experiments at a time in parallel.",
+        metavar="<number>",
+    )
+    parser.add_argument(
+        "--temp",
+        action="store_true",
+        dest="tmp_dir",
+        help=(
+            "Run this experiment in a separate temporary directory instead of "
+            "your workspace."
+        ),
+    )
+    parser.add_argument(
+        "--machine",
+        default=None,
+        help=argparse.SUPPRESS,
+        # help=(
+        #     "Run this experiment on the specified 'dvc machine' instance."
+        # )
+        # metavar="<name>",
+    )

dvc/commands/experiments/show.py

@@ -0,0 +1,669 @@
+import argparse
+import logging
+import os
+import re
+from collections import Counter, OrderedDict, defaultdict
+from datetime import date, datetime
+from typing import TYPE_CHECKING
+
+from funcy import lmap
+
+from dvc.cli.command import CmdBase
+from dvc.cli.utils import append_doc_link
+from dvc.commands import completion
+from dvc.commands.metrics import DEFAULT_PRECISION
+from dvc.exceptions import DvcException, InvalidArgumentError
+from dvc.ui import ui
+from dvc.utils.flatten import flatten
+from dvc.utils.serialize import encode_exception
+
+if TYPE_CHECKING:
+    from dvc.compare import TabularData
+    from dvc.ui import RichText
+
+FILL_VALUE = "-"
+FILL_VALUE_ERRORED = "!"
+
+
+logger = logging.getLogger(__name__)
+
+
+def _update_names(names, items):
+    for name, item in items:
+        item = item.get("data", {})
+        if isinstance(item, dict):
+            item = flatten(item)
+            names[name].update({key: None for key in item})
+
+
+def _collect_names(all_experiments, **kwargs):
+    metric_names = defaultdict(dict)
+    param_names = defaultdict(dict)
+    deps_names = set()
+
+    for _, experiments in all_experiments.items():
+        for exp_data in experiments.values():
+            exp = exp_data.get("data", {})
+            _update_names(metric_names, exp.get("metrics", {}).items())
+            _update_names(param_names, exp.get("params", {}).items())
+            for dep_name in exp.get("deps", {}):
+                deps_names.add(dep_name)
+
+    return metric_names, param_names, sorted(deps_names)
+
+
+experiment_types = {
+    "checkpoint_tip": "│ ╓",
+    "checkpoint_commit": "│ ╟",
+    "checkpoint_base": "├─╨",
+    "branch_commit": "├──",
+    "branch_base": "└──",
+    "baseline": "",
+}
+
+
+def _collect_rows(
+    base_rev,
+    experiments,
+    all_headers,
+    metric_headers,
+    param_headers,
+    metric_names,
+    param_names,
+    deps_names,
+    precision=DEFAULT_PRECISION,
+    sort_by=None,
+    sort_order=None,
+    fill_value=FILL_VALUE,
+    iso=False,
+):
+    from scmrepo.git import Git
+
+    if sort_by:
+        sort_path, sort_name, sort_type = _sort_column(
+            sort_by, metric_names, param_names
+        )
+        reverse = sort_order == "desc"
+        experiments = _sort_exp(
+            experiments, sort_path, sort_name, sort_type, reverse
+        )
+
+    new_checkpoint = True
+    for i, (rev, results) in enumerate(experiments.items()):
+        fill_value = FILL_VALUE_ERRORED if results.get("error") else fill_value
+        row_dict = {k: fill_value for k in all_headers}
+
+        exp = results.get("data", {})
+
+        if exp.get("running"):
+            state = "Running"
+        elif exp.get("queued"):
+            state = "Queued"
+        else:
+            state = fill_value
+
+        is_baseline = rev == "baseline"
+
+        if is_baseline:
+            name_rev = base_rev[:7] if Git.is_sha(base_rev) else base_rev
+        else:
+            name_rev = rev[:7]
+
+        tip = exp.get("checkpoint_tip")
+        parent_rev = exp.get("checkpoint_parent", "")
+        parent_exp = experiments.get(parent_rev, {}).get("data", {})
+        parent_tip = parent_exp.get("checkpoint_tip")
+
+        parent = ""
+        if is_baseline:
+            typ = "baseline"
+        elif tip:
+            if tip == parent_tip:
+                typ = (
+                    "checkpoint_tip" if new_checkpoint else "checkpoint_commit"
+                )
+            elif parent_rev == base_rev:
+                typ = "checkpoint_base"
+            else:
+                typ = "checkpoint_commit"
+                parent = parent_rev[:7]
+        elif i < len(experiments) - 1:
+            typ = "branch_commit"
+        else:
+            typ = "branch_base"
+
+        if not is_baseline:
+            new_checkpoint = not (tip and tip == parent_tip)
+
+        row_dict["Experiment"] = exp.get("name", "")
+        row_dict["rev"] = name_rev
+        row_dict["typ"] = typ
+        row_dict["Created"] = _format_time(
+            exp.get("timestamp"), fill_value, iso
+        )
+        row_dict["parent"] = parent
+        row_dict["State"] = state
+        row_dict["Executor"] = exp.get("executor", fill_value)
+
+        _extend_row(
+            row_dict,
+            metric_names,
+            metric_headers,
+            exp.get("metrics", {}).items(),
+            precision,
+            fill_value=fill_value,
+        )
+        _extend_row(
+            row_dict,
+            param_names,
+            param_headers,
+            exp.get("params", {}).items(),
+            precision,
+            fill_value=fill_value,
+        )
+        for dep in deps_names:
+            hash_info = exp.get("deps", {}).get(dep, {}).get("hash")
+            if hash_info is not None:
+                hash_info = hash_info[:7]
+            row_dict[dep] = hash_info
+        yield list(row_dict.values())
+
+
+def _sort_column(sort_by, metric_names, param_names):
+    path, _, sort_name = sort_by.rpartition(":")
+    matches = set()
+
+    if path:
+        if path in metric_names and sort_name in metric_names[path]:
+            matches.add((path, sort_name, "metrics"))
+        if path in param_names and sort_name in param_names[path]:
+            matches.add((path, sort_name, "params"))
+    else:
+        for path in metric_names:
+            if sort_name in metric_names[path]:
+                matches.add((path, sort_name, "metrics"))
+        for path in param_names:
+            if sort_name in param_names[path]:
+                matches.add((path, sort_name, "params"))
+
+    if len(matches) == 1:
+        return matches.pop()
+    if len(matches) > 1:
+        raise InvalidArgumentError(
+            "Ambiguous sort column '{}' matched '{}'".format(
+                sort_by,
+                ", ".join([f"{path}:{name}" for path, name, _ in matches]),
+            )
+        )
+    raise InvalidArgumentError(f"Unknown sort column '{sort_by}'")
+
+
+def _sort_exp(experiments, sort_path, sort_name, typ, reverse):
+    def _sort(item):
+        rev, exp = item
+        exp_data = exp.get("data", {})
+        tip = exp_data.get("checkpoint_tip")
+        if tip and tip != rev:
+            # Sort checkpoint experiments by tip commit
+            return _sort((tip, experiments[tip]))
+        data = exp_data.get(typ, {}).get(sort_path, {}).get("data", {})
+        val = flatten(data).get(sort_name)
+        return val is None, val
+
+    ret = OrderedDict()
+    if "baseline" in experiments:
+        ret["baseline"] = experiments.pop("baseline")
+
+    ret.update(sorted(experiments.items(), key=_sort, reverse=reverse))
+    return ret
+
+
+def _format_time(datetime_obj, fill_value=FILL_VALUE, iso=False):
+    if datetime_obj is None:
+        return fill_value
+
+    if iso:
+        return datetime_obj.isoformat()
+
+    if datetime_obj.date() == date.today():
+        fmt = "%I:%M %p"
+    else:
+        fmt = "%b %d, %Y"
+    return datetime_obj.strftime(fmt)
+
+
+def _extend_row(row, names, headers, items, precision, fill_value=FILL_VALUE):
+    from dvc.compare import _format_field, with_value
+
+    for fname, data in items:
+        item = data.get("data", {})
+        item = flatten(item) if isinstance(item, dict) else {fname: item}
+        for name in names[fname]:
+            value = with_value(
+                item.get(name),
+                FILL_VALUE_ERRORED if data.get("error", None) else fill_value,
+            )
+            # wrap field data in ui.rich_text, otherwise rich may
+            # interpret unescaped braces from list/dict types as rich
+            # markup tags
+            value = ui.rich_text(str(_format_field(value, precision)))
+            if name in headers:
+                row[name] = value
+            else:
+                row[f"{fname}:{name}"] = value
+
+
+def experiments_table(
+    all_experiments,
+    headers,
+    metric_headers,
+    metric_names,
+    param_headers,
+    param_names,
+    deps_names,
+    sort_by=None,
+    sort_order=None,
+    precision=DEFAULT_PRECISION,
+    fill_value=FILL_VALUE,
+    iso=False,
+) -> "TabularData":
+    from funcy import lconcat
+
+    from dvc.compare import TabularData
+
+    all_headers = lconcat(headers, metric_headers, param_headers, deps_names)
+    td = TabularData(all_headers, fill_value=fill_value)
+    for base_rev, experiments in all_experiments.items():
+        rows = _collect_rows(
+            base_rev,
+            experiments,
+            all_headers,
+            metric_headers,
+            param_headers,
+            metric_names,
+            param_names,
+            deps_names,
+            sort_by=sort_by,
+            sort_order=sort_order,
+            precision=precision,
+            fill_value=fill_value,
+            iso=iso,
+        )
+        td.extend(rows)
+
+    return td
+
+
+def prepare_exp_id(kwargs) -> "RichText":
+    exp_name = kwargs["Experiment"]
+    rev = kwargs["rev"]
+    typ = kwargs.get("typ", "baseline")
+
+    if typ == "baseline" or not exp_name:
+        text = ui.rich_text(exp_name or rev)
+    else:
+        text = ui.rich_text.assemble(rev, " [", (exp_name, "bold"), "]")
+
+    parent = kwargs.get("parent")
+    suff = f" ({parent})" if parent else ""
+    text.append(suff)
+
+    tree = experiment_types[typ]
+    pref = f"{tree} " if tree else ""
+    return ui.rich_text(pref) + text
+
+
+def baseline_styler(typ):
+    return {"style": "bold"} if typ == "baseline" else {}
+
+
+def show_experiments(
+    all_experiments,
+    keep=None,
+    drop=None,
+    pager=True,
+    csv=False,
+    markdown=False,
+    pcp=False,
+    **kwargs,
+):
+    from funcy.seqs import flatten as flatten_list
+
+    metric_names, param_names, deps_names = _collect_names(all_experiments)
+
+    headers = [
+        "Experiment",
+        "rev",
+        "typ",
+        "Created",
+        "parent",
+        "State",
+        "Executor",
+    ]
+
+    names = {**metric_names, **param_names}
+    counter = Counter(flatten_list([list(a.keys()) for a in names.values()]))
+    counter.update(headers)
+    metric_headers = _normalize_headers(metric_names, counter)
+    param_headers = _normalize_headers(param_names, counter)
+
+    td = experiments_table(
+        all_experiments,
+        headers,
+        metric_headers,
+        metric_names,
+        param_headers,
+        param_names,
+        deps_names,
+        kwargs.get("sort_by"),
+        kwargs.get("sort_order"),
+        kwargs.get("precision"),
+        kwargs.get("fill_value"),
+        kwargs.get("iso"),
+    )
+    if keep:
+        for col in td.keys():
+            if re.match(keep, col):
+                td.protect(col)
+
+    for col in ("State", "Executor"):
+        if td.is_empty(col):
+            td.drop(col)
+
+    row_styles = lmap(baseline_styler, td.column("typ"))
+
+    if not csv:
+        merge_headers = ["Experiment", "rev", "typ", "parent"]
+        td.column("Experiment")[:] = map(
+            prepare_exp_id, td.as_dict(merge_headers)
+        )
+        td.drop(*merge_headers[1:])
+
+    headers = {
+        "metrics": metric_headers,
+        "params": param_headers,
+        "deps": deps_names,
+    }
+    styles = {
+        "Experiment": {"no_wrap": True, "header_style": "black on grey93"},
+        "Created": {"header_style": "black on grey93"},
+        "State": {"header_style": "black on grey93"},
+        "Executor": {"header_style": "black on grey93"},
+    }
+    header_bg_colors = {
+        "metrics": "cornsilk1",
+        "params": "light_cyan1",
+        "deps": "plum2",
+    }
+    styles.update(
+        {
+            header: {
+                "justify": "right" if typ == "metrics" else "left",
+                "header_style": f"black on {header_bg_colors[typ]}",
+                "collapse": idx != 0,
+                "no_wrap": typ == "metrics",
+            }
+            for typ, hs in headers.items()
+            for idx, header in enumerate(hs)
+        }
+    )
+
+    if kwargs.get("only_changed", False) or pcp:
+        td.drop_duplicates("cols", ignore_empty=False)
+
+    cols_to_drop = set()
+    if drop is not None:
+        cols_to_drop = {col for col in td.keys() if re.match(drop, col)}
+    if pcp:
+        cols_to_drop.add("Created")
+    td.drop(*cols_to_drop)
+
+    if pcp:
+        subset = {x for x in td.keys() if x != "Experiment"}
+        td.dropna(
+            "rows",
+            how="all",
+            subset=subset,
+        )
+        td.drop_duplicates("rows", subset=subset)
+        td.column("Experiment")[:] = [
+            # remove tree characters
+            str(x).encode("ascii", "ignore").strip().decode()
+            for x in td.column("Experiment")
+        ]
+        out = kwargs.get("out") or "dvc_plots"
+        ui.write(
+            td.to_parallel_coordinates(
+                output_path=os.path.abspath(out),
+                color_by=kwargs.get("sort_by") or "Experiment",
+            )
+        )
+        if kwargs.get("open"):
+            return ui.open_browser(os.path.join(out, "index.html"))
+
+    else:
+        td.render(
+            pager=pager,
+            borders="horizontals",
+            rich_table=True,
+            header_styles=styles,
+            row_styles=row_styles,
+            csv=csv,
+            markdown=markdown,
+        )
+
+
+def _normalize_headers(names, count):
+    return [
+        name if count[name] == 1 else f"{path}:{name}"
+        for path in names
+        for name in names[path]
+    ]
+
+
+def _format_json(item):
+    if isinstance(item, (date, datetime)):
+        return item.isoformat()
+    return encode_exception(item)
+
+
+class CmdExperimentsShow(CmdBase):
+    def run(self):
+        try:
+            all_experiments = self.repo.experiments.show(
+                all_branches=self.args.all_branches,
+                all_tags=self.args.all_tags,
+                all_commits=self.args.all_commits,
+                revs=self.args.rev,
+                num=self.args.num,
+                sha_only=self.args.sha,
+                param_deps=self.args.param_deps,
+            )
+        except DvcException:
+            logger.exception("failed to show experiments")
+            return 1
+
+        if self.args.json:
+            ui.write_json(all_experiments, default=_format_json)
+        else:
+            precision = (
+                self.args.precision or None
+                if self.args.csv
+                else DEFAULT_PRECISION
+            )
+            fill_value = "" if self.args.csv else FILL_VALUE
+            iso = True if self.args.csv else False
+
+            show_experiments(
+                all_experiments,
+                keep=self.args.keep,
+                drop=self.args.drop,
+                sort_by=self.args.sort_by,
+                sort_order=self.args.sort_order,
+                precision=precision,
+                fill_value=fill_value,
+                iso=iso,
+                pager=not self.args.no_pager,
+                csv=self.args.csv,
+                markdown=self.args.markdown,
+                only_changed=self.args.only_changed,
+                pcp=self.args.pcp,
+                out=self.args.out,
+                open=self.args.open,
+            )
+        return 0
+
+
+def add_parser(experiments_subparsers, parent_parser):
+    EXPERIMENTS_SHOW_HELP = "Print experiments."
+    experiments_show_parser = experiments_subparsers.add_parser(
+        "show",
+        parents=[parent_parser],
+        description=append_doc_link(EXPERIMENTS_SHOW_HELP, "exp/show"),
+        help=EXPERIMENTS_SHOW_HELP,
+        formatter_class=argparse.RawDescriptionHelpFormatter,
+    )
+    experiments_show_parser.add_argument(
+        "-a",
+        "--all-branches",
+        action="store_true",
+        default=False,
+        help="Show experiments derived from the tip of all Git branches.",
+    )
+    experiments_show_parser.add_argument(
+        "-T",
+        "--all-tags",
+        action="store_true",
+        default=False,
+        help="Show experiments derived from all Git tags.",
+    )
+    experiments_show_parser.add_argument(
+        "-A",
+        "--all-commits",
+        action="store_true",
+        default=False,
+        help="Show experiments derived from all Git commits.",
+    )
+    experiments_show_parser.add_argument(
+        "--rev",
+        type=str,
+        default=None,
+        help=(
+            "Show experiments derived from the specified revision. "
+            "Defaults to HEAD if none of `--rev`,`-a`,`-A`,`-T` is specified."
+        ),
+        metavar="<rev>",
+    )
+    experiments_show_parser.add_argument(
+        "-n",
+        "--num",
+        type=int,
+        default=1,
+        dest="num",
+        metavar="<num>",
+        help="Show the last `num` commits from <rev>.",
+    )
+    experiments_show_parser.add_argument(
+        "--no-pager",
+        action="store_true",
+        default=False,
+        help="Do not pipe output into a pager.",
+    )
+    experiments_show_parser.add_argument(
+        "--only-changed",
+        action="store_true",
+        default=False,
+        help=(
+            "Only show metrics/params with values varying "
+            "across the selected experiments."
+        ),
+    )
+    experiments_show_parser.add_argument(
+        "--drop",
+        help="Remove the columns matching the specified regex pattern.",
+        metavar="<regex_pattern>",
+    )
+    experiments_show_parser.add_argument(
+        "--keep",
+        help="Preserve the columns matching the specified regex pattern.",
+        metavar="<regex_pattern>",
+    )
+    experiments_show_parser.add_argument(
+        "--param-deps",
+        action="store_true",
+        default=False,
+        help="Show only params that are stage dependencies.",
+    )
+    experiments_show_parser.add_argument(
+        "--sort-by",
+        help="Sort related experiments by the specified metric or param.",
+        metavar="<metric/param>",
+    )
+    experiments_show_parser.add_argument(
+        "--sort-order",
+        help=(
+            "Sort order to use with --sort-by."
+            " Defaults to ascending ('asc')."
+        ),
+        choices=("asc", "desc"),
+        default="asc",
+    )
+    experiments_show_parser.add_argument(
+        "--sha",
+        action="store_true",
+        default=False,
+        help="Always show git commit SHAs instead of branch/tag names.",
+    )
+    experiments_show_parser.add_argument(
+        "--json",
+        "--show-json",
+        action="store_true",
+        default=False,
+        help="Print output in JSON format instead of a human-readable table.",
+    )
+    experiments_show_parser.add_argument(
+        "--csv",
+        "--show-csv",
+        action="store_true",
+        default=False,
+        help="Print output in csv format instead of a human-readable table.",
+    )
+    experiments_show_parser.add_argument(
+        "--md",
+        "--show-md",
+        action="store_true",
+        default=False,
+        dest="markdown",
+        help="Show tabulated output in the Markdown format (GFM).",
+    )
+    experiments_show_parser.add_argument(
+        "--precision",
+        type=int,
+        help=(
+            "Round metrics/params to `n` digits precision after the decimal "
+            f"point. Rounds to {DEFAULT_PRECISION} digits by default."
+        ),
+        metavar="<n>",
+    )
+    experiments_show_parser.add_argument(
+        "--parallel-coordinates-plot",
+        "--pcp",
+        dest="pcp",
+        action="store_true",
+        default=False,
+        help="Generate a Parallel Coordinates Plot from the tabulated output.",
+    )
+    experiments_show_parser.add_argument(
+        "-o",
+        "--out",
+        default=None,
+        help="Destination folder to save the Parallel Coordinates Plot to",
+        metavar="<path>",
+    ).complete = completion.DIR
+    experiments_show_parser.add_argument(
+        "--open",
+        action="store_true",
+        default=False,
+        help="Open the Parallel Coordinates Plot directly in the browser.",
+    )
+    experiments_show_parser.set_defaults(func=CmdExperimentsShow)

dvc/commands/freeze.py

@@ -1,8 +1,9 @@
 import argparse
 import logging
 
-from dvc.command import completion
-from dvc.command.base import CmdBase, append_doc_link
+from dvc.cli.command import CmdBase
+from dvc.cli.utils import append_doc_link
+from dvc.commands import completion
 from dvc.exceptions import DvcException
 
 logger = logging.getLogger(__name__)

dvc/commands/gc.py

@@ -2,7 +2,8 @@ import argparse
 import logging
 import os
 
-from dvc.command.base import CmdBase, append_doc_link
+from dvc.cli.command import CmdBase
+from dvc.cli.utils import append_doc_link
 from dvc.ui import ui
 
 logger = logging.getLogger(__name__)

dvc/commands/get.py

@@ -1,10 +1,11 @@
 import argparse
 import logging
 
+from dvc.cli.command import CmdBaseNoRepo
 from dvc.exceptions import DvcException
 
+from ..cli.utils import append_doc_link
 from . import completion
-from .base import CmdBaseNoRepo, append_doc_link
 
 logger = logging.getLogger(__name__)
 

dvc/commands/get_url.py

@@ -1,10 +1,11 @@
 import argparse
 import logging
 
+from dvc.cli.command import CmdBaseNoRepo
 from dvc.exceptions import DvcException
 
+from ..cli.utils import append_doc_link
 from . import completion
-from .base import CmdBaseNoRepo, append_doc_link
 
 logger = logging.getLogger(__name__)
 

dvc/commands/git_hook.py

@@ -1,7 +1,8 @@
 import logging
 import os
 
-from dvc.command.base import CmdBaseNoRepo, fix_subparsers
+from dvc.cli.command import CmdBaseNoRepo
+from dvc.cli.utils import fix_subparsers
 from dvc.exceptions import NotDvcRepoError
 
 logger = logging.getLogger(__name__)
@@ -25,7 +26,7 @@ class CmdHookBase(CmdBaseNoRepo):
 
 class CmdPreCommit(CmdHookBase):
     def _run(self):
-        from dvc.main import main
+        from dvc.cli import main
 
         return main(["status"])
 
@@ -49,14 +50,14 @@ class CmdPostCheckout(CmdHookBase):
         if os.path.isdir(os.path.join(".git", "rebase-merge")):
             return 0
 
-        from dvc.main import main
+        from dvc.cli import main
 
         return main(["checkout"])
 
 
 class CmdPrePush(CmdHookBase):
     def _run(self):
-        from dvc.main import main
+        from dvc.cli import main
 
         return main(["push"])
 

dvc/commands/imp.py

@@ -1,8 +1,9 @@
 import argparse
 import logging
 
-from dvc.command import completion
-from dvc.command.base import CmdBase, append_doc_link
+from dvc.cli.command import CmdBase
+from dvc.cli.utils import append_doc_link
+from dvc.commands import completion
 from dvc.exceptions import DvcException
 
 logger = logging.getLogger(__name__)

dvc/commands/imp_url.py

@@ -1,8 +1,9 @@
 import argparse
 import logging
 
-from dvc.command import completion
-from dvc.command.base import CmdBase, append_doc_link
+from dvc.cli.command import CmdBase
+from dvc.cli.utils import append_doc_link
+from dvc.commands import completion
 from dvc.exceptions import DvcException
 
 logger = logging.getLogger(__name__)

dvc/commands/init.py

@@ -4,7 +4,8 @@ import logging
 import colorama
 
 from dvc import analytics
-from dvc.command.base import CmdBaseNoRepo, append_doc_link
+from dvc.cli.command import CmdBaseNoRepo
+from dvc.cli.utils import append_doc_link
 from dvc.utils import boxify
 from dvc.utils import format_link as fmt_link
 

dvc/commands/install.py

@@ -1,7 +1,8 @@
 import argparse
 import logging
 
-from dvc.command.base import CmdBase, append_doc_link
+from dvc.cli.command import CmdBase
+from dvc.cli.utils import append_doc_link
 from dvc.exceptions import DvcException
 
 logger = logging.getLogger(__name__)

dvc/commands/live.py

@@ -1,8 +1,8 @@
 import argparse
-from pathlib import Path
 
-from dvc.command import completion
-from dvc.command.base import CmdBase, fix_subparsers
+from dvc.cli.command import CmdBase
+from dvc.cli.utils import fix_subparsers
+from dvc.commands import completion
 from dvc.ui import ui
 
 
@@ -10,13 +10,17 @@ class CmdLive(CmdBase):
     UNINITIALIZED = True
 
     def _run(self, target, revs=None):
+        from dvc.render.utils import match_renderers, render
+
         metrics, plots = self.repo.live.show(target=target, revs=revs)
 
         if plots:
-            html_path = Path.cwd() / (self.args.target + "_html")
-            from dvc.render.utils import render
+            from pathlib import Path
 
-            index_path = render(self.repo, plots, metrics, html_path)
+            html_path = Path.cwd() / (self.args.target + "_html")
+
+            renderers = match_renderers(plots, self.repo.plots.templates)
+            index_path = render(self.repo, renderers, metrics, html_path)
             ui.write(index_path.as_uri())
             return 0
         return 1

dvc/commands/ls/__init__.py

@@ -1,9 +1,10 @@
 import argparse
 import logging
 
-from dvc.command import completion
-from dvc.command.base import CmdBaseNoRepo, append_doc_link
-from dvc.command.ls.ls_colors import LsColors
+from dvc.cli.command import CmdBaseNoRepo
+from dvc.cli.utils import append_doc_link
+from dvc.commands import completion
+from dvc.commands.ls.ls_colors import LsColors
 from dvc.exceptions import DvcException
 from dvc.ui import ui
 

dvc/commands/machine.py

@@ -1,7 +1,8 @@
 import argparse
 
-from dvc.command.base import CmdBase, append_doc_link, fix_subparsers
-from dvc.command.config import CmdConfig
+from dvc.cli.command import CmdBase
+from dvc.cli.utils import append_doc_link, fix_subparsers
+from dvc.commands.config import CmdConfig
 from dvc.compare import TabularData
 from dvc.config import ConfigError
 from dvc.exceptions import DvcException
@@ -241,6 +242,7 @@ class CmdMachineStatus(CmdBase):
         "instance_hdd_size",
         "instance_gpu",
     ]
+    FILL_VALUE = "-"
 
     def _add_row(
         self,
@@ -250,7 +252,11 @@ class CmdMachineStatus(CmdBase):
     ):
 
         if not all_status:
-            row = [name, None, "offline"]
+            row = [
+                name,
+                self.FILL_VALUE,
+                "offline",
+            ]  # back to `None` after #7167
             td.append(row)
         for i, status in enumerate(all_status, start=1):
             row = [name, f"num_{i}", "running" if status else "offline"]
@@ -264,7 +270,7 @@ class CmdMachineStatus(CmdBase):
             raise MachineDisabledError
 
         td = TabularData(
-            self.INSTANCE_FIELD + self.SHOWN_FIELD, fill_value="-"
+            self.INSTANCE_FIELD + self.SHOWN_FIELD, fill_value=self.FILL_VALUE
         )
 
         if self.args.name:
@@ -304,7 +310,7 @@ class CmdMachineSsh(CmdBase):
 
 
 def add_parser(subparsers, parent_parser):
-    from dvc.command.config import parent_config_parser
+    from dvc.commands.config import parent_config_parser
 
     machine_HELP = "Set up and manage cloud machines."
     machine_parser = subparsers.add_parser(

dvc/commands/metrics.py

@@ -1,8 +1,9 @@
 import argparse
 import logging
 
-from dvc.command import completion
-from dvc.command.base import CmdBase, append_doc_link, fix_subparsers
+from dvc.cli.command import CmdBase
+from dvc.cli.utils import append_doc_link, fix_subparsers
+from dvc.commands import completion
 from dvc.exceptions import DvcException
 from dvc.ui import ui
 from dvc.utils.serialize import encode_exception

dvc/commands/move.py

@@ -1,8 +1,9 @@
 import argparse
 import logging
 
-from dvc.command import completion
-from dvc.command.base import CmdBase, append_doc_link
+from dvc.cli.command import CmdBase
+from dvc.cli.utils import append_doc_link
+from dvc.commands import completion
 from dvc.exceptions import DvcException
 
 logger = logging.getLogger(__name__)

dvc/commands/params.py

@@ -1,8 +1,9 @@
 import argparse
 import logging
 
-from dvc.command import completion
-from dvc.command.base import CmdBase, append_doc_link, fix_subparsers
+from dvc.cli.command import CmdBase
+from dvc.cli.utils import append_doc_link, fix_subparsers
+from dvc.commands import completion
 from dvc.exceptions import DvcException
 from dvc.ui import ui
 

dvc/commands/plots.py

@@ -1,16 +1,31 @@
 import argparse
+import json
 import logging
 
-from dvc.command import completion
-from dvc.command.base import CmdBase, append_doc_link, fix_subparsers
+from funcy import first
+
+from dvc.cli.command import CmdBase
+from dvc.cli.utils import append_doc_link, fix_subparsers
+from dvc.commands import completion
 from dvc.exceptions import DvcException
-from dvc.render.utils import find_vega, render
 from dvc.ui import ui
 from dvc.utils import format_link
 
 logger = logging.getLogger(__name__)
 
 
+def _show_json(renderers, path: None):
+    if any(r.needs_output_path for r in renderers) and not path:
+        raise DvcException("Output path ('-o') is required!")
+
+    result = {
+        renderer.filename: json.loads(renderer.as_json(path=path))
+        for renderer in renderers
+    }
+    if result:
+        ui.write_json(result)
+
+
 class CmdPlots(CmdBase):
     def _func(self, *args, **kwargs):
         raise NotImplementedError
@@ -25,6 +40,9 @@ class CmdPlots(CmdBase):
     def run(self):
         from pathlib import Path
 
+        from dvc.render.utils import match_renderers, render
+        from dvc.render.vega import VegaRenderer
+
         if self.args.show_vega:
             if not self.args.targets:
                 logger.error("please specify a target for `--show-vega`")
@@ -34,8 +52,15 @@ class CmdPlots(CmdBase):
                     "you can only specify one target for `--show-vega`"
                 )
                 return 1
+            if self.args.json:
+                logger.error(
+                    "'--show-vega' and '--json' are mutually exclusive "
+                    "options."
+                )
+                return 1
 
         try:
+
             plots_data = self._func(
                 targets=self.args.targets, props=self._props()
             )
@@ -46,25 +71,39 @@ class CmdPlots(CmdBase):
                     "visualization file will not be created."
                 )
 
+            renderers = match_renderers(
+                plots_data=plots_data, templates=self.repo.plots.templates
+            )
+
             if self.args.show_vega:
-                target = self.args.targets[0]
-                plot_json = find_vega(self.repo, plots_data, target)
-                if plot_json:
-                    ui.write(plot_json)
+                renderer = first(
+                    filter(lambda r: isinstance(r, VegaRenderer), renderers)
+                )
+                if renderer:
+                    ui.write_json(renderer.asdict())
+                return 0
+            if self.args.json:
+                _show_json(renderers, self.args.out)
                 return 0
 
             rel: str = self.args.out or "dvc_plots"
             path: Path = (Path.cwd() / rel).resolve()
             index_path = render(
                 self.repo,
-                plots_data,
+                renderers,
                 path=path,
                 html_template_path=self.args.html_template,
             )
 
             ui.write(index_path.as_uri())
-
-            if self.args.open:
+            auto_open = self.repo.config["plots"].get("auto_open", False)
+            if self.args.open or auto_open:
+                if not auto_open:
+                    ui.write(
+                        "To enable auto opening, you can run:\n"
+                        "\n"
+                        "\tdvc config plots.auto_open true"
+                    )
                 return ui.open_browser(index_path)
 
             return 0
@@ -101,6 +140,37 @@ class CmdPlotsModify(CmdPlots):
         return 0
 
 
+class CmdPlotsTemplates(CmdBase):
+    TEMPLATES_CHOICES = [
+        "simple",
+        "linear",
+        "confusion",
+        "confusion_normalized",
+        "scatter",
+        "smooth",
+    ]
+
+    def run(self):
+        import os
+
+        try:
+            out = (
+                os.path.join(os.getcwd(), self.args.out)
+                if self.args.out
+                else self.repo.plots.templates.templates_dir
+            )
+
+            targets = [self.args.target] if self.args.target else None
+            self.repo.plots.templates.init(output=out, targets=targets)
+            templates_path = os.path.relpath(out, os.getcwd())
+            ui.write(f"Templates have been written into '{templates_path}'.")
+
+            return 0
+        except DvcException:
+            logger.exception("")
+            return 1
+
+
 def add_parser(subparsers, parent_parser):
     PLOTS_HELP = (
         "Commands to visualize and compare plot metrics in structured files "
@@ -137,7 +207,8 @@ def add_parser(subparsers, parent_parser):
         "Shows all plots by default.",
     ).complete = completion.FILE
     _add_props_arguments(plots_show_parser)
-    _add_output_arguments(plots_show_parser)
+    _add_output_argument(plots_show_parser)
+    _add_ui_arguments(plots_show_parser)
     plots_show_parser.set_defaults(func=CmdPlotsShow)
 
     PLOTS_DIFF_HELP = (
@@ -172,7 +243,8 @@ def add_parser(subparsers, parent_parser):
         "revisions", nargs="*", default=None, help="Git commits to plot from"
     )
     _add_props_arguments(plots_diff_parser)
-    _add_output_arguments(plots_diff_parser)
+    _add_output_argument(plots_diff_parser)
+    _add_ui_arguments(plots_diff_parser)
     plots_diff_parser.set_defaults(func=CmdPlotsDiff)
 
     PLOTS_MODIFY_HELP = (
@@ -198,6 +270,27 @@ def add_parser(subparsers, parent_parser):
     )
     plots_modify_parser.set_defaults(func=CmdPlotsModify)
 
+    TEMPLATES_HELP = (
+        "Write built-in plots templates to a directory "
+        "(.dvc/plots by default)."
+    )
+    plots_templates_parser = plots_subparsers.add_parser(
+        "templates",
+        parents=[parent_parser],
+        description=append_doc_link(TEMPLATES_HELP, "plots/templates"),
+        help=TEMPLATES_HELP,
+        formatter_class=argparse.RawDescriptionHelpFormatter,
+    )
+    plots_templates_parser.add_argument(
+        "target",
+        default=None,
+        nargs="?",
+        choices=CmdPlotsTemplates.TEMPLATES_CHOICES,
+        help="Template to write. Writes all templates by default.",
+    )
+    _add_output_argument(plots_templates_parser, typ="templates")
+    plots_templates_parser.set_defaults(func=CmdPlotsTemplates)
+
 
 def _add_props_arguments(parser):
     parser.add_argument(
@@ -237,20 +330,30 @@ def _add_props_arguments(parser):
     )
 
 
-def _add_output_arguments(parser):
+def _add_output_argument(parser, typ="plots"):
     parser.add_argument(
         "-o",
         "--out",
         default=None,
-        help="Destination path to save plots to",
+        help=f"Directory to save {typ} to.",
         metavar="<path>",
     ).complete = completion.DIR
+
+
+def _add_ui_arguments(parser):
     parser.add_argument(
         "--show-vega",
         action="store_true",
         default=False,
         help="Show output in Vega format.",
     )
+    parser.add_argument(
+        "--json",
+        "--show-json",
+        action="store_true",
+        default=False,
+        help=argparse.SUPPRESS,
+    )
     parser.add_argument(
         "--open",
         action="store_true",

dvc/commands/remote.py

@@ -1,7 +1,7 @@
 import argparse
 
-from dvc.command.base import append_doc_link, fix_subparsers
-from dvc.command.config import CmdConfig
+from dvc.cli.utils import append_doc_link, fix_subparsers
+from dvc.commands.config import CmdConfig
 from dvc.ui import ui
 from dvc.utils import format_link
 
@@ -151,7 +151,7 @@ class CmdRemoteRename(CmdRemote):
 
 
 def add_parser(subparsers, parent_parser):
-    from dvc.command.config import parent_config_parser
+    from dvc.commands.config import parent_config_parser
 
     REMOTE_HELP = "Set up and manage data remotes."
     remote_parser = subparsers.add_parser(

dvc/commands/remove.py

@@ -1,8 +1,9 @@
 import argparse
 import logging
 
-from dvc.command import completion
-from dvc.command.base import CmdBase, append_doc_link
+from dvc.cli.command import CmdBase
+from dvc.cli.utils import append_doc_link
+from dvc.commands import completion
 from dvc.exceptions import DvcException
 
 logger = logging.getLogger(__name__)

dvc/commands/repro.py

@@ -1,8 +1,9 @@
 import argparse
 
-from dvc.command import completion
-from dvc.command.base import CmdBase, append_doc_link
-from dvc.command.status import CmdDataStatus
+from dvc.cli.command import CmdBase
+from dvc.cli.utils import append_doc_link
+from dvc.commands import completion
+from dvc.commands.status import CmdDataStatus
 
 
 class CmdRepro(CmdBase):

dvc/commands/root.py

@@ -1,7 +1,8 @@
 import argparse
 import logging
 
-from dvc.command.base import CmdBaseNoRepo, append_doc_link
+from dvc.cli.command import CmdBaseNoRepo
+from dvc.cli.utils import append_doc_link
 from dvc.utils import relpath
 
 logger = logging.getLogger(__name__)

dvc/commands/run.py

@@ -1,8 +1,9 @@
 import argparse
 import logging
 
-from dvc.command.base import CmdBase, append_doc_link
-from dvc.command.stage import parse_cmd
+from dvc.cli.command import CmdBase
+from dvc.cli.utils import append_doc_link
+from dvc.commands.stage import parse_cmd
 from dvc.exceptions import DvcException
 
 logger = logging.getLogger(__name__)
@@ -52,7 +53,7 @@ class CmdRun(CmdBase):
 
 
 def add_parser(subparsers, parent_parser):
-    from dvc.command.stage import _add_common_args
+    from dvc.commands.stage import _add_common_args
 
     RUN_HELP = (
         "Generate a dvc.yaml file from a command and execute the command."

dvc/commands/stage.py

@@ -3,8 +3,9 @@ import logging
 from itertools import chain, filterfalse
 from typing import TYPE_CHECKING, Dict, Iterable, List
 
-from dvc.command import completion
-from dvc.command.base import CmdBase, append_doc_link, fix_subparsers
+from dvc.cli.command import CmdBase
+from dvc.cli.utils import append_doc_link, fix_subparsers
+from dvc.commands import completion
 from dvc.utils.cli_parse import parse_params
 from dvc.utils.humanize import truncate_text
 

dvc/commands/status.py

@@ -1,6 +1,6 @@
 import logging
 
-from dvc.command.data_sync import CmdDataBase
+from dvc.commands.data_sync import CmdDataBase
 from dvc.exceptions import DvcException
 from dvc.ui import ui
 from dvc.utils import format_link

dvc/commands/unprotect.py

@@ -1,8 +1,9 @@
 import argparse
 import logging
 
-from dvc.command import completion
-from dvc.command.base import CmdBase, append_doc_link
+from dvc.cli.command import CmdBase
+from dvc.cli.utils import append_doc_link
+from dvc.commands import completion
 from dvc.exceptions import DvcException
 
 logger = logging.getLogger(__name__)

dvc/commands/update.py

@@ -1,8 +1,9 @@
 import argparse
 import logging
 
-from dvc.command import completion
-from dvc.command.base import CmdBase, append_doc_link
+from dvc.cli.command import CmdBase
+from dvc.cli.utils import append_doc_link
+from dvc.commands import completion
 from dvc.exceptions import DvcException
 
 logger = logging.getLogger(__name__)

dvc/commands/version.py

@@ -1,7 +1,8 @@
 import argparse
 import logging
 
-from dvc.command.base import CmdBaseNoRepo, append_doc_link
+from dvc.cli.command import CmdBaseNoRepo
+from dvc.cli.utils import append_doc_link
 from dvc.ui import ui
 
 logger = logging.getLogger(__name__)

dvc/compare.py

@@ -11,6 +11,7 @@ from typing import (
     List,
     Mapping,
     MutableSequence,
+    Optional,
     Sequence,
     Set,
     Tuple,
@@ -21,6 +22,7 @@ from typing import (
 from funcy import reraise
 
 if TYPE_CHECKING:
+    from dvc.types import StrPath
     from dvc.ui.table import CellT
 
 
@@ -37,11 +39,21 @@ class TabularData(MutableSequence[Sequence["CellT"]]):
         self._columns: Dict[str, Column] = {name: Column() for name in columns}
         self._keys: List[str] = list(columns)
         self._fill_value = fill_value
+        self._protected: Set[str] = set()
 
     @property
     def columns(self) -> List[Column]:
         return list(map(self.column, self.keys()))
 
+    def is_protected(self, col_name) -> bool:
+        return col_name in self._protected
+
+    def protect(self, *col_names: str):
+        self._protected.update(col_names)
+
+    def unprotect(self, *col_names: str):
+        self._protected = self._protected.difference(col_names)
+
     def column(self, name: str) -> Column:
         return self._columns[name]
 
@@ -125,9 +137,10 @@ class TabularData(MutableSequence[Sequence["CellT"]]):
         return len(self.columns), len(self)
 
     def drop(self, *col_names: str) -> None:
-        for col_name in col_names:
-            self._keys.remove(col_name)
-            self._columns.pop(col_name)
+        for col in col_names:
+            if not self.is_protected(col):
+                self._keys.remove(col)
+                self._columns.pop(col)
 
     def rename(self, from_col_name: str, to_col_name: str) -> None:
         self._columns[to_col_name] = self._columns.pop(from_col_name)
@@ -153,6 +166,20 @@ class TabularData(MutableSequence[Sequence["CellT"]]):
             writer.writerow(row)
         return buff.getvalue()
 
+    def to_parallel_coordinates(
+        self, output_path: "StrPath", color_by: str = None
+    ) -> str:
+        from dvc.render.html import write
+        from dvc.render.plotly import ParallelCoordinatesRenderer
+
+        index_path = write(
+            output_path,
+            renderers=[
+                ParallelCoordinatesRenderer(self, color_by, self._fill_value)
+            ],
+        )
+        return index_path.as_uri()
+
     def add_column(self, name: str) -> None:
         self._columns[name] = Column([self._fill_value] * len(self))
         self._keys.append(name)
@@ -184,7 +211,12 @@ class TabularData(MutableSequence[Sequence["CellT"]]):
             {k: self._columns[k][i] for k in keys} for i in range(len(self))
         ]
 
-    def dropna(self, axis: str = "rows", how="any"):
+    def dropna(
+        self,
+        axis: str = "rows",
+        how="any",
+        subset: Optional[Iterable[str]] = None,
+    ):
         if axis not in ["rows", "cols"]:
             raise ValueError(
                 f"Invalid 'axis' value {axis}."
@@ -196,13 +228,15 @@ class TabularData(MutableSequence[Sequence["CellT"]]):
             )
 
         match_line: Set = set()
-        match = True
+        match_any = True
         if how == "all":
-            match = False
+            match_any = False
 
         for n_row, row in enumerate(self):
             for n_col, col in enumerate(row):
-                if (col == self._fill_value) is match:
+                if subset and self.keys()[n_col] not in subset:
+                    continue
+                if (col == self._fill_value) is match_any:
                     if axis == "rows":
                         match_line.add(n_row)
                         break
@@ -229,7 +263,12 @@ class TabularData(MutableSequence[Sequence["CellT"]]):
         else:
             self.drop(*to_drop)
 
-    def drop_duplicates(self, axis: str = "rows"):
+    def drop_duplicates(
+        self,
+        axis: str = "rows",
+        subset: Optional[Iterable[str]] = None,
+        ignore_empty: bool = True,
+    ):
         if axis not in ["rows", "cols"]:
             raise ValueError(
                 f"Invalid 'axis' value {axis}."
@@ -239,8 +278,12 @@ class TabularData(MutableSequence[Sequence["CellT"]]):
         if axis == "cols":
             cols_to_drop: List[str] = []
             for n_col, col in enumerate(self.columns):
+                if subset and self.keys()[n_col] not in subset:
+                    continue
                 # Cast to str because Text is not hashable error
-                unique_vals = {str(x) for x in col if x != self._fill_value}
+                unique_vals = {str(x) for x in col}
+                if ignore_empty and self._fill_value in unique_vals:
+                    unique_vals -= {self._fill_value}
                 if len(unique_vals) == 1:
                     cols_to_drop.append(self.keys()[n_col])
             self.drop(*cols_to_drop)
@@ -249,6 +292,13 @@ class TabularData(MutableSequence[Sequence["CellT"]]):
             unique_rows = []
             rows_to_drop: List[int] = []
             for n_row, row in enumerate(self):
+                if subset:
+                    row = [
+                        col
+                        for n_col, col in enumerate(row)
+                        if self.keys()[n_col] in subset
+                    ]
+
                 tuple_row = tuple(row)
                 if tuple_row in unique_rows:
                     rows_to_drop.append(n_row)

dvc/config.py

@@ -94,7 +94,7 @@ class Config(dict):
             try:
                 from dvc.repo import Repo
 
-                self.dvc_dir = os.path.join(Repo.find_dvc_dir())
+                self.dvc_dir = Repo.find_dvc_dir()
             except NotDvcRepoError:
                 self.dvc_dir = None
         else:
@@ -266,7 +266,12 @@ class Config(dict):
                     "key_path": func,
                 }
             },
-            "machine": {str: {"startup_script": func}},
+            "machine": {
+                str: {
+                    "startup_script": func,
+                    "setup_script": func,
+                }
+            },
         }
         return Schema(dirs_schema, extra=ALLOW_EXTRA)(conf)
 

dvc/config_schema.py

@@ -79,13 +79,13 @@ REMOTE_COMMON = {
     "checksum_jobs": All(Coerce(int), Range(1)),
     "jobs": All(Coerce(int), Range(1)),
     Optional("no_traverse"): Bool,  # obsoleted
-    "verify": Bool,
 }
 LOCAL_COMMON = {
     "type": supported_cache_type,
     Optional("protected", default=False): Bool,  # obsoleted
     "shared": All(Lower, Choices("group")),
     Optional("slow_link_warning", default=True): Bool,
+    Optional("verify", default=False): Bool,
 }
 HTTP_COMMON = {
     "auth": All(Lower, Choices("basic", "digest", "custom")),
@@ -95,6 +95,7 @@ HTTP_COMMON = {
     "ask_password": Bool,
     "ssl_verify": Any(Bool, str),
     "method": str,
+    Optional("verify", default=False): Bool,
 }
 WEBDAV_COMMON = {
     "user": str,
@@ -105,6 +106,7 @@ WEBDAV_COMMON = {
     "key_path": str,
     "timeout": Coerce(int),
     "ssl_verify": Any(Bool, str),
+    Optional("verify", default=False): Bool,
 }
 
 SCHEMA = {
@@ -158,11 +160,13 @@ SCHEMA = {
                     "cache_regions": bool,
                     "read_timeout": Coerce(int),
                     "connect_timeout": Coerce(int),
+                    Optional("verify", default=False): Bool,
                     **REMOTE_COMMON,
                 },
                 "gs": {
                     "projectname": str,
                     "credentialpath": str,
+                    Optional("verify", default=False): Bool,
                     **REMOTE_COMMON,
                 },
                 "ssh": {
@@ -175,6 +179,7 @@ SCHEMA = {
                     "timeout": Coerce(int),
                     "gss_auth": Bool,
                     "allow_agent": Bool,
+                    Optional("verify", default=False): Bool,
                     **REMOTE_COMMON,
                 },
                 "hdfs": {"user": str, "kerb_ticket": str, **REMOTE_COMMON},
@@ -185,6 +190,7 @@ SCHEMA = {
                     "ssl_verify": Any(Bool, str),
                     "token": str,
                     "use_https": Bool,
+                    Optional("verify", default=False): Bool,
                     **REMOTE_COMMON,
                 },
                 "azure": {
@@ -200,12 +206,14 @@ SCHEMA = {
                     "exclude_visual_studio_code_credential": Bool,
                     "exclude_shared_token_cache_credential": Bool,
                     "exclude_managed_identity_credential": Bool,
+                    Optional("verify", default=False): Bool,
                     **REMOTE_COMMON,
                 },
                 "oss": {
                     "oss_key_id": str,
                     "oss_key_secret": str,
                     "oss_endpoint": str,
+                    Optional("verify", default=True): Bool,
                     **REMOTE_COMMON,
                 },
                 "gdrive": {
@@ -216,6 +224,7 @@ SCHEMA = {
                     "gdrive_service_account_user_email": str,
                     "gdrive_service_account_json_file_path": str,
                     Optional("gdrive_trash_only", default=False): Bool,
+                    Optional("verify", default=True): Bool,
                     **REMOTE_COMMON,
                 },
                 "http": {**HTTP_COMMON, **REMOTE_COMMON},
@@ -248,6 +257,7 @@ SCHEMA = {
             "instance_gpu": Lower,
             "ssh_private": str,
             "startup_script": str,
+            "setup_script": str,
         },
     },
     # section for experimental features
@@ -256,7 +266,10 @@ SCHEMA = {
         # enabled by default. It's of no use, kept for backward compatibility.
         Optional("parametrization", default=True): Bool,
     },
-    "plots": {"html_template": str},
+    "plots": {
+        "html_template": str,
+        Optional("auto_open", default=False): Bool,
+    },
     "exp": {
         "code": str,
         "data": str,

dvc/daemon.py

@@ -42,7 +42,7 @@ def _spawn_windows(cmd, env):
 
 
 def _spawn_posix(cmd, env):
-    from dvc.main import main
+    from dvc.cli import main
 
     # NOTE: using os._exit instead of sys.exit, because dvc built
     # with PyInstaller has trouble with SystemExit exception and throws

dvc/data/__init__.py

@@ -0,0 +1,33 @@
+import logging
+from typing import TYPE_CHECKING, Iterator, Union
+
+from .tree import Tree
+
+if TYPE_CHECKING:
+    from dvc.hash_info import HashInfo
+    from dvc.objects.db import ObjectDB
+    from dvc.objects.file import HashFile
+
+logger = logging.getLogger(__name__)
+
+
+def check(odb: "ObjectDB", obj: "HashFile", **kwargs):
+    if isinstance(obj, Tree):
+        for _, _, oid in obj:
+            odb.check(oid, **kwargs)
+
+    odb.check(obj.hash_info, **kwargs)
+
+
+def load(odb: "ObjectDB", hash_info: "HashInfo") -> "HashFile":
+    if hash_info.isdir:
+        return Tree.load(odb, hash_info)
+    return odb.get(hash_info)
+
+
+def iterobjs(
+    obj: Union["Tree", "HashFile"]
+) -> Iterator[Union["Tree", "HashFile"]]:
+    if isinstance(obj, Tree):
+        yield from (entry_obj for _, entry_obj in obj)
+    yield obj

dvc/data/checkout.py

@@ -5,14 +5,13 @@ from dvc import prompt
 from dvc.exceptions import CacheLinkError, CheckoutError, ConfirmRemoveError
 from dvc.fs.utils import test_links, transfer
 from dvc.ignore import DvcIgnoreFilter
-from dvc.objects.db.slow_link_detection import (  # type: ignore[attr-defined]
-    slow_link_guard,
-)
-from dvc.objects.diff import ROOT
-from dvc.objects.diff import diff as odiff
-from dvc.objects.stage import stage
 from dvc.types import Optional
 
+from .diff import ROOT
+from .diff import diff as odiff
+from .slow_link_detection import slow_link_guard  # type: ignore[attr-defined]
+from .stage import stage
+
 logger = logging.getLogger(__name__)
 
 

dvc/data/db/__init__.py

@@ -7,20 +7,13 @@ if TYPE_CHECKING:
 
 
 def get_odb(fs, fs_path, **config):
-    from .base import ObjectDB
-    from .gdrive import GDriveObjectDB
+    from dvc.objects.db import ObjectDB
+
     from .local import LocalObjectDB
-    from .oss import OSSObjectDB
 
     if fs.scheme == Schemes.LOCAL:
         return LocalObjectDB(fs, fs_path, **config)
 
-    if fs.scheme == Schemes.GDRIVE:
-        return GDriveObjectDB(fs, fs_path, **config)
-
-    if fs.scheme == Schemes.OSS:
-        return OSSObjectDB(fs, fs_path, **config)
-
     return ObjectDB(fs, fs_path, **config)
 
 
@@ -46,7 +39,6 @@ def get_index(odb) -> "ObjectDBIndexBase":
         hashlib.sha256(
             odb.fs.unstrip_protocol(odb.fs_path).encode("utf-8")
         ).hexdigest(),
-        odb.fs.CHECKSUM_DIR_SUFFIX,
     )
 
 

dvc/data/db/index.py

@@ -1,9 +1,10 @@
 import logging
 import os
 from abc import ABC, abstractmethod
-from typing import TYPE_CHECKING, Iterable, Optional, Set
+from typing import TYPE_CHECKING, Iterable, Set
 
-from ..errors import ObjectDBError
+from dvc.objects.errors import ObjectDBError
+from dvc.utils.decorators import with_diskcache
 
 if TYPE_CHECKING:
     from dvc.types import StrPath
@@ -14,7 +15,9 @@ logger = logging.getLogger(__name__)
 class ObjectDBIndexBase(ABC):
     @abstractmethod
     def __init__(
-        self, tmp_dir: "StrPath", name: str, dir_suffix: Optional[str] = None
+        self,
+        tmp_dir: "StrPath",
+        name: str,
     ):
         pass
 
@@ -50,7 +53,9 @@ class ObjectDBIndexNoop(ObjectDBIndexBase):
     """No-op class for ODBs which are not indexed."""
 
     def __init__(
-        self, tmp_dir: "StrPath", name: str, dir_suffix: Optional[str] = None
+        self,
+        tmp_dir: "StrPath",
+        name: str,
     ):  # pylint: disable=super-init-not-called
         pass
 
@@ -80,9 +85,11 @@ class ObjectDBIndex(ObjectDBIndexBase):
     INDEX_DIR = "index"
 
     def __init__(
-        self, tmp_dir: "StrPath", name: str, dir_suffix: Optional[str] = None
+        self,
+        tmp_dir: "StrPath",
+        name: str,
     ):  # pylint: disable=super-init-not-called
-        from diskcache import Index
+        from diskcache import Cache, Index
 
         from dvc.fs.local import LocalFileSystem
         from dvc.utils.fs import makedirs
@@ -90,25 +97,28 @@ class ObjectDBIndex(ObjectDBIndexBase):
         self.index_dir = os.path.join(tmp_dir, self.INDEX_DIR, name)
         makedirs(self.index_dir, exist_ok=True)
         self.fs = LocalFileSystem()
-        self.index = Index(self.index_dir)
-
-        if not dir_suffix:
-            dir_suffix = self.fs.CHECKSUM_DIR_SUFFIX
-        self.dir_suffix = dir_suffix
+        self.index = Index.fromcache(
+            Cache(
+                self.index_dir,
+                disk_pickle_protocol=4,
+                eviction_policy="none",
+            )
+        )
 
+    @with_diskcache(name="index")
     def __iter__(self):
         return iter(self.index)
 
+    @with_diskcache(name="index")
     def __contains__(self, hash_):
         return hash_ in self.index
 
+    @with_diskcache(name="index")
     def dir_hashes(self):
         """Iterate over .dir hashes stored in the index."""
         yield from (hash_ for hash_, is_dir in self.index.items() if is_dir)
 
-    def is_dir_hash(self, hash_: str):
-        return hash_.endswith(self.dir_suffix)
-
+    @with_diskcache(name="index")
     def clear(self):
         """Clear this index (to force re-indexing later)."""
         from diskcache import Timeout
@@ -118,6 +128,7 @@ class ObjectDBIndex(ObjectDBIndexBase):
         except Timeout as exc:
             raise ObjectDBError("Failed to clear ODB index") from exc
 
+    @with_diskcache(name="index")
     def update(self, dir_hashes: Iterable[str], file_hashes: Iterable[str]):
         """Update this index, adding the specified hashes."""
         from diskcache import Timeout
@@ -132,6 +143,7 @@ class ObjectDBIndex(ObjectDBIndexBase):
         except Timeout as exc:
             raise ObjectDBError("Failed to update ODB index") from exc
 
+    @with_diskcache(name="index")
     def intersection(self, hashes: Set[str]):
         """Iterate over values from `hashes` which exist in the index."""
         yield from hashes.intersection(self.index.keys())

dvc/data/db/local.py

@@ -6,13 +6,12 @@ from funcy import cached_property
 from shortuuid import uuid
 
 from dvc.hash_info import HashInfo
+from dvc.objects.db import ObjectDB
 from dvc.objects.errors import ObjectFormatError
 from dvc.progress import Tqdm
 from dvc.utils import relpath
 from dvc.utils.fs import copyfile, remove, umask, walk_files
 
-from .base import ObjectDB
-
 logger = logging.getLogger(__name__)
 
 

dvc/data/db/reference.py

@@ -3,11 +3,11 @@ import logging
 import os
 from typing import TYPE_CHECKING, Dict
 
+from dvc.objects.db import ObjectDB
+from dvc.objects.errors import ObjectFormatError
 from dvc.scheme import Schemes
 
-from ..errors import ObjectFormatError
 from ..reference import ReferenceHashFile
-from .base import ObjectDB
 
 if TYPE_CHECKING:
     from dvc.fs.base import FileSystem

dvc/data/diff.py

@@ -3,8 +3,7 @@ from typing import TYPE_CHECKING, List, Optional, Tuple
 
 if TYPE_CHECKING:
     from dvc.hash_info import HashInfo
-
-    from .file import HashFile
+    from dvc.objects.file import HashFile
 
 ADD = "add"
 MODIFY = "modify"
@@ -96,7 +95,7 @@ def diff(
         return entry_obj.hash_info if entry_obj else None
 
     def _in_cache(oid, cache):
-        from .errors import ObjectFormatError
+        from dvc.objects.errors import ObjectFormatError
 
         if not oid:
             return False

dvc/data/gc.py

@@ -0,0 +1,40 @@
+def gc(odb, used, jobs=None, cache_odb=None, shallow=True):
+    from dvc.data.tree import Tree
+    from dvc.objects.errors import ObjectDBPermissionError
+
+    if odb.read_only:
+        raise ObjectDBPermissionError("Cannot gc read-only ODB")
+    if not cache_odb:
+        cache_odb = odb
+    used_hashes = set()
+    for hash_info in used:
+        used_hashes.add(hash_info.value)
+        if hash_info.isdir and not shallow:
+            tree = Tree.load(cache_odb, hash_info)
+            used_hashes.update(
+                entry_obj.hash_info.value for _, entry_obj in tree
+            )
+
+    def _is_dir_hash(_hash):
+        from dvc.hash_info import HASH_DIR_SUFFIX
+
+        return _hash.endswith(HASH_DIR_SUFFIX)
+
+    removed = False
+    # hashes must be sorted to ensure we always remove .dir files first
+    for hash_ in sorted(
+        odb.all(jobs, odb.fs_path),
+        key=_is_dir_hash,
+        reverse=True,
+    ):
+        if hash_ in used_hashes:
+            continue
+        fs_path = odb.hash_to_path(hash_)
+        if _is_dir_hash(hash_):
+            # backward compatibility
+            # pylint: disable=protected-access
+            odb._remove_unpacked_dir(hash_)
+        odb.fs.remove(fs_path)
+        removed = True
+
+    return removed

dvc/data/meta.py

@@ -1,15 +1,25 @@
 from collections import OrderedDict
 from dataclasses import dataclass, field
-from typing import Optional
+from typing import TYPE_CHECKING, Optional
+
+if TYPE_CHECKING:
+    from dvc.objects.db import ObjectDB
+    from dvc.objects.file import HashFile
 
 
 @dataclass
 class Meta:
     PARAM_SIZE = "size"
     PARAM_NFILES = "nfiles"
+    PARAM_ISEXEC = "isexec"
 
     size: Optional[int] = field(default=None)
     nfiles: Optional[int] = field(default=None)
+    isexec: Optional[bool] = field(default=False)
+
+    obj: Optional["HashFile"] = field(default=None)
+    odb: Optional["ObjectDB"] = field(default=None)
+    remote: Optional[str] = field(default=None)
 
     @classmethod
     def from_dict(cls, d):
@@ -18,7 +28,9 @@ class Meta:
 
         size = d.pop(cls.PARAM_SIZE, None)
         nfiles = d.pop(cls.PARAM_NFILES, None)
-        return cls(size=size, nfiles=nfiles)
+        isexec = d.pop(cls.PARAM_ISEXEC, False)
+
+        return cls(size=size, nfiles=nfiles, isexec=isexec)
 
     def to_dict(self):
         ret = OrderedDict()
@@ -29,4 +41,7 @@ class Meta:
         if self.nfiles is not None:
             ret[self.PARAM_NFILES] = self.nfiles
 
+        if self.isexec:
+            ret[self.PARAM_ISEXEC] = self.isexec
+
         return ret

dvc/data/reference.py

@@ -4,16 +4,15 @@ import os
 import pickle
 from typing import TYPE_CHECKING, Optional
 
-from .errors import ObjectFormatError
-from .file import HashFile
+from dvc.objects.errors import ObjectFormatError
+from dvc.objects.file import HashFile
 
 if TYPE_CHECKING:
     from dvc.fs.base import FileSystem
     from dvc.hash_info import HashInfo
+    from dvc.objects.db import ObjectDB
     from dvc.types import AnyPath
 
-    from .db.base import ObjectDB
-
 logger = logging.getLogger(__name__)
 
 

dvc/data/stage.py

@@ -8,19 +8,18 @@ from typing import TYPE_CHECKING, Dict, Optional, Tuple
 from dvc.exceptions import DvcIgnoreInCollectedDirError
 from dvc.hash_info import HashInfo
 from dvc.ignore import DvcIgnore
+from dvc.objects.file import HashFile
 from dvc.progress import Tqdm
-from dvc.utils import file_md5
+from dvc.utils import file_md5, is_exec
 
 from .db.reference import ReferenceObjectDB
-from .file import HashFile
 from .meta import Meta
 
 if TYPE_CHECKING:
     from dvc.fs.base import FileSystem
+    from dvc.objects.db import ObjectDB
     from dvc.types import AnyPath
 
-    from .db.base import ObjectDB
-
 logger = logging.getLogger(__name__)
 
 
@@ -73,7 +72,7 @@ def _get_file_hash(fs_path, fs, name):
     else:
         raise NotImplementedError
 
-    meta = Meta(size=info["size"])
+    meta = Meta(size=info["size"], isexec=is_exec(info.get("mode", 0)))
     hash_info = HashInfo(name, hash_value)
     return meta, hash_info
 
@@ -169,7 +168,7 @@ def _build_tree(fs_path, fs, name, **kwargs):
         assert key
         tree.add(key, meta, obj.hash_info)
 
-        tree_meta.size += meta.size
+        tree_meta.size += meta.size or 0
         tree_meta.nfiles += 1
 
     return tree_meta, tree
@@ -245,8 +244,9 @@ def _get_staging(odb: "ObjectDB") -> "ObjectDB":
 
 
 def _load_from_state(odb, staging, fs_path, fs, name):
+    from dvc.objects.errors import ObjectFormatError
+
     from . import check, load
-    from .errors import ObjectFormatError
     from .tree import Tree
 
     state = odb.state

dvc/data/status.py

@@ -7,9 +7,10 @@ from dvc.scheme import Schemes
 from .tree import Tree
 
 if TYPE_CHECKING:
-    from .db.base import ObjectDB
+    from dvc.objects.db import ObjectDB
+    from dvc.objects.file import HashFile
+
     from .db.index import ObjectDBIndexBase
-    from .file import HashFile
 
 logger = logging.getLogger(__name__)
 

dvc/data/transfer.py

@@ -10,8 +10,8 @@ from dvc.utils.threadpool import ThreadPoolExecutor
 
 if TYPE_CHECKING:
     from dvc.hash_info import HashInfo
+    from dvc.objects.db import ObjectDB
 
-    from .db.base import ObjectDB
     from .db.index import ObjectDBIndexBase
     from .tree import Tree
 
@@ -40,7 +40,8 @@ def _log_exceptions(func):
 def find_tree_by_obj_id(
     odbs: Iterable[Optional["ObjectDB"]], obj_id: "HashInfo"
 ) -> Optional["Tree"]:
-    from .errors import ObjectFormatError
+    from dvc.objects.errors import ObjectFormatError
+
     from .tree import Tree
 
     for odb in odbs:

dvc/data/tree.py

@@ -1,22 +1,40 @@
 import json
 import logging
 import posixpath
-from typing import TYPE_CHECKING, Dict, Optional, Tuple
+from typing import TYPE_CHECKING, Dict, Iterable, Optional, Tuple
 
 from funcy import cached_property
 
-from .errors import ObjectFormatError
-from .file import HashFile
+from dvc.objects.errors import ObjectFormatError
+from dvc.objects.file import HashFile
+
 from .stage import get_file_hash
 
 if TYPE_CHECKING:
     from dvc.hash_info import HashInfo
+    from dvc.objects.db import ObjectDB
 
     from .meta import Meta
 
 logger = logging.getLogger(__name__)
 
 
+def _try_load(
+    odbs: Iterable["ObjectDB"],
+    hash_info: "HashInfo",
+) -> Optional["HashFile"]:
+    for odb in odbs:
+        if not odb:
+            continue
+
+        try:
+            return Tree.load(odb, hash_info)
+        except (FileNotFoundError, ObjectFormatError):
+            pass
+
+    return None
+
+
 class Tree(HashFile):
     PARAM_RELPATH = "relpath"
 
@@ -25,7 +43,7 @@ class Tree(HashFile):
         self._dict: Dict[Tuple[str], Tuple["Meta", "HashInfo"]] = {}
 
     @cached_property
-    def trie(self):
+    def _trie(self):
         from pygtrie import Trie
 
         return Trie(self._dict)
@@ -51,6 +69,30 @@ class Tree(HashFile):
             assert self.hash_info.value
             self.hash_info.value += ".dir"
 
+    def _load(self, key, meta, hash_info):
+        if hash_info and hash_info.isdir and not meta.obj:
+            meta.obj = _try_load([meta.odb, meta.remote], hash_info)
+            if meta.obj:
+                for ikey, value in meta.obj.iteritems():
+                    self._trie[key + ikey] = value
+                    self._dict[key + ikey] = value
+
+    def iteritems(self, prefix=None):
+        kwargs = {}
+        if prefix:
+            kwargs = {"prefix": prefix}
+            item = self._trie.longest_prefix(prefix)
+            if item:
+                key, (meta, hash_info) = item
+                self._load(key, meta, hash_info)
+
+        for key, (meta, hash_info) in self._trie.iteritems(**kwargs):
+            self._load(key, meta, hash_info)
+            yield key, (meta, hash_info)
+
+    def shortest_prefix(self, *args, **kwargs):
+        return self._trie.shortest_prefix(*args, **kwargs)
+
     def __len__(self):
         return len(self._dict)
 
@@ -130,7 +172,7 @@ class Tree(HashFile):
         """
         tree = Tree(self.fs_path, self.fs, self.hash_info)
         try:
-            for key, (meta, oid) in self.trie.items(prefix):
+            for key, (meta, oid) in self._trie.items(prefix):
                 tree.add(key, meta, oid)
         except KeyError:
             pass
@@ -148,7 +190,7 @@ class Tree(HashFile):
         tree = Tree(None, None, None)
         depth = len(prefix)
         try:
-            for key, (meta, entry_oid) in self.trie.items(prefix):
+            for key, (meta, entry_oid) in self._trie.items(prefix):
                 tree.add(key[depth:], meta, entry_oid)
         except KeyError:
             return None

dvc/data_cloud.py

@@ -3,11 +3,11 @@
 import logging
 from typing import TYPE_CHECKING, Iterable, Optional
 
-from dvc.objects.db import get_index
+from dvc.data.db import get_index
 
 if TYPE_CHECKING:
     from dvc.hash_info import HashInfo
-    from dvc.objects.db.base import ObjectDB
+    from dvc.objects.db import ObjectDB
 
 logger = logging.getLogger(__name__)
 
@@ -55,8 +55,8 @@ class DataCloud:
         raise NoRemoteError(error_msg)
 
     def _init_odb(self, name):
+        from dvc.data.db import get_odb
         from dvc.fs import get_cloud_fs
-        from dvc.objects.db import get_odb
 
         cls, config, fs_path = get_cloud_fs(self.repo, name=name)
         config["tmp_dir"] = self.repo.index_db_dir
@@ -78,7 +78,7 @@ class DataCloud:
                 By default remote from core.remote config option is used.
             odb: optional ODB to push to. Overrides remote.
         """
-        from dvc.objects.transfer import transfer
+        from dvc.data.transfer import transfer
 
         if not odb:
             odb = self.get_remote_odb(remote, "push")
@@ -107,7 +107,7 @@ class DataCloud:
                 By default remote from core.remote config option is used.
             odb: optional ODB to pull from. Overrides remote.
         """
-        from dvc.objects.transfer import transfer
+        from dvc.data.transfer import transfer
 
         if not odb:
             odb = self.get_remote_odb(remote, "pull")
@@ -141,7 +141,7 @@ class DataCloud:
             log_missing: log warning messages if file doesn't exist
                 neither in cache, neither in cloud.
         """
-        from dvc.objects.status import compare_status
+        from dvc.data.status import compare_status
 
         if not odb:
             odb = self.get_remote_odb(remote, "status")

dvc/dependency/param.py

@@ -1,3 +1,4 @@
+import errno
 import logging
 import os
 from collections import defaultdict
@@ -18,6 +19,14 @@ class MissingParamsError(DvcException):
     pass
 
 
+class MissingParamsFile(DvcException):
+    pass
+
+
+class ParamsIsADirectoryError(DvcException):
+    pass
+
+
 class BadParamFileError(DvcException):
     pass
 
@@ -27,9 +36,9 @@ class ParamsDependency(Dependency):
     PARAM_SCHEMA = {PARAM_PARAMS: Any(dict, list, None)}
     DEFAULT_PARAMS_FILE = "params.yaml"
 
-    def __init__(self, stage, path, params):
+    def __init__(self, stage, path, params=None, repo=None):
         info = {}
-        self.params = []
+        self.params = params or []
         if params:
             if isinstance(params, list):
                 self.params = params
@@ -43,6 +52,7 @@ class ParamsDependency(Dependency):
             path
             or os.path.join(stage.repo.root_dir, self.DEFAULT_PARAMS_FILE),
             info=info,
+            repo=repo,
         )
 
     def dumpd(self):
@@ -88,12 +98,31 @@ class ParamsDependency(Dependency):
     def status(self):
         return self.workspace_status()
 
-    def _read(self):
+    def validate_filepath(self):
         if not self.exists:
-            return {}
+            raise FileNotFoundError(
+                errno.ENOENT, os.strerror(errno.ENOENT), str(self)
+            )
+        if self.isdir():
+            raise IsADirectoryError(
+                errno.EISDIR, os.strerror(errno.EISDIR), str(self)
+            )
+
+    def read_file(self):
+        _, ext = os.path.splitext(self.fs_path)
+        loader = LOADERS[ext]
+
+        try:
+            self.validate_filepath()
+        except FileNotFoundError as exc:
+            raise MissingParamsFile(
+                f"Parameters file '{self}' does not exist"
+            ) from exc
+        except IsADirectoryError as exc:
+            raise ParamsIsADirectoryError(
+                f"'{self}' is a directory, expected a parameters file"
+            ) from exc
 
-        suffix = self.repo.fs.path.suffix(self.fs_path).lower()
-        loader = LOADERS[suffix]
         try:
             return loader(self.fs_path, fs=self.repo.fs)
         except ParseError as exc:
@@ -101,6 +130,12 @@ class ParamsDependency(Dependency):
                 f"Unable to read parameters from '{self}'"
             ) from exc
 
+    def _read(self):
+        try:
+            return self.read_file()
+        except MissingParamsFile:
+            return {}
+
     def read_params_d(self, **kwargs):
         config = self._read()
 

dvc/dependency/repo.py

@@ -9,7 +9,7 @@ from .base import Dependency
 
 if TYPE_CHECKING:
     from dvc.hash_info import HashInfo
-    from dvc.objects.db.base import ObjectDB
+    from dvc.objects.db import ObjectDB
     from dvc.objects.file import HashFile
 
 
@@ -61,7 +61,7 @@ class RepoDependency(Dependency):
         return {self.PARAM_PATH: self.def_path, self.PARAM_REPO: self.def_repo}
 
     def download(self, to, jobs=None):
-        from dvc.objects.checkout import checkout
+        from dvc.data.checkout import checkout
 
         for odb, objs in self.get_used_objs().items():
             self.repo.cloud.pull(objs, jobs=jobs, odb=odb)
@@ -98,9 +98,9 @@ class RepoDependency(Dependency):
         self, obj_only=False, **kwargs
     ) -> Tuple[Dict[Optional["ObjectDB"], Set["HashInfo"]], "HashFile"]:
         from dvc.config import NoRemoteError
+        from dvc.data.stage import stage
+        from dvc.data.tree import Tree
         from dvc.exceptions import NoOutputOrStageError, PathMissingError
-        from dvc.objects.stage import stage
-        from dvc.objects.tree import Tree
 
         local_odb = self.repo.odb.local
         locked = kwargs.pop("locked", True)
@@ -150,10 +150,10 @@ class RepoDependency(Dependency):
             return used_obj_ids, staged_obj
 
     def _check_circular_import(self, odb, obj_ids):
+        from dvc.data.db.reference import ReferenceObjectDB
+        from dvc.data.tree import Tree
         from dvc.exceptions import CircularImportError
         from dvc.fs.repo import RepoFileSystem
-        from dvc.objects.db.reference import ReferenceObjectDB
-        from dvc.objects.tree import Tree
 
         if not isinstance(odb, ReferenceObjectDB):
             return

dvc/exceptions.py

@@ -128,13 +128,6 @@ class NotDvcRepoError(DvcException):
     """Thrown if a directory is not a DVC repo"""
 
 
-class DvcParserError(DvcException):
-    """Base class for CLI parser errors."""
-
-    def __init__(self):
-        super().__init__("parser error")
-
-
 class CyclicGraphError(DvcException):
     def __init__(self, stages):
         assert isinstance(stages, list)
@@ -157,9 +150,9 @@ class InitError(DvcException):
 
 
 class ReproductionError(DvcException):
-    def __init__(self, dvc_file_name):
-        self.path = dvc_file_name
-        super().__init__(f"failed to reproduce '{dvc_file_name}'")
+    def __init__(self, name):
+        self.name = name
+        super().__init__(f"failed to reproduce '{name}'")
 
 
 class BadMetricError(DvcException):

dvc/external_repo.py

@@ -30,6 +30,7 @@ def external_repo(
     from scmrepo.git import Git
 
     from dvc.config import NoRemoteError
+    from dvc.fs.git import GitFileSystem
 
     logger.debug("Creating external repo %s@%s", url, rev)
     path = _cached_clone(url, rev, for_write=for_write)
@@ -45,12 +46,18 @@ def external_repo(
     config = _get_remote_config(url) if os.path.isdir(url) else {}
     config.update(cache_config)
 
-    root_dir = path if for_write else os.path.realpath(path)
+    if for_write:
+        root_dir = path
+        fs = None
+    else:
+        root_dir = os.path.realpath(path)
+        scm = Git(root_dir)
+        fs = GitFileSystem(scm=scm, rev=rev)
+
     repo_kwargs = dict(
         root_dir=root_dir,
         url=url,
-        scm=None if for_write else Git(root_dir),
-        rev=None if for_write else rev,
+        fs=fs,
         config=config,
         repo_factory=erepo_factory(url, cache_config),
         **kwargs,
@@ -86,7 +93,7 @@ def erepo_factory(url, cache_config, *args, **kwargs):
     def make_repo(path, **_kwargs):
         _config = cache_config.copy()
         if os.path.isdir(url):
-            rel = os.path.relpath(path, _kwargs["scm"].root_dir)
+            rel = os.path.relpath(path, _kwargs["fs"].fs_args["scm"].root_dir)
             repo_path = os.path.join(url, rel)
             _config.update(_get_remote_config(repo_path))
         return Repo(path, config=_config, **_kwargs)

dvc/fs/_callback.py

@@ -0,0 +1,49 @@
+import fsspec
+
+from dvc.progress import Tqdm
+
+
+class FsspecCallback(fsspec.Callback):
+    def __init__(self, progress_bar):
+        self.progress_bar = progress_bar
+        super().__init__()
+
+    def set_size(self, size):
+        if size is not None:
+            self.progress_bar.total = size
+            self.progress_bar.refresh()
+            super().set_size(size)
+
+    def relative_update(self, inc=1):
+        self.progress_bar.update(inc)
+        super().relative_update(inc)
+
+    def absolute_update(self, value):
+        self.progress_bar.update_to(value)
+        super().absolute_update(value)
+
+    @staticmethod
+    def wrap_fn(cb, fn):
+        def wrapped(*args, **kwargs):
+            res = fn(*args, **kwargs)
+            cb.relative_update()
+            return res
+
+        return wrapped
+
+
+def tdqm_or_callback_wrapped(
+    fobj, method, total, callback=None, **pbar_kwargs
+):
+    if callback:
+        from funcy import nullcontext
+        from tqdm.utils import CallbackIOWrapper
+
+        callback.set_size(total)
+        wrapper = CallbackIOWrapper(callback.relative_update, fobj, method)
+        return nullcontext(wrapper)
+
+    return Tqdm.wrapattr(fobj, method, total=total, bytes=True, **pbar_kwargs)
+
+
+DEFAULT_CALLBACK = fsspec.callbacks.NoOpCallback()

dvc/fs/_metadata.py

@@ -1,73 +0,0 @@
-from dataclasses import dataclass, field
-from typing import TYPE_CHECKING, List
-
-from dvc.output import Output
-
-if TYPE_CHECKING:
-    from dvc.repo import Repo
-
-
-@dataclass
-class Metadata:
-    """
-    Container for storing metadata for a given path, similar in spirit to
-    `os.stat_result`.
-    """
-
-    # required field
-    fs_path: str
-    repo: "Repo"
-
-    # computed fields
-    is_output: bool = field(init=False, default=False)  # is it an output?
-    # is the path part of an output?
-    part_of_output: bool = field(init=False, default=False)
-    # does the path contain outputs?
-    contains_outputs: bool = field(init=False, default=False)
-    # is the path tracked by dvc? equivalent to (is_output or part_of_output)
-    is_dvc: bool = field(init=False, default=False)
-    # does dvc has any output in that path or inside it? equiv. to bool(outs)
-    output_exists: bool = field(init=False)
-
-    # optional fields
-    isdir: bool = False  # is it a directory?
-    is_exec: bool = False  # is it an executable?
-    outs: List[Output] = field(default_factory=list)  # list of outputs
-
-    def __post_init__(self):
-        self.output_exists = bool(self.outs)
-
-        if not self.output_exists:
-            return
-
-        # it can contain multiple outputs
-        if len(self.outs) > 1:
-            self.contains_outputs = True
-        else:
-            out = self.outs[0]
-            # or, the path itself could be an output
-            self.is_output = self.fs_path == out.fs_path
-
-            # or, a directory must have an output somewhere deep inside it
-            if not self.is_output:
-                self.contains_outputs = self.repo.fs.path.isin(
-                    out.fs_path, self.fs_path
-                )
-
-        # or, the path could be a part of an output, i.e. inside of an output
-        self.part_of_output = not self.is_output and not self.contains_outputs
-
-        self.is_dvc = self.is_output or self.part_of_output
-        # if it contains outputs, it must be a dir
-        self.isdir = self.contains_outputs
-
-    @property
-    def isfile(self):
-        # self.isdir might require additional computation, especially if it's
-        # a dir that's inside a dvc-tracked dir
-        # since, `Metadata()` does not support being used as null (i.e. when
-        # path does not exist), `isfile` becomes `isdir`'s complement.
-        return not self.isdir
-
-    def __str__(self):
-        return f"Metadata: {self.fs_path}"

dvc/fs/azure.py

@@ -41,7 +41,6 @@ def _az_config():
 class AzureFileSystem(CallbackMixin, ObjectFSWrapper):
     scheme = Schemes.AZURE
     PARAM_CHECKSUM = "etag"
-    DETAIL_FIELDS = frozenset(("etag", "size"))
     REQUIRES = {
         "adlfs": "adlfs",
         "knack": "knack",

dvc/fs/base.py

@@ -4,13 +4,13 @@ import os
 from concurrent.futures import ThreadPoolExecutor, as_completed
 from functools import partialmethod
 from multiprocessing import cpu_count
-from typing import ClassVar, Dict, FrozenSet, Optional
+from typing import ClassVar, Dict, Optional
 
 from funcy import cached_property
 from tqdm.utils import CallbackIOWrapper
 
 from dvc.exceptions import DvcException
-from dvc.progress import DEFAULT_CALLBACK, FsspecCallback
+from dvc.fs._callback import DEFAULT_CALLBACK, FsspecCallback
 from dvc.ui import ui
 from dvc.utils import tmp_fname
 from dvc.utils.fs import makedirs, move
@@ -18,14 +18,6 @@ from dvc.utils.fs import makedirs, move
 logger = logging.getLogger(__name__)
 
 
-class RemoteCmdError(DvcException):
-    def __init__(self, remote, cmd, ret, err):
-        super().__init__(
-            "{remote} command '{cmd}' finished with non-zero return code"
-            " {ret}': {err}".format(remote=remote, cmd=cmd, ret=ret, err=err)
-        )
-
-
 class RemoteActionNotImplemented(DvcException):
     def __init__(self, action, scheme):
         m = f"{action} is not supported for {scheme} remotes"
@@ -43,7 +35,6 @@ class FileSystem:
     REQUIRES: ClassVar[Dict[str, str]] = {}
     _JOBS = 4 * cpu_count()
 
-    CHECKSUM_DIR_SUFFIX = ".dir"
     HASH_JOBS = max(1, min(4, cpu_count() // 2))
     LIST_OBJECT_PAGE_SIZE = 1000
     TRAVERSE_WEIGHT_MULTIPLIER = 5
@@ -55,7 +46,6 @@ class FileSystem:
     CHUNK_SIZE = 64 * 1024 * 1024  # 64 MiB
 
     PARAM_CHECKSUM: ClassVar[Optional[str]] = None
-    DETAIL_FIELDS: FrozenSet[str] = frozenset()
 
     def __init__(self, **kwargs):
         self._check_requires(**kwargs)
@@ -160,12 +150,6 @@ class FileSystem:
         """
         return True
 
-    def isexec(self, path):
-        """Optional: Overwrite only if the remote has a way to distinguish
-        between executable and non-executable file.
-        """
-        return False
-
     def iscopy(self, path):
         """Check if this file is an independent copy."""
         return False  # We can't be sure by default
@@ -218,12 +202,6 @@ class FileSystem:
 
     # pylint: enable=unused-argument
 
-    @classmethod
-    def is_dir_hash(cls, hash_):
-        if not hash_:
-            return False
-        return hash_.endswith(cls.CHECKSUM_DIR_SUFFIX)
-
     def upload(
         self,
         from_info,

dvc/fs/dvc.py

@@ -2,15 +2,10 @@ import logging
 import os
 import typing
 
-from dvc.exceptions import OutputNotFoundError
-from dvc.utils import relpath
-
-from ..progress import DEFAULT_CALLBACK
-from ._metadata import Metadata
+from ._callback import DEFAULT_CALLBACK
 from .base import FileSystem
 
 if typing.TYPE_CHECKING:
-    from dvc.output import Output
     from dvc.types import AnyPath
 
 logger = logging.getLogger(__name__)
@@ -36,67 +31,46 @@ class DvcFileSystem(FileSystem):  # pylint:disable=abstract-method
     def config(self):
         raise NotImplementedError
 
-    def _find_outs(self, path, *args, **kwargs):
-        outs = self.repo.find_outs_by_path(path, *args, **kwargs)
+    def _get_key(self, path):
+        from dvc.fs.local import LocalFileSystem
 
-        def _is_cached(out):
-            return out.use_cache
+        from . import get_cloud_fs
 
-        outs = list(filter(_is_cached, outs))
-        if not outs:
-            raise OutputNotFoundError(path, self.repo)
+        cls, kwargs, fs_path = get_cloud_fs(None, url=path)
 
-        return outs
+        if cls != LocalFileSystem or os.path.isabs(path):
+            fs = cls(**kwargs)
+            return (cls.scheme, *fs.path.parts(fs_path))
 
-    def _get_granular_hash(self, path: "AnyPath", out: "Output", remote=None):
-        # NOTE: use string paths here for performance reasons
-        key = tuple(relpath(path, out.fs_path).split(os.sep))
-        out.get_dir_cache(remote=remote)
-        if out.obj is None:
-            raise FileNotFoundError
-        (_, oid) = out.obj.trie.get(key) or (None, None)
-        if oid:
-            return oid
-        raise FileNotFoundError
+        fs_key = "repo"
+        key = self.path.parts(path)
+        if key == (".",):
+            key = ()
+
+        return (fs_key, *key)
 
     def _get_fs_path(self, path: "AnyPath", remote=None):
-        try:
-            outs = self._find_outs(path, strict=False)
-        except OutputNotFoundError as exc:
-            raise FileNotFoundError from exc
+        from dvc.config import NoRemoteError
 
-        if len(outs) != 1 or (
-            outs[0].is_dir_checksum and path == outs[0].fs_path
-        ):
+        info = self.info(path)
+        if info["type"] == "directory":
             raise IsADirectoryError
 
-        out = outs[0]
-
-        if not out.hash_info:
+        value = info.get("md5")
+        if not value:
             raise FileNotFoundError
 
-        if out.changed_cache(filter_info=path):
-            from dvc.config import NoRemoteError
+        cache_path = self.repo.odb.local.hash_to_path(value)
 
-            try:
-                remote_odb = self.repo.cloud.get_remote_odb(remote)
-            except NoRemoteError as exc:
-                raise FileNotFoundError from exc
-            if out.is_dir_checksum:
-                checksum = self._get_granular_hash(path, out).value
-            else:
-                checksum = out.hash_info.value
-            remote_fs_path = remote_odb.hash_to_path(checksum)
-            return remote_odb.fs, remote_fs_path
+        if self.repo.odb.local.fs.exists(cache_path):
+            return self.repo.odb.local.fs, cache_path
 
-        if out.is_dir_checksum:
-            checksum = self._get_granular_hash(path, out).value
-            cache_path = out.odb.fs.unstrip_protocol(
-                out.odb.hash_to_path(checksum)
-            )
-        else:
-            cache_path = out.cache_path
-        return out.odb.fs, cache_path
+        try:
+            remote_odb = self.repo.cloud.get_remote_odb(remote)
+        except NoRemoteError as exc:
+            raise FileNotFoundError from exc
+        remote_fs_path = remote_odb.hash_to_path(value)
+        return remote_odb.fs, remote_fs_path
 
     def open(  # type: ignore
         self, path: str, mode="r", encoding=None, **kwargs
@@ -106,72 +80,44 @@ class DvcFileSystem(FileSystem):  # pylint:disable=abstract-method
 
     def exists(self, path):  # pylint: disable=arguments-renamed
         try:
-            self.metadata(path)
+            self.info(path)
             return True
         except FileNotFoundError:
             return False
 
     def isdir(self, path):  # pylint: disable=arguments-renamed
         try:
-            meta = self.metadata(path)
-            return meta.isdir
+            return self.info(path)["type"] == "directory"
         except FileNotFoundError:
             return False
 
-    def check_isdir(self, path, outs):
-        if len(outs) != 1:
-            return True
-
-        out = outs[0]
-        if not out.is_dir_checksum:
-            return out.fs_path != path
-        if out.fs_path == path:
-            return True
-
-        try:
-            self._get_granular_hash(path, out)
-            return False
-        except FileNotFoundError:
-            return True
-
     def isfile(self, path):  # pylint: disable=arguments-renamed
         try:
-            meta = self.metadata(path)
-            return meta.isfile
+            return self.info(path)["type"] == "file"
         except FileNotFoundError:
             return False
 
-    def _fetch_dir(self, out, **kwargs):
-        # pull dir cache if needed
-        out.get_dir_cache(**kwargs)
-
-        if not out.obj:
-            raise FileNotFoundError
-
-    def _add_dir(self, trie, out, **kwargs):
-        self._fetch_dir(out, **kwargs)
-
-        base = out.fs.path.parts(out.fs_path)
-        for key, _, _ in out.obj:  # noqa: B301
-            trie[base + key] = None
-
-    def _walk(self, root, trie, topdown=True, **kwargs):
+    def _walk(self, root, topdown=True, **kwargs):
         dirs = set()
         files = []
 
-        root_parts = self.path.parts(root)
-        out = trie.get(root_parts)
-        if out and out.is_dir_checksum:
-            self._add_dir(trie, out, **kwargs)
-
+        root_parts = self._get_key(root)
         root_len = len(root_parts)
         try:
-            for key, out in trie.iteritems(prefix=root_parts):  # noqa: B301
+            for key, (meta, hash_info) in self.repo.index.tree.iteritems(
+                prefix=root_parts
+            ):  # noqa: B301
+                if hash_info and hash_info.isdir and meta and not meta.obj:
+                    raise FileNotFoundError
+
                 if key == root_parts:
                     continue
 
+                if hash_info.isdir:
+                    continue
+
                 name = key[root_len]
-                if len(key) > root_len + 1 or (out and out.is_dir_checksum):
+                if len(key) > root_len + 1:
                     dirs.add(name)
                     continue
 
@@ -184,33 +130,23 @@ class DvcFileSystem(FileSystem):  # pylint:disable=abstract-method
         yield root, dirs, files
 
         for dname in dirs:
-            yield from self._walk(self.path.join(root, dname), trie)
+            yield from self._walk(self.path.join(root, dname))
 
     def walk(self, top, topdown=True, onerror=None, **kwargs):
-        from pygtrie import Trie
-
         assert topdown
-        root = os.path.abspath(top)
         try:
-            meta = self.metadata(root)
+            info = self.info(top)
         except FileNotFoundError:
             if onerror is not None:
                 onerror(FileNotFoundError(top))
             return
 
-        if not meta.isdir:
+        if info["type"] != "directory":
             if onerror is not None:
                 onerror(NotADirectoryError(top))
             return
 
-        trie = Trie()
-        for out in meta.outs:
-            trie[out.fs.path.parts(out.fs_path)] = out
-
-            if out.is_dir_checksum and self.path.isin_or_eq(root, out.fs_path):
-                self._add_dir(trie, out, **kwargs)
-
-        yield from self._walk(root, trie, topdown=topdown, **kwargs)
+        yield from self._walk(top, topdown=topdown, **kwargs)
 
     def find(self, path, prefix=None):
         for root, _, files in self.walk(path):
@@ -220,43 +156,58 @@ class DvcFileSystem(FileSystem):  # pylint:disable=abstract-method
 
     def isdvc(self, path, recursive=False, strict=True):
         try:
-            meta = self.metadata(path)
+            info = self.info(path)
         except FileNotFoundError:
             return False
 
         recurse = recursive or not strict
-        return meta.output_exists if recurse else meta.is_output
-
-    def isexec(self, path):  # pylint: disable=unused-argument
-        return False
-
-    def metadata(self, fs_path):
-        abspath = os.path.abspath(fs_path)
-
-        try:
-            outs = self._find_outs(abspath, strict=False, recursive=True)
-        except OutputNotFoundError as exc:
-            raise FileNotFoundError from exc
-
-        meta = Metadata(fs_path=abspath, outs=outs, repo=self.repo)
-        meta.isdir = meta.isdir or self.check_isdir(meta.fs_path, meta.outs)
-        return meta
+        return bool(info.get("outs") if recurse else info.get("isout"))
 
     def info(self, path):
-        meta = self.metadata(path)
-        ret = {"type": "directory" if meta.isdir else "file"}
-        if meta.is_output and len(meta.outs) == 1 and meta.outs[0].hash_info:
-            out = meta.outs[0]
-            ret["size"] = out.meta.size
-            ret[out.hash_info.name] = out.hash_info.value
-        elif meta.part_of_output:
-            (out,) = meta.outs
-            key = self.path.parts(self.path.relpath(path, out.fs_path))
-            (obj_meta, oid) = out.obj.trie.get(key) or (None, None)
-            if oid:
-                ret["size"] = obj_meta.size if obj_meta else 0
-                ret[oid.name] = oid.value
+        from dvc.data.meta import Meta
 
+        key = self._get_key(path)
+
+        try:
+            outs = list(self.repo.index.tree.iteritems(key))
+        except KeyError as exc:
+            raise FileNotFoundError from exc
+
+        ret = {
+            "type": "file",
+            "size": 0,
+            "isexec": False,
+            "isdvc": False,
+            "outs": outs,
+        }
+
+        if len(outs) > 1 and outs[0][0] != key:
+            shortest = self.repo.index.tree.shortest_prefix(key)
+            if shortest:
+                assert shortest[1][1].isdir
+                if len(shortest[0]) <= len(key):
+                    ret["isdvc"] = True
+
+            ret["type"] = "directory"
+            return ret
+
+        item_key, (meta, hash_info) = outs[0]
+
+        meta = meta or Meta()
+
+        if key != item_key:
+            assert item_key[: len(key)] == key
+            ret["type"] = "directory"
+            return ret
+
+        ret["size"] = meta.size
+        ret["isexec"] = meta.isexec
+        ret[hash_info.name] = hash_info.value
+        ret["isdvc"] = True
+        ret["isout"] = True
+        ret["meta"] = meta
+        if hash_info and hash_info.isdir:
+            ret["type"] = "directory"
         return ret
 
     def get_file(

dvc/fs/fsspec_wrapper.py

@@ -5,8 +5,7 @@ from typing import IO, TYPE_CHECKING, Any, Dict, Iterator, Optional, overload
 from funcy import cached_property
 from tqdm.utils import CallbackIOWrapper
 
-from dvc.progress import DEFAULT_CALLBACK
-
+from ._callback import DEFAULT_CALLBACK
 from .base import FileSystem
 
 FSPath = str
@@ -141,7 +140,9 @@ class FSSpecWrapper(FileSystem):
         self.makedirs(self.path.parent(to_info))
         with self.open(to_info, "wb") as fdest:
             shutil.copyfileobj(
-                fobj, fdest, length=getattr(fdest, "blocksize", None)
+                fobj,
+                fdest,
+                length=getattr(fdest, "blocksize", None),  # type: ignore
             )
 
 

dvc/fs/git.py

@@ -51,14 +51,6 @@ class GitFileSystem(FSSpecWrapper):  # pylint:disable=abstract-method
     def rev(self) -> str:
         return self.fs.rev
 
-    def isexec(self, path: AnyFSPath) -> bool:
-        from dvc.utils import is_exec
-
-        try:
-            return is_exec(self.info(path)["mode"])
-        except FileNotFoundError:
-            return False
-
     def isfile(self, path: AnyFSPath) -> bool:
         return self.fs.isfile(path)