diff --git a/dvc/api.py b/dvc/api.py
index a2d648aa1..1202ebec4 100644
--- a/dvc/api.py
+++ b/dvc/api.py
@@ -30,7 +30,7 @@ def get_url(path, repo=None, rev=None, remote=None):
             raise UrlNotDvcRepoError(_repo.url)  # pylint: disable=no-member
         out = _repo.find_out_by_relpath(path)
         remote_obj = _repo.cloud.get_remote(remote)
-        return str(remote_obj.tree.hash_to_path_info(out.checksum))
+        return str(remote_obj.tree.hash_to_path_info(out.hash_info.value))
 
 
 def open(  # noqa, pylint: disable=redefined-builtin
diff --git a/dvc/cache/base.py b/dvc/cache/base.py
index 058812fb5..92d5b5bd0 100644
--- a/dvc/cache/base.py
+++ b/dvc/cache/base.py
@@ -11,6 +11,7 @@ from dvc.exceptions import (
     DvcException,
     MergeError,
 )
+from dvc.hash_info import HashInfo
 from dvc.path_info import WindowsPathInfo
 from dvc.progress import Tqdm
 from dvc.remote.slow_link_detection import slow_link_guard
@@ -56,29 +57,29 @@ class CloudCache:
         self.cache_type_confirmed = False
         self._dir_info = {}
 
-    def get_dir_cache(self, hash_):
-        assert hash_
+    def get_dir_cache(self, hash_info):
+        assert hash_info
 
-        dir_info = self._dir_info.get(hash_)
+        dir_info = self._dir_info.get(hash_info.value)
         if dir_info:
             return dir_info
 
         try:
-            dir_info = self.load_dir_cache(hash_)
+            dir_info = self.load_dir_cache(hash_info)
         except DirCacheError:
             dir_info = []
 
-        self._dir_info[hash_] = dir_info
+        self._dir_info[hash_info.value] = dir_info
         return dir_info
 
-    def load_dir_cache(self, hash_):
-        path_info = self.tree.hash_to_path_info(hash_)
+    def load_dir_cache(self, hash_info):
+        path_info = self.tree.hash_to_path_info(hash_info.value)
 
         try:
             with self.tree.open(path_info, "r") as fobj:
                 d = json.load(fobj)
         except (ValueError, FileNotFoundError) as exc:
-            raise DirCacheError(hash_) from exc
+            raise DirCacheError(hash_info) from exc
 
         if not isinstance(d, list):
             logger.error(
@@ -123,20 +124,21 @@ class CloudCache:
             logger.debug("'%s' doesn't exist.", path_info)
             return True
 
-        hash_ = hash_info.get(self.tree.PARAM_CHECKSUM)
-        if hash_ is None:
+        if not hash_info:
             logger.debug("hash value for '%s' is missing.", path_info)
             return True
 
-        if self.changed_cache(hash_):
-            logger.debug("cache for '%s'('%s') has changed.", path_info, hash_)
+        if self.changed_cache(hash_info):
+            logger.debug(
+                "cache for '%s'('%s') has changed.", path_info, hash_info
+            )
             return True
 
         actual = self.tree.get_hash(path_info)
-        if hash_ != actual.value:
+        if hash_info != actual:
             logger.debug(
                 "hash value '%s' for '%s' has changed (actual '%s').",
-                hash_,
+                hash_info,
                 actual,
                 path_info,
             )
@@ -193,12 +195,12 @@ class CloudCache:
             "Created '%s': %s -> %s", self.cache_types[0], from_info, to_info,
         )
 
-    def _save_file(self, path_info, tree, hash_, save_link=True, **kwargs):
-        assert hash_
+    def _save_file(self, path_info, tree, hash_info, save_link=True, **kwargs):
+        assert hash_info
 
-        cache_info = self.tree.hash_to_path_info(hash_)
+        cache_info = self.tree.hash_to_path_info(hash_info.value)
         if tree == self.tree:
-            if self.changed_cache(hash_):
+            if self.changed_cache(hash_info):
                 self.tree.move(path_info, cache_info, mode=self.CACHE_MODE)
                 self.link(cache_info, path_info)
             elif self.tree.iscopy(path_info) and self._cache_is_copy(
@@ -215,9 +217,9 @@ class CloudCache:
             # we need to update path and cache, since in case of reflink,
             # or copy cache type moving original file results in updates on
             # next executed command, which causes md5 recalculation
-            self.tree.state.save(path_info, hash_)
+            self.tree.state.save(path_info, hash_info.value)
         else:
-            if self.changed_cache(hash_):
+            if self.changed_cache(hash_info):
                 with tree.open(path_info, mode="rb") as fobj:
                     # if tree has fetch enabled, DVC out will be fetched on
                     # open and we do not need to read/copy any data
@@ -229,8 +231,7 @@ class CloudCache:
                 if callback:
                     callback(1)
 
-        self.tree.state.save(cache_info, hash_)
-        return {self.tree.PARAM_CHECKSUM: hash_}
+        self.tree.state.save(cache_info, hash_info.value)
 
     def _cache_is_copy(self, path_info):
         """Checks whether cache uses copies."""
@@ -254,13 +255,15 @@ class CloudCache:
         self.cache_type_confirmed = True
         return self.cache_types[0] == "copy"
 
-    def _save_dir(self, path_info, tree, hash_, save_link=True, **kwargs):
-        dir_info = self.get_dir_cache(hash_)
+    def _save_dir(self, path_info, tree, hash_info, save_link=True, **kwargs):
+        dir_info = self.get_dir_cache(hash_info)
         for entry in Tqdm(
             dir_info, desc="Saving " + path_info.name, unit="file"
         ):
             entry_info = path_info / entry[self.tree.PARAM_RELPATH]
-            entry_hash = entry[self.tree.PARAM_CHECKSUM]
+            entry_hash = HashInfo(
+                self.tree.PARAM_CHECKSUM, entry[self.tree.PARAM_CHECKSUM]
+            )
             self._save_file(
                 entry_info, tree, entry_hash, save_link=False, **kwargs
             )
@@ -268,11 +271,10 @@ class CloudCache:
         if save_link:
             self.tree.state.save_link(path_info)
         if self.tree.exists(path_info):
-            self.tree.state.save(path_info, hash_)
+            self.tree.state.save(path_info, hash_info.value)
 
-        cache_info = self.tree.hash_to_path_info(hash_)
-        self.tree.state.save(cache_info, hash_)
-        return {self.tree.PARAM_CHECKSUM: hash_}
+        cache_info = self.tree.hash_to_path_info(hash_info.value)
+        self.tree.state.save(cache_info, hash_info.value)
 
     def save(self, path_info, tree, hash_info, save_link=True, **kwargs):
         if path_info.scheme != self.tree.scheme:
@@ -281,23 +283,23 @@ class CloudCache:
                 self.tree.scheme,
             )
 
-        hash_ = hash_info[self.tree.PARAM_CHECKSUM]
-        return self._save(path_info, tree, hash_, save_link, **kwargs)
+        self._save(path_info, tree, hash_info, save_link, **kwargs)
 
-    def _save(self, path_info, tree, hash_, save_link=True, **kwargs):
-        to_info = self.tree.hash_to_path_info(hash_)
+    def _save(self, path_info, tree, hash_info, save_link=True, **kwargs):
+        to_info = self.tree.hash_to_path_info(hash_info.value)
         logger.debug("Saving '%s' to '%s'.", path_info, to_info)
 
         if tree.isdir(path_info):
-            return self._save_dir(path_info, tree, hash_, save_link, **kwargs)
-        return self._save_file(path_info, tree, hash_, save_link, **kwargs)
+            self._save_dir(path_info, tree, hash_info, save_link, **kwargs)
+        else:
+            self._save_file(path_info, tree, hash_info, save_link, **kwargs)
 
     # Override to return path as a string instead of PathInfo for clouds
     # which support string paths (see local)
     def hash_to_path(self, hash_):
         return self.tree.hash_to_path_info(hash_)
 
-    def changed_cache_file(self, hash_):
+    def changed_cache_file(self, hash_info):
         """Compare the given hash with the (corresponding) actual one.
 
         - Use `State` as a cache for computed hashes
@@ -310,7 +312,7 @@ class CloudCache:
         - Remove the file from cache if it doesn't match the actual hash
         """
         # Prefer string path over PathInfo when possible due to performance
-        cache_info = self.hash_to_path(hash_)
+        cache_info = self.hash_to_path(hash_info.value)
         if self.tree.is_protected(cache_info):
             logger.debug(
                 "Assuming '%s' is unchanged since it is read-only", cache_info
@@ -320,13 +322,16 @@ class CloudCache:
         actual = self.tree.get_hash(cache_info)
 
         logger.debug(
-            "cache '%s' expected '%s' actual '%s'", cache_info, hash_, actual,
+            "cache '%s' expected '%s' actual '%s'",
+            cache_info,
+            hash_info,
+            actual,
         )
 
-        if not hash_ or not actual:
+        if not hash_info or not actual:
             return True
 
-        if actual.value.split(".")[0] == hash_.split(".")[0]:
+        if actual.value.split(".")[0] == hash_info.value.split(".")[0]:
             # making cache file read-only so we don't need to check it
             # next time
             self.tree.protect(cache_info)
@@ -338,12 +343,14 @@ class CloudCache:
 
         return True
 
-    def _changed_dir_cache(self, hash_, path_info=None, filter_info=None):
-        if self.changed_cache_file(hash_):
+    def _changed_dir_cache(self, hash_info, path_info=None, filter_info=None):
+        if self.changed_cache_file(hash_info):
             return True
 
-        for entry in self.get_dir_cache(hash_):
-            entry_hash = entry[self.tree.PARAM_CHECKSUM]
+        for entry in self.get_dir_cache(hash_info):
+            entry_hash = HashInfo(
+                self.tree.PARAM_CHECKSUM, entry[self.tree.PARAM_CHECKSUM]
+            )
 
             if path_info and filter_info:
                 entry_info = path_info / entry[self.tree.PARAM_RELPATH]
@@ -355,12 +362,12 @@ class CloudCache:
 
         return False
 
-    def changed_cache(self, hash_, path_info=None, filter_info=None):
-        if self.tree.is_dir_hash(hash_):
+    def changed_cache(self, hash_info, path_info=None, filter_info=None):
+        if hash_info.isdir:
             return self._changed_dir_cache(
-                hash_, path_info=path_info, filter_info=filter_info
+                hash_info, path_info=path_info, filter_info=filter_info
             )
-        return self.changed_cache_file(hash_)
+        return self.changed_cache_file(hash_info)
 
     def already_cached(self, path_info):
         _, current = self.tree.get_hash(path_info)
@@ -386,11 +393,11 @@ class CloudCache:
         self.tree.remove(path_info)
 
     def _checkout_file(
-        self, path_info, hash_, force, progress_callback=None, relink=False
+        self, path_info, hash_info, force, progress_callback=None, relink=False
     ):
         """The file is changed we need to checkout a new copy"""
         added, modified = True, False
-        cache_info = self.tree.hash_to_path_info(hash_)
+        cache_info = self.tree.hash_to_path_info(hash_info.value)
         if self.tree.exists(path_info):
             logger.debug("data '%s' will be replaced.", path_info)
             self.safe_remove(path_info, force=force)
@@ -398,7 +405,7 @@ class CloudCache:
 
         self.link(cache_info, path_info)
         self.tree.state.save_link(path_info)
-        self.tree.state.save(path_info, hash_)
+        self.tree.state.save(path_info, hash_info.value)
         if progress_callback:
             progress_callback(str(path_info))
 
@@ -407,7 +414,7 @@ class CloudCache:
     def _checkout_dir(
         self,
         path_info,
-        hash_,
+        hash_info,
         force,
         progress_callback=None,
         relink=False,
@@ -420,7 +427,7 @@ class CloudCache:
             added = True
             self.tree.makedirs(path_info)
 
-        dir_info = self.get_dir_cache(hash_)
+        dir_info = self.get_dir_cache(hash_info)
 
         logger.debug("Linking directory '%s'.", path_info)
 
@@ -433,7 +440,7 @@ class CloudCache:
             if filter_info and not entry_info.isin_or_eq(filter_info):
                 continue
 
-            entry_hash_info = {self.tree.PARAM_CHECKSUM: entry_hash}
+            entry_hash_info = HashInfo(self.tree.PARAM_CHECKSUM, entry_hash)
             if relink or self.changed(entry_info, entry_hash_info):
                 modified = True
                 self.safe_remove(entry_info, force=force)
@@ -448,7 +455,7 @@ class CloudCache:
         )
 
         self.tree.state.save_link(path_info)
-        self.tree.state.save(path_info, hash_)
+        self.tree.state.save(path_info, hash_info.value)
 
         # relink is not modified, assume it as nochange
         return added, not added and modified and not relink
@@ -477,10 +484,9 @@ class CloudCache:
         if path_info.scheme not in ["local", self.tree.scheme]:
             raise NotImplementedError
 
-        hash_ = hash_info.get(self.tree.PARAM_CHECKSUM)
         failed = None
         skip = False
-        if not hash_:
+        if not hash_info:
             logger.warning(
                 "No file hash info found for '%s'. " "It won't be created.",
                 path_info,
@@ -493,11 +499,11 @@ class CloudCache:
             skip = True
 
         elif self.changed_cache(
-            hash_, path_info=path_info, filter_info=filter_info
+            hash_info, path_info=path_info, filter_info=filter_info
         ):
             logger.warning(
                 "Cache '%s' not found. File '%s' won't be created.",
-                hash_,
+                hash_info,
                 path_info,
             )
             self.safe_remove(path_info, force=force)
@@ -508,52 +514,59 @@ class CloudCache:
                 progress_callback(
                     str(path_info),
                     self.get_files_number(
-                        self.tree.path_info, hash_, filter_info
+                        self.tree.path_info, hash_info, filter_info
                     ),
                 )
             if failed:
                 raise CheckoutError([failed])
             return
 
-        logger.debug("Checking out '%s' with cache '%s'.", path_info, hash_)
+        logger.debug(
+            "Checking out '%s' with cache '%s'.", path_info, hash_info
+        )
 
         return self._checkout(
-            path_info, hash_, force, progress_callback, relink, filter_info,
+            path_info,
+            hash_info,
+            force,
+            progress_callback,
+            relink,
+            filter_info,
         )
 
     def _checkout(
         self,
         path_info,
-        hash_,
+        hash_info,
         force=False,
         progress_callback=None,
         relink=False,
         filter_info=None,
     ):
-        if not self.tree.is_dir_hash(hash_):
+        if not hash_info.isdir:
             return self._checkout_file(
-                path_info, hash_, force, progress_callback, relink
+                path_info, hash_info, force, progress_callback, relink
             )
 
         return self._checkout_dir(
-            path_info, hash_, force, progress_callback, relink, filter_info
+            path_info, hash_info, force, progress_callback, relink, filter_info
         )
 
-    def get_files_number(self, path_info, hash_, filter_info):
+    def get_files_number(self, path_info, hash_info, filter_info):
         from funcy.py3 import ilen
 
-        if not hash_:
+        if not hash_info:
             return 0
 
-        if not self.tree.is_dir_hash(hash_):
+        if not hash_info.isdir:
             return 1
 
         if not filter_info:
-            return len(self.get_dir_cache(hash_))
+            return len(self.get_dir_cache(hash_info))
 
         return ilen(
             filter_info.isin_or_eq(path_info / entry[self.tree.PARAM_CHECKSUM])
-            for entry in self.get_dir_cache(hash_)
+            for entry in self.get_dir_cache(hash_info)
         )
 
     def _to_dict(self, dir_info):
@@ -620,16 +633,12 @@ class CloudCache:
         assert their_info
 
         if ancestor_info:
-            ancestor_hash = ancestor_info.value
-            ancestor = self.get_dir_cache(ancestor_hash)
+            ancestor = self.get_dir_cache(ancestor_info)
         else:
             ancestor = []
 
-        our_hash = our_info.value
-        our = self.get_dir_cache(our_hash)
-
-        their_hash = their_info.value
-        their = self.get_dir_cache(their_hash)
+        our = self.get_dir_cache(our_info)
+        their = self.get_dir_cache(their_info)
 
         merged = self._merge_dirs(ancestor, our, their)
         return self.tree.save_dir_info(merged)
diff --git a/dvc/cache/local.py b/dvc/cache/local.py
index 3d63ada81..099086f01 100644
--- a/dvc/cache/local.py
+++ b/dvc/cache/local.py
@@ -7,6 +7,7 @@ from functools import partial, wraps
 from funcy import cached_property, concat
 
 from dvc.exceptions import DownloadError, UploadError
+from dvc.hash_info import HashInfo
 from dvc.path_info import PathInfo
 from dvc.progress import Tqdm
 
@@ -85,20 +86,19 @@ class LocalCache(CloudCache):
                 desc="Querying "
                 + ("cache in " + name if name else "local cache"),
             )
-            if not self.changed_cache_file(hash_)
+            if not self.changed_cache_file(
+                HashInfo(self.tree.PARAM_CHECKSUM, hash_)
+            )
         ]
 
     def already_cached(self, path_info):
         assert path_info.scheme in ["", "local"]
 
         current = self.tree.get_hash(path_info)
-
-        assert current.name == "md5"
-
         if not current:
             return False
 
-        return not self.changed_cache(current.value)
+        return not self.changed_cache(current)
 
     def _verify_link(self, path_info, link_type):
         if link_type == "hardlink" and self.tree.getsize(path_info) == 0:
diff --git a/dvc/external_repo.py b/dvc/external_repo.py
index b698637f5..7ab5e1e2b 100644
--- a/dvc/external_repo.py
+++ b/dvc/external_repo.py
@@ -144,7 +144,7 @@ class BaseExternalRepo:
                 raise PathMissingError(path, self.url)
             hash_info = self.repo_tree.get_hash(
                 path, download_callback=download_update
-            ).to_dict()
+            )
             self.local_cache.save(
                 path,
                 self.repo_tree,
diff --git a/dvc/hash_info.py b/dvc/hash_info.py
index d082ace59..fdf3c914c 100644
--- a/dvc/hash_info.py
+++ b/dvc/hash_info.py
@@ -1,5 +1,7 @@
 from dataclasses import dataclass
 
+HASH_DIR_SUFFIX = ".dir"
+
 
 @dataclass
 class HashInfo:
@@ -18,3 +20,9 @@ class HashInfo:
 
     def to_dict(self):
         return {self.name: self.value} if self else {}
+
+    @property
+    def isdir(self):
+        if not self:
+            return False
+        return self.value.endswith(HASH_DIR_SUFFIX)
diff --git a/dvc/output/base.py b/dvc/output/base.py
index d86793211..c219b436a 100644
--- a/dvc/output/base.py
+++ b/dvc/output/base.py
@@ -163,7 +163,7 @@ class BaseOutput:
 
     @property
     def dir_cache(self):
-        return self.cache.get_dir_cache(self.checksum)
+        return self.cache.get_dir_cache(self.hash_info)
 
     @classmethod
     def supported(cls, url):
@@ -171,34 +171,29 @@ class BaseOutput:
 
     @property
     def cache_path(self):
-        return self.cache.tree.hash_to_path_info(self.checksum).url
-
-    @property
-    def checksum(self):
-        return self.hash_info.value
-
-    def get_checksum(self):
-        return self.get_hash().value
+        return self.cache.tree.hash_to_path_info(self.hash_info.value).url
 
     def get_hash(self):
         return self.tree.get_hash(self.path_info)
 
     @property
     def is_dir_checksum(self):
-        return self.tree.is_dir_hash(self.checksum)
+        return self.hash_info.isdir
 
     @property
     def exists(self):
         return self.tree.exists(self.path_info)
 
     def changed_checksum(self):
-        return self.checksum != self.get_checksum()
+        return self.hash_info != self.get_hash()
 
     def changed_cache(self, filter_info=None):
-        if not self.use_cache or not self.checksum:
+        if not self.use_cache or not self.hash_info:
             return True
 
-        return self.cache.changed_cache(self.checksum, filter_info=filter_info)
+        return self.cache.changed_cache(
+            self.hash_info, filter_info=filter_info
+        )
 
     def workspace_status(self):
         if not self.exists:
@@ -207,13 +202,13 @@ class BaseOutput:
         if self.changed_checksum():
             return {str(self): "modified"}
 
-        if not self.checksum:
+        if not self.hash_info:
             return {str(self): "new"}
 
         return {}
 
     def status(self):
-        if self.checksum and self.use_cache and self.changed_cache():
+        if self.hash_info and self.use_cache and self.changed_cache():
             return {str(self): "not in cache"}
 
         return self.workspace_status()
@@ -286,9 +281,7 @@ class BaseOutput:
     def commit(self):
         assert self.hash_info
         if self.use_cache:
-            self.cache.save(
-                self.path_info, self.cache.tree, self.hash_info.to_dict()
-            )
+            self.cache.save(self.path_info, self.cache.tree, self.hash_info)
 
     def dumpd(self):
         ret = copy(self.hash_info.to_dict())
@@ -340,7 +333,7 @@ class BaseOutput:
 
         return self.cache.checkout(
             self.path_info,
-            self.hash_info.to_dict(),
+            self.hash_info,
             force=force,
             progress_callback=progress_callback,
             relink=relink,
@@ -374,7 +367,7 @@ class BaseOutput:
             return 0
 
         return self.cache.get_files_number(
-            self.path_info, self.checksum, filter_info
+            self.path_info, self.hash_info, filter_info
         )
 
     def unprotect(self):
@@ -384,9 +377,9 @@ class BaseOutput:
     def get_dir_cache(self, **kwargs):
         if not self.is_dir_checksum:
             raise DvcException("cannot get dir cache for file checksum")
-        if self.cache.changed_cache_file(self.checksum):
+        if self.cache.changed_cache_file(self.hash_info):
             self.repo.cloud.pull(
-                NamedCache.make("local", self.checksum, str(self)),
+                NamedCache.make("local", self.hash_info.value, str(self)),
                 show_checksums=False,
                 **kwargs,
             )
@@ -421,7 +414,7 @@ class BaseOutput:
         except DvcException:
             logger.debug(f"failed to pull cache for '{self}'")
 
-        if self.cache.changed_cache_file(self.checksum):
+        if self.cache.changed_cache_file(self.hash_info):
             msg = (
                 "Missing cache for directory '{}'. "
                 "Cache for files inside will be lost. "
@@ -471,7 +464,7 @@ class BaseOutput:
             cache.external[dep.repo_pair].add(dep.def_path)
             return cache
 
-        if not self.checksum:
+        if not self.hash_info:
             msg = (
                 "Output '{}'({}) is missing version info. "
                 "Cache for it will not be collected. "
@@ -490,13 +483,13 @@ class BaseOutput:
             logger.warning(msg)
             return NamedCache()
 
-        ret = NamedCache.make(self.scheme, self.checksum, str(self))
+        ret = NamedCache.make(self.scheme, self.hash_info.value, str(self))
 
         if not self.is_dir_checksum:
             return ret
 
         ret.add_child_cache(
-            self.checksum, self.collect_used_dir_cache(**kwargs),
+            self.hash_info.value, self.collect_used_dir_cache(**kwargs),
         )
 
         return ret
diff --git a/dvc/repo/diff.py b/dvc/repo/diff.py
index f02bf8b9e..75a10877b 100644
--- a/dvc/repo/diff.py
+++ b/dvc/repo/diff.py
@@ -38,7 +38,7 @@ def diff(self, a_rev="HEAD", b_rev=None):
         def _to_checksum(output):
             if on_working_tree:
                 return self.cache.local.tree.get_hash(output.path_info).value
-            return output.checksum
+            return output.hash_info.value
 
         def _exists(output):
             if on_working_tree:
diff --git a/dvc/stage/cache.py b/dvc/stage/cache.py
index 0e312c6d6..ee102723e 100644
--- a/dvc/stage/cache.py
+++ b/dvc/stage/cache.py
@@ -28,7 +28,7 @@ def _get_stage_hash(stage):
         return None
 
     for dep in stage.deps:
-        if not (dep.scheme == "local" and dep.def_path and dep.get_checksum()):
+        if not (dep.scheme == "local" and dep.def_path and dep.get_hash()):
             return None
 
     for out in stage.outs:
diff --git a/dvc/tree/base.py b/dvc/tree/base.py
index af7096f0d..329c90c88 100644
--- a/dvc/tree/base.py
+++ b/dvc/tree/base.py
@@ -343,7 +343,7 @@ class BaseTree:
     def save_dir_info(self, dir_info):
         hash_info, tmp_info = self._get_dir_info_hash(dir_info)
         new_info = self.cache.tree.hash_to_path_info(hash_info.value)
-        if self.cache.changed_cache_file(hash_info.value):
+        if self.cache.changed_cache_file(hash_info):
             self.cache.tree.makedirs(new_info.parent)
             self.cache.tree.move(
                 tmp_info, new_info, mode=self.cache.CACHE_MODE
diff --git a/dvc/tree/dvc.py b/dvc/tree/dvc.py
index 128683b38..3ae13493a 100644
--- a/dvc/tree/dvc.py
+++ b/dvc/tree/dvc.py
@@ -85,7 +85,7 @@ class DvcTree(BaseTree):  # pylint:disable=abstract-method
                 if out.is_dir_checksum:
                     checksum = self._get_granular_checksum(path, out)
                 else:
-                    checksum = out.checksum
+                    checksum = out.hash_info.value
                 try:
                     remote_info = remote_obj.tree.hash_to_path_info(checksum)
                     return remote_obj.tree.open(
@@ -246,7 +246,7 @@ class DvcTree(BaseTree):  # pylint:disable=abstract-method
                 out = outs[0]
                 # other code expects us to fetch the dir at this point
                 self._fetch_dir(out, **kwargs)
-                return HashInfo(out.tree.PARAM_CHECKSUM, out.checksum)
+                return out.hash_info
         except OutputNotFoundError:
             pass
 
@@ -262,7 +262,7 @@ class DvcTree(BaseTree):  # pylint:disable=abstract-method
                 out.tree.PARAM_CHECKSUM,
                 self._get_granular_checksum(path_info, out),
             )
-        return HashInfo(out.tree.PARAM_CHECKSUM, out.checksum)
+        return out.hash_info
 
     def metadata(self, path_info):
         path_info = PathInfo(os.path.abspath(path_info))
diff --git a/tests/func/remote/test_index.py b/tests/func/remote/test_index.py
index d33b452b6..64e28f412 100644
--- a/tests/func/remote/test_index.py
+++ b/tests/func/remote/test_index.py
@@ -39,9 +39,9 @@ def test_indexed_on_status(tmp_dir, dvc, tmp_path_factory, remote):
 
     dvc.status(cloud=True)
     with remote.index:
-        assert {bar.checksum, baz["md5"]} == set(remote.index.hashes())
-        assert [bar.checksum] == list(remote.index.dir_hashes())
-        assert foo.checksum not in remote.index.hashes()
+        assert {bar.hash_info.value, baz["md5"]} == set(remote.index.hashes())
+        assert [bar.hash_info.value] == list(remote.index.dir_hashes())
+        assert foo.hash_info.value not in remote.index.hashes()
 
 
 def test_indexed_on_push(tmp_dir, dvc, tmp_path_factory, remote):
@@ -51,15 +51,15 @@ def test_indexed_on_push(tmp_dir, dvc, tmp_path_factory, remote):
 
     dvc.push()
     with remote.index:
-        assert {bar.checksum, baz["md5"]} == set(remote.index.hashes())
-        assert [bar.checksum] == list(remote.index.dir_hashes())
-        assert foo.checksum not in remote.index.hashes()
+        assert {bar.hash_info.value, baz["md5"]} == set(remote.index.hashes())
+        assert [bar.hash_info.value] == list(remote.index.dir_hashes())
+        assert foo.hash_info.value not in remote.index.hashes()
 
 
 def test_indexed_dir_missing(tmp_dir, dvc, tmp_path_factory, remote):
     bar = tmp_dir.dvc_gen({"bar": {"baz": "baz content"}})[0].outs[0]
     with remote.index:
-        remote.index.update([bar.checksum], [])
+        remote.index.update([bar.hash_info.value], [])
     dvc.status(cloud=True)
     with remote.index:
         assert not list(remote.index.hashes())
diff --git a/tests/func/test_add.py b/tests/func/test_add.py
index e9e1f0d40..9b3003f59 100644
--- a/tests/func/test_add.py
+++ b/tests/func/test_add.py
@@ -72,9 +72,9 @@ def test_add_directory(tmp_dir, dvc):
     assert len(stage.deps) == 0
     assert len(stage.outs) == 1
 
-    md5 = stage.outs[0].hash_info.value
+    hash_info = stage.outs[0].hash_info
 
-    dir_info = dvc.cache.local.load_dir_cache(md5)
+    dir_info = dvc.cache.local.load_dir_cache(hash_info)
     for info in dir_info:
         assert "\\" not in info["relpath"]
 
diff --git a/tests/func/test_cache.py b/tests/func/test_cache.py
index 93a6cdaeb..a28f3a4f1 100644
--- a/tests/func/test_cache.py
+++ b/tests/func/test_cache.py
@@ -6,6 +6,7 @@ import pytest
 
 from dvc.cache import Cache
 from dvc.cache.base import DirCacheError
+from dvc.hash_info import HashInfo
 from dvc.main import main
 from dvc.utils import relpath
 from tests.basic_env import TestDir, TestDvc
@@ -52,14 +53,16 @@ class TestCacheLoadBadDirCache(TestDvc):
         )
         self.create(fname, "<clearly>not,json")
         with pytest.raises(DirCacheError):
-            self.dvc.cache.local.load_dir_cache(dir_hash)
+            self.dvc.cache.local.load_dir_cache(HashInfo("md5", dir_hash))
 
         dir_hash = "234.dir"
         fname = os.fspath(
             self.dvc.cache.local.tree.hash_to_path_info(dir_hash)
         )
         self.create(fname, '{"a": "b"}')
-        self._do_test(self.dvc.cache.local.load_dir_cache(dir_hash))
+        self._do_test(
+            self.dvc.cache.local.load_dir_cache(HashInfo("md5", dir_hash))
+        )
 
 
 class TestExternalCacheDir(TestDvc):
diff --git a/tests/func/test_checkout.py b/tests/func/test_checkout.py
index 67c5198e7..6bcebb4d6 100644
--- a/tests/func/test_checkout.py
+++ b/tests/func/test_checkout.py
@@ -100,7 +100,7 @@ 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 = self.dvc.cache.local.load_dir_cache(out.checksum)[0]
+        entry = self.dvc.cache.local.load_dir_cache(out.hash_info)[0]
         entry_hash = entry[self.dvc.cache.local.tree.PARAM_CHECKSUM]
         cache = os.fspath(
             self.dvc.cache.local.tree.hash_to_path_info(entry_hash)
diff --git a/tests/func/test_data_cloud.py b/tests/func/test_data_cloud.py
index a96683ce4..1fbaa70f0 100644
--- a/tests/func/test_data_cloud.py
+++ b/tests/func/test_data_cloud.py
@@ -27,7 +27,7 @@ def test_cloud(tmp_dir, dvc, remote):  # pylint:disable=unused-argument
     (stage,) = tmp_dir.dvc_gen("foo", "foo")
     out = stage.outs[0]
     cache = out.cache_path
-    md5 = out.checksum
+    md5 = out.hash_info.value
     info = out.get_used_cache()
 
     (stage_dir,) = tmp_dir.dvc_gen(
@@ -41,7 +41,7 @@ def test_cloud(tmp_dir, dvc, remote):  # pylint:disable=unused-argument
     out_dir = stage_dir.outs[0]
     cache_dir = out_dir.cache_path
     name_dir = str(out_dir)
-    md5_dir = out_dir.checksum
+    md5_dir = out_dir.hash_info.value
     info_dir = NamedCache.make(out_dir.scheme, md5_dir, name_dir)
 
     with dvc.state:
diff --git a/tests/func/test_gc.py b/tests/func/test_gc.py
index d19f9a1f4..317683cf5 100644
--- a/tests/func/test_gc.py
+++ b/tests/func/test_gc.py
@@ -370,8 +370,8 @@ def test_gc_external_output(tmp_dir, dvc, workspace):
     (foo_stage,) = dvc.add("remote://workspace/foo")
     (bar_stage,) = dvc.add("remote://workspace/bar")
 
-    foo_hash = foo_stage.outs[0].checksum
-    bar_hash = bar_stage.outs[0].checksum
+    foo_hash = foo_stage.outs[0].hash_info.value
+    bar_hash = bar_stage.outs[0].hash_info.value
 
     assert (
         workspace / "cache" / foo_hash[:2] / foo_hash[2:]
diff --git a/tests/func/test_remote.py b/tests/func/test_remote.py
index 0eb2ded24..a96865830 100644
--- a/tests/func/test_remote.py
+++ b/tests/func/test_remote.py
@@ -193,11 +193,13 @@ def test_partial_push_n_pull(tmp_dir, dvc, tmp_path_factory, local_remote):
 
         remote = dvc.cloud.get_remote("upstream")
         assert not remote.tree.exists(
-            remote.tree.hash_to_path_info(foo.checksum)
+            remote.tree.hash_to_path_info(foo.hash_info.value)
+        )
+        assert remote.tree.exists(
+            remote.tree.hash_to_path_info(bar.hash_info.value)
         )
-        assert remote.tree.exists(remote.tree.hash_to_path_info(bar.checksum))
         assert not remote.tree.exists(
-            remote.tree.hash_to_path_info(baz.checksum)
+            remote.tree.hash_to_path_info(baz.hash_info.value)
         )
 
     # Push everything and delete local cache
@@ -408,7 +410,9 @@ def test_protect_local_remote(tmp_dir, dvc, local_remote):
 
     dvc.push()
     remote = dvc.cloud.get_remote("upstream")
-    remote_cache_file = remote.tree.hash_to_path_info(stage.outs[0].checksum)
+    remote_cache_file = remote.tree.hash_to_path_info(
+        stage.outs[0].hash_info.value
+    )
 
     assert os.path.exists(remote_cache_file)
     assert stat.S_IMODE(os.stat(remote_cache_file).st_mode) == 0o444
@@ -419,7 +423,7 @@ def test_push_incomplete_dir(tmp_dir, dvc, mocker, local_remote):
     remote = dvc.cloud.get_remote("upstream")
 
     cache = dvc.cache.local
-    dir_hash = stage.outs[0].checksum
+    dir_hash = stage.outs[0].hash_info.value
     used = stage.get_used_cache(remote=remote)
 
     # remove one of the cache files for directory
diff --git a/tests/func/test_repro.py b/tests/func/test_repro.py
index 64744eac2..ed65309ad 100644
--- a/tests/func/test_repro.py
+++ b/tests/func/test_repro.py
@@ -674,7 +674,9 @@ class TestReproDataSource(TestReproChangedData):
         stages = self.dvc.reproduce(self.foo_stage.path)
 
         self.assertTrue(filecmp.cmp(self.FOO, self.BAR, shallow=False))
-        self.assertEqual(stages[0].outs[0].checksum, file_md5(self.BAR)[0])
+        self.assertEqual(
+            stages[0].outs[0].hash_info.value, file_md5(self.BAR)[0]
+        )
 
 
 class TestReproChangedDir(SingleStageRun, TestDvc):
@@ -911,7 +913,7 @@ class TestReproAlreadyCached(TestRepro):
             0
         ]
 
-        self.assertNotEqual(run_out.checksum, repro_out.checksum)
+        self.assertNotEqual(run_out.hash_info, repro_out.hash_info)
 
     def test_force_with_dependencies(self):
         run_out = self.dvc.run(
@@ -927,7 +929,7 @@ class TestReproAlreadyCached(TestRepro):
 
         repro_out = Dvcfile(self.dvc, "datetime.dvc").stage.outs[0]
 
-        self.assertNotEqual(run_out.checksum, repro_out.checksum)
+        self.assertNotEqual(run_out.hash_info, repro_out.hash_info)
 
     def test_force_import(self):
         ret = main(["import-url", self.FOO, self.BAR])
diff --git a/tests/func/test_run_single_stage.py b/tests/func/test_run_single_stage.py
index 3f6265ec2..155126459 100644
--- a/tests/func/test_run_single_stage.py
+++ b/tests/func/test_run_single_stage.py
@@ -60,7 +60,7 @@ class TestRun(TestDvc):
         self.assertEqual(len(stage.deps), len(deps))
         self.assertEqual(len(stage.outs), len(outs + outs_no_cache))
         self.assertEqual(stage.outs[0].fspath, outs[0])
-        self.assertEqual(stage.outs[0].checksum, file_md5(self.FOO)[0])
+        self.assertEqual(stage.outs[0].hash_info.value, file_md5(self.FOO)[0])
         self.assertTrue(stage.path, fname)
 
         with self.assertRaises(OutputDuplicationError):
diff --git a/tests/func/test_tree.py b/tests/func/test_tree.py
index 3b835a11b..f89cbf72c 100644
--- a/tests/func/test_tree.py
+++ b/tests/func/test_tree.py
@@ -226,7 +226,7 @@ def test_repotree_cache_save(tmp_dir, dvc, scm, erepo_dir, local_cloud):
         cache = dvc.cache.local
         with cache.tree.state:
             path_info = PathInfo(erepo_dir / "dir")
-            hash_info = cache.tree.get_hash(path_info).to_dict()
+            hash_info = cache.tree.get_hash(path_info)
             cache.save(path_info, tree, hash_info)
 
     for hash_ in expected:
diff --git a/tests/unit/output/test_local.py b/tests/unit/output/test_local.py
index 23ecdb999..c2a2d07c5 100644
--- a/tests/unit/output/test_local.py
+++ b/tests/unit/output/test_local.py
@@ -3,6 +3,7 @@ import os
 from mock import patch
 
 from dvc.cache.local import LocalCache
+from dvc.hash_info import HashInfo
 from dvc.output import LocalOutput
 from dvc.stage import Stage
 from dvc.utils import relpath
@@ -77,7 +78,6 @@ class TestGetFilesNumber(TestDvc):
         o.use_cache = False
         self.assertEqual(0, o.get_files_number())
 
-    @patch.object(LocalOutput, "checksum", "12345678.dir")
     @patch.object(
         LocalCache,
         "get_dir_cache",
@@ -85,12 +85,11 @@ class TestGetFilesNumber(TestDvc):
     )
     def test_return_multiple_for_dir(self, _mock_get_dir_cache):
         o = self._get_output()
-
+        o.hash_info = HashInfo("md5", "12345678.dir")
         self.assertEqual(2, o.get_files_number())
 
-    @patch.object(LocalOutput, "checksum", "12345678")
     @patch.object(LocalOutput, "is_dir_checksum", False)
     def test_return_1_on_single_file_cache(self):
         o = self._get_output()
-
+        o.hash_info = HashInfo("md5", "12345678")
         self.assertEqual(1, o.get_files_number())
diff --git a/tests/unit/output/test_output.py b/tests/unit/output/test_output.py
index 4f2e69e88..8a8240430 100644
--- a/tests/unit/output/test_output.py
+++ b/tests/unit/output/test_output.py
@@ -83,9 +83,6 @@ def test_get_used_cache(exists, expected_message, mocker, caplog):
 
     mocker.patch.object(output, "use_cache", True)
     mocker.patch.object(stage, "is_repo_import", False)
-    mocker.patch.object(
-        BaseOutput, "checksum", new_callable=mocker.PropertyMock
-    ).return_value = None
     mocker.patch.object(
         BaseOutput, "exists", new_callable=mocker.PropertyMock
     ).return_value = exists
diff --git a/tests/unit/stage/test_loader_pipeline_file.py b/tests/unit/stage/test_loader_pipeline_file.py
index f4798e50f..dad059bc8 100644
--- a/tests/unit/stage/test_loader_pipeline_file.py
+++ b/tests/unit/stage/test_loader_pipeline_file.py
@@ -31,7 +31,7 @@ def test_fill_from_lock_deps_outs(dvc, lock_data):
     )
 
     for item in chain(stage.deps, stage.outs):
-        assert not item.checksum and not item.hash_info
+        assert not item.hash_info
 
     StageLoader.fill_from_lock(stage, lock_data)
 
@@ -102,7 +102,7 @@ def test_fill_from_lock_missing_checksums(dvc, lock_data):
 
     assert stage.deps[0].hash_info == HashInfo("md5", "foo_checksum")
     assert stage.outs[0].hash_info == HashInfo("md5", "bar_checksum")
-    assert not stage.deps[1].checksum and not stage.outs[1].checksum
+    assert not stage.deps[1].hash_info and not stage.outs[1].hash_info
 
 
 def test_fill_from_lock_use_appropriate_checksum(dvc, lock_data):
@@ -115,8 +115,8 @@ def test_fill_from_lock_use_appropriate_checksum(dvc, lock_data):
     )
     lock_data["deps"] = [{"path": "s3://dvc-temp/foo", "etag": "e-tag"}]
     StageLoader.fill_from_lock(stage, lock_data)
-    assert stage.deps[0].checksum == "e-tag"
-    assert stage.outs[0].checksum == "bar_checksum"
+    assert stage.deps[0].hash_info == HashInfo("etag", "e-tag")
+    assert stage.outs[0].hash_info == HashInfo("md5", "bar_checksum")
 
 
 def test_fill_from_lock_with_missing_sections(dvc, lock_data):
@@ -126,14 +126,14 @@ def test_fill_from_lock_with_missing_sections(dvc, lock_data):
     lock = deepcopy(lock_data)
     del lock["deps"]
     StageLoader.fill_from_lock(stage, lock)
-    assert not stage.deps[0].checksum
-    assert stage.outs[0].checksum == "bar_checksum"
+    assert not stage.deps[0].hash_info
+    assert stage.outs[0].hash_info == HashInfo("md5", "bar_checksum")
 
     lock = deepcopy(lock_data)
     del lock["outs"]
     StageLoader.fill_from_lock(stage, lock)
-    assert stage.deps[0].checksum == "foo_checksum"
-    assert not stage.outs[0].checksum
+    assert stage.deps[0].hash_info == HashInfo("md5", "foo_checksum")
+    assert not stage.outs[0].hash_info
 
 
 def test_fill_from_lock_empty_data(dvc):
@@ -141,9 +141,9 @@ def test_fill_from_lock_empty_data(dvc):
         PipelineStage, dvc, PIPELINE_FILE, deps=["foo"], outs=["bar"]
     )
     StageLoader.fill_from_lock(stage, None)
-    assert not stage.deps[0].checksum and not stage.outs[0].checksum
+    assert not stage.deps[0].hash_info and not stage.outs[0].hash_info
     StageLoader.fill_from_lock(stage, {})
-    assert not stage.deps[0].checksum and not stage.outs[0].checksum
+    assert not stage.deps[0].hash_info and not stage.outs[0].hash_info
 
 
 def test_load_stage(dvc, stage_data, lock_data):
@@ -155,9 +155,9 @@ def test_load_stage(dvc, stage_data, lock_data):
     assert stage.cmd == "command"
     assert stage.path == os.path.abspath(PIPELINE_FILE)
     assert stage.deps[0].def_path == "foo"
-    assert stage.deps[0].checksum == "foo_checksum"
+    assert stage.deps[0].hash_info == HashInfo("md5", "foo_checksum")
     assert stage.outs[0].def_path == "bar"
-    assert stage.outs[0].checksum == "bar_checksum"
+    assert stage.outs[0].hash_info == HashInfo("md5", "bar_checksum")
 
 
 def test_load_stage_outs_with_flags(dvc, stage_data, lock_data):
@@ -171,8 +171,8 @@ def test_load_stage_no_lock(dvc, stage_data):
     dvcfile = Dvcfile(dvc, PIPELINE_FILE)
     stage = StageLoader.load_stage(dvcfile, "stage-1", stage_data)
     assert stage.deps[0].def_path == "foo" and stage.outs[0].def_path == "bar"
-    assert not stage.deps[0].checksum
-    assert not stage.outs[0].checksum
+    assert not stage.deps[0].hash_info
+    assert not stage.outs[0].hash_info
 
 
 def test_load_stage_with_params(dvc, stage_data, lock_data):
@@ -185,8 +185,8 @@ def test_load_stage_with_params(dvc, stage_data, lock_data):
     assert deps[0].def_path == "foo" and stage.outs[0].def_path == "bar"
     assert params[0].def_path == "params.yaml"
     assert params[0].hash_info == HashInfo("params", {"lorem": "ipsum"})
-    assert deps[0].checksum == "foo_checksum"
-    assert stage.outs[0].checksum == "bar_checksum"
+    assert deps[0].hash_info == HashInfo("md5", "foo_checksum")
+    assert stage.outs[0].hash_info == HashInfo("md5", "bar_checksum")
 
 
 @pytest.mark.parametrize("typ", ["metrics", "plots"])
@@ -196,7 +196,7 @@ def test_load_stage_with_metrics_and_plots(dvc, stage_data, lock_data, typ):
     stage = StageLoader.load_stage(dvcfile, "stage-1", stage_data, lock_data)
 
     assert stage.outs[0].def_path == "bar"
-    assert stage.outs[0].checksum == "bar_checksum"
+    assert stage.outs[0].hash_info == HashInfo("md5", "bar_checksum")
 
 
 def test_load_changed_command(dvc, stage_data, lock_data):