fix(data status): handle missing .dir file from the cache on not_in_remote checks (#10822)

dvc/repo/data.py

@@ -7,6 +7,7 @@ from typing import TYPE_CHECKING, Optional, TypedDict, Union
 from dvc.fs.callbacks import DEFAULT_CALLBACK, Callback, TqdmCallback
 from dvc.log import logger
 from dvc.ui import ui
+from dvc_data.index import DataIndexDirError
 
 if TYPE_CHECKING:
     from dvc.repo import Repo
@@ -328,10 +329,21 @@ def _get_entries_not_in_remote(
     from dvc.repo.worktree import worktree_view
     from dvc_data.index import StorageKeyError
 
+    entries: dict[DataIndexKey, DataIndexEntry] = {}
+
+    def _onerror(entry, exc):
+        if not isinstance(exc, DataIndexDirError):
+            raise exc
+        # We don't have the contents of this dir file, so we will only check this key.
+        entries[entry.key] = entry
+
     # View into the index, with only pushable entries
     index = worktree_view(repo.index, push=True)
     data_index = index.data["repo"]
 
+    orig_data_index_onerror = data_index.onerror
+    data_index.onerror = _onerror
+
     view = filter_index(data_index, filter_keys=filter_keys)  # type: ignore[arg-type]
 
     missing_entries = []
@@ -340,7 +352,6 @@ def _get_entries_not_in_remote(
 
     n = 0
     with TqdmCallback(size=n, desc="Checking remote", unit="entry") as cb:
-        entries: dict[DataIndexKey, DataIndexEntry] = {}
         for key, entry in view.iteritems(shallow=not granular):
             if not (entry and entry.hash_info):
                 continue
@@ -368,6 +379,7 @@ def _get_entries_not_in_remote(
                 pass
             finally:
                 cb.relative_update()
+    data_index.onerror = orig_data_index_onerror
     return missing_entries
 
 

dvc/repo/fetch.py

@@ -10,6 +10,7 @@ from . import locked
 
 if TYPE_CHECKING:
     from dvc.output import Output
+    from dvc.repo import Repo
     from dvc.stage import Stage
 
 logger = logger.getChild(__name__)
@@ -24,7 +25,7 @@ def _make_index_onerror(onerror, rev):
 
 
 def _collect_indexes(  # noqa: PLR0913
-    repo,
+    repo: "Repo",
     targets=None,
     remote=None,
     all_branches=False,
@@ -94,7 +95,7 @@ def _collect_indexes(  # noqa: PLR0913
 
 @locked
 def fetch(  # noqa: PLR0913
-    self,
+    self: "Repo",
     targets=None,
     jobs=None,
     remote=None,

tests/func/test_data_status.py

@@ -918,3 +918,49 @@ def test_compat_legacy_new_cache_types(M, tmp_dir, dvc, scm):
         "committed": {"added": M.unordered("foo", "bar")},
         "git": M.dict(),
     }
+
+
+def test_missing_cache_remote_check(M, tmp_dir, dvc, scm, local_remote):
+    tmp_dir.dvc_gen({"dir": {"foo": "foo", "bar": "bar"}})
+    tmp_dir.dvc_gen("foobar", "foobar")
+    remove(dvc.cache.repo.path)
+
+    assert dvc.data_status(untracked_files="all", not_in_remote=True) == {
+        **EMPTY_STATUS,
+        "untracked": M.unordered("foobar.dvc", "dir.dvc", ".gitignore"),
+        "committed": {"added": M.unordered("foobar", join("dir", ""))},
+        "not_in_cache": M.unordered("foobar", join("dir", "")),
+        "git": M.dict(),
+        "not_in_remote": M.unordered("foobar", join("dir", "")),
+    }
+
+    assert dvc.data_status(
+        granular=True, untracked_files="all", not_in_remote=True
+    ) == {
+        **EMPTY_STATUS,
+        "untracked": M.unordered("foobar.dvc", "dir.dvc", ".gitignore"),
+        "committed": {"added": M.unordered("foobar", join("dir", ""))},
+        "uncommitted": {"unknown": M.unordered(join("dir", "foo"), join("dir", "bar"))},
+        "not_in_cache": M.unordered("foobar", join("dir", "")),
+        "git": M.dict(),
+        "not_in_remote": M.unordered("foobar", join("dir", "")),
+    }
+
+    assert dvc.data_status(["dir"], untracked_files="all", not_in_remote=True) == {
+        **EMPTY_STATUS,
+        "committed": {"added": [join("dir", "")]},
+        "not_in_cache": [join("dir", "")],
+        "git": M.dict(),
+        "not_in_remote": [join("dir", "")],
+    }
+
+    assert dvc.data_status(
+        ["dir"], untracked_files="all", not_in_remote=True, granular=True
+    ) == {
+        **EMPTY_STATUS,
+        "committed": {"added": [join("dir", "")]},
+        "uncommitted": {"unknown": M.unordered(join("dir", "foo"), join("dir", "bar"))},
+        "not_in_cache": [join("dir", "")],
+        "git": M.dict(),
+        "not_in_remote": [join("dir", "")],
+    }