data: status: introduce --not-in-remote

Also supports cloud versioning.

Fixes #8761

dvc/commands/data.py

@@ -21,12 +21,14 @@ logger = logging.getLogger(__name__)
 
 class CmdDataStatus(CmdBase):
     COLORS = {
+        "not_in_remote": "red",
         "not_in_cache": "red",
         "committed": "green",
         "uncommitted": "yellow",
         "untracked": "cyan",
     }
     LABELS = {
+        "not_in_remote": "Not in remote",
         "not_in_cache": "Not in cache",
         "committed": "DVC committed changes",
         "uncommitted": "DVC uncommitted changes",
@@ -34,6 +36,7 @@ class CmdDataStatus(CmdBase):
         "unchanged": "DVC unchanged files",
     }
     HINTS = {
+        "not_in_remote": ('use "dvc push <file>..." to upload files',),
         "not_in_cache": ('use "dvc fetch <file>..." to download files',),
         "committed": ("git commit the corresponding dvc files to update the repo",),
         "uncommitted": (
@@ -114,6 +117,7 @@ class CmdDataStatus(CmdBase):
             status = self.repo.data_status(
                 granular=self.args.granular,
                 untracked_files=self.args.untracked_files,
+                not_in_remote=self.args.not_in_remote,
             )
 
         if not self.args.unchanged:
@@ -239,6 +243,12 @@ def add_parser(subparsers, parent_parser):
         nargs="?",
         help="Show untracked files.",
     )
+    data_status_parser.add_argument(
+        "--not-in-remote",
+        action="store_true",
+        default=False,
+        help="Show files missing from remote.",
+    )
     data_status_parser.set_defaults(func=CmdDataStatus)
 
     DATA_LS_HELP = "List data tracked by DVC with its metadata."

dvc/repo/data.py

@@ -59,7 +59,9 @@ def _diff(
     *,
     granular: bool = False,
     not_in_cache: bool = False,
+    not_in_remote: bool = False,
 ) -> Dict[str, List[str]]:
+    from dvc_data.index import StorageError
     from dvc_data.index.diff import UNCHANGED, UNKNOWN, diff
 
     ret: Dict[str, List[str]] = {}
@@ -102,6 +104,17 @@ def _diff(
             # NOTE: emulating previous behaviour
             _add_change("not_in_cache", change)
 
+        try:
+            if (
+                not_in_remote
+                and change.old
+                and change.old.hash_info
+                and not old.storage_map.remote_exists(change.old)
+            ):
+                _add_change("not_in_remote", change)
+        except StorageError:
+            pass
+
         _add_change(change.typ, change)
 
     return ret
@@ -177,6 +190,7 @@ def _diff_head_to_index(
 
 class Status(TypedDict):
     not_in_cache: List[str]
+    not_in_remote: List[str]
     committed: Dict[str, List[str]]
     uncommitted: Dict[str, List[str]]
     untracked: List[str]
@@ -203,12 +217,20 @@ def _transform_git_paths_to_dvc(repo: "Repo", files: Iterable[str]) -> List[str]
     return [repo.fs.path.relpath(file, start) for file in files]
 
 
-def status(repo: "Repo", untracked_files: str = "no", **kwargs: Any) -> Status:
+def status(
+    repo: "Repo",
+    untracked_files: str = "no",
+    not_in_remote: bool = False,
+    **kwargs: Any,
+) -> Status:
     from dvc.scm import NoSCMError, SCMError
 
     head = kwargs.pop("head", "HEAD")
-    uncommitted_diff = _diff_index_to_wtree(repo, **kwargs)
-    not_in_cache = uncommitted_diff.pop("not_in_cache", [])
+    uncommitted_diff = _diff_index_to_wtree(
+        repo,
+        not_in_remote=not_in_remote,
+        **kwargs,
+    )
     unchanged = set(uncommitted_diff.pop("unchanged", []))
 
     try:
@@ -223,7 +245,8 @@ def status(repo: "Repo", untracked_files: str = "no", **kwargs: Any) -> Status:
     untracked = _transform_git_paths_to_dvc(repo, untracked)
     # order matters here
     return Status(
-        not_in_cache=not_in_cache,
+        not_in_cache=uncommitted_diff.pop("not_in_cache", []),
+        not_in_remote=uncommitted_diff.pop("not_in_remote", []),
         committed=committed_diff,
         uncommitted=uncommitted_diff,
         untracked=untracked,

tests/func/test_data_status.py

@@ -13,6 +13,7 @@ EMPTY_STATUS = {
     "uncommitted": {},
     "git": {},
     "not_in_cache": [],
+    "not_in_remote": [],
     "unchanged": [],
     "untracked": [],
 }
@@ -81,6 +82,7 @@ def test_directory(M, tmp_dir, dvc, scm):
     }
 
     assert dvc.data_status(granular=True, untracked_files="all") == {
+        **EMPTY_STATUS,
         "committed": {
             "added": M.unordered(
                 join("dir", "bar"),
@@ -396,6 +398,42 @@ def test_missing_dir_object_from_index(M, tmp_dir, dvc, scm):
     }
 
 
+def test_missing_remote_cache(M, tmp_dir, dvc, scm, local_remote):
+    tmp_dir.dvc_gen({"dir": {"foo": "foo", "bar": "bar"}})
+    tmp_dir.dvc_gen("foobar", "foobar")
+
+    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_remote": M.unordered("foobar", join("dir", "")),
+        "git": M.dict(),
+    }
+
+    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", ""),
+                join("dir", "foo"),
+                join("dir", "bar"),
+            )
+        },
+        "uncommitted": {},
+        "not_in_remote": M.unordered(
+            "foobar",
+            join("dir", ""),
+            join("dir", "foo"),
+            join("dir", "bar"),
+        ),
+        "git": M.dict(),
+    }
+
+
 def test_root_from_dir_to_file(M, tmp_dir, dvc, scm):
     tmp_dir.dvc_gen({"data": {"foo": "foo", "bar": "bar"}})
     remove("data")

tests/unit/command/test_data_status.py

@@ -51,6 +51,7 @@ def test_cli(dvc, mocker, mocked_status):
     assert cmd.run() == 0
     status.assert_called_once_with(
         untracked_files="all",
+        not_in_remote=False,
         granular=True,
     )