dvc: bump to 1.0.0a9

This one doesn't allow us to release new pre-releases, because pypi
forbids that:

https://travis-ci.com/github/iterative/dvc/jobs/343343967#L2447

Let's install it separately for now, until there is a new official
mock-ssh-server release.

dvc/command/add.py

@@ -18,6 +18,7 @@ class CmdAdd(CmdBase):
                 recursive=self.args.recursive,
                 no_commit=self.args.no_commit,
                 fname=self.args.file,
+                external=self.args.external,
             )
 
         except DvcException:
@@ -49,6 +50,12 @@ def add_parser(subparsers, parent_parser):
         default=False,
         help="Don't put files/directories into cache.",
     )
+    parser.add_argument(
+        "--external",
+        action="store_true",
+        default=False,
+        help="Allow targets that are outside of the DVC repository.",
+    )
     parser.add_argument(
         "-f",
         "--file",

dvc/command/dag.py

@@ -109,7 +109,7 @@ def add_parser(subparsers, parent_parser):
     dag_parser.add_argument(
         "target",
         nargs="?",
-        help="Stage or output to show pipeline for. Optional. "
-        "(Finds all stages in the workspace by default.)",
+        help="Stage or output to show pipeline for (optional). "
+        "Finds all stages in the workspace by default.",
     )
     dag_parser.set_defaults(func=CmdDAG)

dvc/command/plots.py

@@ -197,8 +197,8 @@ def _add_props_arguments(parser):
         help="Provided CSV ot TSV datafile does not have a header.",
     )
     parser.add_argument("--title", default=None, help="Plot title.")
-    parser.add_argument("--xlab", default=None, help="X axis title.")
-    parser.add_argument("--ylab", default=None, help="Y axis title.")
+    parser.add_argument("--x-label", default=None, help="X axis label.")
+    parser.add_argument("--y-label", default=None, help="Y axis lebel.")
 
 
 def _add_output_arguments(parser):
@@ -209,5 +209,5 @@ def _add_output_arguments(parser):
         "--show-vega",
         action="store_true",
         default=False,
-        help="Show output in VEGA format.",
+        help="Show output in Vega format.",
     )

dvc/command/remote.py

@@ -62,7 +62,11 @@ class CmdRemoteModify(CmdRemote):
     def run(self):
         with self.config.edit(self.args.level) as conf:
             self._check_exists(conf)
-            conf["remote"][self.args.name][self.args.option] = self.args.value
+            section = conf["remote"][self.args.name]
+            if self.args.unset:
+                section.pop(self.args.option, None)
+            else:
+                section[self.args.option] = self.args.value
         return 0
 
 

dvc/command/run.py

@@ -53,6 +53,7 @@ class CmdRun(CmdBase):
                 always_changed=self.args.always_changed,
                 name=self.args.name,
                 single_stage=self.args.single_stage,
+                external=self.args.external,
             )
         except DvcException:
             logger.exception("")
@@ -225,6 +226,12 @@ def add_parser(subparsers, parent_parser):
         default=False,
         help=argparse.SUPPRESS,
     )
+    run_parser.add_argument(
+        "--external",
+        action="store_true",
+        default=False,
+        help="Allow outputs that are outside of the DVC repository.",
+    )
     run_parser.add_argument(
         "command", nargs=argparse.REMAINDER, help="Command to execute."
     )

dvc/data_cloud.py

@@ -93,7 +93,7 @@ class DataCloud:
     def _save_pulled_checksums(self, cache):
         for checksum in cache.scheme_keys("local"):
             cache_file = self.repo.cache.local.checksum_to_path_info(checksum)
-            if self.repo.cache.local.exists(cache_file):
+            if self.repo.cache.local.tree.exists(cache_file):
                 # We can safely save here, as existing corrupted files will
                 # be removed upon status, while files corrupted during
                 # download will not be moved from tmp_file

dvc/exceptions.py

@@ -178,6 +178,14 @@ class NoMetricsError(DvcException):
         )
 
 
+class NoPlotsError(DvcException):
+    def __init__(self):
+        super().__init__(
+            "no plots in this repository. Use `--plots/--plots-no-cache` "
+            "options for `dvc run` to mark stage outputs as plots."
+        )
+
+
 class YAMLFileCorruptedError(DvcException):
     def __init__(self, path):
         path = relpath(path)

dvc/output/base.py

@@ -58,8 +58,8 @@ class BaseOutput:
     PARAM_PLOT_TEMPLATE = "template"
     PARAM_PLOT_X = "x"
     PARAM_PLOT_Y = "y"
-    PARAM_PLOT_XLAB = "xlab"
-    PARAM_PLOT_YLAB = "ylab"
+    PARAM_PLOT_X_LABEL = "x_label"
+    PARAM_PLOT_Y_LABEL = "y_label"
     PARAM_PLOT_TITLE = "title"
     PARAM_PLOT_CSV_HEADER = "csv_header"
     PARAM_PERSIST = "persist"
@@ -181,7 +181,7 @@ class BaseOutput:
 
     @property
     def exists(self):
-        return self.remote.exists(self.path_info)
+        return self.remote.tree.exists(self.path_info)
 
     def save_info(self):
         return self.remote.save_info(self.path_info)
@@ -217,13 +217,13 @@ class BaseOutput:
 
     @property
     def is_empty(self):
-        return self.remote.is_empty(self.path_info)
+        return self.remote.tree.is_empty(self.path_info)
 
     def isdir(self):
-        return self.remote.isdir(self.path_info)
+        return self.remote.tree.isdir(self.path_info)
 
     def isfile(self):
-        return self.remote.isfile(self.path_info)
+        return self.remote.tree.isfile(self.path_info)
 
     def ignore(self):
         if not self.use_scm_ignore:
@@ -326,7 +326,7 @@ class BaseOutput:
         )
 
     def remove(self, ignore_remove=False):
-        self.remote.remove(self.path_info)
+        self.remote.tree.remove(self.path_info)
         if self.scheme != "local":
             return
 
@@ -337,7 +337,7 @@ class BaseOutput:
         if self.scheme == "local" and self.use_scm_ignore:
             self.repo.scm.ignore_remove(self.fspath)
 
-        self.remote.move(self.path_info, out.path_info)
+        self.remote.tree.move(self.path_info, out.path_info)
         self.def_path = out.def_path
         self.path_info = out.path_info
         self.save()

dvc/remote/azure.py

@@ -8,12 +8,45 @@ from funcy import cached_property, wrap_prop
 
 from dvc.path_info import CloudURLInfo
 from dvc.progress import Tqdm
-from dvc.remote.base import BaseRemote
+from dvc.remote.base import BaseRemote, BaseRemoteTree
 from dvc.scheme import Schemes
 
 logger = logging.getLogger(__name__)
 
 
+class AzureRemoteTree(BaseRemoteTree):
+    @property
+    def blob_service(self):
+        return self.remote.blob_service
+
+    def _generate_download_url(self, path_info, expires=3600):
+        from azure.storage.blob import BlobPermissions
+
+        expires_at = datetime.utcnow() + timedelta(seconds=expires)
+
+        sas_token = self.blob_service.generate_blob_shared_access_signature(
+            path_info.bucket,
+            path_info.path,
+            permission=BlobPermissions.READ,
+            expiry=expires_at,
+        )
+        download_url = self.blob_service.make_blob_url(
+            path_info.bucket, path_info.path, sas_token=sas_token
+        )
+        return download_url
+
+    def exists(self, path_info):
+        paths = self.remote.list_paths(path_info.bucket, path_info.path)
+        return any(path_info.path == path for path in paths)
+
+    def remove(self, path_info):
+        if path_info.scheme != self.scheme:
+            raise NotImplementedError
+
+        logger.debug(f"Removing {path_info}")
+        self.blob_service.delete_blob(path_info.bucket, path_info.path)
+
+
 class AzureRemote(BaseRemote):
     scheme = Schemes.AZURE
     path_cls = CloudURLInfo
@@ -21,6 +54,7 @@ class AzureRemote(BaseRemote):
     PARAM_CHECKSUM = "etag"
     COPY_POLL_SECONDS = 5
     LIST_OBJECT_PAGE_SIZE = 5000
+    TREE_CLS = AzureRemoteTree
 
     def __init__(self, repo, config):
         super().__init__(repo, config)
@@ -65,14 +99,7 @@ class AzureRemote(BaseRemote):
     def get_file_checksum(self, path_info):
         return self.get_etag(path_info)
 
-    def remove(self, path_info):
-        if path_info.scheme != self.scheme:
-            raise NotImplementedError
-
-        logger.debug(f"Removing {path_info}")
-        self.blob_service.delete_blob(path_info.bucket, path_info.path)
-
-    def _list_paths(self, bucket, prefix, progress_callback=None):
+    def list_paths(self, bucket, prefix, progress_callback=None):
         blob_service = self.blob_service
         next_marker = None
         while True:
@@ -97,7 +124,7 @@ class AzureRemote(BaseRemote):
             )
         else:
             prefix = self.path_info.path
-        return self._list_paths(
+        return self.list_paths(
             self.path_info.bucket, prefix, progress_callback
         )
 
@@ -122,23 +149,3 @@ class AzureRemote(BaseRemote):
                 to_file,
                 progress_callback=pbar.update_to,
             )
-
-    def exists(self, path_info):
-        paths = self._list_paths(path_info.bucket, path_info.path)
-        return any(path_info.path == path for path in paths)
-
-    def _generate_download_url(self, path_info, expires=3600):
-        from azure.storage.blob import BlobPermissions
-
-        expires_at = datetime.utcnow() + timedelta(seconds=expires)
-
-        sas_token = self.blob_service.generate_blob_shared_access_signature(
-            path_info.bucket,
-            path_info.path,
-            permission=BlobPermissions.READ,
-            expiry=expires_at,
-        )
-        download_url = self.blob_service.make_blob_url(
-            path_info.bucket, path_info.path, sas_token=sas_token
-        )
-        return download_url

dvc/remote/base.py

@@ -83,12 +83,99 @@ def index_locked(f):
     return wrapper
 
 
+class BaseRemoteTree:
+    SHARED_MODE_MAP = {None: (None, None), "group": (None, None)}
+
+    def __init__(self, remote, config):
+        self.remote = remote
+        shared = config.get("shared")
+        self._file_mode, self._dir_mode = self.SHARED_MODE_MAP[shared]
+
+    @property
+    def file_mode(self):
+        return self._file_mode
+
+    @property
+    def dir_mode(self):
+        return self._dir_mode
+
+    @property
+    def scheme(self):
+        return self.remote.scheme
+
+    @property
+    def path_cls(self):
+        return self.remote.path_cls
+
+    def open(self, path_info, mode="r", encoding=None):
+        if hasattr(self, "_generate_download_url"):
+            get_url = partial(self._generate_download_url, path_info)
+            return open_url(get_url, mode=mode, encoding=encoding)
+
+        raise RemoteActionNotImplemented("open", self.scheme)
+
+    def exists(self, path_info):
+        raise NotImplementedError
+
+    def isdir(self, path_info):
+        """Optional: Overwrite only if the remote has a way to distinguish
+        between a directory and a file.
+        """
+        return False
+
+    def isfile(self, path_info):
+        """Optional: Overwrite only if the remote has a way to distinguish
+        between a directory and a file.
+        """
+        return True
+
+    def iscopy(self, path_info):
+        """Check if this file is an independent copy."""
+        return False  # We can't be sure by default
+
+    def walk_files(self, path_info):
+        """Return a generator with `PathInfo`s to all the files"""
+        raise NotImplementedError
+
+    def is_empty(self, path_info):
+        return False
+
+    def remove(self, path_info):
+        raise RemoteActionNotImplemented("remove", self.scheme)
+
+    def makedirs(self, path_info):
+        """Optional: Implement only if the remote needs to create
+        directories before copying/linking/moving data
+        """
+
+    def move(self, from_info, to_info, mode=None):
+        assert mode is None
+        self.copy(from_info, to_info)
+        self.remove(from_info)
+
+    def copy(self, from_info, to_info):
+        raise RemoteActionNotImplemented("copy", self.scheme)
+
+    def copy_fobj(self, fobj, to_info):
+        raise RemoteActionNotImplemented("copy_fobj", self.scheme)
+
+    def symlink(self, from_info, to_info):
+        raise RemoteActionNotImplemented("symlink", self.scheme)
+
+    def hardlink(self, from_info, to_info):
+        raise RemoteActionNotImplemented("hardlink", self.scheme)
+
+    def reflink(self, from_info, to_info):
+        raise RemoteActionNotImplemented("reflink", self.scheme)
+
+
 class BaseRemote:
     scheme = "base"
     path_cls = URLInfo
     REQUIRES = {}
     JOBS = 4 * cpu_count()
     INDEX_CLS = RemoteIndex
+    TREE_CLS = BaseRemoteTree
 
     PARAM_RELPATH = "relpath"
     CHECKSUM_DIR_SUFFIX = ".dir"
@@ -102,7 +189,6 @@ class BaseRemote:
     CAN_TRAVERSE = True
 
     CACHE_MODE = None
-    SHARED_MODE_MAP = {None: (None, None), "group": (None, None)}
 
     state = StateNoop()
 
@@ -111,9 +197,6 @@ class BaseRemote:
 
         self._check_requires(config)
 
-        shared = config.get("shared")
-        self._file_mode, self._dir_mode = self.SHARED_MODE_MAP[shared]
-
         self.checksum_jobs = (
             config.get("checksum_jobs")
             or (self.repo and self.repo.config["core"].get("checksum_jobs"))
@@ -134,6 +217,8 @@ class BaseRemote:
         else:
             self.index = RemoteIndexNoop()
 
+        self.tree = self.TREE_CLS(self, config)
+
     @classmethod
     def get_missing_deps(cls):
         import importlib
@@ -218,7 +303,7 @@ class BaseRemote:
         if tree:
             walk_files = tree.walk_files
         else:
-            walk_files = self.walk_files
+            walk_files = self.tree.walk_files
 
         for fname in walk_files(path_info, **kwargs):
             if DvcIgnore.DVCIGNORE_FILE == fname.name:
@@ -273,8 +358,8 @@ class BaseRemote:
         checksum, tmp_info = self._get_dir_info_checksum(dir_info)
         new_info = self.cache.checksum_to_path_info(checksum)
         if self.cache.changed_cache_file(checksum):
-            self.cache.makedirs(new_info.parent)
-            self.cache.move(tmp_info, new_info, mode=self.CACHE_MODE)
+            self.cache.tree.makedirs(new_info.parent)
+            self.cache.tree.move(tmp_info, new_info, mode=self.CACHE_MODE)
 
         if path_info:
             self.state.save(path_info, checksum)
@@ -344,7 +429,7 @@ class BaseRemote:
     def get_checksum(self, path_info):
         assert isinstance(path_info, str) or path_info.scheme == self.scheme
 
-        if not self.exists(path_info):
+        if not self.tree.exists(path_info):
             return None
 
         checksum = self.state.get(path_info)
@@ -355,14 +440,16 @@ class BaseRemote:
         if (
             checksum
             and self.is_dir_checksum(checksum)
-            and not self.exists(self.cache.checksum_to_path_info(checksum))
+            and not self.tree.exists(
+                self.cache.checksum_to_path_info(checksum)
+            )
         ):
             checksum = None
 
         if checksum:
             return checksum
 
-        if self.isdir(path_info):
+        if self.tree.isdir(path_info):
             checksum = self.get_dir_checksum(path_info)
         else:
             checksum = self.get_file_checksum(path_info)
@@ -396,7 +483,7 @@ class BaseRemote:
             "checking if '%s'('%s') has changed.", path_info, checksum_info
         )
 
-        if not self.exists(path_info):
+        if not self.tree.exists(path_info):
             logger.debug("'%s' doesn't exist.", path_info)
             return True
 
@@ -428,9 +515,9 @@ class BaseRemote:
         self._link(from_info, to_info, self.cache_types)
 
     def _link(self, from_info, to_info, link_types):
-        assert self.isfile(from_info)
+        assert self.tree.isfile(from_info)
 
-        self.makedirs(to_info.parent)
+        self.tree.makedirs(to_info.parent)
 
         self._try_links(from_info, to_info, link_types)
 
@@ -438,9 +525,9 @@ class BaseRemote:
         if self.cache_type_confirmed:
             return
 
-        is_link = getattr(self, f"is_{link_type}", None)
+        is_link = getattr(self.tree, f"is_{link_type}", None)
         if is_link and not is_link(path_info):
-            self.remove(path_info)
+            self.tree.remove(path_info)
             raise DvcException(f"failed to verify {link_type}")
 
         self.cache_type_confirmed = True
@@ -448,7 +535,7 @@ class BaseRemote:
     @slow_link_guard
     def _try_links(self, from_info, to_info, link_types):
         while link_types:
-            link_method = getattr(self, link_types[0])
+            link_method = getattr(self.tree, link_types[0])
             try:
                 self._do_link(from_info, to_info, link_method)
                 self._verify_link(to_info, link_types[0])
@@ -463,7 +550,7 @@ class BaseRemote:
         raise DvcException("no possible cache types left to try out.")
 
     def _do_link(self, from_info, to_info, link_method):
-        if self.exists(to_info):
+        if self.tree.exists(to_info):
             raise DvcException(f"Link '{to_info}' already exists!")
 
         link_method(from_info, to_info)
@@ -486,19 +573,21 @@ class BaseRemote:
                     if not (
                         tree.isdvc(path_info, strict=False) and tree.fetch
                     ):
-                        self.copy_fobj(fobj, cache_info)
+                        self.tree.copy_fobj(fobj, cache_info)
                 callback = kwargs.get("download_callback")
                 if callback:
                     callback(1)
         else:
             if self.changed_cache(checksum):
-                self.move(path_info, cache_info, mode=self.CACHE_MODE)
+                self.tree.move(path_info, cache_info, mode=self.CACHE_MODE)
                 self.link(cache_info, path_info)
-            elif self.iscopy(path_info) and self._cache_is_copy(path_info):
+            elif self.tree.iscopy(path_info) and self._cache_is_copy(
+                path_info
+            ):
                 # Default relink procedure involves unneeded copy
                 self.unprotect(path_info)
             else:
-                self.remove(path_info)
+                self.tree.remove(path_info)
                 self.link(cache_info, path_info)
 
             if save_link:
@@ -522,14 +611,14 @@ class BaseRemote:
 
         workspace_file = path_info.with_name("." + uuid())
         test_cache_file = self.path_info / ".cache_type_test_file"
-        if not self.exists(test_cache_file):
-            with self.open(test_cache_file, "wb") as fobj:
+        if not self.tree.exists(test_cache_file):
+            with self.tree.open(test_cache_file, "wb") as fobj:
                 fobj.write(bytes(1))
         try:
             self.link(test_cache_file, workspace_file)
         finally:
-            self.remove(workspace_file)
-            self.remove(test_cache_file)
+            self.tree.remove(workspace_file)
+            self.tree.remove(test_cache_file)
 
         self.cache_type_confirmed = True
         return self.cache_types[0] == "copy"
@@ -561,29 +650,6 @@ class BaseRemote:
             self.state.save(path_info, checksum)
         return {self.PARAM_CHECKSUM: checksum}
 
-    def is_empty(self, path_info):
-        return False
-
-    def isfile(self, path_info):
-        """Optional: Overwrite only if the remote has a way to distinguish
-        between a directory and a file.
-        """
-        return True
-
-    def isdir(self, path_info):
-        """Optional: Overwrite only if the remote has a way to distinguish
-        between a directory and a file.
-        """
-        return False
-
-    def iscopy(self, path_info):
-        """Check if this file is an independent copy."""
-        return False  # We can't be sure by default
-
-    def walk_files(self, path_info):
-        """Return a generator with `PathInfo`s to all the files"""
-        raise NotImplementedError
-
     @staticmethod
     def protect(path_info):
         pass
@@ -617,7 +683,7 @@ class BaseRemote:
             isdir = tree.isdir
             save_link = False
         else:
-            isdir = self.isdir
+            isdir = self.tree.isdir
 
         if isdir(path_info):
             return self._save_dir(
@@ -625,6 +691,9 @@ class BaseRemote:
             )
         return self._save_file(path_info, checksum, save_link, tree, **kwargs)
 
+    def open(self, *args, **kwargs):
+        return self.tree.open(*args, **kwargs)
+
     def _handle_transfer_exception(
         self, from_info, to_info, exception, operation
     ):
@@ -680,13 +749,13 @@ class BaseRemote:
             raise NotImplementedError
 
         if to_info.scheme == self.scheme != "local":
-            self.copy(from_info, to_info)
+            self.tree.copy(from_info, to_info)
             return 0
 
         if to_info.scheme != "local":
             raise NotImplementedError
 
-        if self.isdir(from_info):
+        if self.tree.isdir(from_info):
             return self._download_dir(
                 from_info, to_info, name, no_progress_bar, file_mode, dir_mode
             )
@@ -697,7 +766,7 @@ class BaseRemote:
     def _download_dir(
         self, from_info, to_info, name, no_progress_bar, file_mode, dir_mode
     ):
-        from_infos = list(self.walk_files(from_info))
+        from_infos = list(self.tree.walk_files(from_info))
         to_infos = (
             to_info / info.relative_to(from_info) for info in from_infos
         )
@@ -744,39 +813,6 @@ class BaseRemote:
 
         return 0
 
-    def open(self, path_info, mode="r", encoding=None):
-        if hasattr(self, "_generate_download_url"):
-            get_url = partial(self._generate_download_url, path_info)
-            return open_url(get_url, mode=mode, encoding=encoding)
-
-        raise RemoteActionNotImplemented("open", self.scheme)
-
-    def remove(self, path_info):
-        raise RemoteActionNotImplemented("remove", self.scheme)
-
-    def move(self, from_info, to_info, mode=None):
-        assert mode is None
-        self.copy(from_info, to_info)
-        self.remove(from_info)
-
-    def copy(self, from_info, to_info):
-        raise RemoteActionNotImplemented("copy", self.scheme)
-
-    def copy_fobj(self, fobj, to_info):
-        raise RemoteActionNotImplemented("copy_fobj", self.scheme)
-
-    def symlink(self, from_info, to_info):
-        raise RemoteActionNotImplemented("symlink", self.scheme)
-
-    def hardlink(self, from_info, to_info):
-        raise RemoteActionNotImplemented("hardlink", self.scheme)
-
-    def reflink(self, from_info, to_info):
-        raise RemoteActionNotImplemented("reflink", self.scheme)
-
-    def exists(self, path_info):
-        raise NotImplementedError
-
     def path_to_checksum(self, path):
         parts = self.path_cls(path).parts[-2:]
 
@@ -849,7 +885,7 @@ class BaseRemote:
             if self.is_dir_checksum(checksum):
                 # backward compatibility
                 self._remove_unpacked_dir(checksum)
-            self.remove(path_info)
+            self.tree.remove(path_info)
             removed = True
         if removed:
             self.index.clear()
@@ -896,9 +932,9 @@ class BaseRemote:
             self.protect(cache_info)
             return False
 
-        if self.exists(cache_info):
+        if self.tree.exists(cache_info):
             logger.warning("corrupted cache file '%s'.", cache_info)
-            self.remove(cache_info)
+            self.tree.remove(cache_info)
 
         return True
 
@@ -1142,7 +1178,7 @@ class BaseRemote:
         ) as pbar:
 
             def exists_with_progress(path_info):
-                ret = self.exists(path_info)
+                ret = self.tree.exists(path_info)
                 pbar.update_msg(str(path_info))
                 return ret
 
@@ -1161,7 +1197,7 @@ class BaseRemote:
         return not self.changed_cache(current)
 
     def safe_remove(self, path_info, force=False):
-        if not self.exists(path_info):
+        if not self.tree.exists(path_info):
             return
 
         if not force and not self.already_cached(path_info):
@@ -1173,7 +1209,7 @@ class BaseRemote:
             if not prompt.confirm(msg):
                 raise ConfirmRemoveError(str(path_info))
 
-        self.remove(path_info)
+        self.tree.remove(path_info)
 
     def _checkout_file(
         self, path_info, checksum, force, progress_callback=None, relink=False
@@ -1181,7 +1217,7 @@ class BaseRemote:
         """The file is changed we need to checkout a new copy"""
         added, modified = True, False
         cache_info = self.checksum_to_path_info(checksum)
-        if self.exists(path_info):
+        if self.tree.exists(path_info):
             logger.debug("data '%s' will be replaced.", path_info)
             self.safe_remove(path_info, force=force)
             added, modified = False, True
@@ -1194,11 +1230,6 @@ class BaseRemote:
 
         return added, modified and not relink
 
-    def makedirs(self, path_info):
-        """Optional: Implement only if the remote needs to create
-        directories before copying/linking/moving data
-        """
-
     def _checkout_dir(
         self,
         path_info,
@@ -1211,9 +1242,9 @@ class BaseRemote:
         added, modified = False, False
         # Create dir separately so that dir is created
         # even if there are no files in it
-        if not self.exists(path_info):
+        if not self.tree.exists(path_info):
             added = True
-            self.makedirs(path_info)
+            self.tree.makedirs(path_info)
 
         dir_info = self.get_dir_cache(checksum)
 
@@ -1249,7 +1280,7 @@ class BaseRemote:
         return added, not added and modified and not relink
 
     def _remove_redundant_files(self, path_info, dir_info, force):
-        existing_files = set(self.walk_files(path_info))
+        existing_files = set(self.tree.walk_files(path_info))
 
         needed_files = {
             path_info / entry[self.PARAM_RELPATH] for entry in dir_info

dvc/remote/gdrive.py

@@ -12,7 +12,7 @@ from funcy.py3 import cat
 from dvc.exceptions import DvcException
 from dvc.path_info import CloudURLInfo
 from dvc.progress import Tqdm
-from dvc.remote.base import BaseRemote
+from dvc.remote.base import BaseRemote, BaseRemoteTree
 from dvc.scheme import Schemes
 from dvc.utils import format_link, tmp_fname
 
@@ -87,6 +87,20 @@ class GDriveURLInfo(CloudURLInfo):
         self._spath = re.sub("/{2,}", "/", self._spath.rstrip("/"))
 
 
+class GDriveRemoteTree(BaseRemoteTree):
+    def exists(self, path_info):
+        try:
+            self.remote.get_item_id(path_info)
+        except GDrivePathNotFound:
+            return False
+        else:
+            return True
+
+    def remove(self, path_info):
+        item_id = self.remote.get_item_id(path_info)
+        self.remote.gdrive_delete_file(item_id)
+
+
 class GDriveRemote(BaseRemote):
     scheme = Schemes.GDRIVE
     path_cls = GDriveURLInfo
@@ -95,6 +109,7 @@ class GDriveRemote(BaseRemote):
     # Always prefer traverse for GDrive since API usage quotas are a concern.
     TRAVERSE_WEIGHT_MULTIPLIER = 1
     TRAVERSE_PREFIX_LEN = 2
+    TREE_CLS = GDriveRemoteTree
 
     GDRIVE_CREDENTIALS_DATA = "GDRIVE_CREDENTIALS_DATA"
     DEFAULT_USER_CREDENTIALS_FILE = "gdrive-user-credentials.json"
@@ -281,7 +296,7 @@ class GDriveRemote(BaseRemote):
         cache = {
             "dirs": defaultdict(list),
             "ids": {},
-            "root_id": self._get_item_id(
+            "root_id": self.get_item_id(
                 self.path_info,
                 use_cache=False,
                 hint="Confirm the directory exists and you can access it.",
@@ -394,7 +409,7 @@ class GDriveRemote(BaseRemote):
             gdrive_file.GetContentFile(to_file, callback=pbar.update_to)
 
     @_gdrive_retry
-    def _gdrive_delete_file(self, item_id):
+    def gdrive_delete_file(self, item_id):
         from pydrive2.files import ApiRequestError
 
         param = {"id": item_id}
@@ -498,7 +513,7 @@ class GDriveRemote(BaseRemote):
             [self._create_dir(min(parent_ids), title, path)] if create else []
         )
 
-    def _get_item_id(self, path_info, create=False, use_cache=True, hint=None):
+    def get_item_id(self, path_info, create=False, use_cache=True, hint=None):
         assert path_info.bucket == self._bucket
 
         item_ids = self._path_to_item_ids(path_info.path, create, use_cache)
@@ -508,25 +523,17 @@ class GDriveRemote(BaseRemote):
         assert not create
         raise GDrivePathNotFound(path_info, hint)
 
-    def exists(self, path_info):
-        try:
-            self._get_item_id(path_info)
-        except GDrivePathNotFound:
-            return False
-        else:
-            return True
-
     def _upload(self, from_file, to_info, name=None, no_progress_bar=False):
         dirname = to_info.parent
         assert dirname
-        parent_id = self._get_item_id(dirname, True)
+        parent_id = self.get_item_id(dirname, True)
 
         self._gdrive_upload_file(
             parent_id, to_info.name, no_progress_bar, from_file, name
         )
 
     def _download(self, from_info, to_file, name=None, no_progress_bar=False):
-        item_id = self._get_item_id(from_info)
+        item_id = self.get_item_id(from_info)
         self._gdrive_download_file(item_id, to_file, name, no_progress_bar)
 
     def list_cache_paths(self, prefix=None, progress_callback=None):
@@ -552,12 +559,5 @@ class GDriveRemote(BaseRemote):
                 self._ids_cache["ids"][parent_id], item["title"]
             )
 
-    def remove(self, path_info):
-        item_id = self._get_item_id(path_info)
-        self._gdrive_delete_file(item_id)
-
     def get_file_checksum(self, path_info):
         raise NotImplementedError
-
-    def walk_files(self, path_info):
-        raise NotImplementedError

dvc/remote/gs.py

@@ -10,7 +10,7 @@ from funcy import cached_property, wrap_prop
 from dvc.exceptions import DvcException
 from dvc.path_info import CloudURLInfo
 from dvc.progress import Tqdm
-from dvc.remote.base import BaseRemote
+from dvc.remote.base import BaseRemote, BaseRemoteTree
 from dvc.scheme import Schemes
 
 logger = logging.getLogger(__name__)
@@ -65,11 +65,82 @@ def _upload_to_bucket(
             blob.upload_from_file(wrapped)
 
 
+class GSRemoteTree(BaseRemoteTree):
+    @property
+    def gs(self):
+        return self.remote.gs
+
+    def _generate_download_url(self, path_info, expires=3600):
+        expiration = timedelta(seconds=int(expires))
+
+        bucket = self.gs.bucket(path_info.bucket)
+        blob = bucket.get_blob(path_info.path)
+        if blob is None:
+            raise FileNotFoundError
+        return blob.generate_signed_url(expiration=expiration)
+
+    def exists(self, path_info):
+        """Check if the blob exists. If it does not exist,
+        it could be a part of a directory path.
+
+        eg: if `data/file.txt` exists, check for `data` should return True
+        """
+        return self.isfile(path_info) or self.isdir(path_info)
+
+    def isdir(self, path_info):
+        dir_path = path_info / ""
+        return bool(list(self.remote.list_paths(dir_path, max_items=1)))
+
+    def isfile(self, path_info):
+        if path_info.path.endswith("/"):
+            return False
+
+        blob = self.gs.bucket(path_info.bucket).blob(path_info.path)
+        return blob.exists()
+
+    def walk_files(self, path_info):
+        for fname in self.remote.list_paths(path_info / ""):
+            # skip nested empty directories
+            if fname.endswith("/"):
+                continue
+            yield path_info.replace(fname)
+
+    def remove(self, path_info):
+        if path_info.scheme != "gs":
+            raise NotImplementedError
+
+        logger.debug(f"Removing gs://{path_info}")
+        blob = self.gs.bucket(path_info.bucket).get_blob(path_info.path)
+        if not blob:
+            return
+
+        blob.delete()
+
+    def makedirs(self, path_info):
+        if not path_info.path:
+            return
+
+        self.gs.bucket(path_info.bucket).blob(
+            (path_info / "").path
+        ).upload_from_string("")
+
+    def copy(self, from_info, to_info):
+        from_bucket = self.gs.bucket(from_info.bucket)
+        blob = from_bucket.get_blob(from_info.path)
+        if not blob:
+            msg = f"'{from_info.path}' doesn't exist in the cloud"
+            raise DvcException(msg)
+
+        to_bucket = self.gs.bucket(to_info.bucket)
+        from_bucket.copy_blob(blob, to_bucket, new_name=to_info.path)
+
+
 class GSRemote(BaseRemote):
     scheme = Schemes.GS
     path_cls = CloudURLInfo
     REQUIRES = {"google-cloud-storage": "google.cloud.storage"}
     PARAM_CHECKSUM = "md5"
+    TREE_CLS = GSRemoteTree
 
     def __init__(self, repo, config):
         super().__init__(repo, config)
@@ -105,28 +176,7 @@ class GSRemote(BaseRemote):
         md5 = base64.b64decode(b64_md5)
         return codecs.getencoder("hex")(md5)[0].decode("utf-8")
 
-    def copy(self, from_info, to_info):
-        from_bucket = self.gs.bucket(from_info.bucket)
-        blob = from_bucket.get_blob(from_info.path)
-        if not blob:
-            msg = f"'{from_info.path}' doesn't exist in the cloud"
-            raise DvcException(msg)
-
-        to_bucket = self.gs.bucket(to_info.bucket)
-        from_bucket.copy_blob(blob, to_bucket, new_name=to_info.path)
-
-    def remove(self, path_info):
-        if path_info.scheme != "gs":
-            raise NotImplementedError
-
-        logger.debug(f"Removing gs://{path_info}")
-        blob = self.gs.bucket(path_info.bucket).get_blob(path_info.path)
-        if not blob:
-            return
-
-        blob.delete()
-
-    def _list_paths(
+    def list_paths(
         self, path_info, max_items=None, prefix=None, progress_callback=None
     ):
         if prefix:
@@ -141,44 +191,10 @@ class GSRemote(BaseRemote):
             yield blob.name
 
     def list_cache_paths(self, prefix=None, progress_callback=None):
-        return self._list_paths(
+        return self.list_paths(
             self.path_info, prefix=prefix, progress_callback=progress_callback
         )
 
-    def walk_files(self, path_info):
-        for fname in self._list_paths(path_info / ""):
-            # skip nested empty directories
-            if fname.endswith("/"):
-                continue
-            yield path_info.replace(fname)
-
-    def makedirs(self, path_info):
-        if not path_info.path:
-            return
-
-        self.gs.bucket(path_info.bucket).blob(
-            (path_info / "").path
-        ).upload_from_string("")
-
-    def isdir(self, path_info):
-        dir_path = path_info / ""
-        return bool(list(self._list_paths(dir_path, max_items=1)))
-
-    def isfile(self, path_info):
-        if path_info.path.endswith("/"):
-            return False
-
-        blob = self.gs.bucket(path_info.bucket).blob(path_info.path)
-        return blob.exists()
-
-    def exists(self, path_info):
-        """Check if the blob exists. If it does not exist,
-        it could be a part of a directory path.
-
-        eg: if `data/file.txt` exists, check for `data` should return True
-        """
-        return self.isfile(path_info) or self.isdir(path_info)
-
     def _upload(self, from_file, to_info, name=None, no_progress_bar=False):
         bucket = self.gs.bucket(to_info.bucket)
         _upload_to_bucket(
@@ -201,12 +217,3 @@ class GSRemote(BaseRemote):
                 disable=no_progress_bar,
             ) as wrapped:
                 blob.download_to_file(wrapped)
-
-    def _generate_download_url(self, path_info, expires=3600):
-        expiration = timedelta(seconds=int(expires))
-
-        bucket = self.gs.bucket(path_info.bucket)
-        blob = bucket.get_blob(path_info.path)
-        if blob is None:
-            raise FileNotFoundError
-        return blob.generate_signed_url(expiration=expiration)

dvc/remote/hdfs.py

@@ -11,18 +11,75 @@ from urllib.parse import urlparse
 from dvc.scheme import Schemes
 from dvc.utils import fix_env, tmp_fname
 
-from .base import BaseRemote, RemoteCmdError
+from .base import BaseRemote, BaseRemoteTree, RemoteCmdError
 from .pool import get_connection
 
 logger = logging.getLogger(__name__)
 
 
+class HDFSRemoteTree(BaseRemoteTree):
+    @property
+    def hdfs(self):
+        return self.remote.hdfs
+
+    @contextmanager
+    def open(self, path_info, mode="r", encoding=None):
+        assert mode in {"r", "rt", "rb"}
+
+        try:
+            with self.hdfs(path_info) as hdfs, closing(
+                hdfs.open(path_info.path, mode="rb")
+            ) as fd:
+                if mode == "rb":
+                    yield fd
+                else:
+                    yield io.TextIOWrapper(fd, encoding=encoding)
+        except OSError as e:
+            # Empty .errno and not specific enough error class in pyarrow,
+            # see https://issues.apache.org/jira/browse/ARROW-6248
+            if "file does not exist" in str(e):
+                raise FileNotFoundError(*e.args)
+            raise
+
+    def exists(self, path_info):
+        assert not isinstance(path_info, list)
+        assert path_info.scheme == "hdfs"
+        with self.hdfs(path_info) as hdfs:
+            return hdfs.exists(path_info.path)
+
+    def remove(self, path_info):
+        if path_info.scheme != "hdfs":
+            raise NotImplementedError
+
+        if self.exists(path_info):
+            logger.debug(f"Removing {path_info.path}")
+            with self.hdfs(path_info) as hdfs:
+                hdfs.rm(path_info.path)
+
+    def copy(self, from_info, to_info, **_kwargs):
+        dname = posixpath.dirname(to_info.path)
+        with self.hdfs(to_info) as hdfs:
+            hdfs.mkdir(dname)
+            # NOTE: this is how `hadoop fs -cp` works too: it copies through
+            # your local machine.
+            with hdfs.open(from_info.path, "rb") as from_fobj:
+                tmp_info = to_info.parent / tmp_fname(to_info.name)
+                try:
+                    with hdfs.open(tmp_info.path, "wb") as tmp_fobj:
+                        tmp_fobj.upload(from_fobj)
+                    hdfs.rename(tmp_info.path, to_info.path)
+                except Exception:
+                    self.remove(tmp_info)
+                    raise
+
+
 class HDFSRemote(BaseRemote):
     scheme = Schemes.HDFS
     REGEX = r"^hdfs://((?P<user>.*)@)?.*$"
     PARAM_CHECKSUM = "checksum"
     REQUIRES = {"pyarrow": "pyarrow"}
     TRAVERSE_PREFIX_LEN = 2
+    TREE_CLS = HDFSRemoteTree
 
     def __init__(self, repo, config):
         super().__init__(repo, config)
@@ -91,37 +148,6 @@ class HDFSRemote(BaseRemote):
         )
         return self._group(regex, stdout, "checksum")
 
-    def copy(self, from_info, to_info, **_kwargs):
-        dname = posixpath.dirname(to_info.path)
-        with self.hdfs(to_info) as hdfs:
-            hdfs.mkdir(dname)
-            # NOTE: this is how `hadoop fs -cp` works too: it copies through
-            # your local machine.
-            with hdfs.open(from_info.path, "rb") as from_fobj:
-                tmp_info = to_info.parent / tmp_fname(to_info.name)
-                try:
-                    with hdfs.open(tmp_info.path, "wb") as tmp_fobj:
-                        tmp_fobj.upload(from_fobj)
-                    hdfs.rename(tmp_info.path, to_info.path)
-                except Exception:
-                    self.remove(tmp_info)
-                    raise
-
-    def remove(self, path_info):
-        if path_info.scheme != "hdfs":
-            raise NotImplementedError
-
-        if self.exists(path_info):
-            logger.debug(f"Removing {path_info.path}")
-            with self.hdfs(path_info) as hdfs:
-                hdfs.rm(path_info.path)
-
-    def exists(self, path_info):
-        assert not isinstance(path_info, list)
-        assert path_info.scheme == "hdfs"
-        with self.hdfs(path_info) as hdfs:
-            return hdfs.exists(path_info.path)
-
     def _upload(self, from_file, to_info, **_kwargs):
         with self.hdfs(to_info) as hdfs:
             hdfs.mkdir(posixpath.dirname(to_info.path))
@@ -135,27 +161,8 @@ class HDFSRemote(BaseRemote):
             with open(to_file, "wb+") as fobj:
                 hdfs.download(from_info.path, fobj)
 
-    @contextmanager
-    def open(self, path_info, mode="r", encoding=None):
-        assert mode in {"r", "rt", "rb"}
-
-        try:
-            with self.hdfs(path_info) as hdfs, closing(
-                hdfs.open(path_info.path, mode="rb")
-            ) as fd:
-                if mode == "rb":
-                    yield fd
-                else:
-                    yield io.TextIOWrapper(fd, encoding=encoding)
-        except OSError as e:
-            # Empty .errno and not specific enough error class in pyarrow,
-            # see https://issues.apache.org/jira/browse/ARROW-6248
-            if "file does not exist" in str(e):
-                raise FileNotFoundError(*e.args)
-            raise
-
     def list_cache_paths(self, prefix=None, progress_callback=None):
-        if not self.exists(self.path_info):
+        if not self.tree.exists(self.path_info):
             return
 
         if prefix:

dvc/remote/http.py

@@ -8,7 +8,7 @@ import dvc.prompt as prompt
 from dvc.exceptions import DvcException, HTTPError
 from dvc.path_info import HTTPURLInfo
 from dvc.progress import Tqdm
-from dvc.remote.base import BaseRemote
+from dvc.remote.base import BaseRemote, BaseRemoteTree
 from dvc.scheme import Schemes
 
 logger = logging.getLogger(__name__)
@@ -23,6 +23,11 @@ def ask_password(host, user):
     )
 
 
+class HTTPRemoteTree(BaseRemoteTree):
+    def exists(self, path_info):
+        return bool(self.remote.request("HEAD", path_info.url))
+
+
 class HTTPRemote(BaseRemote):
     scheme = Schemes.HTTP
     path_cls = HTTPURLInfo
@@ -32,6 +37,7 @@ class HTTPRemote(BaseRemote):
     CHUNK_SIZE = 2 ** 16
     PARAM_CHECKSUM = "etag"
     CAN_TRAVERSE = False
+    TREE_CLS = HTTPRemoteTree
 
     def __init__(self, repo, config):
         super().__init__(repo, config)
@@ -52,7 +58,7 @@ class HTTPRemote(BaseRemote):
         self.headers = {}
 
     def _download(self, from_info, to_file, name=None, no_progress_bar=False):
-        response = self._request("GET", from_info.url, stream=True)
+        response = self.request("GET", from_info.url, stream=True)
         if response.status_code != 200:
             raise HTTPError(response.status_code, response.reason)
         with open(to_file, "wb") as fd:
@@ -88,20 +94,17 @@ class HTTPRemote(BaseRemote):
                             break
                         yield chunk
 
-        response = self._request("POST", to_info.url, data=chunks())
+        response = self.request("POST", to_info.url, data=chunks())
         if response.status_code not in (200, 201):
             raise HTTPError(response.status_code, response.reason)
 
-    def exists(self, path_info):
-        return bool(self._request("HEAD", path_info.url))
-
     def _content_length(self, response):
         res = response.headers.get("Content-Length")
         return int(res) if res else None
 
     def get_file_checksum(self, path_info):
         url = path_info.url
-        headers = self._request("HEAD", url).headers
+        headers = self.request("HEAD", url).headers
         etag = headers.get("ETag") or headers.get("Content-MD5")
 
         if not etag:
@@ -155,7 +158,7 @@ class HTTPRemote(BaseRemote):
 
         return session
 
-    def _request(self, method, url, **kwargs):
+    def request(self, method, url, **kwargs):
         import requests
 
         kwargs.setdefault("allow_redirects", True)

dvc/remote/local.py

@@ -17,6 +17,7 @@ from dvc.remote.base import (
     STATUS_MISSING,
     STATUS_NEW,
     BaseRemote,
+    BaseRemoteTree,
     index_locked,
 )
 from dvc.remote.index import RemoteIndexNoop
@@ -36,37 +37,12 @@ from dvc.utils.fs import (
 logger = logging.getLogger(__name__)
 
 
-class LocalRemote(BaseRemote):
-    scheme = Schemes.LOCAL
-    path_cls = PathInfo
-    PARAM_CHECKSUM = "md5"
-    PARAM_PATH = "path"
-    TRAVERSE_PREFIX_LEN = 2
-    INDEX_CLS = RemoteIndexNoop
-
-    UNPACKED_DIR_SUFFIX = ".unpacked"
-
-    DEFAULT_CACHE_TYPES = ["reflink", "copy"]
-
-    CACHE_MODE = 0o444
+class LocalRemoteTree(BaseRemoteTree):
     SHARED_MODE_MAP = {None: (0o644, 0o755), "group": (0o664, 0o775)}
 
-    def __init__(self, repo, config):
-        super().__init__(repo, config)
-        self.cache_dir = config.get("url")
-        self._dir_info = {}
-
     @property
-    def state(self):
-        return self.repo.state
-
-    @property
-    def cache_dir(self):
-        return self.path_info.fspath if self.path_info else None
-
-    @cache_dir.setter
-    def cache_dir(self, value):
-        self.path_info = PathInfo(value) if value else None
+    def repo(self):
+        return self.remote.repo
 
     @cached_property
     def _work_tree(self):
@@ -83,42 +59,9 @@ class LocalRemote(BaseRemote):
             return self._work_tree
         return None
 
-    @classmethod
-    def supported(cls, config):
-        return True
-
-    @cached_property
-    def cache_path(self):
-        return os.path.abspath(self.cache_dir)
-
-    def checksum_to_path(self, checksum):
-        # NOTE: `self.cache_path` is already normalized so we can simply use
-        # `os.sep` instead of `os.path.join`. This results in this helper
-        # being ~5.5 times faster.
-        return (
-            f"{self.cache_path}{os.sep}{checksum[0:2]}{os.sep}{checksum[2:]}"
-        )
-
-    def list_cache_paths(self, prefix=None, progress_callback=None):
-        assert self.path_info is not None
-        if prefix:
-            path_info = self.path_info / prefix[:2]
-            if not self.exists(path_info):
-                return
-        else:
-            path_info = self.path_info
-        if progress_callback:
-            for path in walk_files(path_info):
-                progress_callback()
-                yield path
-        else:
-            yield from walk_files(path_info)
-
-    def get(self, md5):
-        if not md5:
-            return None
-
-        return self.checksum_to_path_info(md5).url
+    @staticmethod
+    def open(path_info, mode="r", encoding=None):
+        return open(path_info, mode=mode, encoding=encoding)
 
     def exists(self, path_info):
         assert isinstance(path_info, str) or path_info.scheme == "local"
@@ -128,36 +71,6 @@ class LocalRemote(BaseRemote):
             return True
         return self.repo.tree.exists(path_info)
 
-    def makedirs(self, path_info):
-        makedirs(path_info, exist_ok=True, mode=self._dir_mode)
-
-    def already_cached(self, path_info):
-        assert path_info.scheme in ["", "local"]
-
-        current_md5 = self.get_checksum(path_info)
-
-        if not current_md5:
-            return False
-
-        return not self.changed_cache(current_md5)
-
-    def _verify_link(self, path_info, link_type):
-        if link_type == "hardlink" and self.getsize(path_info) == 0:
-            return
-
-        super()._verify_link(path_info, link_type)
-
-    def is_empty(self, path_info):
-        path = path_info.fspath
-
-        if self.isfile(path_info) and os.path.getsize(path) == 0:
-            return True
-
-        if self.isdir(path_info) and len(os.listdir(path)) == 0:
-            return True
-
-        return False
-
     def isfile(self, path_info):
         if not self.repo:
             return os.path.isfile(path_info)
@@ -177,10 +90,6 @@ class LocalRemote(BaseRemote):
             System.is_symlink(path_info) or System.is_hardlink(path_info)
         )
 
-    @staticmethod
-    def getsize(path_info):
-        return os.path.getsize(path_info)
-
     def walk_files(self, path_info):
         if self.work_tree:
             tree = self.work_tree
@@ -189,8 +98,16 @@ class LocalRemote(BaseRemote):
         for fname in tree.walk_files(path_info):
             yield PathInfo(fname)
 
-    def get_file_checksum(self, path_info):
-        return file_md5(path_info)[0]
+    def is_empty(self, path_info):
+        path = path_info.fspath
+
+        if self.isfile(path_info) and os.path.getsize(path) == 0:
+            return True
+
+        if self.isdir(path_info) and len(os.listdir(path)) == 0:
+            return True
+
+        return False
 
     def remove(self, path_info):
         if isinstance(path_info, PathInfo):
@@ -203,6 +120,9 @@ class LocalRemote(BaseRemote):
         if self.exists(path):
             remove(path)
 
+    def makedirs(self, path_info):
+        makedirs(path_info, exist_ok=True, mode=self.dir_mode)
+
     def move(self, from_info, to_info, mode=None):
         if from_info.scheme != "local" or to_info.scheme != "local":
             raise NotImplementedError
@@ -211,9 +131,9 @@ class LocalRemote(BaseRemote):
 
         if mode is None:
             if self.isfile(from_info):
-                mode = self._file_mode
+                mode = self.file_mode
             else:
-                mode = self._dir_mode
+                mode = self.dir_mode
 
         move(from_info, to_info, mode=mode)
 
@@ -221,7 +141,7 @@ class LocalRemote(BaseRemote):
         tmp_info = to_info.parent / tmp_fname(to_info.name)
         try:
             System.copy(from_info, tmp_info)
-            os.chmod(tmp_info, self._file_mode)
+            os.chmod(tmp_info, self.file_mode)
             os.rename(tmp_info, to_info)
         except Exception:
             self.remove(tmp_info)
@@ -232,7 +152,7 @@ class LocalRemote(BaseRemote):
         tmp_info = to_info.parent / tmp_fname(to_info.name)
         try:
             copy_fobj_to_file(fobj, tmp_info)
-            os.chmod(tmp_info, self._file_mode)
+            os.chmod(tmp_info, self.file_mode)
             os.rename(tmp_info, to_info)
         except Exception:
             self.remove(tmp_info)
@@ -281,9 +201,102 @@ class LocalRemote(BaseRemote):
         System.reflink(from_info, tmp_info)
         # NOTE: reflink has its own separate inode, so you can set permissions
         # that are different from the source.
-        os.chmod(tmp_info, self._file_mode)
+        os.chmod(tmp_info, self.file_mode)
         os.rename(tmp_info, to_info)
 
+    @staticmethod
+    def getsize(path_info):
+        return os.path.getsize(path_info)
+
+
+class LocalRemote(BaseRemote):
+    scheme = Schemes.LOCAL
+    path_cls = PathInfo
+    PARAM_CHECKSUM = "md5"
+    PARAM_PATH = "path"
+    TRAVERSE_PREFIX_LEN = 2
+    INDEX_CLS = RemoteIndexNoop
+    TREE_CLS = LocalRemoteTree
+
+    UNPACKED_DIR_SUFFIX = ".unpacked"
+
+    DEFAULT_CACHE_TYPES = ["reflink", "copy"]
+
+    CACHE_MODE = 0o444
+
+    def __init__(self, repo, config):
+        super().__init__(repo, config)
+        self.cache_dir = config.get("url")
+        self._dir_info = {}
+
+    @property
+    def state(self):
+        return self.repo.state
+
+    @property
+    def cache_dir(self):
+        return self.path_info.fspath if self.path_info else None
+
+    @cache_dir.setter
+    def cache_dir(self, value):
+        self.path_info = PathInfo(value) if value else None
+
+    @classmethod
+    def supported(cls, config):
+        return True
+
+    @cached_property
+    def cache_path(self):
+        return os.path.abspath(self.cache_dir)
+
+    def checksum_to_path(self, checksum):
+        # NOTE: `self.cache_path` is already normalized so we can simply use
+        # `os.sep` instead of `os.path.join`. This results in this helper
+        # being ~5.5 times faster.
+        return (
+            f"{self.cache_path}{os.sep}{checksum[0:2]}{os.sep}{checksum[2:]}"
+        )
+
+    def list_cache_paths(self, prefix=None, progress_callback=None):
+        assert self.path_info is not None
+        if prefix:
+            path_info = self.path_info / prefix[:2]
+            if not self.tree.exists(path_info):
+                return
+        else:
+            path_info = self.path_info
+        if progress_callback:
+            for path in walk_files(path_info):
+                progress_callback()
+                yield path
+        else:
+            yield from walk_files(path_info)
+
+    def get(self, md5):
+        if not md5:
+            return None
+
+        return self.checksum_to_path_info(md5).url
+
+    def already_cached(self, path_info):
+        assert path_info.scheme in ["", "local"]
+
+        current_md5 = self.get_checksum(path_info)
+
+        if not current_md5:
+            return False
+
+        return not self.changed_cache(current_md5)
+
+    def _verify_link(self, path_info, link_type):
+        if link_type == "hardlink" and self.tree.getsize(path_info) == 0:
+            return
+
+        super()._verify_link(path_info, link_type)
+
+    def get_file_checksum(self, path_info):
+        return file_md5(path_info)[0]
+
     def cache_exists(self, checksums, jobs=None, name=None):
         return [
             checksum
@@ -316,10 +329,6 @@ class LocalRemote(BaseRemote):
             from_info, to_file, no_progress_bar=no_progress_bar, name=name
         )
 
-    @staticmethod
-    def open(path_info, mode="r", encoding=None):
-        return open(path_info, mode=mode, encoding=encoding)
-
     @index_locked
     def status(
         self,
@@ -503,8 +512,8 @@ class LocalRemote(BaseRemote):
         if download:
             func = partial(
                 remote.download,
-                dir_mode=self._dir_mode,
-                file_mode=self._file_mode,
+                dir_mode=self.tree.dir_mode,
+                file_mode=self.tree.file_mode,
             )
             status = STATUS_DELETED
             desc = "Downloading"
@@ -662,7 +671,7 @@ class LocalRemote(BaseRemote):
                 "a symlink or a hardlink.".format(path)
             )
 
-        os.chmod(path, self._file_mode)
+        os.chmod(path, self.tree.file_mode)
 
     def _unprotect_dir(self, path):
         assert is_working_tree(self.repo.tree)
@@ -703,7 +712,7 @@ class LocalRemote(BaseRemote):
     def _remove_unpacked_dir(self, checksum):
         info = self.checksum_to_path_info(checksum)
         path_info = info.with_name(info.name + self.UNPACKED_DIR_SUFFIX)
-        self.remove(path_info)
+        self.tree.remove(path_info)
 
     def is_protected(self, path_info):
         try:

dvc/remote/oss.py

@@ -7,12 +7,34 @@ from funcy import cached_property, wrap_prop
 
 from dvc.path_info import CloudURLInfo
 from dvc.progress import Tqdm
-from dvc.remote.base import BaseRemote
+from dvc.remote.base import BaseRemote, BaseRemoteTree
 from dvc.scheme import Schemes
 
 logger = logging.getLogger(__name__)
 
 
+class OSSRemoteTree(BaseRemoteTree):
+    @property
+    def oss_service(self):
+        return self.remote.oss_service
+
+    def _generate_download_url(self, path_info, expires=3600):
+        assert path_info.bucket == self.remote.path_info.bucket
+
+        return self.oss_service.sign_url("GET", path_info.path, expires)
+
+    def exists(self, path_info):
+        paths = self.remote.list_paths(path_info.path)
+        return any(path_info.path == path for path in paths)
+
+    def remove(self, path_info):
+        if path_info.scheme != self.scheme:
+            raise NotImplementedError
+
+        logger.debug(f"Removing oss://{path_info}")
+        self.oss_service.delete_object(path_info.path)
+
+
 class OSSRemote(BaseRemote):
     """
     oss2 document:
@@ -38,6 +60,7 @@ class OSSRemote(BaseRemote):
     PARAM_CHECKSUM = "etag"
     COPY_POLL_SECONDS = 5
     LIST_OBJECT_PAGE_SIZE = 100
+    TREE_CLS = OSSRemoteTree
 
     def __init__(self, repo, config):
         super().__init__(repo, config)
@@ -83,14 +106,7 @@ class OSSRemote(BaseRemote):
             )
         return bucket
 
-    def remove(self, path_info):
-        if path_info.scheme != self.scheme:
-            raise NotImplementedError
-
-        logger.debug(f"Removing oss://{path_info}")
-        self.oss_service.delete_object(path_info.path)
-
-    def _list_paths(self, prefix, progress_callback=None):
+    def list_paths(self, prefix, progress_callback=None):
         import oss2
 
         for blob in oss2.ObjectIterator(self.oss_service, prefix=prefix):
@@ -105,7 +121,7 @@ class OSSRemote(BaseRemote):
             )
         else:
             prefix = self.path_info.path
-        return self._list_paths(prefix, progress_callback)
+        return self.list_paths(prefix, progress_callback)
 
     def _upload(
         self, from_file, to_info, name=None, no_progress_bar=False, **_kwargs
@@ -122,12 +138,3 @@ class OSSRemote(BaseRemote):
             self.oss_service.get_object_to_file(
                 from_info.path, to_file, progress_callback=pbar.update_to
             )
-
-    def _generate_download_url(self, path_info, expires=3600):
-        assert path_info.bucket == self.path_info.bucket
-
-        return self.oss_service.sign_url("GET", path_info.path, expires)
-
-    def exists(self, path_info):
-        paths = self._list_paths(path_info.path)
-        return any(path_info.path == path for path in paths)

dvc/remote/s3.py

@@ -9,17 +9,197 @@ from dvc.config import ConfigError
 from dvc.exceptions import DvcException, ETagMismatchError
 from dvc.path_info import CloudURLInfo
 from dvc.progress import Tqdm
-from dvc.remote.base import BaseRemote
+from dvc.remote.base import BaseRemote, BaseRemoteTree
 from dvc.scheme import Schemes
 
 logger = logging.getLogger(__name__)
 
 
+class S3RemoteTree(BaseRemoteTree):
+    @property
+    def s3(self):
+        return self.remote.s3
+
+    def _generate_download_url(self, path_info, expires=3600):
+        params = {"Bucket": path_info.bucket, "Key": path_info.path}
+        return self.s3.generate_presigned_url(
+            ClientMethod="get_object", Params=params, ExpiresIn=int(expires)
+        )
+
+    def exists(self, path_info):
+        """Check if the blob exists. If it does not exist,
+        it could be a part of a directory path.
+
+        eg: if `data/file.txt` exists, check for `data` should return True
+        """
+        return self.isfile(path_info) or self.isdir(path_info)
+
+    def isdir(self, path_info):
+        # S3 doesn't have a concept for directories.
+        #
+        # Using `head_object` with a path pointing to a directory
+        # will throw a 404 error.
+        #
+        # A reliable way to know if a given path is a directory is by
+        # checking if there are more files sharing the same prefix
+        # with a `list_objects` call.
+        #
+        # We need to make sure that the path ends with a forward slash,
+        # since we can end with false-positives like the following example:
+        #
+        #       bucket
+        #       └── data
+        #          ├── alice
+        #          └── alpha
+        #
+        # Using `data/al` as prefix will return `[data/alice, data/alpha]`,
+        # While `data/al/` will return nothing.
+        #
+        dir_path = path_info / ""
+        return bool(list(self.remote.list_paths(dir_path, max_items=1)))
+
+    def isfile(self, path_info):
+        from botocore.exceptions import ClientError
+
+        if path_info.path.endswith("/"):
+            return False
+
+        try:
+            self.s3.head_object(Bucket=path_info.bucket, Key=path_info.path)
+        except ClientError as exc:
+            if exc.response["Error"]["Code"] != "404":
+                raise
+            return False
+
+        return True
+
+    def walk_files(self, path_info, max_items=None):
+        for fname in self.remote.list_paths(path_info / "", max_items):
+            if fname.endswith("/"):
+                continue
+
+            yield path_info.replace(path=fname)
+
+    def remove(self, path_info):
+        if path_info.scheme != "s3":
+            raise NotImplementedError
+
+        logger.debug(f"Removing {path_info}")
+        self.s3.delete_object(Bucket=path_info.bucket, Key=path_info.path)
+
+    def makedirs(self, path_info):
+        # We need to support creating empty directories, which means
+        # creating an object with an empty body and a trailing slash `/`.
+        #
+        # We are not creating directory objects for every parent prefix,
+        # as it is not required.
+        if not path_info.path:
+            return
+
+        dir_path = path_info / ""
+        self.s3.put_object(Bucket=path_info.bucket, Key=dir_path.path, Body="")
+
+    def copy(self, from_info, to_info):
+        self._copy(self.s3, from_info, to_info, self.remote.extra_args)
+
+    @classmethod
+    def _copy_multipart(
+        cls, s3, from_info, to_info, size, n_parts, extra_args
+    ):
+        mpu = s3.create_multipart_upload(
+            Bucket=to_info.bucket, Key=to_info.path, **extra_args
+        )
+        mpu_id = mpu["UploadId"]
+
+        parts = []
+        byte_position = 0
+        for i in range(1, n_parts + 1):
+            obj = S3Remote.get_head_object(
+                s3, from_info.bucket, from_info.path, PartNumber=i
+            )
+            part_size = obj["ContentLength"]
+            lastbyte = byte_position + part_size - 1
+            if lastbyte > size:
+                lastbyte = size - 1
+
+            srange = f"bytes={byte_position}-{lastbyte}"
+
+            part = s3.upload_part_copy(
+                Bucket=to_info.bucket,
+                Key=to_info.path,
+                PartNumber=i,
+                UploadId=mpu_id,
+                CopySourceRange=srange,
+                CopySource={"Bucket": from_info.bucket, "Key": from_info.path},
+            )
+            parts.append(
+                {"PartNumber": i, "ETag": part["CopyPartResult"]["ETag"]}
+            )
+            byte_position += part_size
+
+        assert n_parts == len(parts)
+
+        s3.complete_multipart_upload(
+            Bucket=to_info.bucket,
+            Key=to_info.path,
+            UploadId=mpu_id,
+            MultipartUpload={"Parts": parts},
+        )
+
+    @classmethod
+    def _copy(cls, s3, from_info, to_info, extra_args):
+        # NOTE: object's etag depends on the way it was uploaded to s3 or the
+        # way it was copied within the s3. More specifically, it depends on
+        # the chunk size that was used to transfer it, which would affect
+        # whether an object would be uploaded as a single part or as a
+        # multipart.
+        #
+        # If an object's etag looks like '8978c98bb5a48c2fb5f2c4c905768afa',
+        # then it was transferred as a single part, which means that the chunk
+        # size used to transfer it was greater or equal to the ContentLength
+        # of that object. So to preserve that tag over the next transfer, we
+        # could use any value >= ContentLength.
+        #
+        # If an object's etag looks like '50d67013a5e1a4070bef1fc8eea4d5f9-13',
+        # then it was transferred as a multipart, which means that the chunk
+        # size used to transfer it was less than ContentLength of that object.
+        # Unfortunately, in general, it doesn't mean that the chunk size was
+        # the same throughout the transfer, so it means that in order to
+        # preserve etag, we need to transfer each part separately, so the
+        # object is transfered in the same chunks as it was originally.
+        from boto3.s3.transfer import TransferConfig
+
+        obj = S3Remote.get_head_object(s3, from_info.bucket, from_info.path)
+        etag = obj["ETag"].strip('"')
+        size = obj["ContentLength"]
+
+        _, _, parts_suffix = etag.partition("-")
+        if parts_suffix:
+            n_parts = int(parts_suffix)
+            cls._copy_multipart(
+                s3, from_info, to_info, size, n_parts, extra_args=extra_args
+            )
+        else:
+            source = {"Bucket": from_info.bucket, "Key": from_info.path}
+            s3.copy(
+                source,
+                to_info.bucket,
+                to_info.path,
+                ExtraArgs=extra_args,
+                Config=TransferConfig(multipart_threshold=size + 1),
+            )
+
+        cached_etag = S3Remote.get_etag(s3, to_info.bucket, to_info.path)
+        if etag != cached_etag:
+            raise ETagMismatchError(etag, cached_etag)
+
+
 class S3Remote(BaseRemote):
     scheme = Schemes.S3
     path_cls = CloudURLInfo
     REQUIRES = {"boto3": "boto3"}
     PARAM_CHECKSUM = "etag"
+    TREE_CLS = S3RemoteTree
 
     def __init__(self, repo, config):
         super().__init__(repo, config)
@@ -89,107 +269,6 @@ class S3Remote(BaseRemote):
             raise DvcException(f"s3://{bucket}/{path} does not exist") from exc
         return obj
 
-    @classmethod
-    def _copy_multipart(
-        cls, s3, from_info, to_info, size, n_parts, extra_args
-    ):
-        mpu = s3.create_multipart_upload(
-            Bucket=to_info.bucket, Key=to_info.path, **extra_args
-        )
-        mpu_id = mpu["UploadId"]
-
-        parts = []
-        byte_position = 0
-        for i in range(1, n_parts + 1):
-            obj = cls.get_head_object(
-                s3, from_info.bucket, from_info.path, PartNumber=i
-            )
-            part_size = obj["ContentLength"]
-            lastbyte = byte_position + part_size - 1
-            if lastbyte > size:
-                lastbyte = size - 1
-
-            srange = f"bytes={byte_position}-{lastbyte}"
-
-            part = s3.upload_part_copy(
-                Bucket=to_info.bucket,
-                Key=to_info.path,
-                PartNumber=i,
-                UploadId=mpu_id,
-                CopySourceRange=srange,
-                CopySource={"Bucket": from_info.bucket, "Key": from_info.path},
-            )
-            parts.append(
-                {"PartNumber": i, "ETag": part["CopyPartResult"]["ETag"]}
-            )
-            byte_position += part_size
-
-        assert n_parts == len(parts)
-
-        s3.complete_multipart_upload(
-            Bucket=to_info.bucket,
-            Key=to_info.path,
-            UploadId=mpu_id,
-            MultipartUpload={"Parts": parts},
-        )
-
-    @classmethod
-    def _copy(cls, s3, from_info, to_info, extra_args):
-        # NOTE: object's etag depends on the way it was uploaded to s3 or the
-        # way it was copied within the s3. More specifically, it depends on
-        # the chunk size that was used to transfer it, which would affect
-        # whether an object would be uploaded as a single part or as a
-        # multipart.
-        #
-        # If an object's etag looks like '8978c98bb5a48c2fb5f2c4c905768afa',
-        # then it was transferred as a single part, which means that the chunk
-        # size used to transfer it was greater or equal to the ContentLength
-        # of that object. So to preserve that tag over the next transfer, we
-        # could use any value >= ContentLength.
-        #
-        # If an object's etag looks like '50d67013a5e1a4070bef1fc8eea4d5f9-13',
-        # then it was transferred as a multipart, which means that the chunk
-        # size used to transfer it was less than ContentLength of that object.
-        # Unfortunately, in general, it doesn't mean that the chunk size was
-        # the same throughout the transfer, so it means that in order to
-        # preserve etag, we need to transfer each part separately, so the
-        # object is transfered in the same chunks as it was originally.
-        from boto3.s3.transfer import TransferConfig
-
-        obj = cls.get_head_object(s3, from_info.bucket, from_info.path)
-        etag = obj["ETag"].strip('"')
-        size = obj["ContentLength"]
-
-        _, _, parts_suffix = etag.partition("-")
-        if parts_suffix:
-            n_parts = int(parts_suffix)
-            cls._copy_multipart(
-                s3, from_info, to_info, size, n_parts, extra_args=extra_args
-            )
-        else:
-            source = {"Bucket": from_info.bucket, "Key": from_info.path}
-            s3.copy(
-                source,
-                to_info.bucket,
-                to_info.path,
-                ExtraArgs=extra_args,
-                Config=TransferConfig(multipart_threshold=size + 1),
-            )
-
-        cached_etag = cls.get_etag(s3, to_info.bucket, to_info.path)
-        if etag != cached_etag:
-            raise ETagMismatchError(etag, cached_etag)
-
-    def copy(self, from_info, to_info):
-        self._copy(self.s3, from_info, to_info, self.extra_args)
-
-    def remove(self, path_info):
-        if path_info.scheme != "s3":
-            raise NotImplementedError
-
-        logger.debug(f"Removing {path_info}")
-        self.s3.delete_object(Bucket=path_info.bucket, Key=path_info.path)
-
     def _list_objects(
         self, path_info, max_items=None, prefix=None, progress_callback=None
     ):
@@ -211,7 +290,7 @@ class S3Remote(BaseRemote):
             else:
                 yield from contents
 
-    def _list_paths(
+    def list_paths(
         self, path_info, max_items=None, prefix=None, progress_callback=None
     ):
         return (
@@ -222,69 +301,10 @@ class S3Remote(BaseRemote):
         )
 
     def list_cache_paths(self, prefix=None, progress_callback=None):
-        return self._list_paths(
+        return self.list_paths(
             self.path_info, prefix=prefix, progress_callback=progress_callback
         )
 
-    def isfile(self, path_info):
-        from botocore.exceptions import ClientError
-
-        if path_info.path.endswith("/"):
-            return False
-
-        try:
-            self.s3.head_object(Bucket=path_info.bucket, Key=path_info.path)
-        except ClientError as exc:
-            if exc.response["Error"]["Code"] != "404":
-                raise
-            return False
-
-        return True
-
-    def exists(self, path_info):
-        """Check if the blob exists. If it does not exist,
-        it could be a part of a directory path.
-
-        eg: if `data/file.txt` exists, check for `data` should return True
-        """
-        return self.isfile(path_info) or self.isdir(path_info)
-
-    def makedirs(self, path_info):
-        # We need to support creating empty directories, which means
-        # creating an object with an empty body and a trailing slash `/`.
-        #
-        # We are not creating directory objects for every parent prefix,
-        # as it is not required.
-        if not path_info.path:
-            return
-
-        dir_path = path_info / ""
-        self.s3.put_object(Bucket=path_info.bucket, Key=dir_path.path, Body="")
-
-    def isdir(self, path_info):
-        # S3 doesn't have a concept for directories.
-        #
-        # Using `head_object` with a path pointing to a directory
-        # will throw a 404 error.
-        #
-        # A reliable way to know if a given path is a directory is by
-        # checking if there are more files sharing the same prefix
-        # with a `list_objects` call.
-        #
-        # We need to make sure that the path ends with a forward slash,
-        # since we can end with false-positives like the following example:
-        #
-        #       bucket
-        #       └── data
-        #          ├── alice
-        #          └── alpha
-        #
-        # Using `data/al` as prefix will return `[data/alice, data/alpha]`,
-        # While `data/al/` will return nothing.
-        #
-        dir_path = path_info / ""
-        return bool(list(self._list_paths(dir_path, max_items=1)))
-
     def _upload(self, from_file, to_info, name=None, no_progress_bar=False):
         total = os.path.getsize(from_file)
         with Tqdm(
@@ -312,19 +332,6 @@ class S3Remote(BaseRemote):
                 from_info.bucket, from_info.path, to_file, Callback=pbar.update
             )
 
-    def _generate_download_url(self, path_info, expires=3600):
-        params = {"Bucket": path_info.bucket, "Key": path_info.path}
-        return self.s3.generate_presigned_url(
-            ClientMethod="get_object", Params=params, ExpiresIn=int(expires)
-        )
-
-    def walk_files(self, path_info, max_items=None):
-        for fname in self._list_paths(path_info / "", max_items):
-            if fname.endswith("/"):
-                continue
-
-            yield path_info.replace(path=fname)
-
     def _append_aws_grants_to_extra_args(self, config):
         # Keys for extra_args can be one of the following list:
         # https://boto3.amazonaws.com/v1/documentation/api/latest/reference/customizations/s3.html#boto3.s3.transfer.S3Transfer.ALLOWED_UPLOAD_ARGS

dvc/remote/ssh/__init__.py

@@ -14,7 +14,7 @@ from funcy import first, memoize, silent, wrap_with
 
 import dvc.prompt as prompt
 from dvc.progress import Tqdm
-from dvc.remote.base import BaseRemote
+from dvc.remote.base import BaseRemote, BaseRemoteTree
 from dvc.remote.pool import get_connection
 from dvc.scheme import Schemes
 from dvc.utils import to_chunks
@@ -33,6 +33,105 @@ def ask_password(host, user, port):
     )
 
 
+class SSHRemoteTree(BaseRemoteTree):
+    @property
+    def ssh(self):
+        return self.remote.ssh
+
+    @contextmanager
+    def open(self, path_info, mode="r", encoding=None):
+        assert mode in {"r", "rt", "rb", "wb"}
+
+        with self.ssh(path_info) as ssh, closing(
+            ssh.sftp.open(path_info.path, mode)
+        ) as fd:
+            if "b" in mode:
+                yield fd
+            else:
+                yield io.TextIOWrapper(fd, encoding=encoding)
+
+    def exists(self, path_info):
+        with self.ssh(path_info) as ssh:
+            return ssh.exists(path_info.path)
+
+    def isdir(self, path_info):
+        with self.ssh(path_info) as ssh:
+            return ssh.isdir(path_info.path)
+
+    def isfile(self, path_info):
+        with self.ssh(path_info) as ssh:
+            return ssh.isfile(path_info.path)
+
+    def walk_files(self, path_info):
+        with self.ssh(path_info) as ssh:
+            for fname in ssh.walk_files(path_info.path):
+                yield path_info.replace(path=fname)
+
+    def remove(self, path_info):
+        if path_info.scheme != self.scheme:
+            raise NotImplementedError
+
+        with self.ssh(path_info) as ssh:
+            ssh.remove(path_info.path)
+
+    def makedirs(self, path_info):
+        with self.ssh(path_info) as ssh:
+            ssh.makedirs(path_info.path)
+
+    def move(self, from_info, to_info, mode=None):
+        assert mode is None
+        if from_info.scheme != self.scheme or to_info.scheme != self.scheme:
+            raise NotImplementedError
+
+        with self.ssh(from_info) as ssh:
+            ssh.move(from_info.path, to_info.path)
+
+    def copy(self, from_info, to_info):
+        if not from_info.scheme == to_info.scheme == self.scheme:
+            raise NotImplementedError
+
+        with self.ssh(from_info) as ssh:
+            ssh.atomic_copy(from_info.path, to_info.path)
+
+    def symlink(self, from_info, to_info):
+        if not from_info.scheme == to_info.scheme == self.scheme:
+            raise NotImplementedError
+
+        with self.ssh(from_info) as ssh:
+            ssh.symlink(from_info.path, to_info.path)
+
+    def hardlink(self, from_info, to_info):
+        if not from_info.scheme == to_info.scheme == self.scheme:
+            raise NotImplementedError
+
+        # See dvc/remote/local/__init__.py - hardlink()
+        if self.getsize(from_info) == 0:
+
+            with self.ssh(to_info) as ssh:
+                ssh.sftp.open(to_info.path, "w").close()
+
+            logger.debug(
+                "Created empty file: {src} -> {dest}".format(
+                    src=str(from_info), dest=str(to_info)
+                )
+            )
+            return
+
+        with self.ssh(from_info) as ssh:
+            ssh.hardlink(from_info.path, to_info.path)
+
+    def reflink(self, from_info, to_info):
+        if from_info.scheme != self.scheme or to_info.scheme != self.scheme:
+            raise NotImplementedError
+
+        with self.ssh(from_info) as ssh:
+            ssh.reflink(from_info.path, to_info.path)
+
+    def getsize(self, path_info):
+        with self.ssh(path_info) as ssh:
+            return ssh.getsize(path_info.path)
+
+
 class SSHRemote(BaseRemote):
     scheme = Schemes.SSH
     REQUIRES = {"paramiko": "paramiko"}
@@ -46,6 +145,7 @@ class SSHRemote(BaseRemote):
     # We use conservative setting of 4 instead to not exhaust max sessions.
     CHECKSUM_JOBS = 4
     TRAVERSE_PREFIX_LEN = 2
+    TREE_CLS = SSHRemoteTree
 
     DEFAULT_CACHE_TYPES = ["copy"]
 
@@ -148,10 +248,6 @@ class SSHRemote(BaseRemote):
             sock=self.sock,
         )
 
-    def exists(self, path_info):
-        with self.ssh(path_info) as ssh:
-            return ssh.exists(path_info.path)
-
     def get_file_checksum(self, path_info):
         if path_info.scheme != self.scheme:
             raise NotImplementedError
@@ -159,74 +255,6 @@ class SSHRemote(BaseRemote):
         with self.ssh(path_info) as ssh:
             return ssh.md5(path_info.path)
 
-    def isdir(self, path_info):
-        with self.ssh(path_info) as ssh:
-            return ssh.isdir(path_info.path)
-
-    def isfile(self, path_info):
-        with self.ssh(path_info) as ssh:
-            return ssh.isfile(path_info.path)
-
-    def getsize(self, path_info):
-        with self.ssh(path_info) as ssh:
-            return ssh.getsize(path_info.path)
-
-    def copy(self, from_info, to_info):
-        if not from_info.scheme == to_info.scheme == self.scheme:
-            raise NotImplementedError
-
-        with self.ssh(from_info) as ssh:
-            ssh.atomic_copy(from_info.path, to_info.path)
-
-    def symlink(self, from_info, to_info):
-        if not from_info.scheme == to_info.scheme == self.scheme:
-            raise NotImplementedError
-
-        with self.ssh(from_info) as ssh:
-            ssh.symlink(from_info.path, to_info.path)
-
-    def hardlink(self, from_info, to_info):
-        if not from_info.scheme == to_info.scheme == self.scheme:
-            raise NotImplementedError
-
-        # See dvc/remote/local/__init__.py - hardlink()
-        if self.getsize(from_info) == 0:
-
-            with self.ssh(to_info) as ssh:
-                ssh.sftp.open(to_info.path, "w").close()
-
-            logger.debug(
-                "Created empty file: {src} -> {dest}".format(
-                    src=str(from_info), dest=str(to_info)
-                )
-            )
-            return
-
-        with self.ssh(from_info) as ssh:
-            ssh.hardlink(from_info.path, to_info.path)
-
-    def reflink(self, from_info, to_info):
-        if from_info.scheme != self.scheme or to_info.scheme != self.scheme:
-            raise NotImplementedError
-
-        with self.ssh(from_info) as ssh:
-            ssh.reflink(from_info.path, to_info.path)
-
-    def remove(self, path_info):
-        if path_info.scheme != self.scheme:
-            raise NotImplementedError
-
-        with self.ssh(path_info) as ssh:
-            ssh.remove(path_info.path)
-
-    def move(self, from_info, to_info, mode=None):
-        assert mode is None
-        if from_info.scheme != self.scheme or to_info.scheme != self.scheme:
-            raise NotImplementedError
-
-        with self.ssh(from_info) as ssh:
-            ssh.move(from_info.path, to_info.path)
-
     def _download(self, from_info, to_file, name=None, no_progress_bar=False):
         assert from_info.isin(self.path_info)
         with self.ssh(self.path_info) as ssh:
@@ -247,18 +275,6 @@ class SSHRemote(BaseRemote):
                 no_progress_bar=no_progress_bar,
             )
 
-    @contextmanager
-    def open(self, path_info, mode="r", encoding=None):
-        assert mode in {"r", "rt", "rb", "wb"}
-
-        with self.ssh(path_info) as ssh, closing(
-            ssh.sftp.open(path_info.path, mode)
-        ) as fd:
-            if "b" in mode:
-                yield fd
-            else:
-                yield io.TextIOWrapper(fd, encoding=encoding)
-
     def list_cache_paths(self, prefix=None, progress_callback=None):
         if prefix:
             root = posixpath.join(self.path_info.path, prefix[:2])
@@ -275,15 +291,6 @@ class SSHRemote(BaseRemote):
             else:
                 yield from ssh.walk_files(root)
 
-    def walk_files(self, path_info):
-        with self.ssh(path_info) as ssh:
-            for fname in ssh.walk_files(path_info.path):
-                yield path_info.replace(path=fname)
-
-    def makedirs(self, path_info):
-        with self.ssh(path_info) as ssh:
-            ssh.makedirs(path_info.path)
-
     def batch_exists(self, path_infos, callback):
         def _exists(chunk_and_channel):
             chunk, channel = chunk_and_channel

dvc/repo/__init__.py

@@ -527,14 +527,13 @@ class Repo:
 
         for root, dirs, files in self.tree.walk(self.root_dir):
             for file_name in filter(is_valid_filename, files):
-                path = os.path.join(root, file_name)
-                stages.extend(self.get_stages(path))
+                new_stages = self.get_stages(os.path.join(root, file_name))
+                stages.extend(new_stages)
                 outs.update(
                     out.fspath
-                    for stage in stages
-                    for out in (
-                        out for out in stage.outs if out.scheme == "local"
-                    )
+                    for stage in new_stages
+                    for out in stage.outs
+                    if out.scheme == "local"
                 )
             dirs[:] = [d for d in dirs if os.path.join(root, d) not in outs]
         return stages

dvc/repo/add.py

@@ -21,7 +21,9 @@ logger = logging.getLogger(__name__)
 
 @locked
 @scm_context
-def add(repo, targets, recursive=False, no_commit=False, fname=None):
+def add(
+    repo, targets, recursive=False, no_commit=False, fname=None, external=False
+):
     if recursive and fname:
         raise RecursiveAddingWhileUsingFilename()
 
@@ -52,7 +54,9 @@ def add(repo, targets, recursive=False, no_commit=False, fname=None):
                     )
                 )
 
-            stages = _create_stages(repo, sub_targets, fname, pbar=pbar)
+            stages = _create_stages(
+                repo, sub_targets, fname, pbar=pbar, external=external
+            )
 
             try:
                 repo.check_modified_graph(stages)
@@ -120,7 +124,7 @@ def _find_all_targets(repo, target, recursive):
     return [target]
 
 
-def _create_stages(repo, targets, fname, pbar=None):
+def _create_stages(repo, targets, fname, pbar=None, external=False):
     from dvc.stage import Stage, create_stage
 
     stages = []
@@ -132,7 +136,14 @@ def _create_stages(repo, targets, fname, pbar=None):
         unit="file",
     ):
         path, wdir, out = resolve_paths(repo, out)
-        stage = create_stage(Stage, repo, fname or path, wdir=wdir, outs=[out])
+        stage = create_stage(
+            Stage,
+            repo,
+            fname or path,
+            wdir=wdir,
+            outs=[out],
+            external=external,
+        )
         if stage:
             Dvcfile(repo, stage.path).remove_with_prompt(force=True)
 

dvc/repo/plots/show.py

@@ -4,7 +4,7 @@ import os
 
 from funcy import first, last, project
 
-from dvc.exceptions import DvcException
+from dvc.exceptions import DvcException, NoPlotsError
 from dvc.repo import locked
 from dvc.schema import PLOT_PROPS
 
@@ -108,6 +108,9 @@ def _show(repo, datafile=None, revs=None, props=None):
     if revs is None:
         revs = ["workspace"]
 
+    if props is None:
+        props = {}
+
     if not datafile and not props.get("template"):
         raise NoDataOrTemplateProvided()
 
@@ -201,7 +204,11 @@ def show(repo, targets=None, revs=None, props=None) -> dict:
         if targets:
             raise NoMetricInHistoryError(", ".join(targets))
 
-        datafile, plot = _show(repo, datafile=None, revs=revs, props=props)
+        try:
+            datafile, plot = _show(repo, datafile=None, revs=revs, props=props)
+        except NoDataOrTemplateProvided:
+            raise NoPlotsError()
+
         return {datafile: plot}
 
     return {

dvc/repo/plots/template.py

@@ -38,8 +38,8 @@ class Template:
     X_ANCHOR = "<DVC_METRIC_X>"
     Y_ANCHOR = "<DVC_METRIC_Y>"
     TITLE_ANCHOR = "<DVC_METRIC_TITLE>"
-    X_TITLE_ANCHOR = "<DVC_METRIC_X_TITLE>"
-    Y_TITLE_ANCHOR = "<DVC_METRIC_Y_TITLE>"
+    X_LABEL_ANCHOR = "<DVC_METRIC_X_LABEL>"
+    Y_LABEL_ANCHOR = "<DVC_METRIC_Y_LABEL>"
 
     def __init__(self, templates_dir):
         self.plot_templates_dir = templates_dir
@@ -119,15 +119,15 @@ class Template:
     @staticmethod
     def _replace_metadata_anchors(result_content, props):
         props.setdefault("title", "")
-        props.setdefault("xlab", props.get("x"))
-        props.setdefault("ylab", props.get("y"))
+        props.setdefault("x_label", props.get("x"))
+        props.setdefault("y_label", props.get("y"))
 
         replace_pairs = [
             (Template.TITLE_ANCHOR, "title"),
             (Template.X_ANCHOR, "x"),
             (Template.Y_ANCHOR, "y"),
-            (Template.X_TITLE_ANCHOR, "xlab"),
-            (Template.Y_TITLE_ANCHOR, "ylab"),
+            (Template.X_LABEL_ANCHOR, "x_label"),
+            (Template.Y_LABEL_ANCHOR, "y_label"),
         ]
         for anchor, key in replace_pairs:
             value = props.get(key)
@@ -170,12 +170,12 @@ class DefaultLinearTemplate(Template):
             "x": {
                 "field": Template.X_ANCHOR,
                 "type": "quantitative",
-                "title": Template.X_TITLE_ANCHOR,
+                "title": Template.X_LABEL_ANCHOR,
             },
             "y": {
                 "field": Template.Y_ANCHOR,
                 "type": "quantitative",
-                "title": Template.Y_TITLE_ANCHOR,
+                "title": Template.Y_LABEL_ANCHOR,
                 "scale": {"zero": False},
             },
             "color": {"field": "rev", "type": "nominal"},
@@ -195,13 +195,13 @@ class DefaultConfusionTemplate(Template):
                 "field": Template.X_ANCHOR,
                 "type": "nominal",
                 "sort": "ascending",
-                "title": Template.X_TITLE_ANCHOR,
+                "title": Template.X_LABEL_ANCHOR,
             },
             "y": {
                 "field": Template.Y_ANCHOR,
                 "type": "nominal",
                 "sort": "ascending",
-                "title": Template.Y_TITLE_ANCHOR,
+                "title": Template.Y_LABEL_ANCHOR,
             },
             "color": {"aggregate": "count", "type": "quantitative"},
             "facet": {"field": "rev", "type": "nominal"},
@@ -220,12 +220,12 @@ class DefaultScatterTemplate(Template):
             "x": {
                 "field": Template.X_ANCHOR,
                 "type": "quantitative",
-                "title": Template.X_TITLE_ANCHOR,
+                "title": Template.X_LABEL_ANCHOR,
             },
             "y": {
                 "field": Template.Y_ANCHOR,
                 "type": "quantitative",
-                "title": Template.Y_TITLE_ANCHOR,
+                "title": Template.Y_LABEL_ANCHOR,
                 "scale": {"zero": False},
             },
             "color": {"field": "rev", "type": "nominal"},

dvc/schema.py

@@ -34,8 +34,8 @@ PLOT_PROPS = {
     BaseOutput.PARAM_PLOT_TEMPLATE: str,
     BaseOutput.PARAM_PLOT_X: str,
     BaseOutput.PARAM_PLOT_Y: str,
-    BaseOutput.PARAM_PLOT_XLAB: str,
-    BaseOutput.PARAM_PLOT_YLAB: str,
+    BaseOutput.PARAM_PLOT_X_LABEL: str,
+    BaseOutput.PARAM_PLOT_Y_LABEL: str,
     BaseOutput.PARAM_PLOT_TITLE: str,
     BaseOutput.PARAM_PLOT_CSV_HEADER: bool,
 }

dvc/stage/__init__.py

@@ -19,6 +19,7 @@ from .utils import (
     check_circular_dependency,
     check_duplicated_arguments,
     check_missing_outputs,
+    check_no_externals,
     check_stage_path,
     compute_md5,
     fill_stage_dependencies,
@@ -52,7 +53,7 @@ def loads_from(cls, repo, path, wdir, data):
     return cls(**kw)
 
 
-def create_stage(cls, repo, path, **kwargs):
+def create_stage(cls, repo, path, external=False, **kwargs):
     from dvc.dvcfile import check_dvc_filename
 
     wdir = os.path.abspath(kwargs.get("wdir", None) or os.curdir)
@@ -63,6 +64,8 @@ def create_stage(cls, repo, path, **kwargs):
 
     stage = loads_from(cls, repo, path, wdir, kwargs)
     fill_stage_outputs(stage, **kwargs)
+    if not external:
+        check_no_externals(stage)
     fill_stage_dependencies(
         stage, **project(kwargs, ["deps", "erepo", "params"])
     )

dvc/stage/cache.py

@@ -102,6 +102,7 @@ class StageCache:
             params=params,
             deps=[dep["path"] for dep in cache.get("deps", [])],
             outs=[out["path"] for out in cache["outs"]],
+            external=True,
         )
         StageLoader.fill_from_lock(stage, cache)
         return stage

dvc/stage/exceptions.py

@@ -58,6 +58,10 @@ class StageCommitError(DvcException):
     pass
 
 
+class StageExternalOutputsError(DvcException):
+    pass
+
+
 class StageUpdateError(DvcException):
     def __init__(self, path):
         super().__init__(

dvc/stage/utils.py

@@ -6,9 +6,10 @@ from dvc import dependency, output
 from dvc.utils.fs import path_isin
 
 from ..remote import LocalRemote, S3Remote
-from ..utils import dict_md5, relpath
+from ..utils import dict_md5, format_link, relpath
 from .exceptions import (
     MissingDataSource,
+    StageExternalOutputsError,
     StagePathNotDirectoryError,
     StagePathNotFoundError,
     StagePathOutsideError,
@@ -68,6 +69,31 @@ def fill_stage_dependencies(stage, deps=None, erepo=None, params=None):
     stage.deps += dependency.loads_params(stage, params or [])
 
 
+def check_no_externals(stage):
+    from urllib.parse import urlparse
+
+    # NOTE: preventing users from accidentally using external outputs. See
+    # https://github.com/iterative/dvc/issues/1545 for more details.
+
+    def _is_external(out):
+        # NOTE: in case of `remote://` notation, the user clearly knows that
+        # this is an advanced feature and so we shouldn't error-out.
+        if out.is_in_repo or urlparse(out.def_path).scheme == "remote":
+            return False
+        return True
+
+    outs = [str(out) for out in stage.outs if _is_external(out)]
+    if not outs:
+        return
+
+    str_outs = ", ".join(outs)
+    link = format_link("https://dvc.org/doc/user-guide/managing-external-data")
+    raise StageExternalOutputsError(
+        f"Output(s) outside of DVC project: {str_outs}. "
+        f"See {link} for more info."
+    )
+
+
 def check_circular_dependency(stage):
     from dvc.exceptions import CircularDependencyError
 

dvc/version.py

@@ -6,7 +6,7 @@
 import os
 import subprocess
 
-_BASE_VERSION = "1.0.0a7"
+_BASE_VERSION = "1.0.0a9"
 
 
 def _generate_version(base_version):

scripts/ci/install.sh

@@ -10,6 +10,8 @@ scriptdir="$(dirname $0)"
 # work or not.
 $scriptdir/retry.sh pip install .[all,tests]
 
+$scriptdir/retry.sh pip install git+https://github.com/iterative/mock-ssh-server.git
+
 git config --global user.email "[email protected]"
 git config --global user.name "DVC Tester"
 

scripts/completion/dvc.bash

@@ -12,7 +12,7 @@ _dvc_commands='add cache checkout commit config dag destroy diff fetch freeze \
 _dvc_options='-h --help -V --version'
 _dvc_global_options='-h --help -q --quiet -v --verbose'
 
-_dvc_add='-R --recursive -f --file --no-commit'
+_dvc_add='-R --recursive -f --file --no-commit --external'
 _dvc_add_COMPGEN=_dvc_compgen_files
 _dvc_cache='dir'
 _dvc_cache_dir=' --global --system --local -u --unset'
@@ -47,8 +47,9 @@ _dvc_move_COMPGEN=_dvc_compgen_files
 _dvc_params='diff'
 _dvc_params_diff='--all --show-json --show-md --no-path'
 _dvc_plots='show diff'
-_dvc_plots_show='-t --template -o --out -x -y --show-json --no-csv-header --title --xlab --ylab'
-_dvc_plots_diff='-t --template --targets -o --out -x -y --show-json --no-csv-header --title --xlab --ylab'
+_dvc_plots_show='-t --template -o --out -x -y --show-vega --no-csv-header --title --x-label --y-label'
+_dvc_plots_diff='-t --template --targets -o --out -x -y --show-vega --no-csv-header --title --x-label --y-label'
+_dvc_plots_modify='-t --template -x -y --no-csv-header --title --x-label --y-label'
 _dvc_pull='-j --jobs -r --remote -a --all-branches -T --all-tags -f --force -d --with-deps -R --recursive'
 _dvc_pull_COMPGEN=_dvc_compgen_DVCFiles
 _dvc_push='-j --jobs -r --remote -a --all-branches -T --all-tags -d --with-deps -R --recursive'
@@ -65,7 +66,7 @@ _dvc_remove_COMPGEN=_dvc_compgen_DVCFiles
 _dvc_repro='-f --force -s --single-item -c --cwd -m --metrics --dry -i --interactive -p --pipeline -P --all-pipelines --no-run-cache --force-downstream --no-commit -R --recursive --downstream'
 _dvc_repro_COMPGEN=_dvc_compgen_DVCFiles
 _dvc_root=''
-_dvc_run='--no-exec -f --file -d --deps -o --outs -O --outs-no-cache --outs-persist --outs-persist-no-cache -m --metrics -M --metrics-no-cache --overwrite-dvcfile --no-run-cache --no-commit -w --wdir'
+_dvc_run='--no-exec -f --file -d --deps -o --outs -O --outs-no-cache --outs-persist --outs-persist-no-cache -m --metrics -M --metrics-no-cache --overwrite-dvcfile --no-run-cache --no-commit -w --wdir --external'
 _dvc_run_COMPGEN=_dvc_compgen_DVCFiles
 _dvc_status='-j --jobs -r --remote -a --all-branches -T --all-tags -d --with-deps -c --cloud'
 _dvc_status_COMPGEN=_dvc_compgen_DVCFiles

scripts/completion/dvc.zsh

@@ -66,6 +66,7 @@ _dvc_add=(
   {-R,--recursive}"[Recursively add each file under the directory.]"
   "--no-commit[Don't put files/directories into cache.]"
   {-f,--file}"[Specify name of the DVC-file it generates.]:File:_files"
+  "--external[Allow targets that are outside of the DVC project.]"
   "1:File:_files"
 )
 
@@ -216,7 +217,7 @@ _dvc_push=(
 )
 
 _dvc_plots=(
-  "1:Sub command:(show diff)"
+  "1:Sub command:(show diff modify)"
 )
 
 _dvc_remote=(
@@ -267,6 +268,7 @@ _dvc_run=(
   "--no-commit[Don't put files/directories into cache.]"
   "--outs-persist[Declare output file or directory that will not be removed upon repro.]:Output persistent:_files"
   "--outs-persist-no-cache[Declare output file or directory that will not be removed upon repro (do not put into DVC cache).]:Output persistent regular:_files"
+  "--external[Allow outputs that are outside of the DVC project.]"
 )
 
 _dvc_status=(

setup.py

@@ -121,7 +121,7 @@ tests_requirements = [
     "flake8-docstrings",
     "pydocstyle<4.0",
     "jaraco.windows==3.9.2",
-    "mock-ssh-server@git+https://github.com/iterative/mock-ssh-server.git",
+    "mock-ssh-server",
     "moto==1.3.14.dev464",
     "rangehttpserver==1.2.0",
     "beautifulsoup4==4.4.0",

tests/func/plots/test_plots.py

@@ -14,7 +14,6 @@ from dvc.repo.plots.data import (
     PlotData,
     PlotMetricTypeError,
 )
-from dvc.repo.plots.show import NoDataOrTemplateProvided
 from dvc.repo.plots.template import (
     NoDataForTemplateError,
     NoFieldInDataError,
@@ -51,8 +50,8 @@ def test_plot_csv_one_column(tmp_dir, scm, dvc, run_copy_metrics):
 
     props = {
         "csv_header": False,
-        "xlab": "x_title",
-        "ylab": "y_title",
+        "x_label": "x_title",
+        "y_label": "y_title",
         "title": "mytitle",
     }
     plot_string = dvc.plots.show(props=props)["metric.csv"]
@@ -457,8 +456,10 @@ def test_plot_override_specified_data_source(
     assert plot_content["encoding"]["y"]["field"] == "b"
 
 
-def test_should_raise_on_no_template_and_datafile(tmp_dir, dvc):
-    with pytest.raises(NoDataOrTemplateProvided):
+def test_no_plots(tmp_dir, dvc):
+    from dvc.exceptions import NoPlotsError
+
+    with pytest.raises(NoPlotsError):
         dvc.plots.show()
 
 

tests/func/test_add.py

@@ -21,7 +21,7 @@ from dvc.exceptions import (
 )
 from dvc.main import main
 from dvc.output.base import OutputAlreadyTrackedError, OutputIsStageFileError
-from dvc.remote import LocalRemote
+from dvc.remote.local import LocalRemote, LocalRemoteTree
 from dvc.repo import Repo as DvcRepo
 from dvc.stage import Stage
 from dvc.system import System
@@ -187,11 +187,16 @@ def test_add_file_in_dir(tmp_dir, dvc):
 
 class TestAddExternalLocalFile(TestDvc):
     def test(self):
+        from dvc.stage.exceptions import StageExternalOutputsError
+
         dname = TestDvc.mkdtemp()
         fname = os.path.join(dname, "foo")
         shutil.copyfile(self.FOO, fname)
 
-        stages = self.dvc.add(fname)
+        with self.assertRaises(StageExternalOutputsError):
+            self.dvc.add(fname)
+
+        stages = self.dvc.add(fname, external=True)
         self.assertEqual(len(stages), 1)
         stage = stages[0]
         self.assertNotEqual(stage, None)
@@ -592,7 +597,7 @@ def test_should_not_checkout_when_adding_cached_copy(tmp_dir, dvc, mocker):
 
     shutil.copy("bar", "foo")
 
-    copy_spy = mocker.spy(dvc.cache.local, "copy")
+    copy_spy = mocker.spy(dvc.cache.local.tree, "copy")
 
     dvc.add("foo")
 
@@ -618,7 +623,7 @@ def test_should_relink_on_repeated_add(
     tmp_dir.dvc_gen({"foo": "foo", "bar": "bar"})
 
     os.remove("foo")
-    getattr(dvc.cache.local, link)(PathInfo("bar"), PathInfo("foo"))
+    getattr(dvc.cache.local.tree, link)(PathInfo("bar"), PathInfo("foo"))
 
     dvc.cache.local.cache_types = [new_link]
 
@@ -698,7 +703,7 @@ def test_add_empty_files(tmp_dir, dvc, link):
 def test_add_optimization_for_hardlink_on_empty_files(tmp_dir, dvc, mocker):
     dvc.cache.local.cache_types = ["hardlink"]
     tmp_dir.gen({"foo": "", "bar": "", "lorem": "lorem", "ipsum": "ipsum"})
-    m = mocker.spy(LocalRemote, "is_hardlink")
+    m = mocker.spy(LocalRemoteTree, "is_hardlink")
     stages = dvc.add(["foo", "bar", "lorem", "ipsum"])
 
     assert m.call_count == 1

tests/func/test_checkout.py

@@ -763,12 +763,12 @@ def test_checkout_for_external_outputs(tmp_dir, dvc):
         Bucket=remote.path_info.bucket, Key=file_path.path, Body="foo"
     )
 
-    dvc.add(str(remote.path_info / "foo"))
+    dvc.add(str(remote.path_info / "foo"), external=True)
 
-    remote.remove(file_path)
+    remote.tree.remove(file_path)
     stats = dvc.checkout(force=True)
     assert stats == {**empty_checkout, "added": [str(file_path)]}
-    assert remote.exists(file_path)
+    assert remote.tree.exists(file_path)
 
     remote.s3.put_object(
         Bucket=remote.path_info.bucket, Key=file_path.path, Body="foo\nfoo"

tests/func/test_data_cloud.py

@@ -189,26 +189,22 @@ class TestS3Remote(S3, TestDataCloudBase):
         return S3Remote
 
 
-def setup_gdrive_cloud(remote_url, dvc):
-    config = copy.deepcopy(TEST_CONFIG)
-    config["remote"][TEST_REMOTE] = {
-        "url": remote_url,
-        "gdrive_service_account_email": "test",
-        "gdrive_service_account_p12_file_path": "test.p12",
-        "gdrive_use_service_account": True,
-    }
-
-    dvc.config = config
-    remote = DataCloud(dvc).get_remote()
-    remote._gdrive_create_dir("root", remote.path_info.path)
-
-
 class TestGDriveRemote(GDrive, TestDataCloudBase):
     def _setup_cloud(self):
         self._ensure_should_run()
 
-        setup_gdrive_cloud(self.get_url(), self.dvc)
+        url = self.get_url()
+        self.create_dir(self.dvc, url)
 
+        config = copy.deepcopy(TEST_CONFIG)
+        config["remote"][TEST_REMOTE] = {
+            "url": url,
+            "gdrive_service_account_email": "test",
+            "gdrive_service_account_p12_file_path": "test.p12",
+            "gdrive_use_service_account": True,
+        }
+
+        self.dvc.config = config
         self.cloud = DataCloud(self.dvc)
         remote = self.cloud.get_remote()
         self.assertIsInstance(remote, self._get_cloud_class())
@@ -421,12 +417,11 @@ class TestS3RemoteCLI(S3, TestDataCloudCLIBase):
 
 
 class TestGDriveRemoteCLI(GDrive, TestDataCloudCLIBase):
-    def _setup_cloud(self):
-        setup_gdrive_cloud(self.get_url(), self.dvc)
-
     def _test(self):
         url = self.get_url()
 
+        self.create_dir(self.dvc, url)
+
         self.main(["remote", "add", TEST_REMOTE, url])
         self.main(
             [
@@ -434,7 +429,7 @@ class TestGDriveRemoteCLI(GDrive, TestDataCloudCLIBase):
                 "modify",
                 TEST_REMOTE,
                 "gdrive_service_account_email",
-                "modified",
+                "test",
             ]
         )
         self.main(
@@ -443,7 +438,7 @@ class TestGDriveRemoteCLI(GDrive, TestDataCloudCLIBase):
                 "modify",
                 TEST_REMOTE,
                 "gdrive_service_account_p12_file_path",
-                "modified.p12",
+                "test.p12",
             ]
         )
         self.main(

tests/func/test_gc.py

@@ -8,7 +8,7 @@ from git import Repo
 
 from dvc.exceptions import CollectCacheError
 from dvc.main import main
-from dvc.remote.local import LocalRemote
+from dvc.remote.local import LocalRemote, LocalRemoteTree
 from dvc.repo import Repo as DvcRepo
 from dvc.utils.fs import remove
 from tests.basic_env import TestDir, TestDvcGit
@@ -153,7 +153,7 @@ class TestGCMultipleDvcRepos(TestDvcGit):
         cwd = os.getcwd()
         os.chdir(self.additional_path)
         # ADD FILE ONLY IN SECOND PROJECT
-        fname = os.path.join(self.additional_path, "only_in_second")
+        fname = "only_in_second"
         with open(fname, "w+") as fobj:
             fobj.write("only in additional repo")
 
@@ -161,7 +161,7 @@ class TestGCMultipleDvcRepos(TestDvcGit):
         self.assertEqual(len(stages), 1)
 
         # ADD FILE IN SECOND PROJECT THAT IS ALSO IN MAIN PROJECT
-        fname = os.path.join(self.additional_path, "in_both")
+        fname = "in_both"
         with open(fname, "w+") as fobj:
             fobj.write("in both repos")
 
@@ -321,7 +321,9 @@ def test_gc_cloud_remove_order(tmp_dir, scm, dvc, tmp_path_factory, mocker):
     dvc.remove(dir2.relpath)
     dvc.gc(workspace=True)
 
-    mocked_remove = mocker.patch.object(LocalRemote, "remove", autospec=True)
+    mocked_remove = mocker.patch.object(
+        LocalRemoteTree, "remove", autospec=True
+    )
     dvc.gc(workspace=True, cloud=True)
     assert len(mocked_remove.mock_calls) == 8
     # dir (and unpacked dir) should be first 4 checksums removed from

tests/func/test_get.py

@@ -101,7 +101,8 @@ def test_get_full_dvc_path(tmp_dir, erepo_dir, tmp_path_factory):
     external_data.write_text("ext_data")
 
     with erepo_dir.chdir():
-        erepo_dir.dvc_add(os.fspath(external_data), commit="add external data")
+        erepo_dir.dvc.add(os.fspath(external_data), external=True)
+        erepo_dir.scm_add("ext_data.dvc", commit="add external data")
 
     Repo.get(
         os.fspath(erepo_dir), os.fspath(external_data), "ext_data_imported"

tests/func/test_ignore.py

@@ -140,7 +140,7 @@ def test_match_nested(tmp_dir, dvc):
     )
 
     remote = LocalRemote(dvc, {})
-    result = {os.fspath(f) for f in remote.walk_files(".")}
+    result = {os.fspath(f) for f in remote.tree.walk_files(".")}
     assert result == {".dvcignore", "foo"}
 
 
@@ -150,7 +150,7 @@ def test_ignore_external(tmp_dir, scm, dvc, tmp_path_factory):
     ext_dir.gen({"y.backup": "y", "tmp": "ext tmp"})
 
     remote = LocalRemote(dvc, {})
-    result = {relpath(f, ext_dir) for f in remote.walk_files(ext_dir)}
+    result = {relpath(f, ext_dir) for f in remote.tree.walk_files(ext_dir)}
     assert result == {"y.backup", "tmp"}
 
 

tests/func/test_remote.py

@@ -194,9 +194,13 @@ def test_partial_push_n_pull(tmp_dir, dvc, tmp_path_factory, setup_remote):
         assert upload_error_info.value.amount == 3
 
         remote = dvc.cloud.get_remote("upstream")
-        assert not remote.exists(remote.checksum_to_path_info(foo.checksum))
-        assert remote.exists(remote.checksum_to_path_info(bar.checksum))
-        assert not remote.exists(remote.checksum_to_path_info(baz.checksum))
+        assert not remote.tree.exists(
+            remote.checksum_to_path_info(foo.checksum)
+        )
+        assert remote.tree.exists(remote.checksum_to_path_info(bar.checksum))
+        assert not remote.tree.exists(
+            remote.checksum_to_path_info(baz.checksum)
+        )
 
     # Push everything and delete local cache
     dvc.push()
@@ -273,6 +277,29 @@ def test_remote_modify_validation(dvc):
     assert unsupported_config not in config[f'remote "{remote_name}"']
 
 
+def test_remote_modify_unset(dvc):
+    assert main(["remote", "add", "-d", "myremote", "gdrive://test/test"]) == 0
+    config = configobj.ConfigObj(dvc.config.files["repo"])
+    assert config['remote "myremote"'] == {"url": "gdrive://test/test"}
+
+    assert (
+        main(["remote", "modify", "myremote", "gdrive_client_id", "something"])
+        == 0
+    )
+    config = configobj.ConfigObj(dvc.config.files["repo"])
+    assert config['remote "myremote"'] == {
+        "url": "gdrive://test/test",
+        "gdrive_client_id": "something",
+    }
+
+    assert (
+        main(["remote", "modify", "myremote", "gdrive_client_id", "--unset"])
+        == 0
+    )
+    config = configobj.ConfigObj(dvc.config.files["repo"])
+    assert config['remote "myremote"'] == {"url": "gdrive://test/test"}
+
+
 def test_remote_modify_default(dvc):
     remote_repo = "repo_level"
     remote_local = "local_level"

tests/func/test_repro.py

@@ -963,6 +963,7 @@ class TestReproExternalBase(SingleStageRun, TestDvc):
             deps=[out_foo_path],
             cmd=self.cmd(foo_path, bar_path),
             name="external-base",
+            external=True,
         )
 
         self.assertEqual(self.dvc.status([cmd_stage.addressing]), {})

tests/func/test_s3.py

@@ -5,7 +5,7 @@ import moto.s3.models as s3model
 import pytest
 from moto import mock_s3
 
-from dvc.remote.s3 import S3Remote
+from dvc.remote.s3 import S3Remote, S3RemoteTree
 from tests.remotes import S3
 
 # from https://github.com/spulec/moto/blob/v1.3.5/tests/test_s3/test_s3.py#L40
@@ -42,7 +42,7 @@ def test_copy_singlepart_preserve_etag():
     s3.create_bucket(Bucket=from_info.bucket)
     s3.put_object(Bucket=from_info.bucket, Key=from_info.path, Body="data")
 
-    S3Remote._copy(s3, from_info, to_info, {})
+    S3RemoteTree._copy(s3, from_info, to_info, {})
 
 
 @mock_s3
@@ -58,15 +58,15 @@ def test_link_created_on_non_nested_path(base_info, tmp_dir, dvc, scm):
     )
     remote.link(base_info / "from", base_info / "to")
 
-    assert remote.exists(base_info / "from")
-    assert remote.exists(base_info / "to")
+    assert remote.tree.exists(base_info / "from")
+    assert remote.tree.exists(base_info / "to")
 
 
 @mock_s3
 def test_makedirs_doesnot_try_on_top_level_paths(tmp_dir, dvc, scm):
     base_info = S3Remote.path_cls("s3://bucket/")
     remote = S3Remote(dvc, {"url": str(base_info)})
-    remote.makedirs(base_info)
+    remote.tree.makedirs(base_info)
 
 
 def _upload_multipart(s3, Bucket, Key):
@@ -107,4 +107,4 @@ def test_copy_multipart_preserve_etag():
     s3 = boto3.client("s3")
     s3.create_bucket(Bucket=from_info.bucket)
     _upload_multipart(s3, from_info.bucket, from_info.path)
-    S3Remote._copy(s3, from_info, to_info, {})
+    S3RemoteTree._copy(s3, from_info, to_info, {})

tests/remotes.py

@@ -141,10 +141,21 @@ class GDrive:
     def should_test():
         return os.getenv(GDriveRemote.GDRIVE_CREDENTIALS_DATA) is not None
 
-    def get_url(self):
-        if not getattr(self, "_remote_url", None):
-            self._remote_url = "gdrive://root/" + str(uuid.uuid4())
-        return self._remote_url
+    @staticmethod
+    def create_dir(dvc, url):
+        config = {
+            "url": url,
+            "gdrive_service_account_email": "test",
+            "gdrive_service_account_p12_file_path": "test.p12",
+            "gdrive_use_service_account": True,
+        }
+        remote = GDriveRemote(dvc, config)
+        remote._gdrive_create_dir("root", remote.path_info.path)
+
+    @staticmethod
+    def get_url():
+        # NOTE: `get_url` should always return new random url
+        return "gdrive://root/" + str(uuid.uuid4())
 
 
 class Azure:

tests/unit/command/test_plots.py

@@ -20,9 +20,9 @@ def test_metrics_diff(dvc, mocker):
             "y_field",
             "--title",
             "my_title",
-            "--xlab",
+            "--x-label",
             "x_title",
-            "--ylab",
+            "--y-label",
             "y_title",
             "HEAD",
             "tag1",
@@ -47,8 +47,8 @@ def test_metrics_diff(dvc, mocker):
             "x": "x_field",
             "y": "y_field",
             "title": "my_title",
-            "xlab": "x_title",
-            "ylab": "y_title",
+            "x_label": "x_title",
+            "y_label": "y_title",
         },
     )
 

tests/unit/command/test_run.py

@@ -39,6 +39,7 @@ def test_run(mocker, dvc):
             "file:param1,param2",
             "--params",
             "param3",
+            "--external",
             "command",
         ]
     )
@@ -70,6 +71,7 @@ def test_run(mocker, dvc):
         cmd="command",
         name="nam",
         single_stage=False,
+        external=True,
     )
 
 
@@ -99,6 +101,7 @@ def test_run_args_from_cli(mocker, dvc):
         cmd="echo foo",
         name=None,
         single_stage=False,
+        external=False,
     )
 
 
@@ -128,4 +131,5 @@ def test_run_args_with_spaces(mocker, dvc):
         cmd='echo "foo bar"',
         name=None,
         single_stage=False,
+        external=False,
     )

tests/unit/dependency/test_local.py

@@ -15,6 +15,6 @@ class TestLocalDependency(TestDvc):
 
     def test_save_missing(self):
         d = self._get_dependency()
-        with mock.patch.object(d.remote, "exists", return_value=False):
+        with mock.patch.object(d.remote.tree, "exists", return_value=False):
             with self.assertRaises(d.DoesNotExistError):
                 d.save()

tests/unit/output/test_local.py

@@ -19,7 +19,7 @@ class TestLocalOutput(TestDvc):
 
     def test_save_missing(self):
         o = self._get_output()
-        with patch.object(o.remote, "exists", return_value=False):
+        with patch.object(o.remote.tree, "exists", return_value=False):
             with self.assertRaises(o.DoesNotExistError):
                 o.save()
 

tests/unit/remote/ssh/test_ssh.py

@@ -207,5 +207,5 @@ def test_hardlink_optimization(dvc, tmp_dir, ssh_server):
     else:
         link_path = to_info.path
 
-    remote.hardlink(from_info, to_info)
+    remote.tree.hardlink(from_info, to_info)
     assert not System.is_hardlink(link_path)

tests/unit/remote/test_azure.py

@@ -41,7 +41,7 @@ def test_get_file_checksum(tmp_dir):
     remote = AzureRemote(None, {})
     to_info = remote.path_cls(Azure.get_url())
     remote.upload(PathInfo("foo"), to_info)
-    assert remote.exists(to_info)
+    assert remote.tree.exists(to_info)
     checksum = remote.get_file_checksum(to_info)
     assert checksum
     assert isinstance(checksum, str)

tests/unit/remote/test_base.py

@@ -33,12 +33,12 @@ def test_cmd_error(dvc):
     err = "sed: expression #1, char 2: extra characters after command"
 
     with mock.patch.object(
-        REMOTE_CLS,
+        REMOTE_CLS.TREE_CLS,
         "remove",
         side_effect=RemoteCmdError("base", cmd, ret, err),
     ):
         with pytest.raises(RemoteCmdError):
-            REMOTE_CLS(dvc, config).remove("file")
+            REMOTE_CLS(dvc, config).tree.remove("file")
 
 
 @mock.patch.object(BaseRemote, "_cache_checksums_traverse")

tests/unit/remote/test_local.py

@@ -34,7 +34,7 @@ def test_status_download_optimization(mocker, dvc):
 @pytest.mark.parametrize("link_name", ["hardlink", "symlink"])
 def test_is_protected(tmp_dir, dvc, link_name):
     remote = LocalRemote(dvc, {})
-    link_method = getattr(remote, link_name)
+    link_method = getattr(remote.tree, link_name)
 
     (tmp_dir / "foo").write_text("foo")
 

tests/unit/remote/test_remote.py

@@ -37,5 +37,5 @@ def test_makedirs_not_create_for_top_level_path(remote_cls, dvc, mocker):
     # we use remote clients with same name as scheme to interact with remote
     mocker.patch.object(remote_cls, remote.scheme, mocked_client)
 
-    remote.makedirs(remote.path_info)
+    remote.tree.makedirs(remote.path_info)
     assert not mocked_client.called

tests/unit/remote/test_remote_tree.py

@@ -48,7 +48,7 @@ def test_isdir(remote):
     ]
 
     for expected, path in test_cases:
-        assert remote.isdir(remote.path_info / path) == expected
+        assert remote.tree.isdir(remote.path_info / path) == expected
 
 
 @pytest.mark.parametrize("remote", remotes, indirect=True)
@@ -68,7 +68,7 @@ def test_exists(remote):
     ]
 
     for expected, path in test_cases:
-        assert remote.exists(remote.path_info / path) == expected
+        assert remote.tree.exists(remote.path_info / path) == expected
 
 
 @pytest.mark.parametrize("remote", remotes, indirect=True)
@@ -83,7 +83,7 @@ def test_walk_files(remote):
         remote.path_info / "data/subdir/empty_file",
     ]
 
-    assert list(remote.walk_files(remote.path_info / "data")) == files
+    assert list(remote.tree.walk_files(remote.path_info / "data")) == files
 
 
 @pytest.mark.parametrize("remote", [S3Mocked], indirect=True)
@@ -96,7 +96,7 @@ def test_copy_preserve_etag_across_buckets(remote, dvc):
     from_info = remote.path_info / "foo"
     to_info = another.path_info / "foo"
 
-    remote.copy(from_info, to_info)
+    remote.tree.copy(from_info, to_info)
 
     from_etag = S3Remote.get_etag(s3, from_info.bucket, from_info.path)
     to_etag = S3Remote.get_etag(s3, "another", "foo")
@@ -106,12 +106,13 @@ def test_copy_preserve_etag_across_buckets(remote, dvc):
 
 @pytest.mark.parametrize("remote", remotes, indirect=True)
 def test_makedirs(remote):
+    tree = remote.tree
     empty_dir = remote.path_info / "empty_dir" / ""
-    remote.remove(empty_dir)
-    assert not remote.exists(empty_dir)
-    remote.makedirs(empty_dir)
-    assert remote.exists(empty_dir)
-    assert remote.isdir(empty_dir)
+    tree.remove(empty_dir)
+    assert not tree.exists(empty_dir)
+    tree.makedirs(empty_dir)
+    assert tree.exists(empty_dir)
+    assert tree.isdir(empty_dir)
 
 
 @pytest.mark.parametrize("remote", [GCP, S3Mocked], indirect=True)
@@ -133,7 +134,7 @@ def test_isfile(remote):
     ]
 
     for expected, path in test_cases:
-        assert remote.isfile(remote.path_info / path) == expected
+        assert remote.tree.isfile(remote.path_info / path) == expected
 
 
 @pytest.mark.parametrize("remote", remotes, indirect=True)

tests/unit/stage/test_stage.py

@@ -102,3 +102,21 @@ def test_always_changed(dvc):
     with dvc.lock:
         assert stage.changed()
         assert stage.status()["path"] == ["always changed"]
+
+
+def test_external_outs(tmp_path_factory, dvc):
+    from dvc.stage import create_stage
+    from dvc.stage.exceptions import StageExternalOutputsError
+
+    tmp_path = tmp_path_factory.mktemp("external-outs")
+    foo = tmp_path / "foo"
+    foo.write_text("foo")
+
+    with pytest.raises(StageExternalOutputsError):
+        create_stage(Stage, dvc, "path.dvc", outs=[os.fspath(foo)])
+
+    with dvc.config.edit() as conf:
+        conf["remote"]["myremote"] = {"url": os.fspath(tmp_path)}
+
+    create_stage(Stage, dvc, "path.dvc", outs=["remote://myremote/foo"])
+    create_stage(Stage, dvc, "path.dvc", outs=[os.fspath(foo)], external=True)