diff --git a/dvc/checkout.py b/dvc/checkout.py
index 6a827e64a..b22df2d24 100644
--- a/dvc/checkout.py
+++ b/dvc/checkout.py
@@ -1,4 +1,5 @@
 import logging
+from itertools import chain
 
 from shortuuid import uuid
 
@@ -10,52 +11,19 @@ from dvc.exceptions import (
     DvcException,
 )
 from dvc.ignore import DvcIgnoreFilter
-from dvc.objects import check, load
 from dvc.objects.db.slow_link_detection import (  # type: ignore[attr-defined]
     slow_link_guard,
 )
-from dvc.objects.errors import ObjectFormatError
+from dvc.objects.diff import ROOT
+from dvc.objects.diff import diff as odiff
 from dvc.objects.stage import stage
+from dvc.objects.tree import Tree
 from dvc.types import Optional
 
 logger = logging.getLogger(__name__)
 
 
-def _changed(path_info, fs, obj, cache, state=None):
-    logger.trace("checking if '%s'('%s') has changed.", path_info, obj)
-
-    try:
-        check(cache, obj)
-    except (FileNotFoundError, ObjectFormatError):
-        logger.debug(
-            "cache for '%s'('%s') has changed.", path_info, obj.hash_info
-        )
-        return True
-
-    actual = state.get(path_info, fs) if state else None
-    if actual is None:
-        try:
-            actual = stage(
-                cache, path_info, fs, obj.hash_info.name, dry_run=True
-            )[1].hash_info
-        except FileNotFoundError:
-            logger.debug("'%s' doesn't exist.", path_info)
-            return True
-
-    if obj.hash_info != actual:
-        logger.debug(
-            "hash value '%s' for '%s' has changed (actual '%s').",
-            obj.hash_info,
-            actual,
-            path_info,
-        )
-        return True
-
-    logger.trace("'%s' hasn't changed.", path_info)
-    return False
-
-
-def _remove(path_info, fs, cache, force=False):
+def _remove(path_info, fs, in_cache, force=False):
     if not fs.exists(path_info):
         return
 
@@ -63,13 +31,7 @@ def _remove(path_info, fs, cache, force=False):
         fs.remove(path_info)
         return
 
-    current = stage(cache, path_info, fs, fs.PARAM_CHECKSUM, dry_run=True)[
-        1
-    ].hash_info
-    try:
-        obj = load(cache, current)
-        check(cache, obj)
-    except (FileNotFoundError, ObjectFormatError):
+    if not in_cache:
         msg = (
             f"file/directory '{path_info}' is going to be removed. "
             "Are you sure you want to proceed?"
@@ -98,7 +60,7 @@ def _verify_link(cache, path_info, link_type):
 
 def _do_link(cache, from_info, to_info, link_method):
     if cache.fs.exists(to_info):
-        raise DvcException(f"Link '{to_info}' already exists!")
+        cache.fs.remove(to_info)  # broken symlink
 
     link_method(from_info, to_info)
 
@@ -130,7 +92,7 @@ def _link(cache, from_info, to_info):
     try:
         _try_links(cache, from_info, to_info, cache.cache_types)
     except FileNotFoundError as exc:
-        raise CheckoutError([to_info]) from exc
+        raise CheckoutError([str(to_info)]) from exc
 
 
 def _cache_is_copy(cache, path_info):
@@ -160,7 +122,7 @@ def _cache_is_copy(cache, path_info):
 def _checkout_file(
     path_info,
     fs,
-    obj,
+    change,
     cache,
     force,
     progress_callback=None,
@@ -169,24 +131,24 @@ def _checkout_file(
 ):
     """The file is changed we need to checkout a new copy"""
     modified = False
-    cache_info = cache.hash_to_path_info(obj.hash_info.value)
-    if fs.exists(path_info):
-        if not relink and _changed(path_info, fs, obj, cache, state=state):
-            modified = True
-            _remove(path_info, fs, cache, force=force)
-            _link(cache, cache_info, path_info)
-        else:
+    cache_info = cache.hash_to_path_info(change.new.obj.hash_info.value)
+    if change.old.obj:
+        if relink:
             if fs.iscopy(path_info) and _cache_is_copy(cache, path_info):
                 cache.unprotect(path_info)
             else:
-                _remove(path_info, fs, cache, force=force)
+                _remove(path_info, fs, change.old.in_cache, force=force)
                 _link(cache, cache_info, path_info)
+        else:
+            modified = True
+            _remove(path_info, fs, change.old.in_cache, force=force)
+            _link(cache, cache_info, path_info)
     else:
         _link(cache, cache_info, path_info)
         modified = True
 
     if state:
-        state.save(path_info, fs, obj.hash_info)
+        state.save(path_info, fs, change.new.obj.hash_info)
 
     if progress_callback:
         progress_callback(str(path_info))
@@ -194,109 +156,83 @@ def _checkout_file(
     return modified
 
 
-def _remove_redundant_files(
-    path_info, fs, obj, cache, force, dvcignore: Optional[DvcIgnoreFilter]
-):
-    if dvcignore:
-        existing_files = set(dvcignore.walk_files(fs, path_info))
-    else:
-        existing_files = set(fs.walk_files(path_info))
-
-    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)
-
-    return bool(redundant_files)
-
-
-def _checkout_dir(
+def _diff(
     path_info,
     fs,
     obj,
     cache,
-    force,
-    progress_callback=None,
     relink=False,
     dvcignore: Optional[DvcIgnoreFilter] = None,
-    state=None,
 ):
-    modified = False
-    # Create dir separately so that dir is created
-    # even if there are no files in it
-    if not fs.exists(path_info):
-        modified = True
-        fs.makedirs(path_info)
-
-    logger.debug("Linking directory '%s'.", path_info)
-
-    failed = []
-    for entry_key, entry_obj in obj:
-        try:
-            entry_modified = _checkout_file(
-                path_info.joinpath(*entry_key),
-                fs,
-                entry_obj,
-                cache,
-                force,
-                progress_callback,
-                relink,
-                state=None,
-            )
-            if entry_modified:
-                modified = True
-        except CheckoutError as exc:
-            failed.extend(exc.target_infos)
-
-    modified = (
-        _remove_redundant_files(
-            path_info, fs, obj, cache, force, dvcignore=dvcignore
+    old = None
+    try:
+        _, old = stage(
+            cache,
+            path_info,
+            fs,
+            obj.hash_info.name if obj else cache.fs.PARAM_CHECKSUM,
+            dry_run=True,
+            dvcignore=dvcignore,
         )
-        or modified
-    )
+    except FileNotFoundError:
+        pass
 
-    if failed:
-        raise CheckoutError(failed)
+    diff = odiff(old, obj, cache)
 
-    if state:
-        state.save(path_info, fs, obj.hash_info)
+    if relink:
+        diff.modified.extend(diff.unchanged)
 
-    # relink is not modified, assume it as nochange
-    return modified and not relink
+    return diff
 
 
 def _checkout(
+    diff,
     path_info,
     fs,
-    obj,
     cache,
     force=False,
     progress_callback=None,
     relink=False,
-    dvcignore: Optional[DvcIgnoreFilter] = None,
     state=None,
 ):
-    if not obj.hash_info.isdir:
-        ret = _checkout_file(
-            path_info, fs, obj, cache, force, progress_callback, relink, state
-        )
-    else:
-        ret = _checkout_dir(
-            path_info,
-            fs,
-            obj,
-            cache,
-            force,
-            progress_callback,
-            relink,
-            dvcignore=dvcignore,
-            state=state,
-        )
+    if not diff:
+        return
 
-    if state:
-        state.save_link(path_info, fs)
+    for change in diff.deleted:
+        entry_path = (
+            path_info.joinpath(*change.old.key)
+            if change.old.key != ROOT
+            else path_info
+        )
+        _remove(entry_path, fs, change.old.in_cache, force=force)
 
-    return ret
+    failed = []
+    for change in chain(diff.added, diff.modified):
+        entry_path = (
+            path_info.joinpath(*change.new.key)
+            if change.new.key != ROOT
+            else path_info
+        )
+        if isinstance(change.new.obj, Tree):
+            fs.makedirs(entry_path)
+            continue
+
+        try:
+            _checkout_file(
+                entry_path,
+                fs,
+                change,
+                cache,
+                force,
+                progress_callback,
+                relink,
+                state=state,
+            )
+        except CheckoutError as exc:
+            failed.extend(exc.target_infos)
+
+    if failed:
+        raise CheckoutError(failed)
 
 
 def checkout(
@@ -311,41 +247,54 @@ def checkout(
     dvcignore: Optional[DvcIgnoreFilter] = None,
     state=None,
 ):
+
     if path_info.scheme not in ["local", cache.fs.scheme]:
         raise NotImplementedError
 
-    failed = None
-    skip = False
+    diff = _diff(
+        path_info,
+        fs,
+        obj,
+        cache,
+        relink=relink,
+        dvcignore=dvcignore,
+    )
+
+    failed = []
     if not obj:
         if not quiet:
             logger.warning(
                 "No file hash info found for '%s'. It won't be created.",
                 path_info,
             )
-        _remove(path_info, fs, cache, force=force)
-        failed = path_info
-
-    elif not relink and not _changed(path_info, fs, obj, cache, state=state):
+        failed.append(str(path_info))
+    elif not diff:
         logger.trace("Data '%s' didn't change.", path_info)  # type: ignore
-        skip = True
 
-    if failed or skip:
+    try:
+        _checkout(
+            diff,
+            path_info,
+            fs,
+            cache,
+            force=force,
+            progress_callback=progress_callback,
+            relink=relink,
+            state=state,
+        )
+    except CheckoutError as exc:
+        failed.extend(exc.target_infos)
+
+    if diff and state:
+        state.save_link(path_info, fs)
+        if not failed:
+            state.save(path_info, fs, obj.hash_info)
+
+    if failed or not diff:
         if progress_callback and obj:
             progress_callback(str(path_info), len(obj))
         if failed:
-            raise CheckoutError([failed])
+            raise CheckoutError(failed)
         return
 
-    logger.debug("Checking out '%s' with cache '%s'.", path_info, obj)
-
-    return _checkout(
-        path_info,
-        fs,
-        obj,
-        cache,
-        force,
-        progress_callback,
-        relink,
-        dvcignore=dvcignore,
-        state=state,
-    )
+    return bool(diff) and not relink
diff --git a/dvc/fs/webhdfs.py b/dvc/fs/webhdfs.py
index 63cabde29..80a2682d4 100644
--- a/dvc/fs/webhdfs.py
+++ b/dvc/fs/webhdfs.py
@@ -1,3 +1,4 @@
+import errno
 import logging
 import os
 import posixpath
@@ -131,14 +132,19 @@ class WebHDFSFileSystem(BaseFileSystem):  # pylint:disable=abstract-method
         return status is not None
 
     def info(self, path_info):
-        st = self.hdfs_client.status(path_info.path)
+        st = self.hdfs_client.status(path_info.path, strict=False)
+        if not st:
+            raise FileNotFoundError(
+                errno.ENOENT, os.strerror(errno.ENOENT), path_info.path
+            )
         return {"size": st["length"], "type": "file"}
 
     def checksum(self, path_info):
+        size = self.info(path_info)["size"]
         return HashInfo(
             "checksum",
             self.hdfs_client.checksum(path_info.path)["bytes"],
-            size=self.hdfs_client.status(path_info.path)["length"],
+            size=size,
         )
 
     def copy(self, from_info, to_info, **_kwargs):
diff --git a/dvc/objects/diff.py b/dvc/objects/diff.py
new file mode 100644
index 000000000..458e19636
--- /dev/null
+++ b/dvc/objects/diff.py
@@ -0,0 +1,135 @@
+from dataclasses import dataclass, field
+from typing import TYPE_CHECKING, List, Optional, Tuple
+
+if TYPE_CHECKING:
+    from .file import HashFile
+
+ADD = "add"
+MODIFY = "modify"
+DELETE = "delete"
+UNCHANGED = "unchanged"
+
+
+@dataclass
+class TreeEntry:
+    in_cache: bool
+    key: Tuple[str]
+    obj: Optional["HashFile"] = field(default=None)
+
+    def __bool__(self):
+        return bool(self.obj)
+
+    def __eq__(self, other):
+        if not isinstance(other, TreeEntry):
+            return False
+
+        if self.key != other.key or bool(self.obj) != bool(other.obj):
+            return False
+
+        if not self.obj:
+            return False
+
+        return self.obj.hash_info == other.obj.hash_info
+
+
+@dataclass
+class Change:
+    old: TreeEntry
+    new: TreeEntry
+
+    @property
+    def typ(self):
+        if not self.old and not self.new:
+            return UNCHANGED
+
+        if self.old and not self.new:
+            return DELETE
+
+        if not self.old and self.new:
+            return ADD
+
+        if self.old != self.new:
+            return MODIFY
+
+        return UNCHANGED
+
+    def __bool__(self):
+        return self.typ != UNCHANGED
+
+
+@dataclass
+class DiffResult:
+    added: List[Change] = field(default_factory=list, compare=True)
+    modified: List[Change] = field(default_factory=list, compare=True)
+    deleted: List[Change] = field(default_factory=list, compare=True)
+    unchanged: List[Change] = field(default_factory=list, compare=True)
+
+    def __bool__(self):
+        return bool(self.added or self.modified or self.deleted)
+
+
+ROOT = ("",)
+
+
+def diff(
+    old: Optional["HashFile"], new: Optional["HashFile"], cache
+) -> DiffResult:
+    from .tree import Tree
+
+    if old is None and new is None:
+        return DiffResult()
+
+    def _get_keys(obj):
+        if not obj:
+            return []
+        return [ROOT] + (
+            [key for key, _ in obj] if isinstance(obj, Tree) else []
+        )
+
+    old_keys = set(_get_keys(old))
+    new_keys = set(_get_keys(new))
+
+    def _get_obj(obj, key):
+        if not obj or key == ROOT:
+            return obj
+
+        return obj.get(key)
+
+    def _in_cache(obj, cache):
+        from . import check
+        from .errors import ObjectFormatError
+
+        if not obj:
+            return False
+
+        try:
+            check(cache, obj)
+            return True
+        except (FileNotFoundError, ObjectFormatError):
+            return False
+
+    ret = DiffResult()
+    for key in old_keys | new_keys:
+        old_obj = _get_obj(old, key)
+        new_obj = _get_obj(new, key)
+
+        change = Change(
+            old=TreeEntry(_in_cache(old_obj, cache), key, old_obj),
+            new=TreeEntry(_in_cache(new_obj, cache), key, new_obj),
+        )
+
+        if change.typ == ADD:
+            ret.added.append(change)
+        elif change.typ == MODIFY:
+            ret.modified.append(change)
+        elif change.typ == DELETE:
+            ret.deleted.append(change)
+        else:
+            assert change.typ == UNCHANGED
+            if not change.new.in_cache and not isinstance(
+                change.new.obj, Tree
+            ):
+                ret.modified.append(change)
+            else:
+                ret.unchanged.append(change)
+    return ret
diff --git a/tests/func/test_add.py b/tests/func/test_add.py
index 070207fe0..52352858f 100644
--- a/tests/func/test_add.py
+++ b/tests/func/test_add.py
@@ -496,27 +496,27 @@ def test_should_update_state_entry_for_directory_after_add(
 
     ret = main(["add", "data"])
     assert ret == 0
-    assert file_md5_counter.mock.call_count == 3
+    assert file_md5_counter.mock.call_count == 4
 
     ret = main(["status"])
     assert ret == 0
-    assert file_md5_counter.mock.call_count == 3
+    assert file_md5_counter.mock.call_count == 4
 
     ls = "dir" if os.name == "nt" else "ls"
     ret = main(
         ["run", "--single-stage", "-d", "data", "{} {}".format(ls, "data")]
     )
     assert ret == 0
-    assert file_md5_counter.mock.call_count == 3
+    assert file_md5_counter.mock.call_count == 4
 
     os.rename("data", "data" + ".back")
     ret = main(["checkout"])
     assert ret == 0
-    assert file_md5_counter.mock.call_count == 3
+    assert file_md5_counter.mock.call_count == 4
 
     ret = main(["status"])
     assert ret == 0
-    assert file_md5_counter.mock.call_count == 3
+    assert file_md5_counter.mock.call_count == 4
 
 
 class TestAddCommit(TestDvc):
@@ -537,13 +537,15 @@ def test_should_collect_dir_cache_only_once(mocker, tmp_dir, dvc):
     counter = mocker.spy(dvc_module.objects.stage, "_get_tree_obj")
     ret = main(["add", "data"])
     assert ret == 0
+    assert counter.mock.call_count == 2
 
     ret = main(["status"])
     assert ret == 0
+    assert counter.mock.call_count == 2
 
     ret = main(["status"])
     assert ret == 0
-    assert counter.mock.call_count == 1
+    assert counter.mock.call_count == 2
 
 
 class TestShouldPlaceStageInDataDirIfRepositoryBelowSymlink(TestDvc):