gc: Add --not-in-remote arg.

Closes #2036

dvc/commands/gc.py

@@ -22,6 +22,8 @@ class CmdGC(CmdBase):
             workspace=self.args.workspace,
             rev=self.args.rev,
             num=self.args.num,
+            cloud=self.args.cloud,
+            not_in_remote=self.args.not_in_remote,
         )
 
         if self.args.rev:
@@ -74,6 +76,7 @@ class CmdGC(CmdBase):
             workspace=self.args.workspace,
             rev=self.args.rev,
             num=self.args.num,
+            not_in_remote=self.args.not_in_remote,
         )
         return 0
 
@@ -157,6 +160,12 @@ def add_parser(subparsers, parent_parser):
         default=False,
         help="Keep data files for all experiments.",
     )
+    gc_parser.add_argument(
+        "--not-in-remote",
+        action="store_true",
+        default=False,
+        help="Keep data files that are not present in the remote.",
+    )
     gc_parser.add_argument(
         "-c",
         "--cloud",

dvc/repo/gc.py

@@ -12,6 +12,12 @@ logger = logging.getLogger(__name__)
 
 
 def _validate_args(**kwargs):
+    not_in_remote = kwargs.pop("not_in_remote", None)
+    cloud = kwargs.pop("cloud", None)
+    if not_in_remote and cloud:
+        raise InvalidArgumentError(
+            "`--not-in-remote` and `--cloud` are mutually exclusive"
+        )
     if not any(kwargs.values()):
         raise InvalidArgumentError(
             "Either of `-w|--workspace`, `-a|--all-branches`, `-T|--all-tags` "
@@ -23,7 +29,7 @@ def _validate_args(**kwargs):
 
 
 @locked
-def gc(  # noqa: PLR0913
+def gc(  # noqa: PLR0913, C901
     self: "Repo",
     all_branches: bool = False,
     cloud: bool = False,
@@ -39,6 +45,7 @@ def gc(  # noqa: PLR0913
     commit_date: Optional[str] = None,
     rev: Optional[str] = None,
     num: Optional[int] = None,
+    not_in_remote: bool = False,
 ):
     # require `workspace` to be true to come into effect.
     # assume `workspace` to be enabled if any of `all_tags`, `all_commits`,
@@ -52,6 +59,8 @@ def gc(  # noqa: PLR0913
         commit_date=commit_date,
         rev=rev,
         num=num,
+        cloud=cloud,
+        not_in_remote=not_in_remote,
     )
 
     from contextlib import ExitStack
@@ -85,6 +94,11 @@ def gc(  # noqa: PLR0913
             ).values():
                 used_obj_ids.update(obj_ids)
 
+    if not_in_remote:
+        cloud_odb = self.cloud.get_remote_odb(remote, "gc --not-in-remote")
+        remote_hashes = list(cloud_odb.all(jobs=jobs))
+        used_obj_ids = {x for x in used_obj_ids if x.value not in remote_hashes}
+
     for scheme, odb in self.cache.by_scheme():
         if not odb:
             continue
@@ -96,9 +110,9 @@ def gc(  # noqa: PLR0913
     if not cloud:
         return
 
-    odb = self.cloud.get_remote_odb(remote, "gc -c")
-    removed = ogc(odb, used_obj_ids, jobs=jobs)
+    cloud_odb = self.cloud.get_remote_odb(remote, "gc -c")
+    removed = ogc(cloud_odb, used_obj_ids, jobs=jobs)
     if removed:
-        get_index(odb).clear()
+        get_index(cloud_odb).clear()
     else:
         logger.info("No unused cache to remove from remote.")

tests/func/test_gc.py

@@ -2,11 +2,12 @@ import datetime
 import logging
 import os
 import shutil
+import textwrap
 
 import pytest
 
 from dvc.cli import main
-from dvc.exceptions import CollectCacheError
+from dvc.exceptions import CollectCacheError, InvalidArgumentError
 from dvc.fs import LocalFileSystem
 from dvc.utils.fs import remove
 from dvc_data.hashfile.db.local import LocalHashFileDB
@@ -334,3 +335,100 @@ def test_date(tmp_dir, scm, dvc):
     assert dvc.cache.local.exists(
         "3bcf3b1be3e794a97a5a6b93a005784c"
     )  # "modified, again"
+
+
+def test_gc_not_in_remote(tmp_dir, scm, dvc, tmp_path_factory, mocker):
+    storage = os.fspath(tmp_path_factory.mktemp("test_remote_base"))
+    dvc.config["remote"]["local_remote"] = {"url": storage}
+    dvc.config["core"]["remote"] = "local_remote"
+
+    (standalone, dir1, dir2) = tmp_dir.dvc_gen(
+        {
+            "file1": "standalone",
+            "dir1": {"file2": "file2"},
+            "dir2": {"file3": "file3", "file4": "file4"},
+        }
+    )
+    mocked_remove = mocker.spy(LocalFileSystem, "remove")
+    dvc.gc(workspace=True)
+    assert not mocked_remove.call_args_list
+
+    dvc.push(["file1", "dir1"])
+
+    dvc.gc(workspace=True, not_in_remote=True)
+
+    assert len(mocked_remove.mock_calls) == 3
+
+    arg_list = mocked_remove.call_args_list
+
+    standalone_hash = standalone.outs[0].hash_info.value
+    dir1_hash = dir1.outs[0].hash_info.value
+    assert f"{dir1_hash[2:]}.unpacked" in arg_list[0][0][1]
+    assert f"{dir1_hash[2:]}" in arg_list[1][0][1][0]
+    # We expect 2 calls: standalone_hash and dir1/file2/file2
+    assert len(arg_list[2][0][1]) == 2
+    # Order is not guaranteed here.
+    assert (
+        f"{standalone_hash[2:]}" in arg_list[2][0][1][0]
+        or f"{standalone_hash[2:]}" in arg_list[2][0][1][1]
+    )
+
+
+def test_gc_not_in_remote_remote_arg(tmp_dir, scm, dvc, tmp_path_factory, mocker):
+    storage = os.fspath(tmp_path_factory.mktemp("test_remote_base"))
+    dvc.config["remote"]["local_remote"] = {"url": storage}
+    dvc.config["core"]["remote"] = "local_remote"
+    other_storage = os.fspath(tmp_path_factory.mktemp("test_remote_other"))
+    dvc.config["remote"]["other_remote"] = {"url": other_storage}
+
+    tmp_dir.dvc_gen(
+        {
+            "file1": "standalone",
+            "dir1": {"file2": "file2"},
+            "dir2": {"file3": "file3", "file4": "file4"},
+        }
+    )
+    mocked_remove = mocker.spy(LocalFileSystem, "remove")
+
+    dvc.push(["file1", "dir1"], remote="other_remote")
+
+    dvc.gc(workspace=True, not_in_remote=True)
+
+    assert not mocked_remove.mock_calls
+
+    dvc.gc(workspace=True, not_in_remote=True, remote="other_remote")
+
+    assert len(mocked_remove.mock_calls) == 3
+
+
+def test_gc_not_in_remote_cloud(tmp_dir, scm, dvc):
+    with pytest.raises(
+        InvalidArgumentError,
+        match="`--not-in-remote` and `--cloud` are mutually exclusive",
+    ):
+        dvc.gc(workspace=True, not_in_remote=True, cloud=True)
+
+
+@pytest.mark.xfail(reason="--cloud Doesn't respect `remote` field.")
+def test_gc_remote_field(tmp_dir, scm, dvc, tmp_path_factory, mocker):
+    storage = os.fspath(tmp_path_factory.mktemp("test_remote_base"))
+    dvc.config["remote"]["local_remote"] = {"url": storage}
+    dvc.config["core"]["remote"] = "local_remote"
+    other_storage = os.fspath(tmp_path_factory.mktemp("test_remote_other"))
+    dvc.config["remote"]["other_remote"] = {"url": other_storage}
+
+    text = textwrap.dedent(
+        """\
+        outs:
+        - path: foo
+          remote: other_remote
+    """
+    )
+    tmp_dir.gen("foo.dvc", text)
+    (foo,) = tmp_dir.dvc_gen("foo", "foo")
+    dvc.push()
+    dvc.remove(foo.relpath)
+
+    mocked_remove = mocker.spy(LocalFileSystem, "remove")
+    dvc.gc(workspace=True, cloud=True)
+    assert len(mocked_remove.mock_calls) == 2

tests/unit/command/test_gc.py

@@ -48,6 +48,7 @@ def test_(dvc, scm, mocker):
         repos=["project1", "project2"],
         rev=None,
         num=None,
+        not_in_remote=False,
     )
 
     cli_args = parse_args(["gc"])