commit: support granularity (#4297) (#5088)

* commit: support granularity (#4297)

* commit: support granularity

Co-authored-by: Ruslan Kuprieiev <[email protected]>

dvc/cache/base.py

@@ -307,7 +307,15 @@ class CloudCache:
 
         return hi
 
-    def _save_dir(self, path_info, tree, hash_info, save_link=True, **kwargs):
+    def _save_dir(
+        self,
+        path_info,
+        tree,
+        hash_info,
+        save_link=True,
+        filter_info=None,
+        **kwargs,
+    ):
         if not hash_info.dir_info:
             hash_info.dir_info = tree.cache.get_dir_cache(hash_info)
         hi = self.save_dir_info(hash_info.dir_info, hash_info)
@@ -316,6 +324,9 @@ class CloudCache:
             desc="Saving " + path_info.name,
             unit="file",
         ):
+            if filter_info and not entry_info.isin_or_eq(filter_info):
+                continue
+
             self._save_file(
                 entry_info, tree, entry_hash, save_link=False, **kwargs
             )

dvc/output/base.py

@@ -303,13 +303,19 @@ class BaseOutput:
         if self.isfile() and self.isexec:
             self.tree.set_exec(self.path_info)
 
-    def commit(self):
+    def commit(self, filter_info=None):
         if not self.exists:
             raise self.DoesNotExistError(self)
 
         assert self.hash_info
+
         if self.use_cache:
-            self.cache.save(self.path_info, self.cache.tree, self.hash_info)
+            self.cache.save(
+                self.path_info,
+                self.cache.tree,
+                self.hash_info,
+                filter_info=filter_info,
+            )
             self.set_exec()
 
     def dumpd(self):

dvc/repo/commit.py

@@ -41,15 +41,16 @@ def commit(
 ):
     from dvc.dvcfile import Dvcfile
 
-    stages = self.stage.collect(
+    stages_info = self.stage.collect_granular(
         target, with_deps=with_deps, recursive=recursive
     )
-    for stage in stages:
+    for stage_info in stages_info:
+        stage = stage_info.stage
         changes = stage.changed_entries()
         if any(changes):
             prompt_to_commit(stage, changes, force=force)
             stage.save()
-        stage.commit()
+        stage.commit(filter_info=stage_info.filter_info)
 
         Dvcfile(self, stage.path).dump(stage, update_pipeline=False)
-    return stages
+    return [s.stage for s in stages_info]

dvc/stage/__init__.py

@@ -479,13 +479,13 @@ class Stage(params.StageParams):
         )
 
     @rwlocked(write=["outs"])
-    def commit(self, allow_missing=False):
+    def commit(self, allow_missing=False, filter_info=None):
         from dvc.output.base import OutputDoesNotExistError
 
         link_failures = []
-        for out in self.outs:
+        for out in self.filter_outs(filter_info):
             try:
-                out.commit()
+                out.commit(filter_info=filter_info)
             except OutputDoesNotExistError:
                 if not (allow_missing or out.checkpoint):
                     raise

tests/func/test_commit.py

@@ -1,4 +1,4 @@
-from os import fspath
+import os
 
 import pytest
 
@@ -83,6 +83,77 @@ def test_commit_no_exec(tmp_dir, dvc):
     assert dvc.status(stage.path) == {}
 
 
+def test_commit_granular_output(tmp_dir, dvc):
+    dvc.run(
+        name="mystage",
+        cmd=["echo foo>foo", "echo bar>bar"],
+        outs=["foo", "bar"],
+        no_commit=True,
+    )
+
+    cache = tmp_dir / ".dvc" / "cache"
+    assert not list(cache.glob("*/*"))
+
+    dvc.commit("foo")
+    assert list(cache.glob("*/*")) == [
+        cache / "d3" / "b07384d113edec49eaa6238ad5ff00"
+    ]
+
+
+def test_commit_granular_output_file(tmp_dir, dvc):
+    tmp_dir.gen("foo", "foo")
+    dvc.add("foo", no_commit=True)
+    dvc.commit("foo")
+    assert dvc.status() == {}
+
+
+def test_commit_granular_output_dir(tmp_dir, dvc):
+    tmp_dir.gen(
+        {
+            "data": {
+                "foo": "foo",
+                "bar": "bar",
+                "subdir": {"subfoo": "subfoo", "subbar": "subbar"},
+            }
+        }
+    )
+    dvc.add("data", no_commit=True)
+    dvc.commit("data")
+    assert dvc.status() == {}
+
+
+def test_commit_granular_dir(tmp_dir, dvc):
+    tmp_dir.gen(
+        {
+            "data": {
+                "foo": "foo",
+                "bar": "bar",
+                "subdir": {"subfoo": "subfoo", "subbar": "subbar"},
+            }
+        }
+    )
+    dvc.add("data", no_commit=True)
+
+    cache = tmp_dir / ".dvc" / "cache"
+    assert set(cache.glob("*/*")) == {
+        cache / "1a" / "ca2c799df82929bbdd976557975546.dir"
+    }
+
+    dvc.commit(os.path.join("data", "foo"))
+    assert set(cache.glob("*/*")) == {
+        cache / "1a" / "ca2c799df82929bbdd976557975546.dir",
+        cache / "ac" / "bd18db4cc2f85cedef654fccc4a4d8",
+    }
+
+    dvc.commit(os.path.join("data", "subdir"))
+    assert set(cache.glob("*/*")) == {
+        cache / "1a" / "ca2c799df82929bbdd976557975546.dir",
+        cache / "ac" / "bd18db4cc2f85cedef654fccc4a4d8",
+        cache / "4c" / "e8d2a2cf314a52fa7f315ca37ca445",
+        cache / "68" / "dde2c3c4e7953c2290f176bbdc9a54",
+    }
+
+
 def test_commit_no_exec_missing_dep(tmp_dir, dvc):
     stage = dvc.run(
         name="my", cmd="mycmd", deps=["dep"], outs=["out"], no_exec=True
@@ -118,6 +189,6 @@ def test_imported_entries_unchanged(tmp_dir, dvc, erepo_dir):
     with erepo_dir.chdir():
         erepo_dir.dvc_gen("file", "file content", "initial commit")
 
-    stage = dvc.imp(fspath(erepo_dir), "file")
+    stage = dvc.imp(os.fspath(erepo_dir), "file")
 
     assert stage.changed_entries() == ([], [], None)