dvc: bump to 2.0.7

* Use existing level if defined in addLoggingLevel

Fixes https://github.com/iterative/dvc/issues/4546

* Add addLoggingLevel test

* Format fix.

.github/ISSUE_TEMPLATE/bug_report.md

@@ -57,5 +57,8 @@ $ dvc doctor
 **Additional Information (if any):**
 
 <!--
+Please check https://github.com/iterative/dvc/wiki/Debugging-DVC on ways to gather more information regarding the issue.
+
 If applicable, please also provide a `--verbose` output of the command, eg: `dvc add --verbose`.
+If the issue is regarding the performance, please attach the profiling information and the benchmark comparisons.
 -->

.github/workflows/tests.yaml

@@ -94,14 +94,3 @@ jobs:
       with:
         file: ./coverage.xml
         fail_ci_if_error: false
-    - name: Post TAP
-      # Running custom fork with an option, that does not fail CI on error
-      uses: skshetry/flaky-service/packages/action@master
-      if: always() && github.ref == 'refs/heads/master'
-      with:
-        file-path: ${{github.workspace}}/testresults.tap
-        repo-token: ${{ secrets.GITHUB_TOKEN }}
-        os: ${{runner.os}}
-        matrix: ${{toJson(matrix)}}
-        repo-description: ${{github.event.repository.description}}
-        fail_ci_if_error: false

.pylintrc

@@ -15,10 +15,13 @@ ignore-patterns=
 
 # Python code to execute, usually for sys.path manipulation such as
 # pygtk.require().
-init-hook="import sys; \
-           from pathlib import Path; \
-           from pylint.config import find_pylintrc; \
-           sys.path.append(Path(find_pylintrc()).parent / 'tests');"
+init-hook="import sys;
+           from pathlib import Path;
+           from pylint.config import find_pylintrc;
+
+           tests_folder = Path(find_pylintrc()).parent / 'tests'
+           sys.path.append(str(tests_folder));
+           "
 
 
 # Use multiple processes to speed up Pylint. Specifying 0 will auto-detect the
@@ -178,7 +181,10 @@ disable=format,
         # we use it to speedup/optimize
         import-outside-toplevel,
         fixme,
-
+        # we use cause of the exception for error message output
+        # so some of the times we don't use `raise from` to not show the cause
+        # of the error
+        raise-missing-from,
 
 # Enable the message, report, category or checker with the given id(s). You can
 # either give multiple identifier separated by comma (,) or put this option

README.rst

@@ -222,7 +222,7 @@ Citation
 Iterative, *DVC: Data Version Control - Git for Data & Models* (2020)
 `DOI:10.5281/zenodo.012345 <https://doi.org/10.5281/zenodo.3677553>`_.
 
-Barrak, A., Eghan, E.E. and Adams, B. On the Co-evolution of ML Pipelines and Source Code - Empirical Study of DVC Projects, in Proceedings of the 28th IEEE International Conference on Software Analysis, Evolution, and Reengineering, SANER 2021. Hawaii, USA, to appear.
+Barrak, A., Eghan, E.E. and Adams, B. `On the Co-evolution of ML Pipelines and Source Code - Empirical Study of DVC Projects <https://mcis.cs.queensu.ca/publications/2021/saner.pdf>`_ , in Proceedings of the 28th IEEE International Conference on Software Analysis, Evolution, and Reengineering, SANER 2021. Hawaii, USA.
 
 
 .. |Banner| image:: https://dvc.org/img/logo-github-readme.png

dvc/api.py

@@ -4,7 +4,6 @@ from contextlib import _GeneratorContextManager as GCM
 from funcy import reraise
 
 from dvc.exceptions import OutputNotFoundError, PathMissingError
-from dvc.objects.stage import get_hash
 from dvc.path_info import PathInfo
 from dvc.repo import Repo
 
@@ -29,8 +28,8 @@ def get_url(path, repo=None, rev=None, remote=None):
             raise OutputNotFoundError(path, repo)
 
         cloud = metadata.repo.cloud
-        hash_info = get_hash(path_info, _repo.repo_fs, "md5", _repo.odb)
-        return cloud.get_url_for(remote, checksum=hash_info.value)
+        md5 = metadata.repo.dvcfs.info(path_info)["md5"]
+        return cloud.get_url_for(remote, checksum=md5)
 
 
 def open(  # noqa, pylint: disable=redefined-builtin

dvc/checkout.py

@@ -9,8 +9,9 @@ from dvc.exceptions import (
     ConfirmRemoveError,
     DvcException,
 )
-from dvc.objects import ObjectFormatError, check, load
-from dvc.objects.stage import get_hash
+from dvc.objects import check, load
+from dvc.objects.errors import ObjectFormatError
+from dvc.objects.stage import stage
 from dvc.remote.slow_link_detection import (  # type: ignore[attr-defined]
     slow_link_guard,
 )
@@ -30,7 +31,7 @@ def _changed(path_info, fs, obj, cache):
         return True
 
     try:
-        actual = get_hash(path_info, fs, obj.hash_info.name, cache)
+        actual = stage(cache, path_info, fs, obj.hash_info.name).hash_info
     except FileNotFoundError:
         logger.debug("'%s' doesn't exist.", path_info)
         return True
@@ -56,7 +57,7 @@ def _remove(path_info, fs, cache, force=False):
         fs.remove(path_info)
         return
 
-    current = get_hash(path_info, fs, fs.PARAM_CHECKSUM, cache)
+    current = stage(cache, path_info, fs, fs.PARAM_CHECKSUM).hash_info
     try:
         obj = load(cache, current)
         check(cache, obj)
@@ -174,10 +175,10 @@ def _checkout_file(
     return modified
 
 
-def _remove_redundant_files(path_info, fs, dir_info, cache, force):
+def _remove_redundant_files(path_info, fs, obj, cache, force):
     existing_files = set(fs.walk_files(path_info))
 
-    needed_files = {path_info.joinpath(*key) for key, _ in dir_info.items()}
+    needed_files = {path_info.joinpath(*key) for key, _ in obj}
     redundant_files = existing_files - needed_files
     for path in redundant_files:
         _remove(path, fs, cache, force)
@@ -197,11 +198,11 @@ def _checkout_dir(
 
     logger.debug("Linking directory '%s'.", path_info)
 
-    for entry_key, entry_hash_info in obj.hash_info.dir_info.items():
+    for entry_key, entry_obj in obj:
         entry_modified = _checkout_file(
             path_info.joinpath(*entry_key),
             fs,
-            cache.get(entry_hash_info),
+            entry_obj,
             cache,
             force,
             progress_callback,
@@ -211,10 +212,7 @@ def _checkout_dir(
             modified = True
 
     modified = (
-        _remove_redundant_files(
-            path_info, fs, obj.hash_info.dir_info, cache, force
-        )
-        or modified
+        _remove_redundant_files(path_info, fs, obj, cache, force) or modified
     )
 
     fs.repo.state.save(path_info, fs, obj.hash_info)

dvc/command/experiments.py

@@ -514,6 +514,11 @@ class CmdExperimentsRun(CmdRepro):
     def run(self):
         from dvc.command.metrics import _show_metrics
 
+        if self.args.reset and self.args.checkpoint_resume:
+            raise InvalidArgumentError(
+                "--reset and --rev are mutually exclusive."
+            )
+
         if self.args.reset:
             logger.info("Any existing checkpoints will be reset and re-run.")
 

dvc/command/plots.py

@@ -45,13 +45,20 @@ class CmdPlots(CmdBase):
             path = os.path.join(os.getcwd(), path)
 
             write(path, plots)
-
-            logger.info(f"file://{path}")
-
+            url = f"file://{path}"
         except DvcException:
             logger.exception("")
             return 1
 
+        logger.info(url)
+        if self.args.open:
+            import webbrowser
+
+            opened = webbrowser.open(url)
+            if not opened:
+                logger.error("Failed to open. Please try opening it manually.")
+                return 1
+
         return 0
 
 
@@ -229,3 +236,9 @@ def _add_output_arguments(parser):
         default=False,
         help="Show output in Vega format.",
     )
+    parser.add_argument(
+        "--open",
+        action="store_true",
+        default=False,
+        help="Open plot file directly in the browser.",
+    )

dvc/command/stage.py

@@ -255,7 +255,8 @@ def _add_common_args(parser):
         "--checkpoints",
         action="append",
         default=[],
-        help=argparse.SUPPRESS,
+        help="Declare checkpoint output file or directory for 'dvc exp run'. "
+        "Not compatible with 'dvc repro'.",
         metavar="<filename>",
     ).complete = completion.FILE
     parser.add_argument(

dvc/dependency/repo.py

@@ -55,6 +55,7 @@ class RepoDependency(LocalDependency):
                 self.repo.odb.local,
                 path_info,
                 repo.repo_fs,
+                self.repo.odb.local.fs.PARAM_CHECKSUM,
                 follow_subrepos=False,
             ).hash_info
 
@@ -94,7 +95,12 @@ class RepoDependency(LocalDependency):
             except (NoOutputOrStageError, NoRemoteError):
                 pass
             obj = stage(
-                odb, path_info, repo.repo_fs, jobs=jobs, follow_subrepos=False,
+                odb,
+                path_info,
+                repo.repo_fs,
+                odb.fs.PARAM_CHECKSUM,
+                jobs=jobs,
+                follow_subrepos=False,
             )
             save(odb, obj, jobs=jobs)
 

dvc/dir_info.py

@@ -1,117 +0,0 @@
-import posixpath
-from operator import itemgetter
-
-from funcy import cached_property
-from pygtrie import Trie
-
-from .hash_info import HashInfo
-
-
-def _diff(ancestor, other, allow_removed=False):
-    from dictdiffer import diff
-
-    from dvc.exceptions import MergeError
-
-    allowed = ["add"]
-    if allow_removed:
-        allowed.append("remove")
-
-    result = list(diff(ancestor, other))
-    for typ, _, _ in result:
-        if typ not in allowed:
-            raise MergeError(
-                "unable to auto-merge directories with diff that contains "
-                f"'{typ}'ed files"
-            )
-    return result
-
-
-def _merge(ancestor, our, their):
-    import copy
-
-    from dictdiffer import patch
-
-    our_diff = _diff(ancestor, our)
-    if not our_diff:
-        return copy.deepcopy(their)
-
-    their_diff = _diff(ancestor, their)
-    if not their_diff:
-        return copy.deepcopy(our)
-
-    # make sure there are no conflicting files
-    _diff(our, their, allow_removed=True)
-
-    return patch(our_diff + their_diff, ancestor)
-
-
-class DirInfo:
-    PARAM_RELPATH = "relpath"
-
-    def __init__(self):
-        self._dict = {}
-
-    def as_dict(self):
-        return self._dict.copy()
-
-    def add(self, key, hash_info):
-        self.__dict__.pop("trie", None)
-        self._dict[key] = hash_info
-
-    def get(self, key, default=None):
-        return self._dict.get(key, default)
-
-    @cached_property
-    def trie(self):
-        return Trie(self._dict)
-
-    @property
-    def size(self):
-        try:
-            return sum(hash_info.size for _, hash_info in self._dict.items())
-        except TypeError:
-            return None
-
-    @property
-    def nfiles(self):
-        return len(self._dict)
-
-    def items(self):
-        for key, hash_info in self._dict.items():
-            yield key, hash_info
-
-    @classmethod
-    def from_list(cls, lst):
-        ret = DirInfo()
-        for _entry in lst:
-            entry = _entry.copy()
-            relpath = entry.pop(cls.PARAM_RELPATH)
-            parts = tuple(relpath.split(posixpath.sep))
-            hash_info = HashInfo.from_dict(entry)
-            ret.add(parts, hash_info)
-        return ret
-
-    def to_list(self):
-        # Sorting the list by path to ensure reproducibility
-        return sorted(
-            (
-                {
-                    # NOTE: not using hash_info.to_dict() because we don't want
-                    # size/nfiles fields at this point.
-                    hash_info.name: hash_info.value,
-                    self.PARAM_RELPATH: posixpath.sep.join(parts),
-                }
-                for parts, hash_info in self._dict.items()  # noqa: B301
-            ),
-            key=itemgetter(self.PARAM_RELPATH),
-        )
-
-    def merge(self, ancestor, their):
-        merged_dict = _merge(
-            ancestor.as_dict(), self.as_dict(), their.as_dict(),
-        )
-
-        merged = DirInfo()
-        for key, hi in merged_dict.items():
-            merged.add(key, hi)
-        return merged

dvc/fs/azure.py

@@ -48,8 +48,6 @@ class AzureFileSystem(FSSpecWrapper):
         self.path_info = self.PATH_CLS(url)
         self.bucket = self.path_info.bucket
 
-        self.login_method, self.login_info = self._prepare_credentials(config)
-
     def _prepare_credentials(self, config):
         from azure.identity.aio import DefaultAzureCredential
 
@@ -90,7 +88,8 @@ class AzureFileSystem(FSSpecWrapper):
         else:
             login_method = None
 
-        return login_method, login_info
+        self.login_method = login_method
+        return login_info
 
     @cached_property
     def _az_config(self):
@@ -111,7 +110,7 @@ class AzureFileSystem(FSSpecWrapper):
         from azure.core.exceptions import AzureError
 
         try:
-            file_system = AzureBlobFileSystem(**self.login_info)
+            file_system = AzureBlobFileSystem(**self.fs_args)
             if self.bucket not in [
                 container.rstrip("/") for container in file_system.ls("/")
             ]:

dvc/fs/base.py

@@ -283,6 +283,9 @@ class BaseFileSystem:
         self, from_info, to_info, name, no_progress_bar, jobs, **kwargs,
     ):
         from_infos = list(self.walk_files(from_info, **kwargs))
+        if not from_infos:
+            makedirs(to_info, exist_ok=True)
+            return None
         to_infos = (
             to_info / info.relative_to(from_info) for info in from_infos
         )

dvc/fs/dvc.py

@@ -48,9 +48,9 @@ class DvcFileSystem(BaseFileSystem):  # pylint:disable=abstract-method
         # NOTE: use string paths here for performance reasons
         key = tuple(relpath(path_info, out.path_info).split(os.sep))
         out.get_dir_cache(remote=remote)
-        file_hash = out.hash_info.dir_info.get(key)
-        if file_hash:
-            return file_hash
+        obj = out.obj.trie.get(key)
+        if obj:
+            return obj.hash_info
         raise FileNotFoundError
 
     def open(  # type: ignore
@@ -131,21 +131,14 @@ class DvcFileSystem(BaseFileSystem):  # pylint:disable=abstract-method
         # pull dir cache if needed
         out.get_dir_cache(**kwargs)
 
-        dir_cache = out.dir_cache
-        if not dir_cache:
-            raise FileNotFoundError
-
-        from dvc.objects import Tree
-
-        hash_info = Tree.save_dir_info(out.odb, dir_cache)
-        if hash_info != out.hash_info:
+        if not out.obj:
             raise FileNotFoundError
 
     def _add_dir(self, trie, out, **kwargs):
         self._fetch_dir(out, **kwargs)
 
         base = out.path_info.parts
-        for key, _ in out.dir_cache.items():  # noqa: B301
+        for key, _ in out.obj:  # noqa: B301
             trie[base + key] = None
 
     def _walk(self, root, trie, topdown=True, **kwargs):
@@ -241,9 +234,9 @@ class DvcFileSystem(BaseFileSystem):  # pylint:disable=abstract-method
         elif meta.part_of_output:
             (out,) = meta.outs
             key = path_info.relative_to(out.path_info).parts
-            hash_info = out.hash_info.dir_info.trie.get(key)
-            if hash_info:
-                ret["size"] = hash_info.size
-                ret[hash_info.name] = hash_info.value
+            obj = out.obj.trie.get(key)
+            if obj:
+                ret["size"] = obj.size
+                ret[obj.hash_info.name] = obj.hash_info.value
 
         return ret

dvc/fs/fsspec_wrapper.py

@@ -1,5 +1,6 @@
 import os
 import shutil
+from functools import lru_cache
 
 from dvc.progress import Tqdm
 
@@ -8,17 +9,29 @@ from .base import BaseFileSystem
 
 # pylint: disable=no-member
 class FSSpecWrapper(BaseFileSystem):
+    def __init__(self, repo, config):
+        super().__init__(repo, config)
+        self.fs_args = {"skip_instance_cache": True}
+        self.fs_args.update(self._prepare_credentials(config))
+
     def fs(self):
         raise NotImplementedError
 
+    @lru_cache(512)
     def _with_bucket(self, path):
         if isinstance(path, self.PATH_CLS):
             return f"{path.bucket}/{path.path}"
         return path
 
     def _strip_bucket(self, entry):
-        _, entry = entry.split("/", 1)
-        return entry
+        try:
+            bucket, path = entry.split("/", 1)
+        except ValueError:
+            # If there is no path attached, only returns
+            # the bucket (top-level).
+            bucket, path = entry, None
+
+        return path or bucket
 
     def _strip_buckets(self, entries, detail, prefix=None):
         for entry in entries:
@@ -36,11 +49,38 @@ class FSSpecWrapper(BaseFileSystem):
         entries within info() and ls(detail=True) calls"""
         return entry
 
+    def _prepare_credentials(self, config):  # pylint: disable=unused-argument
+        """Prepare the arguments for authentication to the
+        host filesystem"""
+        return {}
+
     def isdir(self, path_info):
-        return self.fs.isdir(self._with_bucket(path_info))
+        if not self.exists(path_info):
+            return False
+
+        # Directory in object storages are interpreted differently
+        # among different fsspec providers, so this logic is a temporary
+        # measure for us to adapt as of now. It checks whether it is a
+        # directory (as in a prefix with contents) or whether it is an empty
+        # file where it's name ends with a forward slash
+        entry = self.info(path_info)
+        return entry["type"] == "directory" or (
+            entry["size"] == 0
+            and entry["type"] == "file"
+            and entry["name"].endswith("/")
+        )
 
     def isfile(self, path_info):
-        return self.fs.isfile(self._with_bucket(path_info))
+        if not self.exists(path_info):
+            return False
+
+        return not self.isdir(path_info)
+
+    def is_empty(self, path_info):
+        entry = self.info(path_info)
+        if entry["type"] == "directory":
+            return not self.fs.ls(self._with_bucket(path_info))
+        return entry["size"] == 0
 
     def open(
         self, path_info, mode="r", **kwargs
@@ -56,6 +96,9 @@ class FSSpecWrapper(BaseFileSystem):
     def ls(
         self, path_info, detail=False, recursive=False
     ):  # pylint: disable=arguments-differ
+        if self.isdir(path_info) and self.is_empty(path_info):
+            return None
+
         path = self._with_bucket(path_info)
         if recursive:
             for root, _, files in self.fs.walk(path, detail=detail):

dvc/fs/gs.py

@@ -22,16 +22,15 @@ class GSFileSystem(FSSpecWrapper):
         url = config.get("url", "gs://")
         self.path_info = self.PATH_CLS(url)
 
-        self.login_info = self._prepare_credentials(config)
-
     def _prepare_credentials(self, config):
-        login_info = {}
+        login_info = {"consistency": None}
         login_info["project"] = config.get("projectname")
         login_info["token"] = config.get("credentialpath")
         return login_info
 
     def _entry_hook(self, entry):
-        entry["etag"] = base64.b64decode(entry["etag"]).hex()
+        if "etag" in entry:
+            entry["etag"] = base64.b64decode(entry["etag"]).hex()
         return entry
 
     @wrap_prop(threading.Lock())
@@ -39,4 +38,4 @@ class GSFileSystem(FSSpecWrapper):
     def fs(self):
         from gcsfs import GCSFileSystem
 
-        return GCSFileSystem(**self.login_info)
+        return GCSFileSystem(**self.fs_args)

dvc/fs/http.py

@@ -1,6 +1,7 @@
 import logging
 import os.path
 import threading
+from typing import Optional
 
 from funcy import cached_property, memoize, wrap_prop, wrap_with
 
@@ -146,10 +147,9 @@ class HTTPFileSystem(BaseFileSystem):  # pylint:disable=abstract-method
         raise HTTPError(res.status_code, res.reason)
 
     def info(self, path_info):
-        url = path_info.url
-        headers = self._head(url).headers
-        etag = headers.get("ETag") or headers.get("Content-MD5")
-        size = int(headers.get("Content-Length"))
+        resp = self._head(path_info.url)
+        etag = resp.headers.get("ETag") or resp.headers.get("Content-MD5")
+        size = self._content_length(resp)
         return {"etag": etag, "size": size}
 
     def _upload_fobj(self, fobj, to_info):
@@ -195,6 +195,6 @@ class HTTPFileSystem(BaseFileSystem):  # pylint:disable=abstract-method
             )
 
     @staticmethod
-    def _content_length(response):
+    def _content_length(response) -> Optional[int]:
         res = response.headers.get("Content-Length")
         return int(res) if res else None

dvc/fs/memory.py

@@ -13,13 +13,16 @@ class MemoryFileSystem(BaseFileSystem):
         self.fs = MemFS()
 
     def exists(self, path_info, use_dvcignore=True):
-        return self.fs.exists(path_info.path)
+        return self.fs.exists(path_info.fspath)
 
     def open(self, path_info, mode="r", encoding=None, **kwargs):
         return self.fs.open(
             path_info.fspath, mode=mode, encoding=encoding, **kwargs
         )
 
+    def info(self, path_info):
+        return self.fs.info(path_info.fspath)
+
     def stat(self, path_info):
         import os
 

dvc/hash_info.py

@@ -1,11 +1,9 @@
 from collections import OrderedDict
 from dataclasses import dataclass, field
-from typing import Dict, Optional
+from typing import Optional
 
 HASH_DIR_SUFFIX = ".dir"
 
-DirInfo = Dict[str, str]
-
 
 @dataclass
 class HashInfo:
@@ -14,7 +12,6 @@ class HashInfo:
 
     name: Optional[str]
     value: Optional[str]
-    dir_info: Optional[DirInfo] = field(default=None, compare=False)
     size: Optional[int] = field(default=None, compare=False)
     nfiles: Optional[int] = field(default=None, compare=False)
 

dvc/logger.py

@@ -23,14 +23,19 @@ def addLoggingLevel(levelName, levelNum, methodName=None):
     Adds a new logging level to the `logging` module and the
     currently configured logging class.
 
+    Uses the existing numeric levelNum if already defined.
+
     Based on https://stackoverflow.com/questions/2183233
     """
     if methodName is None:
         methodName = levelName.lower()
 
-    assert not hasattr(logging, levelName)
-    assert not hasattr(logging, methodName)
-    assert not hasattr(logging.getLoggerClass(), methodName)
+    # If the level name is already defined as a top-level `logging`
+    # constant, then adopt the existing numeric level.
+    if hasattr(logging, levelName):
+        existingLevelNum = getattr(logging, levelName)
+        assert isinstance(existingLevelNum, int)
+        levelNum = existingLevelNum
 
     def logForLevel(self, message, *args, **kwargs):
         if self.isEnabledFor(levelNum):
@@ -40,10 +45,19 @@ def addLoggingLevel(levelName, levelNum, methodName=None):
     def logToRoot(message, *args, **kwargs):
         logging.log(levelNum, message, *args, **kwargs)
 
-    logging.addLevelName(levelNum, levelName)
-    setattr(logging, levelName, levelNum)
-    setattr(logging.getLoggerClass(), methodName, logForLevel)
-    setattr(logging, methodName, logToRoot)
+    # getLevelName resolves the numeric log level if already defined,
+    # otherwise returns a string
+    if not isinstance(logging.getLevelName(levelName), int):
+        logging.addLevelName(levelNum, levelName)
+
+    if not hasattr(logging, levelName):
+        setattr(logging, levelName, levelNum)
+
+    if not hasattr(logging.getLoggerClass(), methodName):
+        setattr(logging.getLoggerClass(), methodName, logForLevel)
+
+    if not hasattr(logging, methodName):
+        setattr(logging, methodName, logToRoot)
 
 
 class LoggingException(Exception):

dvc/objects/__init__.py

@@ -1,315 +1,28 @@
-import itertools
-import json
 import logging
-from concurrent import futures
 
-from dvc.dir_info import DirInfo
-from dvc.exceptions import DvcException
 from dvc.progress import Tqdm
 
-from .stage import get_hash
+from .tree import Tree
 
 logger = logging.getLogger(__name__)
 
 
-class ObjectError(DvcException):
-    pass
-
-
-class ObjectFormatError(ObjectError):
-    pass
-
-
-class HashFile:
-    def __init__(self, path_info, fs, hash_info):
-        self.path_info = path_info
-        self.fs = fs
-        self.hash_info = hash_info
-
-    def __len__(self):
-        return 1
-
-    def __str__(self):
-        return f"object {self.hash_info}"
-
-    def __bool__(self):
-        return bool(self.hash_info)
-
-    def check(self, odb):
-        actual = get_hash(self.path_info, self.fs, odb.fs.PARAM_CHECKSUM, odb)
-
-        logger.trace(
-            "cache '%s' expected '%s' actual '%s'",
-            self.path_info,
-            self.hash_info,
-            actual,
-        )
-
-        assert actual.name == self.hash_info.name
-        if actual.value.split(".")[0] != self.hash_info.value.split(".")[0]:
-            raise ObjectFormatError(f"{self} is corrupted")
-
-    @classmethod
-    def load(cls, odb, hash_info):
-        return odb.get(hash_info)
-
-    def save(self, odb, **kwargs):
-        odb.add(self.path_info, self.fs, self.hash_info, **kwargs)
-
-
-class File(HashFile):
-    def __init__(self, *args, **kwargs):
-        super().__init__(*args, **kwargs)
-        self.src = None
-
-    @classmethod
-    def stage(cls, odb, path_info, fs, **kwargs):
-        hash_info = get_hash(
-            path_info, fs, odb.fs.PARAM_CHECKSUM, odb, **kwargs
-        )
-        raw = odb.get(hash_info)
-        obj = cls(raw.path_info, raw.fs, hash_info)
-        obj.src = HashFile(path_info, fs, hash_info)
-        return obj
-
-    def save(self, odb, **kwargs):
-        self.src.save(odb, **kwargs)
-
-
-class Tree(HashFile):
-    def __init__(self, *args, **kwargs):
-        super().__init__(*args, **kwargs)
-        self.src_hash_info = None
-        self.src_path_info = None
-        self.src_fs = None
-
-    def __len__(self):
-        return self.hash_info.dir_info.nfiles
-
-    def __iter__(self):
-        yield from self.hash_info.dir_info.items()
-
-    @classmethod
-    def save_dir_info(cls, odb, dir_info, hash_info=None):
-        if hash_info and hash_info.name == odb.fs.PARAM_CHECKSUM:
-            try:
-                odb.check(hash_info)
-                assert hash_info.dir_info == dir_info
-                return hash_info
-            except (FileNotFoundError, ObjectFormatError):
-                pass
-
-        from dvc.fs.memory import MemoryFileSystem
-        from dvc.path_info import PathInfo
-        from dvc.utils import tmp_fname
-
-        fs = MemoryFileSystem(None, {})
-        path_info = PathInfo(tmp_fname(""))
-        with fs.open(path_info, "w") as fobj:
-            json.dump(dir_info.to_list(), fobj, sort_keys=True)
-
-        tmp_info = odb.fs.path_info / tmp_fname("")
-        with fs.open(path_info, "rb") as fobj:
-            odb.fs.upload_fobj(fobj, tmp_info)
-
-        hash_info = get_hash(tmp_info, odb.fs, odb.fs.PARAM_CHECKSUM, odb)
-        hash_info.value += odb.fs.CHECKSUM_DIR_SUFFIX
-        hash_info.dir_info = dir_info
-        hash_info.nfiles = dir_info.nfiles
-
-        odb.add(tmp_info, odb.fs, hash_info)
-
-        return hash_info
-
-    @classmethod
-    def stage(cls, odb, path_info, fs, **kwargs):
-        hash_info = get_hash(
-            path_info, fs, odb.fs.PARAM_CHECKSUM, odb, **kwargs
-        )
-        hi = cls.save_dir_info(odb, hash_info.dir_info, hash_info)
-        hi.size = hash_info.size
-        raw = odb.get(hi)
-        obj = cls(raw.path_info, raw.fs, hi)
-        obj.src_hash_info = hash_info
-        obj.src_path_info = path_info
-        obj.src_fs = fs
-        return obj
-
-    @classmethod
-    def load(cls, odb, hash_info):
-
-        obj = odb.get(hash_info)
-
-        try:
-            with obj.fs.open(obj.path_info, "r") as fobj:
-                raw = json.load(fobj)
-        except ValueError as exc:
-            raise ObjectFormatError(f"{obj} is corrupted") from exc
-
-        if not isinstance(raw, list):
-            logger.error(
-                "dir cache file format error '%s' [skipping the file]",
-                obj.path_info,
-            )
-            raise ObjectFormatError(f"{obj} is corrupted")
-
-        dir_info = DirInfo.from_list(raw)
-        hash_info.dir_info = dir_info
-        hash_info.nfiles = dir_info.nfiles
-
-        return cls(obj.path_info, obj.fs, hash_info)
-
-    def save(self, odb, **kwargs):
-        assert self.src_hash_info.dir_info
-        hi = self.save_dir_info(
-            odb, self.src_hash_info.dir_info, self.hash_info
-        )
-        for entry_key, entry_hash in Tqdm(
-            hi.dir_info.items(),
-            desc="Saving " + self.src_path_info.name,
-            unit="file",
-        ):
-            entry_info = self.src_path_info.joinpath(*entry_key)
-            entry_obj = HashFile(entry_info, self.src_fs, entry_hash)
-            entry_obj.save(odb, **kwargs)
-        self.src_fs.repo.state.save(self.src_path_info, self.src_fs, hi)
-
-    def filter(self, odb, prefix):
-        hash_info = self.hash_info.dir_info.get(prefix)
-        if hash_info:
-            return load(odb, hash_info)
-
-        depth = len(prefix)
-        dir_info = DirInfo()
-        try:
-            for key, value in self.hash_info.dir_info.trie.items(prefix):
-                dir_info.add(key[depth:], value)
-        except KeyError:
-            return None
-
-        return load(odb, self.save_dir_info(odb, dir_info))
-
-
 def save(odb, obj, **kwargs):
-    obj.save(odb, **kwargs)
+    if isinstance(obj, Tree):
+        for _, entry in Tqdm(obj):
+            odb.add(entry.path_info, entry.fs, entry.hash_info, **kwargs)
+    odb.add(obj.path_info, obj.fs, obj.hash_info, **kwargs)
 
 
 def check(odb, obj):
     odb.check(obj.hash_info)
 
     if isinstance(obj, Tree):
-        for _, hash_info in obj:
-            odb.check(hash_info)
+        for _, entry in obj:
+            odb.check(entry.hash_info)
 
 
 def load(odb, hash_info):
     if hash_info.isdir:
         return Tree.load(odb, hash_info)
-    return File.load(odb, hash_info)
-
-
-def _get_dir_size(odb, dir_info):
-    try:
-        return sum(
-            odb.fs.getsize(odb.hash_to_path_info(hi.value))
-            for _, hi in dir_info.items()
-        )
-    except FileNotFoundError:
-        return None
-
-
-def merge(odb, ancestor_info, our_info, their_info):
-    assert our_info
-    assert their_info
-
-    if ancestor_info:
-        ancestor = load(odb, ancestor_info).hash_info.dir_info
-    else:
-        ancestor = DirInfo()
-
-    our = load(odb, our_info).hash_info.dir_info
-    their = load(odb, their_info).hash_info.dir_info
-
-    merged = our.merge(ancestor, their)
-    hash_info = Tree.save_dir_info(odb, merged)
-    hash_info.size = _get_dir_size(odb, merged)
-    return hash_info
-
-
-def _transfer_file(odb, from_fs, from_info):
-    from dvc.utils import tmp_fname
-    from dvc.utils.stream import HashedStreamReader
-
-    tmp_info = odb.fs.path_info / tmp_fname()
-    with from_fs.open(
-        from_info, mode="rb", chunk_size=from_fs.CHUNK_SIZE
-    ) as stream:
-        stream_reader = HashedStreamReader(stream)
-        # Since we don't know the hash beforehand, we'll
-        # upload it to a temporary location and then move
-        # it.
-        odb.fs.upload_fobj(
-            stream_reader,
-            tmp_info,
-            total=from_fs.getsize(from_info),
-            desc=from_info.name,
-        )
-
-    hash_info = stream_reader.hash_info
-    return tmp_info, hash_info
-
-
-def _transfer_directory_contents(odb, from_fs, from_info, jobs, pbar):
-    rel_path_infos = {}
-    from_infos = from_fs.walk_files(from_info)
-
-    def create_tasks(executor, amount):
-        for entry_info in itertools.islice(from_infos, amount):
-            pbar.total += 1
-            task = executor.submit(
-                pbar.wrap_fn(_transfer_file), odb, from_fs, entry_info
-            )
-            rel_path_infos[task] = entry_info.relative_to(from_info)
-            yield task
-
-    pbar.total = 0
-    with futures.ThreadPoolExecutor(max_workers=jobs) as executor:
-        tasks = set(create_tasks(executor, jobs * 5))
-
-        while tasks:
-            done, tasks = futures.wait(
-                tasks, return_when=futures.FIRST_COMPLETED
-            )
-            tasks.update(create_tasks(executor, len(done)))
-            for task in done:
-                yield rel_path_infos.pop(task), task.result()
-
-
-def _transfer_directory(odb, from_fs, from_info, jobs, no_progress_bar=False):
-    dir_info = DirInfo()
-
-    with Tqdm(total=1, unit="Files", disable=no_progress_bar) as pbar:
-        for (
-            entry_info,
-            (entry_tmp_info, entry_hash),
-        ) in _transfer_directory_contents(odb, from_fs, from_info, jobs, pbar):
-            odb.add(entry_tmp_info, odb.fs, entry_hash)
-            dir_info.add(entry_info.parts, entry_hash)
-
-    return Tree.save_dir_info(odb, dir_info)
-
-
-def transfer(odb, from_fs, from_info, jobs=None, no_progress_bar=False):
-    jobs = jobs or min((from_fs.jobs, odb.fs.jobs))
-
-    if from_fs.isdir(from_info):
-        return _transfer_directory(
-            odb,
-            from_fs,
-            from_info,
-            jobs=jobs,
-            no_progress_bar=no_progress_bar,
-        )
-    tmp_info, hash_info = _transfer_file(odb, from_fs, from_info)
-    odb.add(tmp_info, odb.fs, hash_info)
-    return hash_info
+    return odb.get(hash_info)

dvc/objects/db/base.py

@@ -4,10 +4,10 @@ from concurrent.futures import ThreadPoolExecutor
 from copy import copy
 from typing import Optional
 
+from dvc.objects.errors import ObjectFormatError
+from dvc.objects.file import HashFile
 from dvc.progress import Tqdm
 
-from .. import HashFile, ObjectFormatError
-
 logger = logging.getLogger(__name__)
 
 

dvc/objects/db/local.py

@@ -6,7 +6,7 @@ from funcy import cached_property
 from shortuuid import uuid
 
 from dvc.hash_info import HashInfo
-from dvc.objects import ObjectFormatError
+from dvc.objects.errors import ObjectFormatError
 from dvc.path_info import PathInfo
 from dvc.progress import Tqdm
 from dvc.utils import relpath

dvc/objects/errors.py

@@ -0,0 +1,9 @@
+from dvc.exceptions import DvcException
+
+
+class ObjectError(DvcException):
+    pass
+
+
+class ObjectFormatError(ObjectError):
+    pass

dvc/objects/file.py

@@ -0,0 +1,53 @@
+import logging
+
+from .errors import ObjectFormatError
+from .stage import get_file_hash
+
+logger = logging.getLogger(__name__)
+
+
+class HashFile:
+    def __init__(self, path_info, fs, hash_info):
+        self.path_info = path_info
+        self.fs = fs
+        self.hash_info = hash_info
+
+    @property
+    def size(self):
+        if not (self.path_info and self.fs):
+            return None
+        return self.fs.getsize(self.path_info)
+
+    def __len__(self):
+        return 1
+
+    def __str__(self):
+        return f"object {self.hash_info}"
+
+    def __bool__(self):
+        return bool(self.hash_info)
+
+    def __eq__(self, other):
+        if not isinstance(other, HashFile):
+            return False
+        return (
+            self.path_info == other.path_info
+            and self.fs == other.fs
+            and self.hash_info == other.hash_info
+        )
+
+    def check(self, odb):
+        actual = get_file_hash(
+            self.path_info, self.fs, self.hash_info.name, odb.repo.state
+        )
+
+        logger.trace(
+            "cache '%s' expected '%s' actual '%s'",
+            self.path_info,
+            self.hash_info,
+            actual,
+        )
+
+        assert actual.name == self.hash_info.name
+        if actual.value.split(".")[0] != self.hash_info.value.split(".")[0]:
+            raise ObjectFormatError(f"{self} is corrupted")

dvc/objects/stage.py

@@ -9,7 +9,7 @@ from dvc.progress import Tqdm
 from dvc.utils import file_md5
 
 
-def get_file_hash(path_info, fs, name):
+def _get_file_hash(path_info, fs, name):
     info = fs.info(path_info)
     if name in info:
         assert not info[name].endswith(".dir")
@@ -27,20 +27,36 @@ def get_file_hash(path_info, fs, name):
     raise NotImplementedError
 
 
-def _calculate_hashes(path_info, fs, name, state, **kwargs):
-    def _get_file_hash(file_info):
+def get_file_hash(path_info, fs, name, state=None):
+    if state:
         hash_info = state.get(  # pylint: disable=assignment-from-none
-            file_info, fs,
+            path_info, fs,
         )
-        if not hash_info:
-            hash_info = get_file_hash(file_info, fs, name)
-            state.save(file_info, fs, hash_info)
-        return file_info, hash_info
+        if hash_info:
+            return hash_info
+
+    if not fs.exists(path_info):
+        raise FileNotFoundError(
+            errno.ENOENT, os.strerror(errno.ENOENT), path_info
+        )
+
+    hash_info = _get_file_hash(path_info, fs, name)
+
+    if state:
+        assert ".dir" not in hash_info.value
+        state.save(path_info, fs, hash_info)
+
+    return hash_info
+
+
+def _calculate_hashes(path_info, fs, name, state, **kwargs):
+    def _hash(file_info):
+        return file_info, get_file_hash(file_info, fs, name, state)
 
     with Tqdm(
         unit="md5", desc="Computing file/dir hashes (only done once)",
     ) as pbar:
-        worker = pbar.wrap_fn(_get_file_hash)
+        worker = pbar.wrap_fn(_hash)
         with ThreadPoolExecutor(max_workers=fs.hash_jobs) as executor:
             pairs = executor.map(worker, fs.walk_files(path_info, **kwargs))
             return dict(pairs)
@@ -60,14 +76,17 @@ def _iter_hashes(path_info, fs, name, state, **kwargs):
     yield from _calculate_hashes(path_info, fs, name, state, **kwargs).items()
 
 
-def _collect_dir(path_info, fs, name, state, **kwargs):
-    from dvc.dir_info import DirInfo
+def _build_tree(path_info, fs, name, state, **kwargs):
+    from .file import HashFile
+    from .tree import Tree
 
-    dir_info = DirInfo()
+    tree = Tree(None, None, None)
     for fi, hi in _iter_hashes(path_info, fs, name, state, **kwargs):
         if DvcIgnore.DVCIGNORE_FILE == fi.name:
             raise DvcIgnoreInCollectedDirError(fi.parent)
 
+        obj = HashFile(fi, fs, hi)
+
         # NOTE: this is lossy transformation:
         #   "hey\there" -> "hey/there"
         #   "hey/there" -> "hey/there"
@@ -76,31 +95,44 @@ def _collect_dir(path_info, fs, name, state, **kwargs):
         #
         # Yes, this is a BUG, as long as we permit "/" in
         # filenames on Windows and "\" on Unix
-        dir_info.add(fi.relative_to(path_info).parts, hi)
+        tree.add(fi.relative_to(path_info).parts, obj)
 
-    return dir_info
+    tree.digest()
+
+    return tree
 
 
-def get_dir_hash(path_info, fs, name, odb, state, **kwargs):
-    from . import Tree
+def _get_tree_obj(path_info, fs, name, odb, state, **kwargs):
+    from .tree import Tree
 
     value = fs.info(path_info).get(name)
     if value:
         hash_info = HashInfo(name, value)
         try:
-            Tree.load(odb, hash_info)
-            return hash_info
+            return Tree.load(odb, hash_info)
         except FileNotFoundError:
             pass
 
-    dir_info = _collect_dir(path_info, fs, name, state, **kwargs)
-    hash_info = Tree.save_dir_info(fs.repo.odb.local, dir_info)
-    hash_info.size = dir_info.size
-    hash_info.dir_info = dir_info
-    return hash_info
+    tree = _build_tree(path_info, fs, name, state, **kwargs)
+
+    odb.add(tree.path_info, tree.fs, tree.hash_info)
+    if name != "md5":
+        # NOTE: used only for external outputs. Initial reasoning was to be
+        # able to validate .dir files right in the workspace (e.g. check s3
+        # etag), but could be dropped for manual validation with regular md5,
+        # that would be universal for all clouds.
+        raw = odb.get(tree.hash_info)
+        hash_info = get_file_hash(raw.path_info, raw.fs, name, state)
+        tree.hash_info.name = hash_info.name
+        tree.hash_info.value = hash_info.value
+        if not tree.hash_info.value.endswith(".dir"):
+            tree.hash_info.value += ".dir"
+        odb.add(tree.path_info, tree.fs, tree.hash_info)
+
+    return tree
 
 
-def get_hash(path_info, fs, name, odb, **kwargs):
+def stage(odb, path_info, fs, name, **kwargs):
     assert path_info and (
         isinstance(path_info, str) or path_info.scheme == fs.scheme
     )
@@ -125,28 +157,31 @@ def get_hash(path_info, fs, name, odb, **kwargs):
         hash_info = None
 
     if hash_info:
-        if hash_info.isdir:
-            from . import Tree
+        from . import load
+        from .tree import Tree
 
-            # NOTE: loading the fs will restore hash_info.dir_info
-            Tree.load(odb, hash_info)
-        assert hash_info.name == name
-        return hash_info
+        obj = load(odb, hash_info)
+        if isinstance(obj, Tree):
+            obj.hash_info.nfiles = len(obj)
+            for key, entry in obj:
+                entry.fs = fs
+                entry.path_info = path_info.joinpath(*key)
+        else:
+            obj.fs = fs
+            obj.path_info = path_info
+        assert obj.hash_info.name == name
+        obj.hash_info.size = hash_info.size
+        return obj
 
     if fs.isdir(path_info):
-        hash_info = get_dir_hash(path_info, fs, name, odb, state, **kwargs)
+        obj = _get_tree_obj(path_info, fs, name, odb, state, **kwargs)
     else:
-        hash_info = get_file_hash(path_info, fs, name)
+        from .file import HashFile
 
-    if hash_info and fs.exists(path_info):
-        state.save(path_info, fs, hash_info)
+        hash_info = get_file_hash(path_info, fs, name, state)
+        obj = HashFile(path_info, fs, hash_info)
 
-    return hash_info
+    if obj.hash_info and fs.exists(path_info):
+        state.save(path_info, fs, obj.hash_info)
 
-
-def stage(odb, path_info, fs, **kwargs):
-    from . import File, Tree
-
-    if fs.isdir(path_info):
-        return Tree.stage(odb, path_info, fs, **kwargs)
-    return File.stage(odb, path_info, fs, **kwargs)
+    return obj

dvc/objects/transfer.py

@@ -0,0 +1,92 @@
+import itertools
+import logging
+from concurrent import futures
+
+from dvc.progress import Tqdm
+
+from . import save
+from .file import HashFile
+from .tree import Tree
+
+logger = logging.getLogger(__name__)
+
+
+def _transfer_file(odb, from_fs, from_info):
+    from dvc.utils import tmp_fname
+    from dvc.utils.stream import HashedStreamReader
+
+    tmp_info = odb.fs.path_info / tmp_fname()
+    with from_fs.open(
+        from_info, mode="rb", chunk_size=from_fs.CHUNK_SIZE
+    ) as stream:
+        stream_reader = HashedStreamReader(stream)
+        # Since we don't know the hash beforehand, we'll
+        # upload it to a temporary location and then move
+        # it.
+        odb.fs.upload_fobj(
+            stream_reader,
+            tmp_info,
+            total=from_fs.getsize(from_info),
+            desc=from_info.name,
+        )
+
+    hash_info = stream_reader.hash_info
+    return tmp_info, hash_info
+
+
+def _transfer_directory_contents(odb, from_fs, from_info, jobs, pbar):
+    rel_path_infos = {}
+    from_infos = from_fs.walk_files(from_info)
+
+    def create_tasks(executor, amount):
+        for entry_info in itertools.islice(from_infos, amount):
+            pbar.total += 1
+            task = executor.submit(
+                pbar.wrap_fn(_transfer_file), odb, from_fs, entry_info
+            )
+            rel_path_infos[task] = entry_info.relative_to(from_info)
+            yield task
+
+    pbar.total = 0
+    with futures.ThreadPoolExecutor(max_workers=jobs) as executor:
+        tasks = set(create_tasks(executor, jobs * 5))
+
+        while tasks:
+            done, tasks = futures.wait(
+                tasks, return_when=futures.FIRST_COMPLETED
+            )
+            tasks.update(create_tasks(executor, len(done)))
+            for task in done:
+                yield rel_path_infos.pop(task), task.result()
+
+
+def _transfer_directory(odb, from_fs, from_info, jobs, no_progress_bar=False):
+    tree = Tree(None, None, None)
+
+    with Tqdm(total=1, unit="Files", disable=no_progress_bar) as pbar:
+        for (
+            entry_info,
+            (entry_tmp_info, entry_hash),
+        ) in _transfer_directory_contents(odb, from_fs, from_info, jobs, pbar):
+            obj = HashFile(entry_tmp_info, odb.fs, entry_hash)
+            tree.add(entry_info.parts, obj)
+
+    tree.digest()
+    save(odb, tree)
+    return tree.hash_info
+
+
+def transfer(odb, from_fs, from_info, jobs=None, no_progress_bar=False):
+    jobs = jobs or min((from_fs.jobs, odb.fs.jobs))
+
+    if from_fs.isdir(from_info):
+        return _transfer_directory(
+            odb,
+            from_fs,
+            from_info,
+            jobs=jobs,
+            no_progress_bar=no_progress_bar,
+        )
+    tmp_info, hash_info = _transfer_file(odb, from_fs, from_info)
+    odb.add(tmp_info, odb.fs, hash_info)
+    return hash_info

dvc/objects/tree.py

@@ -0,0 +1,211 @@
+import json
+import logging
+import posixpath
+
+from funcy import cached_property
+
+from .errors import ObjectFormatError
+from .file import HashFile
+from .stage import get_file_hash
+
+logger = logging.getLogger(__name__)
+
+
+class Tree(HashFile):
+    PARAM_RELPATH = "relpath"
+
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self._dict = {}
+
+    @cached_property
+    def trie(self):
+        from pygtrie import Trie
+
+        return Trie(self._dict)
+
+    @property
+    def size(self):
+        try:
+            return sum(obj.size for _, obj in self)
+        except TypeError:
+            return None
+
+    def add(self, key, obj):
+        self.__dict__.pop("trie", None)
+        self._dict[key] = obj
+
+    def digest(self):
+        from dvc.fs.memory import MemoryFileSystem
+        from dvc.path_info import PathInfo
+        from dvc.utils import tmp_fname
+
+        memfs = MemoryFileSystem(None, {})
+        path_info = PathInfo(tmp_fname(""))
+        with memfs.open(path_info, "wb") as fobj:
+            fobj.write(self.as_bytes())
+        self.fs = memfs
+        self.path_info = path_info
+        self.hash_info = get_file_hash(path_info, memfs, "md5")
+        self.hash_info.value += ".dir"
+        self.hash_info.size = self.size
+        self.hash_info.nfiles = len(self)
+
+    def __len__(self):
+        return len(self._dict)
+
+    def __iter__(self):
+        yield from self._dict.items()
+
+    def as_dict(self):
+        return self._dict.copy()
+
+    def as_list(self):
+        from operator import itemgetter
+
+        # Sorting the list by path to ensure reproducibility
+        return sorted(
+            (
+                {
+                    # NOTE: not using hash_info.to_dict() because we don't want
+                    # size/nfiles fields at this point.
+                    obj.hash_info.name: obj.hash_info.value,
+                    self.PARAM_RELPATH: posixpath.sep.join(parts),
+                }
+                for parts, obj in self._dict.items()  # noqa: B301
+            ),
+            key=itemgetter(self.PARAM_RELPATH),
+        )
+
+    def as_bytes(self):
+        return json.dumps(self.as_list(), sort_keys=True).encode("utf-8")
+
+    @classmethod
+    def from_list(cls, lst):
+        from dvc.hash_info import HashInfo
+
+        tree = cls(None, None, None)
+        for _entry in lst:
+            entry = _entry.copy()
+            relpath = entry.pop(cls.PARAM_RELPATH)
+            parts = tuple(relpath.split(posixpath.sep))
+            hash_info = HashInfo.from_dict(entry)
+            obj = HashFile(None, None, hash_info)
+            tree.add(parts, obj)
+        return tree
+
+    @classmethod
+    def load(cls, odb, hash_info):
+
+        obj = odb.get(hash_info)
+
+        try:
+            with obj.fs.open(obj.path_info, "r") as fobj:
+                raw = json.load(fobj)
+        except ValueError as exc:
+            raise ObjectFormatError(f"{obj} is corrupted") from exc
+
+        if not isinstance(raw, list):
+            logger.error(
+                "dir cache file format error '%s' [skipping the file]",
+                obj.path_info,
+            )
+            raise ObjectFormatError(f"{obj} is corrupted")
+
+        tree = cls.from_list(raw)
+        tree.path_info = obj.path_info
+        tree.fs = obj.fs
+        tree.hash_info = hash_info
+
+        return tree
+
+    def filter(self, odb, prefix):
+        obj = self._dict.get(prefix)
+        if obj:
+            return obj
+
+        depth = len(prefix)
+        tree = Tree(None, None, None)
+        try:
+            for key, obj in self.trie.items(prefix):
+                tree.add(key[depth:], obj)
+        except KeyError:
+            return None
+        tree.digest()
+        odb.add(tree.path_info, tree.fs, tree.hash_info)
+        return tree
+
+
+def _get_dir_size(odb, tree):
+    try:
+        return sum(
+            odb.fs.getsize(odb.hash_to_path_info(obj.hash_info.value))
+            for _, obj in tree
+        )
+    except FileNotFoundError:
+        return None
+
+
+def _diff(ancestor, other, allow_removed=False):
+    from dictdiffer import diff
+
+    from dvc.exceptions import MergeError
+
+    allowed = ["add"]
+    if allow_removed:
+        allowed.append("remove")
+
+    result = list(diff(ancestor, other))
+    for typ, _, _ in result:
+        if typ not in allowed:
+            raise MergeError(
+                "unable to auto-merge directories with diff that contains "
+                f"'{typ}'ed files"
+            )
+    return result
+
+
+def _merge(ancestor, our, their):
+    import copy
+
+    from dictdiffer import patch
+
+    our_diff = _diff(ancestor, our)
+    if not our_diff:
+        return copy.deepcopy(their)
+
+    their_diff = _diff(ancestor, their)
+    if not their_diff:
+        return copy.deepcopy(our)
+
+    # make sure there are no conflicting files
+    _diff(our, their, allow_removed=True)
+
+    return patch(our_diff + their_diff, ancestor)
+
+
+def merge(odb, ancestor_info, our_info, their_info):
+    from . import load
+
+    assert our_info
+    assert their_info
+
+    if ancestor_info:
+        ancestor = load(odb, ancestor_info)
+    else:
+        ancestor = Tree(None, None, None)
+
+    our = load(odb, our_info)
+    their = load(odb, their_info)
+
+    merged_dict = _merge(ancestor.as_dict(), our.as_dict(), their.as_dict(),)
+
+    merged = Tree(None, None, None)
+    for key, hi in merged_dict.items():
+        merged.add(key, hi)
+    merged.digest()
+
+    odb.add(merged.path_info, merged.fs, merged.hash_info)
+    hash_info = merged.hash_info
+    hash_info.size = _get_dir_size(odb, merged)
+    return hash_info

dvc/output/base.py

@@ -18,7 +18,7 @@ from dvc.exceptions import (
 )
 from dvc.hash_info import HashInfo
 from dvc.objects.db import NamedCache
-from dvc.objects.stage import get_hash
+from dvc.objects.errors import ObjectFormatError
 from dvc.objects.stage import stage as ostage
 
 from ..fs.base import BaseFileSystem
@@ -183,23 +183,21 @@ class BaseOutput:
     def odb(self):
         return getattr(self.repo.odb, self.scheme)
 
-    @property
-    def dir_cache(self):
-        return self.hash_info.dir_info
-
     @property
     def cache_path(self):
         return self.odb.hash_to_path_info(self.hash_info.value).url
 
     def get_hash(self):
         if not self.use_cache:
-            return get_hash(
+            return ostage(
+                self.repo.odb.local,
                 self.path_info,
                 self.fs,
                 self.fs.PARAM_CHECKSUM,
-                self.repo.odb.local,
-            )
-        return ostage(self.odb, self.path_info, self.fs).hash_info
+            ).hash_info
+        return ostage(
+            self.odb, self.path_info, self.fs, self.odb.fs.PARAM_CHECKSUM
+        ).hash_info
 
     @property
     def is_dir_checksum(self):
@@ -223,7 +221,7 @@ class BaseOutput:
         try:
             objects.check(self.odb, obj)
             return False
-        except (FileNotFoundError, objects.ObjectFormatError):
+        except (FileNotFoundError, ObjectFormatError):
             return True
 
     def workspace_status(self):
@@ -307,7 +305,9 @@ class BaseOutput:
             logger.debug("Output '%s' didn't change. Skipping saving.", self)
             return
 
-        self.obj = ostage(self.odb, self.path_info, self.fs)
+        self.obj = ostage(
+            self.odb, self.path_info, self.fs, self.odb.fs.PARAM_CHECKSUM
+        )
         self.hash_info = self.obj.hash_info
         self.isexec = self.isfile() and self.fs.isexec(self.path_info)
 
@@ -322,7 +322,12 @@ class BaseOutput:
         assert self.hash_info
 
         if self.use_cache:
-            obj = ostage(self.odb, filter_info or self.path_info, self.fs)
+            obj = ostage(
+                self.odb,
+                filter_info or self.path_info,
+                self.fs,
+                self.odb.fs.PARAM_CHECKSUM,
+            )
             objects.save(self.odb, obj)
             checkout(
                 filter_info or self.path_info,
@@ -403,6 +408,7 @@ class BaseOutput:
         relink=False,
         filter_info=None,
         allow_missing=False,
+        checkpoint_reset=False,
         **kwargs,
     ):
         if not self.use_cache:
@@ -417,6 +423,11 @@ class BaseOutput:
             # backward compatibility
             return None
 
+        if self.checkpoint and checkpoint_reset:
+            if self.exists:
+                self.remove()
+            return None
+
         added = not self.exists
 
         try:
@@ -460,8 +471,6 @@ class BaseOutput:
             self.repo.scm.ignore(self.fspath)
 
     def get_files_number(self, filter_info=None):
-        from funcy import ilen
-
         if not self.use_cache or not self.hash_info:
             return 0
 
@@ -471,13 +480,8 @@ class BaseOutput:
         if not filter_info or filter_info == self.path_info:
             return self.hash_info.nfiles or 0
 
-        if not self.hash_info.dir_info:
-            return 0
-
-        return ilen(
-            filter_info.isin_or_eq(self.path_info.joinpath(*relpath))
-            for relpath, _ in self.hash_info.dir_info.items()
-        )
+        obj = self.get_obj(filter_info=filter_info)
+        return len(obj) if obj else 0
 
     def unprotect(self):
         if self.exists:
@@ -490,7 +494,7 @@ class BaseOutput:
 
         try:
             objects.check(self.odb, self.odb.get(self.hash_info))
-        except (FileNotFoundError, objects.ObjectFormatError):
+        except (FileNotFoundError, ObjectFormatError):
             self.repo.cloud.pull(
                 NamedCache.make("local", self.hash_info.value, str(self)),
                 show_checksums=False,
@@ -498,12 +502,11 @@ class BaseOutput:
             )
 
         try:
-            objects.load(self.odb, self.hash_info)
-            assert self.hash_info.dir_info
-        except (objects.ObjectFormatError, FileNotFoundError):
-            self.hash_info.dir_info = None
+            self.obj = objects.load(self.odb, self.hash_info)
+        except (FileNotFoundError, ObjectFormatError):
+            self.obj = None
 
-        return self.dir_cache
+        return self.obj
 
     def collect_used_dir_cache(
         self, remote=None, force=False, jobs=None, filter_info=None
@@ -536,7 +539,7 @@ class BaseOutput:
 
         try:
             objects.check(self.odb, self.odb.get(self.hash_info))
-        except (FileNotFoundError, objects.ObjectFormatError):
+        except (FileNotFoundError, ObjectFormatError):
             msg = (
                 "Missing cache for directory '{}'. "
                 "Cache for files inside will be lost. "
@@ -551,14 +554,14 @@ class BaseOutput:
 
         path = str(self.path_info)
         filter_path = str(filter_info) if filter_info else None
-        for entry_key, entry_hash_info in self.dir_cache.items():
+        for entry_key, entry_obj in self.obj:
             entry_path = os.path.join(path, *entry_key)
             if (
                 not filter_path
                 or entry_path == filter_path
                 or entry_path.startswith(filter_path + os.sep)
             ):
-                cache.add(self.scheme, entry_hash_info.value, entry_path)
+                cache.add(self.scheme, entry_obj.hash_info.value, entry_path)
 
         return cache
 
@@ -651,6 +654,8 @@ class BaseOutput:
             )
 
     def merge(self, ancestor, other):
+        from dvc.objects.tree import merge
+
         assert other
 
         if ancestor:
@@ -662,6 +667,6 @@ class BaseOutput:
         self._check_can_merge(self)
         self._check_can_merge(other)
 
-        self.hash_info = objects.merge(
+        self.hash_info = merge(
             self.odb, ancestor_info, self.hash_info, other.hash_info
         )

dvc/remote/base.py

@@ -506,7 +506,7 @@ class Remote:
         return ret
 
     def transfer(self, from_fs, from_info, jobs=None, no_progress_bar=False):
-        from dvc.objects import transfer
+        from dvc.objects.transfer import transfer
 
         return transfer(
             self.odb,

dvc/repo/__init__.py

@@ -463,6 +463,12 @@ class Repo:
         path_parts = os.path.normpath(path).split(os.path.sep)
         return self.DVC_DIR in path_parts
 
+    @cached_property
+    def dvcfs(self):
+        from dvc.fs.dvc import DvcFileSystem
+
+        return DvcFileSystem(self)
+
     @cached_property
     def repo_fs(self):
         from dvc.fs.repo import RepoFileSystem

dvc/repo/add.py

@@ -173,7 +173,7 @@ def _process_stages(
             )
         else:
             from dvc.fs import get_cloud_fs
-            from dvc.objects import transfer
+            from dvc.objects.transfer import transfer
 
             from_fs = get_cloud_fs(repo, url=target)
             out.hash_info = transfer(

dvc/repo/diff.py

@@ -1,5 +1,7 @@
 import logging
 import os
+from collections import defaultdict
+from typing import Dict, List
 
 from dvc.exceptions import PathMissingError
 from dvc.objects.stage import get_file_hash
@@ -130,7 +132,7 @@ def _output_paths(repo, repo_fs, targets):
     def _to_checksum(output):
         if on_working_fs:
             return ostage(
-                repo.odb.local, output.path_info, repo.odb.local.fs,
+                repo.odb.local, output.path_info, repo.odb.local.fs, "md5",
             ).hash_info.value
         return output.hash_info.value
 
@@ -190,29 +192,23 @@ def _targets_to_path_infos(repo_fs, targets):
 
 
 def _calculate_renamed(new, old, added, deleted):
-    old_inverted = {}
+    old_inverted: Dict[str, List[str]] = defaultdict(list)
     # It is needed to be dict of lists to cover cases
     # when repo has paths with same hash
     for path, path_hash in old.items():
-        bucket = old_inverted.get(path_hash, None)
-        if bucket is None:
-            old_inverted[path_hash] = [path]
-        else:
-            bucket.append(path)
+        old_inverted[path_hash].append(path)
 
     renamed = []
     for path in added:
         path_hash = new[path]
-        old_paths = old_inverted.get(path_hash, None)
-        if not old_paths:
-            continue
-        old_path = None
-        for tmp_old_path in old_paths:
-            if tmp_old_path in deleted:
-                old_path = tmp_old_path
-                break
-        if not old_path:
+        old_paths = old_inverted[path_hash]
+        try:
+            iterator = enumerate(old_paths)
+            index = next(idx for idx, path in iterator if path in deleted)
+        except StopIteration:
             continue
+
+        old_path = old_paths.pop(index)
         renamed.append(
             {"path": {"old": old_path, "new": path}, "hash": path_hash}
         )

dvc/repo/experiments/__init__.py

@@ -182,8 +182,6 @@ class Experiments:
                         )
                         self.scm.reset()
 
-                    self._prune_lockfiles()
-
                     # update experiment params from command line
                     if params:
                         self._update_params(params)
@@ -244,34 +242,6 @@ class Experiments:
 
         return stash_rev
 
-    def _prune_lockfiles(self):
-        from dvc.dvcfile import is_lock_file
-
-        # NOTE: dirty DVC lock files must be restored to index state to
-        # avoid checking out incorrect persist or checkpoint outs
-        fs = self.scm.get_fs("HEAD")
-        lock_files = [
-            str(fname)
-            for fname in fs.walk_files(self.scm.root_dir)
-            if is_lock_file(fname)
-        ]
-        if lock_files:
-
-            self.scm.reset(paths=lock_files)
-            self.scm.checkout_index(paths=lock_files, force=True)
-
-    def _prune_untracked_lockfiles(self):
-        from dvc.dvcfile import is_lock_file
-        from dvc.utils.fs import remove
-
-        untracked = [
-            fname
-            for fname in self.scm.untracked_files()
-            if is_lock_file(fname)
-        ]
-        for fname in untracked:
-            remove(fname)
-
     def _stash_msg(
         self,
         rev: str,
@@ -345,9 +315,17 @@ class Experiments:
         queue: bool = False,
         tmp_dir: bool = False,
         checkpoint_resume: Optional[str] = None,
+        reset: bool = False,
         **kwargs,
     ):
         """Reproduce and checkout a single experiment."""
+        if queue and not checkpoint_resume:
+            reset = True
+
+        if reset:
+            self.reset_checkpoints()
+            kwargs["force"] = True
+
         if not (queue or tmp_dir):
             staged, _, _ = self.scm.status()
             if staged:
@@ -370,7 +348,9 @@ class Experiments:
         else:
             checkpoint_resume = self._workspace_resume_rev()
 
-        stash_rev = self.new(checkpoint_resume=checkpoint_resume, **kwargs)
+        stash_rev = self.new(
+            checkpoint_resume=checkpoint_resume, reset=reset, **kwargs
+        )
         if queue:
             logger.info(
                 "Queued experiment '%s' for future execution.", stash_rev[:7],
@@ -709,7 +689,6 @@ class Experiments:
         # result in conflict between workspace params and stashed CLI params).
         self.scm.reset(hard=True)
         with self.scm.detach_head(entry.rev):
-            self._prune_untracked_lockfiles()
             rev = self.stash.pop()
             self.scm.set_ref(EXEC_BASELINE, entry.baseline_rev)
             if entry.branch:
@@ -723,12 +702,17 @@ class Experiments:
                     rev,
                     name=entry.name,
                     rel_cwd=relpath(os.getcwd(), self.scm.root_dir),
+                    log_errors=False,
                 )
 
-                if not exec_result.exp_hash or not exec_result.ref_info:
+                if not exec_result.exp_hash:
                     raise DvcException(
                         f"Failed to reproduce experiment '{rev[:7]}'"
                     )
+                if not exec_result.ref_info:
+                    # repro succeeded but result matches baseline
+                    # (no experiment generated or applied)
+                    return {}
                 exp_rev = self.scm.get_ref(str(exec_result.ref_info))
                 self.scm.set_ref(EXEC_APPLY, exp_rev)
                 return {exp_rev: exec_result.exp_hash}

dvc/repo/experiments/executor/base.py

@@ -181,16 +181,16 @@ class BaseExecutor(ABC):
                 refs.append(ref)
 
         def on_diverged_ref(orig_ref: str, new_rev: str):
-            orig_rev = dest_scm.get_ref(orig_ref)
-            if dest_scm.diff(orig_rev, new_rev):
-                if force:
-                    logger.debug(
-                        "Replacing existing experiment '%s'", orig_ref,
-                    )
-                    return True
-                if on_diverged:
-                    checkpoint = self.scm.get_ref(EXEC_CHECKPOINT) is not None
-                    on_diverged(orig_ref, checkpoint)
+            if force:
+                logger.debug(
+                    "Replacing existing experiment '%s'", orig_ref,
+                )
+                return True
+
+            checkpoint = self.scm.get_ref(EXEC_CHECKPOINT) is not None
+            self._raise_ref_conflict(dest_scm, orig_ref, new_rev, checkpoint)
+            if on_diverged:
+                on_diverged(orig_ref, checkpoint)
             logger.debug("Reproduced existing experiment '%s'", orig_ref)
             return False
 
@@ -218,6 +218,7 @@ class BaseExecutor(ABC):
         queue: Optional["Queue"] = None,
         rel_cwd: Optional[str] = None,
         name: Optional[str] = None,
+        log_errors: bool = True,
         log_level: Optional[int] = None,
     ) -> "ExecutorResult":
         """Run dvc repro and return the result.
@@ -234,7 +235,7 @@ class BaseExecutor(ABC):
 
         if queue is not None:
             queue.put((rev, os.getpid()))
-        if log_level is not None:
+        if log_errors and log_level is not None:
             cls._set_log_level(log_level)
 
         def filter_pipeline(stages):
@@ -246,7 +247,7 @@ class BaseExecutor(ABC):
         exp_ref: Optional["ExpRefInfo"] = None
         repro_force: bool = False
 
-        with cls._repro_dvc(dvc_dir, rel_cwd) as dvc:
+        with cls._repro_dvc(dvc_dir, rel_cwd, log_errors) as dvc:
             args, kwargs = cls._repro_args(dvc)
             if args:
                 targets: Optional[Union[list, str]] = args[0]
@@ -258,17 +259,19 @@ class BaseExecutor(ABC):
                 "Executor repro with force = '%s'", str(repro_force)
             )
 
-            # NOTE: for checkpoint experiments we handle persist outs slightly
-            # differently than normal:
+            # NOTE: checkpoint outs are handled as a special type of persist
+            # out:
             #
             # - checkpoint out may not yet exist if this is the first time this
             #   experiment has been run, this is not an error condition for
             #   experiments
-            # - at the start of a repro run, we need to remove the persist out
-            #   and restore it to its last known (committed) state (which may
-            #   be removed/does not yet exist) so that our executor workspace
-            #   is not polluted with the (persistent) out from an unrelated
-            #   experiment run
+            # - if experiment was run with --reset, the checkpoint out will be
+            #   removed at the start of the experiment (regardless of any
+            #   dvc.lock entry for the checkpoint out)
+            # - if run without --reset, the checkpoint out will be checked out
+            #   using any hash present in dvc.lock (or removed if no entry
+            #   exists in dvc.lock)
+            checkpoint_reset = kwargs.pop("reset", False)
             dvc_checkout(
                 dvc,
                 targets=targets,
@@ -276,6 +279,7 @@ class BaseExecutor(ABC):
                 force=True,
                 quiet=True,
                 allow_missing=True,
+                checkpoint_reset=checkpoint_reset,
             )
 
             checkpoint_func = partial(
@@ -321,7 +325,9 @@ class BaseExecutor(ABC):
 
     @classmethod
     @contextmanager
-    def _repro_dvc(cls, dvc_dir: Optional[str], rel_cwd: Optional[str]):
+    def _repro_dvc(
+        cls, dvc_dir: Optional[str], rel_cwd: Optional[str], log_errors: bool
+    ):
         from dvc.repo import Repo
 
         dvc = Repo(dvc_dir)
@@ -342,10 +348,12 @@ class BaseExecutor(ABC):
         except CheckpointKilledError:
             raise
         except DvcException:
-            logger.exception("")
+            if log_errors:
+                logger.exception("")
             raise
         except Exception:
-            logger.exception("unexpected error")
+            if log_errors:
+                logger.exception("unexpected error")
             raise
         finally:
             dvc.close()
@@ -358,6 +366,8 @@ class BaseExecutor(ABC):
         if os.path.exists(args_path):
             args, kwargs = cls.unpack_repro_args(args_path)
             remove(args_path)
+            # explicitly git rm/unstage the args file
+            dvc.scm.add([args_path])
         else:
             args = []
             kwargs = {}
@@ -396,6 +406,7 @@ class BaseExecutor(ABC):
             logger.debug("No changes to commit")
             raise UnchangedExperimentError(rev)
 
+        check_conflict = False
         branch = scm.get_ref(EXEC_BRANCH, follow=False)
         if branch:
             old_ref = rev
@@ -406,21 +417,41 @@ class BaseExecutor(ABC):
             ref_info = ExpRefInfo(baseline_rev, name)
             branch = str(ref_info)
             old_ref = None
-            if not force and scm.get_ref(branch):
-                if checkpoint:
-                    raise CheckpointExistsError(ref_info.name)
-                raise ExperimentExistsError(ref_info.name)
-            logger.debug("Commit to new experiment branch '%s'", branch)
+            if scm.get_ref(branch):
+                if not force:
+                    check_conflict = True
+                logger.debug(
+                    "%s existing experiment branch '%s'",
+                    "Replace" if force else "Reuse",
+                    branch,
+                )
+            else:
+                logger.debug("Commit to new experiment branch '%s'", branch)
 
         scm.add([], update=True)
         scm.commit(f"dvc: commit experiment {exp_hash}", no_verify=True)
         new_rev = scm.get_rev()
-        scm.set_ref(branch, new_rev, old_ref=old_ref)
+        if check_conflict:
+            new_rev = cls._raise_ref_conflict(scm, branch, new_rev, checkpoint)
+        else:
+            scm.set_ref(branch, new_rev, old_ref=old_ref)
         scm.set_ref(EXEC_BRANCH, branch, symbolic=True)
         if checkpoint:
             scm.set_ref(EXEC_CHECKPOINT, new_rev)
         return new_rev
 
+    @staticmethod
+    def _raise_ref_conflict(scm, ref, new_rev, checkpoint):
+        # If this commit is a duplicate of the existing commit at 'ref', return
+        # the existing commit. Otherwise, error out and require user to re-run
+        # with --force as needed
+        orig_rev = scm.get_ref(ref)
+        if scm.diff(orig_rev, new_rev):
+            if checkpoint:
+                raise CheckpointExistsError(ref)
+            raise ExperimentExistsError(ref)
+        return orig_rev
+
     @staticmethod
     def _set_log_level(level):
         from dvc.logger import disable_other_loggers

dvc/repo/experiments/executor/local.py

@@ -3,13 +3,29 @@ import os
 from tempfile import TemporaryDirectory
 from typing import Optional
 
-from dvc.utils.fs import remove
-
 from .base import BaseExecutor
 
 logger = logging.getLogger(__name__)
 
 
+class ExpTemporaryDirectory(TemporaryDirectory):
+    # Python's TemporaryDirectory cleanup shutil.rmtree wrapper does not handle
+    # git read-only dirs cleanly in Windows on Python <3.8, so we use our own
+    # remove(). See:
+    # https://github.com/iterative/dvc/pull/5425
+    # https://bugs.python.org/issue26660
+
+    @classmethod
+    def _rmtree(cls, name):
+        from dvc.utils.fs import remove
+
+        remove(name)
+
+    def cleanup(self):
+        if self._finalizer.detach():
+            self._rmtree(self.name)
+
+
 class BaseLocalExecutor(BaseExecutor):
     """Base local machine executor."""
 
@@ -37,7 +53,7 @@ class TempDirExecutor(BaseLocalExecutor):
         cache_dir: Optional[str] = None,
         **kwargs,
     ):
-        self._tmp_dir = TemporaryDirectory(dir=tmp_dir)
+        self._tmp_dir = ExpTemporaryDirectory(dir=tmp_dir)
         kwargs["root_dir"] = self._tmp_dir.name
         super().__init__(*args, **kwargs)
         if cache_dir:
@@ -55,4 +71,4 @@ class TempDirExecutor(BaseLocalExecutor):
     def cleanup(self):
         super().cleanup()
         logger.debug("Removing tmpdir '%s'", self._tmp_dir)
-        remove(self._tmp_dir.name)
+        self._tmp_dir.cleanup()

dvc/repo/experiments/run.py

@@ -15,7 +15,6 @@ def run(
     run_all: bool = False,
     jobs: int = 1,
     tmp_dir: bool = False,
-    reset: bool = False,
     **kwargs,
 ) -> dict:
     """Reproduce the specified targets as an experiment.
@@ -25,10 +24,6 @@ def run(
     Returns a dict mapping new experiment SHAs to the results
     of `repro` for that experiment.
     """
-    if reset:
-        repo.experiments.reset_checkpoints()
-        kwargs["force"] = True
-
     if run_all:
         return repo.experiments.reproduce_queued(jobs=jobs)
 

dvc/repo/fetch.py

@@ -115,6 +115,7 @@ def _fetch_external(self, repo_url, repo_rev, files, jobs):
                     odb,
                     path_info,
                     repo.repo_fs,
+                    "md5",
                     jobs=jobs,
                     follow_subrepos=False,
                 )

dvc/scm/git/__init__.py

@@ -272,19 +272,10 @@ class Git(Base):
         self.files_to_track = set()
 
     def remind_to_track(self):
-        def _filter(files):
-            if not files:
-                return set()
-            staged, unstaged, untracked = self.status()
-            return files.intersection(
-                staged.get("modify", []) + unstaged + untracked
-            )
-
-        files_to_track = _filter(self.files_to_track)
-        if self.quiet or not files_to_track:
+        if self.quiet or not self.files_to_track:
             return
 
-        files = " ".join(shlex.quote(path) for path in sorted(files_to_track))
+        files = " ".join(shlex.quote(path) for path in self.files_to_track)
 
         logger.info(
             "\n"

dvc/scm/git/backend/base.py

@@ -67,7 +67,11 @@ class BaseGitBackend(ABC):
 
     @abstractmethod
     def checkout(
-        self, branch: str, create_new: Optional[bool] = False, **kwargs,
+        self,
+        branch: str,
+        create_new: Optional[bool] = False,
+        force: bool = False,
+        **kwargs,
     ):
         pass
 

dvc/scm/git/backend/dulwich.py

@@ -189,7 +189,11 @@ class DulwichBackend(BaseGitBackend):  # pylint:disable=abstract-method
             raise SCMError("Git commit failed") from exc
 
     def checkout(
-        self, branch: str, create_new: Optional[bool] = False, **kwargs,
+        self,
+        branch: str,
+        create_new: Optional[bool] = False,
+        force: bool = False,
+        **kwargs,
     ):
         raise NotImplementedError
 

dvc/scm/git/backend/gitpython.py

@@ -217,12 +217,16 @@ class GitPythonBackend(BaseGitBackend):  # pylint:disable=abstract-method
             raise SCMError("Git pre-commit hook failed") from exc
 
     def checkout(
-        self, branch: str, create_new: Optional[bool] = False, **kwargs
+        self,
+        branch: str,
+        create_new: Optional[bool] = False,
+        force: bool = False,
+        **kwargs,
     ):
         if create_new:
-            self.repo.git.checkout("HEAD", b=branch, **kwargs)
+            self.repo.git.checkout("HEAD", b=branch, force=force, **kwargs)
         else:
-            self.repo.git.checkout(branch, **kwargs)
+            self.repo.git.checkout(branch, force=force, **kwargs)
 
     def pull(self, **kwargs):
         infos = self.repo.remote().pull(**kwargs)

dvc/scm/git/backend/pygit2.py

@@ -100,14 +100,20 @@ class Pygit2Backend(BaseGitBackend):  # pylint:disable=abstract-method
         raise NotImplementedError
 
     def checkout(
-        self, branch: str, create_new: Optional[bool] = False, **kwargs,
+        self,
+        branch: str,
+        create_new: Optional[bool] = False,
+        force: bool = False,
+        **kwargs,
     ):
-        from pygit2 import GitError
+        from pygit2 import GIT_CHECKOUT_FORCE, GitError
+
+        checkout_strategy = GIT_CHECKOUT_FORCE if force else None
 
         if create_new:
             commit = self.repo.revparse_single("HEAD")
             new_branch = self.repo.branches.local.create(branch, commit)
-            self.repo.checkout(new_branch)
+            self.repo.checkout(new_branch, strategy=checkout_strategy)
         else:
             if branch == "-":
                 branch = "@{-1}"
@@ -115,7 +121,7 @@ class Pygit2Backend(BaseGitBackend):  # pylint:disable=abstract-method
                 commit, ref = self.repo.resolve_refish(branch)
             except (KeyError, GitError):
                 raise RevError(f"unknown Git revision '{branch}'")
-            self.repo.checkout_tree(commit)
+            self.repo.checkout_tree(commit, strategy=checkout_strategy)
             detach = kwargs.get("detach", False)
             if ref and not detach:
                 self.repo.set_head(ref.name)

dvc/stage/imports.py

@@ -12,7 +12,7 @@ def _update_import_on_remote(stage, remote, jobs):
             "be updated with --to-remote"
         )
 
-    url = stage.deps[0].path_info.url
+    url = stage.deps[0].def_path
     stage.outs[0].hash_info = stage.repo.cloud.transfer(
         url, jobs=jobs, remote=remote, command="update"
     )

dvc/version.py

@@ -6,7 +6,7 @@
 import os
 import subprocess
 
-_BASE_VERSION = "2.0.4"
+_BASE_VERSION = "2.0.7"
 
 
 def _generate_version(base_version):

scripts/hooks/hook-googleapiclient.model.py

@@ -0,0 +1,8 @@
+"""Temporary fix for https://github.com/iterative/dvc/issues/5618."""
+from PyInstaller.utils.hooks import (  # pylint:disable=import-error
+    collect_data_files,
+    copy_metadata,
+)
+
+datas = collect_data_files("googleapiclient.discovery")
+datas += copy_metadata("google_api_python_client")

setup.py

@@ -118,7 +118,6 @@ tests_requirements = [
     "wheel>=0.31.1",
     # Test requirements:
     "pytest>=6.0.1,<6.2.2",
-    "pytest-cov",
     "pytest-docker>=0.7.2",
     "pytest-timeout>=1.3.3",
     "pytest-cov>=2.6.1",
@@ -145,10 +144,9 @@ tests_requirements = [
     "urllib3<1.26.0",
     "rangehttpserver==1.2.0",
     "beautifulsoup4==4.4.0",
-    "pylint==2.5.3",
-    "astroid<2.5",  # temporary required for pylint, see #5469
-    "pylint-pytest>=0.3.0",
-    "pylint-plugin-utils",
+    "pylint==2.7.2",
+    "pylint-pytest==0.3.0",
+    "pylint-plugin-utils==0.6",
     "wget",
     "filelock",
     "mypy",
@@ -192,6 +190,7 @@ setup(
         "Programming Language :: Python :: 3.6",
         "Programming Language :: Python :: 3.7",
         "Programming Language :: Python :: 3.8",
+        "Programming Language :: Python :: 3.9",
     ],
     packages=find_packages(exclude=["tests"]),
     include_package_data=True,

tests/func/experiments/test_checkpoints.py

@@ -88,10 +88,6 @@ def test_reset_checkpoint(
         checkpoint_stage.addressing, name="foo", tmp_dir=not workspace,
     )
 
-    if workspace:
-        scm.reset(hard=True)
-        scm.gitpython.repo.git.clean(force=True)
-
     results = dvc.experiments.run(
         checkpoint_stage.addressing,
         params=["foo=2"],

tests/func/experiments/test_experiments.py

@@ -418,28 +418,6 @@ def test_untracked(tmp_dir, scm, dvc, caplog, workspace):
         assert fobj.read().strip() == "foo: 2"
 
 
-@pytest.mark.parametrize("workspace", [True, False])
-def test_dirty_lockfile(tmp_dir, scm, dvc, exp_stage, workspace):
-    from dvc.dvcfile import LockfileCorruptedError
-
-    tmp_dir.gen("dvc.lock", "foo")
-
-    with pytest.raises(LockfileCorruptedError):
-        dvc.reproduce(exp_stage.addressing)
-
-    results = dvc.experiments.run(
-        exp_stage.addressing, params=["foo=2"], tmp_dir=not workspace
-    )
-    exp = first(results)
-
-    fs = scm.get_fs(exp)
-    with fs.open(tmp_dir / "metrics.yaml") as fobj:
-        assert fobj.read().strip() == "foo: 2"
-
-    if not workspace:
-        assert (tmp_dir / "dvc.lock").read_text() == "foo"
-
-
 def test_packed_args_exists(tmp_dir, scm, dvc, exp_stage, caplog):
     from dvc.repo.experiments.executor.base import BaseExecutor
 

tests/func/params/test_diff.py

@@ -1,7 +1,3 @@
-import os
-
-import pytest
-
 from dvc.utils import relpath
 from dvc.utils.serialize import dump_yaml
 
@@ -18,132 +14,6 @@ def test_diff_no_changes(tmp_dir, scm, dvc):
     assert dvc.params.diff() == {}
 
 
-@pytest.mark.parametrize("commit_last", [True, False])
-def test_diff_add_similar_files(tmp_dir, scm, dvc, commit_last):
-    if commit_last:
-        last_commit_msg = "commit #2"
-        a_rev = "HEAD~1"
-    else:
-        last_commit_msg = None
-        a_rev = "HEAD"
-
-    tmp_dir.dvc_gen(
-        {"dir": {"file": "text1", "subdir": {"file2": "text2"}}},
-        commit="commit #1",
-    )
-    tmp_dir.dvc_gen(
-        {"dir2": {"file": "text1", "subdir": {"file2": "text2"}}},
-        commit=last_commit_msg,
-    )
-    assert dvc.diff(a_rev) == {
-        "added": [
-            {
-                "path": os.path.join("dir2", ""),
-                "hash": "cb58ee07cb01044db229e4d6121a0dfc.dir",
-            },
-            {
-                "path": os.path.join("dir2", "file"),
-                "hash": "cef7ccd89dacf1ced6f5ec91d759953f",
-            },
-            {
-                "path": os.path.join("dir2", "subdir", "file2"),
-                "hash": "fe6123a759017e4a2af4a2d19961ed71",
-            },
-        ],
-        "deleted": [],
-        "modified": [],
-        "renamed": [],
-        "not in cache": [],
-    }
-
-
-@pytest.mark.parametrize("commit_last", [True, False])
-def test_diff_rename_folder(tmp_dir, scm, dvc, commit_last):
-    if commit_last:
-        last_commit_msg = "commit #2"
-        a_rev = "HEAD~1"
-    else:
-        last_commit_msg = None
-        a_rev = "HEAD"
-
-    tmp_dir.dvc_gen(
-        {"dir": {"file": "text1", "subdir": {"file2": "text2"}}},
-        commit="commit #1",
-    )
-    (tmp_dir / "dir").replace(tmp_dir / "dir2")
-    tmp_dir.dvc_add("dir2", commit=last_commit_msg)
-    assert dvc.diff(a_rev) == {
-        "added": [],
-        "deleted": [],
-        "modified": [],
-        "renamed": [
-            {
-                "path": {
-                    "old": os.path.join("dir", ""),
-                    "new": os.path.join("dir2", ""),
-                },
-                "hash": "cb58ee07cb01044db229e4d6121a0dfc.dir",
-            },
-            {
-                "path": {
-                    "old": os.path.join("dir", "file"),
-                    "new": os.path.join("dir2", "file"),
-                },
-                "hash": "cef7ccd89dacf1ced6f5ec91d759953f",
-            },
-            {
-                "path": {
-                    "old": os.path.join("dir", "subdir", "file2"),
-                    "new": os.path.join("dir2", "subdir", "file2"),
-                },
-                "hash": "fe6123a759017e4a2af4a2d19961ed71",
-            },
-        ],
-        "not in cache": [],
-    }
-
-
-@pytest.mark.parametrize("commit_last", [True, False])
-def test_diff_rename_file(tmp_dir, scm, dvc, commit_last):
-    if commit_last:
-        last_commit_msg = "commit #2"
-        a_rev = "HEAD~1"
-    else:
-        last_commit_msg = None
-        a_rev = "HEAD"
-
-    paths = tmp_dir.gen(
-        {"dir": {"file": "text1", "subdir": {"file2": "text2"}}}
-    )
-    tmp_dir.dvc_add(paths, commit="commit #1")
-    (tmp_dir / "dir" / "file").replace(tmp_dir / "dir" / "subdir" / "file3")
-
-    tmp_dir.dvc_add(paths, commit=last_commit_msg)
-    assert dvc.diff(a_rev) == {
-        "added": [],
-        "deleted": [],
-        "modified": [
-            {
-                "path": os.path.join("dir", ""),
-                "hash": {
-                    "old": "cb58ee07cb01044db229e4d6121a0dfc.dir",
-                    "new": "a4ac9c339aacc60b6a3152e362c319c8.dir",
-                },
-            }
-        ],
-        "renamed": [
-            {
-                "path": {
-                    "old": os.path.join("dir", "file"),
-                    "new": os.path.join("dir", "subdir", "file3"),
-                },
-                "hash": "cef7ccd89dacf1ced6f5ec91d759953f",
-            }
-        ],
-        "not in cache": [],
-    }
-
-
 def test_diff(tmp_dir, scm, dvc):
     tmp_dir.gen("params.yaml", "foo: bar")
     dvc.run(cmd="echo params.yaml", params=["foo"], single_stage=True)

tests/func/remote/test_index.py

@@ -32,7 +32,7 @@ def remote(tmp_dir, dvc, tmp_path_factory, mocker):
 def test_indexed_on_status(tmp_dir, dvc, tmp_path_factory, remote):
     foo = tmp_dir.dvc_gen({"foo": "foo content"})[0].outs[0]
     bar = tmp_dir.dvc_gen({"bar": {"baz": "baz content"}})[0].outs[0]
-    baz_hash = bar.dir_cache.trie.get(("baz",))
+    baz_hash = bar.obj.trie.get(("baz",)).hash_info
     dvc.push()
     with remote.index:
         remote.index.clear()
@@ -49,7 +49,7 @@ def test_indexed_on_status(tmp_dir, dvc, tmp_path_factory, remote):
 def test_indexed_on_push(tmp_dir, dvc, tmp_path_factory, remote):
     foo = tmp_dir.dvc_gen({"foo": "foo content"})[0].outs[0]
     bar = tmp_dir.dvc_gen({"bar": {"baz": "baz content"}})[0].outs[0]
-    baz_hash = bar.dir_cache.trie.get(("baz",))
+    baz_hash = bar.obj.trie.get(("baz",)).hash_info
 
     dvc.push()
     with remote.index:

tests/func/test_add.py

@@ -109,9 +109,10 @@ def test_add_directory(tmp_dir, dvc):
 
     hash_info = stage.outs[0].hash_info
 
-    dir_info = load(dvc.odb.local, hash_info).hash_info.dir_info
-    for path, _ in dir_info.trie.items():
-        assert "\\" not in path
+    obj = load(dvc.odb.local, hash_info)
+    for key, _ in obj:
+        for part in key:
+            assert "\\" not in part
 
 
 class TestAddDirectoryRecursive(TestDvc):
@@ -516,7 +517,7 @@ class TestAddCommit(TestDvc):
 
 def test_should_collect_dir_cache_only_once(mocker, tmp_dir, dvc):
     tmp_dir.gen({"data/data": "foo"})
-    get_dir_hash_counter = mocker.spy(dvc_module.objects.stage, "get_dir_hash")
+    counter = mocker.spy(dvc_module.objects.stage, "_get_tree_obj")
     ret = main(["add", "data"])
     assert ret == 0
 
@@ -525,7 +526,7 @@ def test_should_collect_dir_cache_only_once(mocker, tmp_dir, dvc):
 
     ret = main(["status"])
     assert ret == 0
-    assert get_dir_hash_counter.mock.call_count == 1
+    assert counter.mock.call_count == 1
 
 
 class TestShouldPlaceStageInDataDirIfRepositoryBelowSymlink(TestDvc):

tests/func/test_checkout.py

@@ -100,11 +100,10 @@ class TestCheckoutCorruptedCacheDir(TestDvc):
 
         # NOTE: modifying cache file for one of the files inside the directory
         # to check if dvc will detect that the cache is corrupted.
-        _, entry_hash = next(
-            load(self.dvc.odb.local, out.hash_info).hash_info.dir_info.items()
-        )
+        obj = load(self.dvc.odb.local, out.hash_info)
+        _, entry_obj = list(obj)[0]
         cache = os.fspath(
-            self.dvc.odb.local.hash_to_path_info(entry_hash.value)
+            self.dvc.odb.local.hash_to_path_info(entry_obj.hash_info.value)
         )
 
         os.chmod(cache, 0o644)

tests/func/test_data_cloud.py

@@ -287,7 +287,7 @@ def test_verify_hashes(
     remove("dir")
     remove(dvc.odb.local.cache_dir)
 
-    hash_spy = mocker.spy(dvc_module.objects.stage, "get_file_hash")
+    hash_spy = mocker.spy(dvc_module.objects.stage, "_get_file_hash")
 
     dvc.pull()
     assert hash_spy.call_count == 0

tests/func/test_diff.py

@@ -423,3 +423,175 @@ def test_targets_single_file_in_dir_with_file(tmp_dir, scm, dvc):
         "not in cache": [],
         "renamed": [],
     }
+
+
+@pytest.mark.parametrize("commit_last", [True, False])
+def test_diff_add_similar_files(tmp_dir, scm, dvc, commit_last):
+    if commit_last:
+        last_commit_msg = "commit #2"
+        a_rev = "HEAD~1"
+    else:
+        last_commit_msg = None
+        a_rev = "HEAD"
+
+    tmp_dir.dvc_gen(
+        {"dir": {"file": "text1", "subdir": {"file2": "text2"}}},
+        commit="commit #1",
+    )
+    tmp_dir.dvc_gen(
+        {"dir2": {"file": "text1", "subdir": {"file2": "text2"}}},
+        commit=last_commit_msg,
+    )
+    assert dvc.diff(a_rev) == {
+        "added": [
+            {
+                "path": os.path.join("dir2", ""),
+                "hash": "cb58ee07cb01044db229e4d6121a0dfc.dir",
+            },
+            {
+                "path": os.path.join("dir2", "file"),
+                "hash": "cef7ccd89dacf1ced6f5ec91d759953f",
+            },
+            {
+                "path": os.path.join("dir2", "subdir", "file2"),
+                "hash": "fe6123a759017e4a2af4a2d19961ed71",
+            },
+        ],
+        "deleted": [],
+        "modified": [],
+        "renamed": [],
+        "not in cache": [],
+    }
+
+
+@pytest.mark.parametrize("commit_last", [True, False])
+def test_diff_rename_folder(tmp_dir, scm, dvc, commit_last):
+    if commit_last:
+        last_commit_msg = "commit #2"
+        a_rev = "HEAD~1"
+    else:
+        last_commit_msg = None
+        a_rev = "HEAD"
+
+    tmp_dir.dvc_gen(
+        {"dir": {"file": "text1", "subdir": {"file2": "text2"}}},
+        commit="commit #1",
+    )
+    (tmp_dir / "dir").replace(tmp_dir / "dir2")
+    tmp_dir.dvc_add("dir2", commit=last_commit_msg)
+    assert dvc.diff(a_rev) == {
+        "added": [],
+        "deleted": [],
+        "modified": [],
+        "renamed": [
+            {
+                "path": {
+                    "old": os.path.join("dir", ""),
+                    "new": os.path.join("dir2", ""),
+                },
+                "hash": "cb58ee07cb01044db229e4d6121a0dfc.dir",
+            },
+            {
+                "path": {
+                    "old": os.path.join("dir", "file"),
+                    "new": os.path.join("dir2", "file"),
+                },
+                "hash": "cef7ccd89dacf1ced6f5ec91d759953f",
+            },
+            {
+                "path": {
+                    "old": os.path.join("dir", "subdir", "file2"),
+                    "new": os.path.join("dir2", "subdir", "file2"),
+                },
+                "hash": "fe6123a759017e4a2af4a2d19961ed71",
+            },
+        ],
+        "not in cache": [],
+    }
+
+
+@pytest.mark.parametrize("commit_last", [True, False])
+def test_diff_rename_file(tmp_dir, scm, dvc, commit_last):
+    if commit_last:
+        last_commit_msg = "commit #2"
+        a_rev = "HEAD~1"
+    else:
+        last_commit_msg = None
+        a_rev = "HEAD"
+
+    paths = tmp_dir.gen(
+        {"dir": {"file": "text1", "subdir": {"file2": "text2"}}}
+    )
+    tmp_dir.dvc_add(paths, commit="commit #1")
+    (tmp_dir / "dir" / "file").replace(tmp_dir / "dir" / "subdir" / "file3")
+
+    tmp_dir.dvc_add(paths, commit=last_commit_msg)
+    assert dvc.diff(a_rev) == {
+        "added": [],
+        "deleted": [],
+        "modified": [
+            {
+                "path": os.path.join("dir", ""),
+                "hash": {
+                    "old": "cb58ee07cb01044db229e4d6121a0dfc.dir",
+                    "new": "a4ac9c339aacc60b6a3152e362c319c8.dir",
+                },
+            }
+        ],
+        "renamed": [
+            {
+                "path": {
+                    "old": os.path.join("dir", "file"),
+                    "new": os.path.join("dir", "subdir", "file3"),
+                },
+                "hash": "cef7ccd89dacf1ced6f5ec91d759953f",
+            }
+        ],
+        "not in cache": [],
+    }
+
+
+def test_rename_multiple_files_same_hashes(tmp_dir, scm, dvc):
+    """Test diff by renaming >=2 instances of file with same hashes.
+
+    DVC should be able to detect that they are renames, and should not include
+    them in either of the `added` or the `deleted` section.
+    """
+    tmp_dir.dvc_gen(
+        {"dir": {"foo": "foo", "subdir": {"foo": "foo"}}}, commit="commit #1",
+    )
+    remove(tmp_dir / "dir")
+    # changing foo and subdir/foo to bar and subdir/bar respectively
+    tmp_dir.dvc_gen(
+        {"dir": {"bar": "foo", "subdir": {"bar": "foo"}}}, commit="commit #2"
+    )
+    assert dvc.diff("HEAD~") == {
+        "added": [],
+        "deleted": [],
+        "modified": [
+            {
+                "hash": {
+                    "new": "31b36b3ea5f4485e27f10578c47183b0.dir",
+                    "old": "c7684c8b3b0d28cf80d5305e2d856bfc.dir",
+                },
+                "path": os.path.join("dir", ""),
+            }
+        ],
+        "not in cache": [],
+        "renamed": [
+            {
+                "hash": "acbd18db4cc2f85cedef654fccc4a4d8",
+                "path": {
+                    "new": os.path.join("dir", "bar"),
+                    "old": os.path.join("dir", "foo"),
+                },
+            },
+            {
+                "hash": "acbd18db4cc2f85cedef654fccc4a4d8",
+                "path": {
+                    "new": os.path.join("dir", "subdir", "bar"),
+                    "old": os.path.join("dir", "subdir", "foo"),
+                },
+            },
+        ],
+    }

tests/func/test_external_repo.py

@@ -205,6 +205,7 @@ def test_subrepos_are_ignored(tmp_dir, erepo_dir):
             repo.odb.local,
             PathInfo(repo.root_dir) / "dir",
             repo.repo_fs,
+            "md5",
             follow_subrepos=False,
         )
         save(repo.odb.local, obj)

tests/func/test_fs.py

@@ -349,3 +349,23 @@ def test_fs_ls_with_etag(dvc, cloud):
             fs.info(path_info.replace(path=details["name"]))["etag"]
             == details["etag"]
         )
+
+
+@pytest.mark.parametrize(
+    "cloud", [pytest.lazy_fixture("azure"), pytest.lazy_fixture("gs")]
+)
+def test_fs_fsspec_path_management(dvc, cloud):
+    cloud.gen({"foo": "foo", "data": {"bar": "bar", "baz": {"foo": "foo"}}})
+    fs = get_cloud_fs(dvc, **cloud.config)
+
+    root = cloud.parents[len(cloud.parents) - 1]
+    bucket_details = fs.info(root)
+
+    # special conditions: name always points to the bucket name
+    assert bucket_details["name"] == root.bucket
+    assert bucket_details["type"] == "directory"
+
+    data = cloud / "data"
+    data_details = fs.info(data)
+    assert data_details["name"].rstrip("/") == data.path
+    assert data_details["type"] == "directory"

tests/func/test_import_url.py

@@ -353,3 +353,42 @@ def test_import_url_to_remote_absolute(
 def test_import_url_to_remote_invalid_combinations(dvc):
     with pytest.raises(InvalidArgumentError, match="--no-exec"):
         dvc.imp_url("s3://bucket/foo", no_exec=True, to_remote=True)
+
+
+empty_xfail = pytest.mark.xfail(
+    reason="https://github.com/iterative/dvc/issues/5521"
+)
+
+
+@pytest.mark.parametrize(
+    "workspace",
+    [
+        pytest.lazy_fixture("s3"),
+        pytest.lazy_fixture("hdfs"),
+        pytest.param(pytest.lazy_fixture("gs"), marks=empty_xfail),
+        pytest.param(pytest.lazy_fixture("azure"), marks=empty_xfail),
+        pytest.param(
+            pytest.lazy_fixture("ssh"),
+            marks=pytest.mark.skipif(
+                os.name == "nt", reason="disabled on windows"
+            ),
+        ),
+    ],
+    indirect=True,
+)
+def test_import_url_empty_directory(tmp_dir, dvc, workspace):
+    # prefix based storage services (e.g s3) doesn't have the real concept
+    # of directories. So instead we create an empty file that ends with a
+    # trailing slash in order to actually support this operation
+    if workspace.IS_OBJECT_STORAGE:
+        contents = ""
+    else:
+        contents = {}
+
+    workspace.gen({"empty_dir/": contents})
+
+    dvc.imp_url("remote://workspace/empty_dir/")
+
+    empty_dir = tmp_dir / "empty_dir"
+    assert empty_dir.is_dir()
+    assert tuple(empty_dir.iterdir()) == ()

tests/func/test_odb.py

@@ -7,8 +7,8 @@ import pytest
 
 from dvc.hash_info import HashInfo
 from dvc.main import main
-from dvc.objects import ObjectFormatError
 from dvc.objects.db import ODBManager
+from dvc.objects.errors import ObjectFormatError
 from dvc.utils import relpath
 from tests.basic_env import TestDir, TestDvc
 

tests/func/test_remote.py

@@ -7,7 +7,6 @@ import pytest
 from mock import patch
 
 from dvc.config import Config
-from dvc.dir_info import DirInfo
 from dvc.exceptions import DownloadError, RemoteCacheRequiredError, UploadError
 from dvc.fs.local import LocalFileSystem
 from dvc.main import main
@@ -147,26 +146,31 @@ class TestRemoteShouldHandleUppercaseRemoteName(TestDvc):
 
 def test_dir_hash_should_be_key_order_agnostic(tmp_dir, dvc):
     from dvc.objects.stage import stage
+    from dvc.objects.tree import Tree
 
     tmp_dir.gen({"data": {"1": "1 content", "2": "2 content"}})
 
     path_info = PathInfo("data")
 
-    dir_info = DirInfo.from_list(
+    tree = Tree.from_list(
         [{"relpath": "1", "md5": "1"}, {"relpath": "2", "md5": "2"}]
     )
     with patch(
-        "dvc.objects.stage._collect_dir", return_value=dir_info,
+        "dvc.objects.stage._get_tree_obj", return_value=tree,
     ):
-        hash1 = stage(dvc.odb.local, path_info, dvc.odb.local.fs).hash_info
+        hash1 = stage(
+            dvc.odb.local, path_info, dvc.odb.local.fs, "md5"
+        ).hash_info
 
-    dir_info = DirInfo.from_list(
+    tree = Tree.from_list(
         [{"md5": "1", "relpath": "1"}, {"md5": "2", "relpath": "2"}]
     )
     with patch(
-        "dvc.objects.stage._collect_dir", return_value=dir_info,
+        "dvc.objects.stage._get_tree_obj", return_value=tree,
     ):
-        hash2 = stage(dvc.odb.local, path_info, dvc.odb.local.fs).hash_info
+        hash2 = stage(
+            dvc.odb.local, path_info, dvc.odb.local.fs, "md5"
+        ).hash_info
 
     assert hash1 == hash2
 
@@ -278,7 +282,7 @@ def test_push_order(tmp_dir, dvc, tmp_path_factory, mocker, local_remote):
     remote = dvc.cloud.get_remote("upstream")
     foo_path = remote.odb.hash_to_path_info(foo.hash_info.value)
     bar_path = remote.odb.hash_to_path_info(
-        foo.hash_info.dir_info.trie[("bar",)].value
+        foo.obj.trie[("bar",)].hash_info.value
     )
     paths = [args[1] for args, _ in mocked_upload.call_args_list]
     assert paths.index(foo_path) > paths.index(bar_path)

tests/remotes/azure.py

@@ -22,6 +22,7 @@ TEST_AZURE_CONNECTION_STRING = (
 
 class Azure(Base, CloudURLInfo):
 
+    IS_OBJECT_STORAGE = True
     CONNECTION_STRING = None
 
     @cached_property

tests/remotes/base.py

@@ -7,6 +7,9 @@ from dvc.path_info import URLInfo
 
 
 class Base(URLInfo):
+
+    IS_OBJECT_STORAGE = False
+
     def is_file(self):
         raise NotImplementedError
 

tests/remotes/gs.py

@@ -23,6 +23,9 @@ os.environ["GOOGLE_APPLICATION_CREDENTIALS"] = TEST_GCP_CREDS_FILE
 
 
 class GCP(Base, CloudURLInfo):
+
+    IS_OBJECT_STORAGE = True
+
     @staticmethod
     def should_test():
         do_test = env2bool("DVC_TEST_GCP", undefined=None)

tests/remotes/oss.py

@@ -16,6 +16,9 @@ EMULATOR_OSS_ACCESS_KEY_SECRET = "AccessKeySecret"
 
 
 class OSS(Base, CloudURLInfo):
+
+    IS_OBJECT_STORAGE = True
+
     @staticmethod
     def get_url():
         return f"oss://{TEST_OSS_REPO_BUCKET}/{uuid.uuid4()}"

tests/remotes/s3.py

@@ -15,6 +15,9 @@ TEST_AWS_REPO_BUCKET = os.environ.get("DVC_TEST_AWS_REPO_BUCKET", "dvc-temp")
 
 
 class S3(Base, CloudURLInfo):
+
+    IS_OBJECT_STORAGE = True
+
     @staticmethod
     def should_test():
         do_test = env2bool("DVC_TEST_AWS", undefined=None)

tests/unit/command/test_plots.py

@@ -1,3 +1,5 @@
+import logging
+
 from dvc.cli import parse_args
 from dvc.command.plots import CmdPlotsDiff, CmdPlotsShow
 
@@ -55,7 +57,7 @@ def test_plots_diff(dvc, mocker):
     )
 
 
-def test_plots_show(dvc, mocker):
+def test_plots_show_vega(dvc, mocker):
     cli_args = parse_args(
         [
             "plots",
@@ -85,7 +87,7 @@ def test_plots_show(dvc, mocker):
     )
 
 
-def test_plots_show_vega(dvc, mocker, caplog):
+def test_plots_diff_vega(dvc, mocker, caplog):
     cli_args = parse_args(
         [
             "plots",
@@ -103,3 +105,39 @@ def test_plots_show_vega(dvc, mocker, caplog):
     )
     assert cmd.run() == 0
     assert "plothtml" in caplog.text
+
+
+def test_plots_diff_open(tmp_dir, dvc, mocker, caplog):
+    mocked_open = mocker.patch("webbrowser.open", return_value=True)
+    cli_args = parse_args(["plots", "diff", "--targets", "datafile", "--open"])
+    cmd = cli_args.func(cli_args)
+    mocker.patch(
+        "dvc.repo.plots.diff.diff", return_value={"datafile": "filledtemplate"}
+    )
+
+    assert cmd.run() == 0
+
+    expected_url = f"file://{tmp_dir / 'plots.html'}"
+    assert expected_url in caplog.text
+
+    mocked_open.assert_called_once_with(expected_url)
+
+
+def test_plots_diff_open_failed(tmp_dir, dvc, mocker, caplog):
+    mocked_open = mocker.patch("webbrowser.open", return_value=False)
+    cli_args = parse_args(["plots", "diff", "--targets", "datafile", "--open"])
+    cmd = cli_args.func(cli_args)
+    mocker.patch(
+        "dvc.repo.plots.diff.diff", return_value={"datafile": "filledtemplate"}
+    )
+
+    assert cmd.run() == 1
+
+    expected_url = f"file://{tmp_dir / 'plots.html'}"
+    mocked_open.assert_called_once_with(expected_url)
+
+    error_message = "Failed to open. Please try opening it manually."
+    assert caplog.record_tuples == [
+        ("dvc.command.plots", logging.INFO, expected_url),
+        ("dvc.command.plots", logging.ERROR, error_message),
+    ]

tests/unit/fs/test_dvc.py

@@ -229,15 +229,15 @@ def test_get_hash_granular(tmp_dir, dvc):
     fs = DvcFileSystem(dvc)
     subdir = PathInfo(tmp_dir) / "dir" / "subdir"
     assert fs.info(subdir).get("md5") is None
-    assert stage(dvc.odb.local, subdir, fs).hash_info == HashInfo(
+    assert stage(dvc.odb.local, subdir, fs, "md5").hash_info == HashInfo(
         "md5", "af314506f1622d107e0ed3f14ec1a3b5.dir",
     )
     assert (
         fs.info(subdir / "data")["md5"] == "8d777f385d3dfec8815d20f7496026dc"
     )
-    assert stage(dvc.odb.local, subdir / "data", fs).hash_info == HashInfo(
-        "md5", "8d777f385d3dfec8815d20f7496026dc",
-    )
+    assert stage(
+        dvc.odb.local, subdir / "data", fs, "md5"
+    ).hash_info == HashInfo("md5", "8d777f385d3dfec8815d20f7496026dc",)
 
 
 def test_get_hash_dirty_file(tmp_dir, dvc):
@@ -248,7 +248,7 @@ def test_get_hash_dirty_file(tmp_dir, dvc):
     expected = "8c7dd922ad47494fc02c388e12c00eac"
     assert fs.info(PathInfo(tmp_dir) / "file").get("md5") == expected
     assert stage(
-        dvc.odb.local, PathInfo(tmp_dir) / "file", fs
+        dvc.odb.local, PathInfo(tmp_dir) / "file", fs, "md5"
     ).hash_info == HashInfo("md5", expected)
 
 
@@ -260,5 +260,5 @@ def test_get_hash_dirty_dir(tmp_dir, dvc):
     expected = "5ea40360f5b4ec688df672a4db9c17d1.dir"
     assert fs.info(PathInfo(tmp_dir) / "dir").get("md5") == expected
     assert stage(
-        dvc.odb.local, PathInfo(tmp_dir) / "dir", fs
+        dvc.odb.local, PathInfo(tmp_dir) / "dir", fs, "md5"
     ).hash_info == HashInfo("md5", expected)

tests/unit/fs/test_repo.py

@@ -483,7 +483,7 @@ def test_get_hash_cached_file(tmp_dir, dvc, mocker):
     expected = "acbd18db4cc2f85cedef654fccc4a4d8"
     assert fs.info(PathInfo(tmp_dir) / "foo").get("md5") is None
     assert stage(
-        dvc.odb.local, PathInfo(tmp_dir) / "foo", fs
+        dvc.odb.local, PathInfo(tmp_dir) / "foo", fs, "md5",
     ).hash_info == HashInfo("md5", expected,)
     (tmp_dir / "foo").unlink()
     assert fs.info(PathInfo(tmp_dir) / "foo")["md5"] == expected
@@ -497,13 +497,13 @@ def test_get_hash_cached_dir(tmp_dir, dvc, mocker):
     expected = "8761c4e9acad696bee718615e23e22db.dir"
     assert fs.info(PathInfo(tmp_dir) / "dir").get("md5") is None
     assert stage(
-        dvc.odb.local, PathInfo(tmp_dir) / "dir", fs
+        dvc.odb.local, PathInfo(tmp_dir) / "dir", fs, "md5",
     ).hash_info == HashInfo("md5", "8761c4e9acad696bee718615e23e22db.dir",)
 
     shutil.rmtree(tmp_dir / "dir")
     assert fs.info(PathInfo(tmp_dir) / "dir")["md5"] == expected
     assert stage(
-        dvc.odb.local, PathInfo(tmp_dir) / "dir", fs
+        dvc.odb.local, PathInfo(tmp_dir) / "dir", fs, "md5",
     ).hash_info == HashInfo("md5", "8761c4e9acad696bee718615e23e22db.dir",)
 
 
@@ -514,13 +514,13 @@ def test_get_hash_cached_granular(tmp_dir, dvc, mocker):
     fs = RepoFileSystem(dvc)
     subdir = PathInfo(tmp_dir) / "dir" / "subdir"
     assert fs.info(subdir).get("md5") is None
-    assert stage(dvc.odb.local, subdir, fs).hash_info == HashInfo(
+    assert stage(dvc.odb.local, subdir, fs, "md5").hash_info == HashInfo(
         "md5", "af314506f1622d107e0ed3f14ec1a3b5.dir",
     )
     assert fs.info(subdir / "data").get("md5") is None
-    assert stage(dvc.odb.local, subdir / "data", fs).hash_info == HashInfo(
-        "md5", "8d777f385d3dfec8815d20f7496026dc",
-    )
+    assert stage(
+        dvc.odb.local, subdir / "data", fs, "md5"
+    ).hash_info == HashInfo("md5", "8d777f385d3dfec8815d20f7496026dc",)
     (tmp_dir / "dir" / "subdir" / "data").unlink()
     assert (
         fs.info(subdir / "data")["md5"] == "8d777f385d3dfec8815d20f7496026dc"
@@ -540,7 +540,9 @@ def test_get_hash_mixed_dir(tmp_dir, scm, dvc):
     tmp_dir.scm.commit("add dir")
 
     fs = RepoFileSystem(dvc)
-    actual = stage(dvc.odb.local, PathInfo(tmp_dir) / "dir", fs).hash_info
+    actual = stage(
+        dvc.odb.local, PathInfo(tmp_dir) / "dir", fs, "md5"
+    ).hash_info
     expected = HashInfo("md5", "e1d9e8eae5374860ae025ec84cfd85c7.dir")
     assert actual == expected
 
@@ -551,7 +553,9 @@ def test_get_hash_dirty_file(tmp_dir, dvc):
 
     fs = RepoFileSystem(dvc)
     assert fs.info(PathInfo(tmp_dir) / "file").get("md5") is None
-    actual = stage(dvc.odb.local, PathInfo(tmp_dir) / "file", fs).hash_info
+    actual = stage(
+        dvc.odb.local, PathInfo(tmp_dir) / "file", fs, "md5"
+    ).hash_info
     expected = HashInfo("md5", "437b930db84b8079c2dd804a71936b5f")
     assert actual == expected
 
@@ -560,7 +564,9 @@ def test_get_hash_dirty_file(tmp_dir, dvc):
         fs.info(PathInfo(tmp_dir) / "file")["md5"]
         == "8c7dd922ad47494fc02c388e12c00eac"
     )
-    actual = stage(dvc.odb.local, PathInfo(tmp_dir) / "file", fs).hash_info
+    actual = stage(
+        dvc.odb.local, PathInfo(tmp_dir) / "file", fs, "md5"
+    ).hash_info
     expected = HashInfo("md5", "8c7dd922ad47494fc02c388e12c00eac")
     assert actual == expected
 
@@ -570,10 +576,12 @@ def test_get_hash_dirty_dir(tmp_dir, dvc):
     (tmp_dir / "dir" / "baz").write_text("baz")
 
     fs = RepoFileSystem(dvc)
-    actual = stage(dvc.odb.local, PathInfo(tmp_dir) / "dir", fs).hash_info
+    actual = stage(
+        dvc.odb.local, PathInfo(tmp_dir) / "dir", fs, "md5"
+    ).hash_info
     expected = HashInfo("md5", "ba75a2162ca9c29acecb7957105a0bc2.dir")
     assert actual == expected
-    assert actual.dir_info.nfiles == 3
+    assert actual.nfiles == 3
 
 
 @pytest.mark.parametrize("traverse_subrepos", [True, False])

tests/unit/objects/test_tree.py

@@ -2,8 +2,9 @@ from operator import itemgetter
 
 import pytest
 
-from dvc.dir_info import DirInfo, _merge
 from dvc.hash_info import HashInfo
+from dvc.objects.file import HashFile
+from dvc.objects.tree import Tree, _merge
 
 
 @pytest.mark.parametrize(
@@ -18,10 +19,10 @@ from dvc.hash_info import HashInfo
                 {"md5": "456", "relpath": "bar"},
             ],
             {
-                ("zzz",): HashInfo("md5", "def"),
-                ("foo",): HashInfo("md5", "123"),
-                ("bar",): HashInfo("md5", "456"),
-                ("aaa",): HashInfo("md5", "abc"),
+                ("zzz",): HashFile(None, None, HashInfo("md5", "def")),
+                ("foo",): HashFile(None, None, HashInfo("md5", "123")),
+                ("bar",): HashFile(None, None, HashInfo("md5", "456")),
+                ("aaa",): HashFile(None, None, HashInfo("md5", "abc")),
             },
         ),
         (
@@ -37,23 +38,29 @@ from dvc.hash_info import HashInfo
                 {"md5": "pqr", "relpath": "dir/subdir/a"},
             ],
             {
-                ("dir", "b"): HashInfo("md5", "123"),
-                ("dir", "z"): HashInfo("md5", "456"),
-                ("dir", "a"): HashInfo("md5", "789"),
-                ("b",): HashInfo("md5", "abc"),
-                ("a",): HashInfo("md5", "def"),
-                ("z",): HashInfo("md5", "ghi"),
-                ("dir", "subdir", "b"): HashInfo("md5", "jkl"),
-                ("dir", "subdir", "z"): HashInfo("md5", "mno"),
-                ("dir", "subdir", "a"): HashInfo("md5", "pqr"),
+                ("dir", "b"): HashFile(None, None, HashInfo("md5", "123")),
+                ("dir", "z"): HashFile(None, None, HashInfo("md5", "456")),
+                ("dir", "a"): HashFile(None, None, HashInfo("md5", "789")),
+                ("b",): HashFile(None, None, HashInfo("md5", "abc")),
+                ("a",): HashFile(None, None, HashInfo("md5", "def")),
+                ("z",): HashFile(None, None, HashInfo("md5", "ghi")),
+                ("dir", "subdir", "b"): HashFile(
+                    None, None, HashInfo("md5", "jkl")
+                ),
+                ("dir", "subdir", "z"): HashFile(
+                    None, None, HashInfo("md5", "mno")
+                ),
+                ("dir", "subdir", "a"): HashFile(
+                    None, None, HashInfo("md5", "pqr")
+                ),
             },
         ),
     ],
 )
 def test_list(lst, trie_dict):
-    dir_info = DirInfo.from_list(lst)
-    assert dir_info._dict == trie_dict
-    assert dir_info.to_list() == sorted(lst, key=itemgetter("relpath"))
+    tree = Tree.from_list(lst)
+    assert tree._dict == trie_dict
+    assert tree.as_list() == sorted(lst, key=itemgetter("relpath"))
 
 
 @pytest.mark.parametrize(
@@ -81,9 +88,9 @@ def test_list(lst, trie_dict):
     ],
 )
 def test_size(trie_dict, size):
-    dir_info = DirInfo()
-    dir_info._dict = trie_dict
-    assert dir_info.size == size
+    tree = Tree(None, None, None)
+    tree._dict = trie_dict
+    assert tree.size == size
 
 
 @pytest.mark.parametrize(
@@ -111,9 +118,9 @@ def test_size(trie_dict, size):
     ],
 )
 def test_nfiles(trie_dict, nfiles):
-    dir_info = DirInfo()
-    dir_info._dict = trie_dict
-    assert dir_info.nfiles == nfiles
+    tree = Tree(None, None, None)
+    tree._dict = trie_dict
+    assert len(tree) == nfiles
 
 
 @pytest.mark.parametrize(
@@ -134,9 +141,9 @@ def test_nfiles(trie_dict, nfiles):
     ],
 )
 def test_items(trie_dict):
-    dir_info = DirInfo()
-    dir_info._dict = trie_dict
-    assert list(dir_info.items()) == list(trie_dict.items())
+    tree = Tree(None, None, None)
+    tree._dict = trie_dict
+    assert list(tree) == list(trie_dict.items())
 
 
 @pytest.mark.parametrize(

tests/unit/output/test_local.py

@@ -2,7 +2,6 @@ import os
 
 from mock import patch
 
-from dvc.dir_info import DirInfo
 from dvc.hash_info import HashInfo
 from dvc.output import LocalOutput
 from dvc.stage import Stage
@@ -80,17 +79,7 @@ class TestGetFilesNumber(TestDvc):
 
     def test_return_multiple_for_dir(self):
         o = self._get_output()
-        o.hash_info = HashInfo(
-            "md5",
-            "12345678.dir",
-            nfiles=2,
-            dir_info=DirInfo.from_list(
-                [
-                    {"relpath": "foo", "md5": "asdf"},
-                    {"relpath": "bar", "md5": "qwe"},
-                ]
-            ),
-        )
+        o.hash_info = HashInfo("md5", "12345678.dir", nfiles=2,)
         self.assertEqual(2, o.get_files_number())
 
     @patch.object(LocalOutput, "is_dir_checksum", False)

tests/unit/remote/test_http.py

@@ -1,7 +1,11 @@
+import io
+
 import pytest
+import requests
 
 from dvc.exceptions import HTTPError
 from dvc.fs.http import HTTPFileSystem
+from dvc.path_info import URLInfo
 
 
 def test_download_fails_on_error_code(dvc, http):
@@ -128,12 +132,6 @@ def test_http_method(dvc):
 
 
 def test_exists(mocker):
-    import io
-
-    import requests
-
-    from dvc.path_info import URLInfo
-
     res = requests.Response()
     # need to add `raw`, as `exists()` fallbacks to a streaming GET requests
     # on HEAD request failure.
@@ -153,3 +151,20 @@ def test_exists(mocker):
     res.status_code = 403
     with pytest.raises(HTTPError):
         fs.exists(url)
+
+
+@pytest.mark.parametrize(
+    "headers, expected_size", [({"Content-Length": "3"}, 3), ({}, None)]
+)
+def test_content_length(mocker, headers, expected_size):
+    res = requests.Response()
+    res.headers.update(headers)
+    res.status_code = 200
+
+    fs = HTTPFileSystem(None, {})
+    mocker.patch.object(fs, "request", return_value=res)
+
+    url = URLInfo("https://example.org/file.txt")
+
+    assert fs.info(url) == {"etag": None, "size": expected_size}
+    assert fs._content_length(res) == expected_size

tests/unit/scm/test_git.py

@@ -528,20 +528,10 @@ def test_reset(tmp_dir, scm, git):
     assert len(unstaged) == 2
 
 
-def test_remind_to_track(tmp_dir, scm, caplog):
-    files = ["fname with spaces.txt", "тест", "foo"]
-    tmp_dir.gen({name: "foo" for name in files})
-    scm.files_to_track.update(files)
-
+def test_remind_to_track(scm, caplog):
+    scm.files_to_track = ["fname with spaces.txt", "тест", "foo"]
     scm.remind_to_track()
-    assert "git add 'fname with spaces.txt' foo 'тест'" in caplog.text
-
-    scm.add(["foo"])
-    caplog.clear()
-
-    scm.remind_to_track()
-    assert "git add 'fname with spaces.txt' 'тест'" in caplog.text
-    assert "foo" not in caplog.text
+    assert "git add 'fname with spaces.txt' 'тест' foo" in caplog.text
 
 
 def test_add(tmp_dir, scm, git):

tests/unit/test_logger.py

@@ -248,3 +248,31 @@ def test_info_with_debug_loglevel_shows_no_datetime(caplog, dt):
         logger.info("message")
 
         assert "message" == formatter.format(caplog.records[0])
+
+
+def test_add_existing_level(caplog, dt):
+    # Common pattern to configure logging level in external libraries
+    # eg:
+    # https://github.com/bokeh/bokeh/blob/04bb30fef2e72e64baaa8b2f330806d5bfdd3b11/
+    # bokeh/util/logconfig.py#L79-L85
+    TRACE2 = 4
+    logging.addLevelName(TRACE2, "TRACE2")
+    logging.TRACE2 = TRACE2
+
+    dvc.logger.addLoggingLevel("TRACE2", 2)
+
+    # DVC sets all expected entrypoints, but doesn't override the level
+    assert logging.TRACE2 == 4
+    assert hasattr(logging, "trace2")
+    assert hasattr(logger, "trace2")
+    assert logging.getLevelName("TRACE2") == 4
+
+    # The TRACE2 logging level uses the original, higher logging level
+    with caplog.at_level(logging.TRACE2, logger="dvc"):
+        logger.trace2("TRACE2")
+    assert len(caplog.records) == 1
+
+    (record,) = caplog.records
+    assert record.levelno == 4
+    assert record.levelname == "TRACE2"
+    assert record.message == "TRACE2"