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]>

.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: >-

.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

@@ -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.
 
 

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

@@ -1,10 +1,15 @@
 import argparse
 
-from dvc.command.base import append_doc_link, fix_subparsers
-from dvc.command.experiments import (
+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,
@@ -19,6 +24,7 @@ SUB_COMMANDS = [
     apply,
     branch,
     diff,
+    exec_run,
     gc,
     init,
     ls,
@@ -51,3 +57,4 @@ def add_parser(subparsers, parent_parser):
     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

@@ -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/experiments/branch.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/experiments/diff.py

@@ -1,9 +1,10 @@
 import argparse
 import logging
 
-from dvc.command import completion
-from dvc.command.base import CmdBase, append_doc_link
-from dvc.command.metrics import DEFAULT_PRECISION
+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
 

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

@@ -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 InvalidArgumentError
 from dvc.ui import ui
 

dvc/commands/experiments/init.py

@@ -3,7 +3,8 @@ import logging
 
 from funcy import compact
 
-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 InvalidArgumentError
 from dvc.ui import ui
 
@@ -31,7 +32,7 @@ class CmdExperimentsInit(CmdBase):
     EXP_LINK = "https://s.dvc.org/g/exp/run"
 
     def run(self):
-        from dvc.command.stage import parse_cmd
+        from dvc.commands.stage import parse_cmd
 
         cmd = parse_cmd(self.args.command)
         if not self.args.interactive and not cmd:
@@ -97,7 +98,7 @@ class CmdExperimentsInit(CmdBase):
 
 def add_parser(experiments_subparsers, parent_parser):
 
-    EXPERIMENTS_INIT_HELP = "Initialize experiments."
+    EXPERIMENTS_INIT_HELP = "Quickly setup any project to use experiments."
     experiments_init_parser = experiments_subparsers.add_parser(
         "init",
         parents=[parent_parser],

dvc/commands/experiments/ls.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/experiments/pull.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.ui import ui
 
 logger = logging.getLogger(__name__)
@@ -87,6 +88,9 @@ def add_parser(experiments_subparsers, parent_parser):
         metavar="<git_remote>",
     )
     experiments_pull_parser.add_argument(
-        "experiment", help="Experiment to pull.", metavar="<experiment>"
+        "experiment",
+        nargs="+",
+        help="Experiments to pull.",
+        metavar="<experiment>",
     )
     experiments_pull_parser.set_defaults(func=CmdExperimentsPull)

dvc/commands/experiments/push.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.ui import ui
 
 logger = logging.getLogger(__name__)
@@ -88,6 +89,9 @@ def add_parser(experiments_subparsers, parent_parser):
         metavar="<git_remote>",
     )
     experiments_push_parser.add_argument(
-        "experiment", help="Experiment to push.", metavar="<experiment>"
+        "experiment",
+        nargs="+",
+        help="Experiments to push.",
+        metavar="<experiment>",
     ).complete = completion.EXPERIMENT
     experiments_push_parser.set_defaults(func=CmdExperimentsPush)

dvc/commands/experiments/remove.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/experiments/run.py

@@ -1,10 +1,10 @@
 import argparse
 import logging
 
-from dvc.command import completion
-from dvc.command.base import append_doc_link
-from dvc.command.repro import CmdRepro
-from dvc.command.repro import add_arguments as add_repro_arguments
+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
 
@@ -38,6 +38,7 @@ class CmdExperimentsRun(CmdRepro):
             checkpoint_resume=self.args.checkpoint_resume,
             reset=self.args.reset,
             tmp_dir=self.args.tmp_dir,
+            machine=self.args.machine,
             **self._repro_kwargs,
         )
 
@@ -130,3 +131,12 @@ def _add_run_common(parser):
             "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

@@ -1,16 +1,17 @@
 import argparse
 import logging
+import os
+import re
 from collections import Counter, OrderedDict, defaultdict
 from datetime import date, datetime
-from fnmatch import fnmatch
-from pathlib import Path
-from typing import TYPE_CHECKING, Dict, Iterable, Optional
+from typing import TYPE_CHECKING
 
 from funcy import lmap
 
-from dvc.command import completion
-from dvc.command.base import CmdBase, append_doc_link
-from dvc.command.metrics import DEFAULT_PRECISION
+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
@@ -27,64 +28,6 @@ FILL_VALUE_ERRORED = "!"
 logger = logging.getLogger(__name__)
 
 
-def _filter_name(names, label, filter_strs):
-    ret = defaultdict(dict)
-    path_filters = defaultdict(list)
-
-    for filter_s in filter_strs:
-        path, _, name = filter_s.rpartition(":")
-        path_filters[path].append(name)
-
-    for path, filters in path_filters.items():
-        if path:
-            match_paths = [path]
-        else:
-            match_paths = names.keys()
-        for match_path in match_paths:
-            for f in filters:
-                matches = [
-                    name for name in names[match_path] if fnmatch(name, f)
-                ]
-                if not matches:
-                    raise InvalidArgumentError(
-                        f"'{f}' does not match any known {label}"
-                    )
-                ret[match_path].update({match: None for match in matches})
-
-    return ret
-
-
-def _filter_names(
-    names: Dict[str, Dict[str, None]],
-    label: str,
-    include: Optional[Iterable],
-    exclude: Optional[Iterable],
-):
-    if include and exclude:
-        intersection = set(include) & set(exclude)
-        if intersection:
-            values = ", ".join(intersection)
-            raise InvalidArgumentError(
-                f"'{values}' specified in both --include-{label} and"
-                f" --exclude-{label}"
-            )
-
-    if include:
-        ret = _filter_name(names, label, include)
-    else:
-        ret = names
-
-    if exclude:
-        to_remove = _filter_name(names, label, exclude)
-        for path in to_remove:
-            if path in ret:
-                for key in to_remove[path]:
-                    if key in ret[path]:
-                        del ret[path][key]
-
-    return ret
-
-
 def _update_names(names, items):
     for name, item in items:
         item = item.get("data", {})
@@ -96,26 +39,17 @@ def _update_names(names, items):
 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())
-    metric_names = _filter_names(
-        metric_names,
-        "metrics",
-        kwargs.get("include_metrics"),
-        kwargs.get("exclude_metrics"),
-    )
-    param_names = _filter_names(
-        param_names,
-        "params",
-        kwargs.get("include_params"),
-        kwargs.get("exclude_params"),
-    )
+            for dep_name in exp.get("deps", {}):
+                deps_names.add(dep_name)
 
-    return metric_names, param_names
+    return metric_names, param_names, sorted(deps_names)
 
 
 experiment_types = {
@@ -131,8 +65,12 @@ experiment_types = {
 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,
@@ -152,14 +90,18 @@ def _collect_rows(
 
     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
-        executor = exp.get("executor", fill_value)
+
         is_baseline = rev == "baseline"
 
         if is_baseline:
@@ -167,9 +109,7 @@ def _collect_rows(
         else:
             name_rev = rev[:7]
 
-        exp_name = exp.get("name", "")
         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")
@@ -195,32 +135,38 @@ def _collect_rows(
         if not is_baseline:
             new_checkpoint = not (tip and tip == parent_tip)
 
-        row = [
-            exp_name,
-            name_rev,
-            typ,
-            _format_time(exp.get("timestamp"), fill_value, iso),
-            parent,
-            state,
-            executor,
-        ]
-        fill_value = FILL_VALUE_ERRORED if results.get("error") else fill_value
+        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,
+            row_dict,
             metric_names,
+            metric_headers,
             exp.get("metrics", {}).items(),
             precision,
             fill_value=fill_value,
         )
         _extend_row(
-            row,
+            row_dict,
             param_names,
+            param_headers,
             exp.get("params", {}).items(),
             precision,
             fill_value=fill_value,
         )
-
-        yield row
+        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):
@@ -286,13 +232,9 @@ def _format_time(datetime_obj, fill_value=FILL_VALUE, iso=False):
     return datetime_obj.strftime(fmt)
 
 
-def _extend_row(row, names, items, precision, fill_value=FILL_VALUE):
+def _extend_row(row, names, headers, items, precision, fill_value=FILL_VALUE):
     from dvc.compare import _format_field, with_value
 
-    if not items:
-        row.extend(fill_value for keys in names.values() for _ in keys)
-        return
-
     for fname, data in items:
         item = data.get("data", {})
         item = flatten(item) if isinstance(item, dict) else {fname: item}
@@ -304,18 +246,11 @@ def _extend_row(row, names, items, precision, fill_value=FILL_VALUE):
             # wrap field data in ui.rich_text, otherwise rich may
             # interpret unescaped braces from list/dict types as rich
             # markup tags
-            row.append(ui.rich_text(str(_format_field(value, precision))))
-
-
-def _parse_filter_list(param_list):
-    ret = []
-    for param_str in param_list:
-        path, _, param_str = param_str.rpartition(":")
-        if path:
-            ret.extend(f"{path}:{param}" for param in param_str.split(","))
-        else:
-            ret.extend(param_str.split(","))
-    return ret
+            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(
@@ -325,6 +260,7 @@ def experiments_table(
     metric_names,
     param_headers,
     param_names,
+    deps_names,
     sort_by=None,
     sort_order=None,
     precision=DEFAULT_PRECISION,
@@ -335,15 +271,18 @@ def experiments_table(
 
     from dvc.compare import TabularData
 
-    td = TabularData(
-        lconcat(headers, metric_headers, param_headers), fill_value=fill_value
-    )
+    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,
@@ -375,32 +314,22 @@ def prepare_exp_id(kwargs) -> "RichText":
 
 
 def baseline_styler(typ):
-    return {"style": "bold"} if typ == "baseline" else {"style": "default"}
+    return {"style": "bold"} if typ == "baseline" else {}
 
 
 def show_experiments(
     all_experiments,
+    keep=None,
+    drop=None,
     pager=True,
-    no_timestamp=False,
     csv=False,
     markdown=False,
-    html=False,
+    pcp=False,
     **kwargs,
 ):
     from funcy.seqs import flatten as flatten_list
 
-    include_metrics = _parse_filter_list(kwargs.pop("include_metrics", []))
-    exclude_metrics = _parse_filter_list(kwargs.pop("exclude_metrics", []))
-    include_params = _parse_filter_list(kwargs.pop("include_params", []))
-    exclude_params = _parse_filter_list(kwargs.pop("exclude_params", []))
-
-    metric_names, param_names = _collect_names(
-        all_experiments,
-        include_metrics=include_metrics,
-        exclude_metrics=exclude_metrics,
-        include_params=include_params,
-        exclude_params=exclude_params,
-    )
+    metric_names, param_names, deps_names = _collect_names(all_experiments)
 
     headers = [
         "Experiment",
@@ -425,24 +354,23 @@ def show_experiments(
         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 no_timestamp or html:
-        td.drop("Created")
+    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"))
-    for n, row_style in enumerate(row_styles):
-        if n % 2 != 0:
-            row_style["style"] += " on grey23"
 
     if not csv:
         merge_headers = ["Experiment", "rev", "typ", "parent"]
@@ -451,14 +379,22 @@ def show_experiments(
         )
         td.drop(*merge_headers[1:])
 
-    headers = {"metrics": metric_headers, "params": param_headers}
+    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"}
+    header_bg_colors = {
+        "metrics": "cornsilk1",
+        "params": "light_cyan1",
+        "deps": "plum2",
+    }
     styles.update(
         {
             header: {
@@ -472,35 +408,49 @@ def show_experiments(
         }
     )
 
-    if kwargs.get("only_changed", False) or html:
-        td.drop_duplicates("cols")
+    if kwargs.get("only_changed", False) or pcp:
+        td.drop_duplicates("cols", ignore_empty=False)
 
-    html_args = {}
-    if html:
-        td.dropna("rows", how="all")
+    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"
-        html_args["output_path"] = (Path.cwd() / out).resolve()
-        html_args["color_by"] = kwargs.get("sort_by") or "Experiment"
+        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"))
 
-    td.render(
-        pager=pager,
-        borders="horizontals",
-        rich_table=True,
-        header_styles=styles,
-        row_styles=row_styles,
-        csv=csv,
-        markdown=markdown,
-        html=html,
-        **html_args,
-    )
-
-    if html and kwargs.get("open"):
-        return ui.open_browser(Path(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):
@@ -517,14 +467,6 @@ def _format_json(item):
     return encode_exception(item)
 
 
-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 CmdExperimentsShow(CmdBase):
     def run(self):
         try:
@@ -532,8 +474,9 @@ class CmdExperimentsShow(CmdBase):
                 all_branches=self.args.all_branches,
                 all_tags=self.args.all_tags,
                 all_commits=self.args.all_commits,
-                sha_only=self.args.sha,
+                revs=self.args.rev,
                 num=self.args.num,
+                sha_only=self.args.sha,
                 param_deps=self.args.param_deps,
             )
         except DvcException:
@@ -553,11 +496,8 @@ class CmdExperimentsShow(CmdBase):
 
             show_experiments(
                 all_experiments,
-                include_metrics=self.args.include_metrics,
-                exclude_metrics=self.args.exclude_metrics,
-                include_params=self.args.include_params,
-                exclude_params=self.args.exclude_params,
-                no_timestamp=self.args.no_timestamp,
+                keep=self.args.keep,
+                drop=self.args.drop,
                 sort_by=self.args.sort_by,
                 sort_order=self.args.sort_order,
                 precision=precision,
@@ -567,7 +507,7 @@ class CmdExperimentsShow(CmdBase):
                 csv=self.args.csv,
                 markdown=self.args.markdown,
                 only_changed=self.args.only_changed,
-                html=self.args.html,
+                pcp=self.args.pcp,
                 out=self.args.out,
                 open=self.args.open,
             )
@@ -604,6 +544,16 @@ def add_parser(experiments_subparsers, parent_parser):
         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",
@@ -611,7 +561,7 @@ def add_parser(experiments_subparsers, parent_parser):
         default=1,
         dest="num",
         metavar="<num>",
-        help="Show the last `num` commits from HEAD.",
+        help="Show the last `num` commits from <rev>.",
     )
     experiments_show_parser.add_argument(
         "--no-pager",
@@ -620,32 +570,23 @@ def add_parser(experiments_subparsers, parent_parser):
         help="Do not pipe output into a pager.",
     )
     experiments_show_parser.add_argument(
-        "--include-metrics",
-        action="append",
-        default=[],
-        help="Include the specified metrics in output table.",
-        metavar="<metrics_list>",
+        "--only-changed",
+        action="store_true",
+        default=False,
+        help=(
+            "Only show metrics/params with values varying "
+            "across the selected experiments."
+        ),
     )
     experiments_show_parser.add_argument(
-        "--exclude-metrics",
-        action="append",
-        default=[],
-        help="Exclude the specified metrics from output table.",
-        metavar="<metrics_list>",
+        "--drop",
+        help="Remove the columns matching the specified regex pattern.",
+        metavar="<regex_pattern>",
     )
     experiments_show_parser.add_argument(
-        "--include-params",
-        action="append",
-        default=[],
-        help="Include the specified params in output table.",
-        metavar="<params_list>",
-    )
-    experiments_show_parser.add_argument(
-        "--exclude-params",
-        action="append",
-        default=[],
-        help="Exclude the specified params from output table.",
-        metavar="<params_list>",
+        "--keep",
+        help="Preserve the columns matching the specified regex pattern.",
+        metavar="<regex_pattern>",
     )
     experiments_show_parser.add_argument(
         "--param-deps",
@@ -667,12 +608,6 @@ def add_parser(experiments_subparsers, parent_parser):
         choices=("asc", "desc"),
         default="asc",
     )
-    experiments_show_parser.add_argument(
-        "--no-timestamp",
-        action="store_true",
-        default=False,
-        help="Do not show experiment timestamps.",
-    )
     experiments_show_parser.add_argument(
         "--sha",
         action="store_true",
@@ -711,31 +646,24 @@ def add_parser(experiments_subparsers, parent_parser):
         metavar="<n>",
     )
     experiments_show_parser.add_argument(
-        "--only-changed",
+        "--parallel-coordinates-plot",
+        "--pcp",
+        dest="pcp",
         action="store_true",
         default=False,
-        help=(
-            "Only show metrics/params with values varying "
-            "across the selected experiments."
-        ),
-    )
-    experiments_show_parser.add_argument(
-        "--html",
-        action="store_true",
-        default=False,
-        help="Generate a parallel coordinates plot from the tabulated output.",
+        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 HTML to",
+        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 HTML directly in the browser.",
+        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,9 +1,8 @@
 import argparse
-from pathlib import Path
 
-from dvc.command import completion
-from dvc.command.base import CmdBase, fix_subparsers
-from dvc.render.utils import match_renderers
+from dvc.cli.command import CmdBase
+from dvc.cli.utils import fix_subparsers
+from dvc.commands import completion
 from dvc.ui import ui
 
 
@@ -11,11 +10,14 @@ 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:
+            from pathlib import Path
+
             html_path = Path.cwd() / (self.args.target + "_html")
-            from dvc.render.utils import render
 
             renderers = match_renderers(plots, self.repo.plots.templates)
             index_path = render(self.repo, renderers, metrics, html_path)

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

@@ -4,11 +4,10 @@ import logging
 
 from funcy import first
 
-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.render.utils import match_renderers, render
-from dvc.render.vega import VegaRenderer
 from dvc.ui import ui
 from dvc.utils import format_link
 
@@ -16,7 +15,7 @@ logger = logging.getLogger(__name__)
 
 
 def _show_json(renderers, path: None):
-    if any(lambda r: r.needs_output_path for r in renderers) and not path:
+    if any(r.needs_output_path for r in renderers) and not path:
         raise DvcException("Output path ('-o') is required!")
 
     result = {
@@ -41,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`")
@@ -94,8 +96,14 @@ class CmdPlots(CmdBase):
             )
 
             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
@@ -132,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 "
@@ -168,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 = (
@@ -203,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 = (
@@ -229,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(
@@ -268,14 +330,17 @@ 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",

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,7 +166,9 @@ class TabularData(MutableSequence[Sequence["CellT"]]):
             writer.writerow(row)
         return buff.getvalue()
 
-    def to_parallel_coordinates(self, output_path, color_by):
+    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
 
@@ -185,14 +200,6 @@ class TabularData(MutableSequence[Sequence["CellT"]]):
 
         if kwargs.pop("csv", False):
             ui.write(self.to_csv(), end="")
-
-        elif kwargs.pop("html", False):
-            ui.write(
-                self.to_parallel_coordinates(
-                    kwargs["output_path"], kwargs.get("color_by")
-                )
-            )
-
         else:
             ui.table(self, headers=self.keys(), **kwargs)
 
@@ -204,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}."
@@ -216,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
@@ -249,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}."
@@ -259,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)
@@ -269,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)
 
 

dvc/data/db/index.py

@@ -3,7 +3,8 @@ import os
 from abc import ABC, abstractmethod
 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
@@ -88,7 +89,7 @@ class ObjectDBIndex(ObjectDBIndexBase):
         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
@@ -96,18 +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)
+        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)
 
+    @with_diskcache(name="index")
     def clear(self):
         """Clear this index (to force re-indexing later)."""
         from diskcache import Timeout
@@ -117,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
@@ -131,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/base.py

@@ -10,7 +10,7 @@ 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"
@@ -158,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

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

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)
 

dvc/fs/http.py

@@ -4,9 +4,9 @@ from typing import Any
 from funcy import cached_property, memoize, wrap_with
 
 from dvc import prompt
-from dvc.progress import DEFAULT_CALLBACK
 from dvc.scheme import Schemes
 
+from ._callback import DEFAULT_CALLBACK
 from .fsspec_wrapper import AnyFSPath, FSSpecWrapper, NoDirectoriesMixin
 
 

dvc/fs/local.py

@@ -3,10 +3,10 @@ import os
 
 from dvc.scheme import Schemes
 from dvc.system import System
-from dvc.utils import is_exec, tmp_fname
+from dvc.utils import tmp_fname
 from dvc.utils.fs import copy_fobj_to_file, copyfile, makedirs, move, remove
 
-from ..progress import DEFAULT_CALLBACK
+from ._callback import DEFAULT_CALLBACK
 from .base import FileSystem
 
 logger = logging.getLogger(__name__)
@@ -83,10 +83,6 @@ class LocalFileSystem(FileSystem):
     def makedirs(self, path, **kwargs):
         makedirs(path, exist_ok=kwargs.pop("exist_ok", True))
 
-    def isexec(self, path):
-        mode = self.info(path)["mode"]
-        return is_exec(mode)
-
     def move(self, from_info, to_info):
         self.makedirs(self.path.parent(to_info))
         move(from_info, to_info)

dvc/fs/path.py

@@ -15,7 +15,7 @@ class Path:
         return self.flavour.join(*parts)
 
     def parts(self, path):
-        drive, path = self.flavour.splitdrive(path)
+        drive, path = self.flavour.splitdrive(path.rstrip(self.flavour.sep))
 
         ret = []
         while True:

dvc/fs/repo.py

@@ -1,14 +1,12 @@
-import errno
 import logging
 import os
 import threading
-from contextlib import suppress
 from itertools import takewhile
 from typing import TYPE_CHECKING, Callable, Optional, Tuple, Type, Union
 
 from funcy import lfilter, wrap_with
 
-from ..progress import DEFAULT_CALLBACK
+from ._callback import DEFAULT_CALLBACK
 from .base import FileSystem
 from .dvc import DvcFileSystem
 
@@ -20,6 +18,11 @@ logger = logging.getLogger(__name__)
 RepoFactory = Union[Callable[[str], "Repo"], Type["Repo"]]
 
 
+def _wrap_walk(dvc_fs, *args, **kwargs):
+    for root, dnames, fnames in dvc_fs.walk(*args, **kwargs):
+        yield dvc_fs.path.join(dvc_fs.repo.root_dir, root), dnames, fnames
+
+
 class RepoFileSystem(FileSystem):  # pylint:disable=abstract-method
     """DVC + git-tracked files fs.
 
@@ -66,7 +69,7 @@ class RepoFileSystem(FileSystem):  # pylint:disable=abstract-method
 
         self._main_repo = repo
         self.hash_jobs = repo.fs.hash_jobs
-        self.root_dir = repo.root_dir
+        self.root_dir: str = repo.root_dir
         self._traverse_subrepos = subrepos
 
         self._subrepos_trie = PathStringTrie()
@@ -177,8 +180,7 @@ class RepoFileSystem(FileSystem):  # pylint:disable=abstract-method
             if self._is_dvc_repo(d):
                 repo = self.repo_factory(
                     d,
-                    scm=self._main_repo.scm,
-                    rev=self._main_repo.get_rev(),
+                    fs=self._main_repo.fs,
                     repo_factory=self.repo_factory,
                 )
                 self._dvcfss[repo.root_dir] = DvcFileSystem(repo=repo)
@@ -194,7 +196,9 @@ class RepoFileSystem(FileSystem):  # pylint:disable=abstract-method
         repo_path = os.path.join(dir_path, Repo.DVC_DIR)
         return self._main_repo.fs.isdir(repo_path)
 
-    def _get_fs_pair(self, path) -> Tuple[FileSystem, Optional[DvcFileSystem]]:
+    def _get_fs_pair(
+        self, path
+    ) -> Tuple[FileSystem, Optional[DvcFileSystem], str]:
         """
         Returns a pair of fss based on repo the path falls in, using prefix.
         """
@@ -205,7 +209,13 @@ class RepoFileSystem(FileSystem):  # pylint:disable=abstract-method
         repo = self._get_repo(path) or self._main_repo
 
         dvc_fs = self._dvcfss.get(repo.root_dir)
-        return repo.fs, dvc_fs
+
+        if path.startswith(repo.root_dir):
+            dvc_path = path[len(repo.root_dir) + 1 :]
+        else:
+            dvc_path = path
+
+        return repo.fs, dvc_fs, dvc_path
 
     def open(
         self, path, mode="r", encoding="utf-8", **kwargs
@@ -213,17 +223,19 @@ class RepoFileSystem(FileSystem):  # pylint:disable=abstract-method
         if "b" in mode:
             encoding = None
 
-        fs, dvc_fs = self._get_fs_pair(path)
+        fs, dvc_fs, dvc_path = self._get_fs_pair(path)
         try:
             return fs.open(path, mode=mode, encoding=encoding)
         except FileNotFoundError:
             if not dvc_fs:
                 raise
 
-        return dvc_fs.open(path, mode=mode, encoding=encoding, **kwargs)
+        return dvc_fs.open(dvc_path, mode=mode, encoding=encoding, **kwargs)
 
     def exists(self, path) -> bool:
-        fs, dvc_fs = self._get_fs_pair(path)
+        path = os.path.abspath(path)
+
+        fs, dvc_fs, dvc_path = self._get_fs_pair(path)
 
         if not dvc_fs:
             return fs.exists(path)
@@ -234,19 +246,22 @@ class RepoFileSystem(FileSystem):  # pylint:disable=abstract-method
         if fs.exists(path):
             return True
 
-        try:
-            meta = dvc_fs.metadata(path)
-        except FileNotFoundError:
+        if not dvc_fs.exists(dvc_path):
             return False
 
-        for out in meta.outs:
-            if fs.exists(out.fs_path):
-                return False
+        for p in self.path.parents(path):
+            try:
+                if fs.info(p)["type"] != "directory":
+                    return False
+            except FileNotFoundError:
+                continue
 
         return True
 
     def isdir(self, path):  # pylint: disable=arguments-renamed
-        fs, dvc_fs = self._get_fs_pair(path)
+        path = os.path.abspath(path)
+
+        fs, dvc_fs, dvc_path = self._get_fs_pair(path)
 
         if dvc_fs and dvc_fs.repo.dvcignore.is_ignored_dir(path):
             return False
@@ -262,22 +277,27 @@ class RepoFileSystem(FileSystem):  # pylint:disable=abstract-method
             return False
 
         try:
-            meta = dvc_fs.metadata(path)
+            info = dvc_fs.info(dvc_path)
         except FileNotFoundError:
             return False
 
-        for out in meta.outs:
-            if fs.exists(out.fs_path):
-                return False
+        for p in self.path.parents(path):
+            try:
+                if fs.info(p)["type"] != "directory":
+                    return False
+            except FileNotFoundError:
+                continue
 
-        return meta.isdir
+        return info["type"] == "directory"
 
     def isdvc(self, path, **kwargs):
-        _, dvc_fs = self._get_fs_pair(path)
-        return dvc_fs is not None and dvc_fs.isdvc(path, **kwargs)
+        _, dvc_fs, dvc_path = self._get_fs_pair(path)
+        return dvc_fs is not None and dvc_fs.isdvc(dvc_path, **kwargs)
 
     def isfile(self, path):  # pylint: disable=arguments-renamed
-        fs, dvc_fs = self._get_fs_pair(path)
+        path = os.path.abspath(path)
+
+        fs, dvc_fs, dvc_path = self._get_fs_pair(path)
 
         if dvc_fs and dvc_fs.repo.dvcignore.is_ignored_file(path):
             return False
@@ -293,21 +313,18 @@ class RepoFileSystem(FileSystem):  # pylint:disable=abstract-method
             return False
 
         try:
-            meta = dvc_fs.metadata(path)
+            info = dvc_fs.info(dvc_path)
         except FileNotFoundError:
             return False
 
-        (out,) = meta.outs
-        assert len(meta.outs) == 1
-        if fs.exists(out.fs_path):
-            return False
-        return meta.isfile
+        for p in self.path.parents(path):
+            try:
+                if fs.info(p)["type"] != "directory":
+                    return False
+            except FileNotFoundError:
+                continue
 
-    def isexec(self, path):
-        fs, dvc_fs = self._get_fs_pair(path)
-        if dvc_fs and dvc_fs.exists(path):
-            return dvc_fs.isexec(path)
-        return fs.isexec(path)
+        return info["type"] == "file"
 
     def _dvc_walk(self, walk):
         try:
@@ -324,10 +341,10 @@ class RepoFileSystem(FileSystem):  # pylint:disable=abstract-method
         NOTE: subrepo will only be discovered when walking if
         ignore_subrepos is set to False.
         """
-        fs, dvc_fs = self._get_fs_pair(dir_path)
+        fs, dvc_fs, dvc_path = self._get_fs_pair(dir_path)
         fs_walk = fs.walk(dir_path, topdown=True)
         if dvc_fs:
-            dvc_walk = dvc_fs.walk(dir_path, topdown=True, **kwargs)
+            dvc_walk = _wrap_walk(dvc_fs, dvc_path, topdown=True, **kwargs)
         else:
             dvc_walk = None
         yield from self._walk(fs_walk, dvc_walk, **kwargs)
@@ -416,26 +433,26 @@ class RepoFileSystem(FileSystem):  # pylint:disable=abstract-method
         repo = self._get_repo(os.path.abspath(top))
         dvcfiles = kwargs.pop("dvcfiles", False)
 
-        fs, dvc_fs = self._get_fs_pair(top)
+        fs, dvc_fs, dvc_path = self._get_fs_pair(top)
         repo_exists = fs.exists(top)
 
         repo_walk = repo.dvcignore.walk(
             fs, top, topdown=topdown, onerror=onerror, **kwargs
         )
 
-        if not dvc_fs or (repo_exists and dvc_fs.isdvc(top)):
+        if not dvc_fs or (repo_exists and dvc_fs.isdvc(dvc_path)):
             yield from self._walk(repo_walk, None, dvcfiles=dvcfiles)
             return
 
         if not repo_exists:
-            yield from dvc_fs.walk(
-                top, topdown=topdown, onerror=onerror, **kwargs
+            yield from _wrap_walk(
+                dvc_fs, dvc_path, topdown=topdown, onerror=onerror, **kwargs
             )
 
         dvc_walk = None
-        if dvc_fs.exists(top):
-            dvc_walk = dvc_fs.walk(
-                top, topdown=topdown, onerror=onerror, **kwargs
+        if dvc_fs.exists(dvc_path):
+            dvc_walk = _wrap_walk(
+                dvc_fs, dvc_path, topdown=topdown, onerror=onerror, **kwargs
             )
 
         yield from self._walk(repo_walk, dvc_walk, dvcfiles=dvcfiles)
@@ -448,7 +465,7 @@ class RepoFileSystem(FileSystem):  # pylint:disable=abstract-method
     def get_file(
         self, from_info, to_file, callback=DEFAULT_CALLBACK, **kwargs
     ):
-        fs, dvc_fs = self._get_fs_pair(from_info)
+        fs, dvc_fs, dvc_path = self._get_fs_pair(from_info)
         try:
             fs.get_file(  # pylint: disable=protected-access
                 from_info, to_file, callback=callback, **kwargs
@@ -459,55 +476,49 @@ class RepoFileSystem(FileSystem):  # pylint:disable=abstract-method
                 raise
 
         dvc_fs.get_file(  # pylint: disable=protected-access
-            from_info, to_file, callback=callback, **kwargs
+            dvc_path, to_file, callback=callback, **kwargs
         )
 
-    def metadata(self, path):
-        fs_path = os.path.abspath(path)
-        fs, dvc_fs = self._get_fs_pair(fs_path)
-
-        dvc_meta = None
-        if dvc_fs:
-            with suppress(FileNotFoundError):
-                dvc_meta = dvc_fs.metadata(fs_path)
-
-        info_result = None
-        with suppress(FileNotFoundError):
-            info_result = fs.info(fs_path)
-
-        if not info_result and not dvc_meta:
-            raise FileNotFoundError(
-                errno.ENOENT, os.strerror(errno.ENOENT), fs_path
-            )
-
-        from ._metadata import Metadata
-
-        meta = dvc_meta or Metadata(
-            fs_path=fs_path,
-            repo=self._get_repo(fs_path) or self._main_repo,
-        )
-
-        isdir = bool(info_result) and info_result["type"] == "directory"
-        meta.isdir = meta.isdir or isdir
-
-        if not dvc_meta:
-            from dvc.utils import is_exec
-
-            meta.is_exec = bool(info_result) and is_exec(info_result["mode"])
-        return meta
-
     def info(self, path):
-        fs, dvc_fs = self._get_fs_pair(path)
+        fs, dvc_fs, dvc_path = self._get_fs_pair(path)
 
         try:
-            return fs.info(path)
+            dvc_info = dvc_fs.info(dvc_path)
         except FileNotFoundError:
-            return dvc_fs.info(path)
+            dvc_info = None
+
+        try:
+            from dvc.utils import is_exec
+
+            fs_info = fs.info(path)
+            fs_info["repo"] = dvc_fs.repo
+            fs_info["isout"] = (
+                dvc_info.get("isout", False) if dvc_info else False
+            )
+            fs_info["outs"] = dvc_info["outs"] if dvc_info else None
+            fs_info["isdvc"] = dvc_info["isdvc"] if dvc_info else False
+            fs_info["meta"] = dvc_info.get("meta") if dvc_info else None
+
+            isexec = False
+            if dvc_info:
+                isexec = dvc_info["isexec"]
+            elif fs_info["type"] == "file":
+                isexec = is_exec(fs_info["mode"])
+            fs_info["isexec"] = isexec
+            return fs_info
+
+        except FileNotFoundError:
+            if not dvc_info:
+                raise
+
+            dvc_info["repo"] = dvc_fs.repo
+            dvc_info["isdvc"] = True
+            return dvc_info
 
     def checksum(self, path):
-        fs, dvc_fs = self._get_fs_pair(path)
+        fs, dvc_fs, dvc_path = self._get_fs_pair(path)
 
         try:
             return fs.checksum(path)
         except FileNotFoundError:
-            return dvc_fs.checksum(path)
+            return dvc_fs.checksum(dvc_path)

dvc/fs/s3.py

@@ -5,9 +5,9 @@ from collections import defaultdict
 
 from funcy import cached_property, wrap_prop
 
-from dvc.progress import DEFAULT_CALLBACK
 from dvc.scheme import Schemes
 
+from ._callback import DEFAULT_CALLBACK
 from .fsspec_wrapper import ObjectFSWrapper
 
 _AWS_CONFIG_PATH = os.path.join(os.path.expanduser("~"), ".aws", "config")

dvc/fs/ssh.py

@@ -8,7 +8,7 @@ from dvc import prompt
 from dvc.scheme import Schemes
 from dvc.utils.fs import as_atomic
 
-from ..progress import DEFAULT_CALLBACK
+from ._callback import DEFAULT_CALLBACK
 from .fsspec_wrapper import FSSpecWrapper
 
 _SSH_TIMEOUT = 60 * 30

dvc/fs/utils.py

@@ -153,6 +153,7 @@ def test_links(
     from_fs.makedirs(from_fs.path.parent(from_file))
     with from_fs.open(from_file, "wb") as fobj:
         fobj.write(b"test")
+    to_fs.makedirs(to_fs.path.parent(to_file))
 
     ret = []
     try: