run: introduce --always-changed

This is an extension of our callback stages (no deps, so always
considered as changed) for stages with dependencies, so that these
stages could be used properly in the middle of the DAG. See attached
issue for more info.

Related to #2378

Author: efiop

dvc/command/run.py

@@ -51,6 +51,7 @@ def run(self):
                 no_commit=self.args.no_commit,
                 outs_persist=self.args.outs_persist,
                 outs_persist_no_cache=self.args.outs_persist_no_cache,
+                always_changed=self.args.always_changed,
             )
         except DvcException:
             logger.exception("failed to run command")
@@ -189,6 +190,12 @@ def add_parser(subparsers, parent_parser):
         help="Declare output file or directory that will not be "
         "removed upon repro (do not put into DVC cache).",
     )
+    run_parser.add_argument(
+        "--always-changed",
+        action="store_true",
+        default=False,
+        help="Always consider this DVC-file as changed.",
+    )
     run_parser.add_argument(
         "command", nargs=argparse.REMAINDER, help="Command to execute."
     )

dvc/stage.py

@@ -139,6 +139,7 @@ class Stage(object):
     PARAM_OUTS = "outs"
     PARAM_LOCKED = "locked"
     PARAM_META = "meta"
+    PARAM_ALWAYS_CHANGED = "always_changed"
 
     SCHEMA = {
         Optional(PARAM_MD5): Or(str, None),
@@ -148,6 +149,7 @@ class Stage(object):
         Optional(PARAM_OUTS): Or(And(list, Schema([output.SCHEMA])), None),
         Optional(PARAM_LOCKED): bool,
         Optional(PARAM_META): object,
+        Optional(PARAM_ALWAYS_CHANGED): bool,
     }
 
     TAG_REGEX = r"^(?P<path>.*)@(?P<tag>[^\\/@:]*)$"
@@ -164,6 +166,7 @@ def __init__(
         locked=False,
         tag=None,
         state=None,
+        always_changed=False,
     ):
         if deps is None:
             deps = []
@@ -179,6 +182,7 @@ def __init__(
         self.md5 = md5
         self.locked = locked
         self.tag = tag
+        self.always_changed = always_changed
         self._state = state or {}
 
     def __repr__(self):
@@ -244,6 +248,9 @@ def _changed_deps(self):
             )
             return True
 
+        if self.always_changed:
+            return True
+
         for dep in self.deps:
             status = dep.status()
             if status:
@@ -457,6 +464,7 @@ def create(repo, **kwargs):
             wdir=wdir,
             cmd=kwargs.get("cmd", None),
             locked=kwargs.get("locked", False),
+            always_changed=kwargs.get("always_changed", False),
         )
 
         Stage._fill_stage_outputs(stage, **kwargs)
@@ -515,6 +523,7 @@ def create(repo, **kwargs):
                 not ignore_build_cache
                 and stage.is_cached
                 and not stage.is_callback
+                and not stage.always_changed
             ):
                 logger.info("Stage is cached, skipping.")
                 return None
@@ -619,6 +628,7 @@ def load(repo, fname):
             md5=d.get(Stage.PARAM_MD5),
             locked=d.get(Stage.PARAM_LOCKED, False),
             tag=tag,
+            always_changed=d.get(Stage.PARAM_ALWAYS_CHANGED, False),
             state=state,
         )
 
@@ -639,6 +649,7 @@ def dumpd(self):
                 Stage.PARAM_DEPS: [d.dumpd() for d in self.deps],
                 Stage.PARAM_OUTS: [o.dumpd() for o in self.outs],
                 Stage.PARAM_META: self._state.get("meta"),
+                Stage.PARAM_ALWAYS_CHANGED: self.always_changed,
             }.items()
             if value
         }
@@ -839,6 +850,7 @@ def run(self, dry=False, no_commit=False, force=False):
                 if (
                     not force
                     and not self.is_callback
+                    and not self.always_changed
                     and self._already_cached()
                 ):
                     self.checkout()
@@ -885,7 +897,7 @@ def status(self):
         if self.changed_md5():
             ret.append("changed checksum")
 
-        if self.is_callback:
+        if self.is_callback or self.always_changed:
             ret.append("always changed")
 
         if ret:

tests/unit/command/test_run.py

@@ -32,6 +32,7 @@ def test_run(mocker, dvc_repo):
             "outs-persist",
             "--outs-persist-no-cache",
             "outs-persist-no-cache",
+            "--always-changed",
             "command",
         ]
     )
@@ -58,6 +59,7 @@ def test_run(mocker, dvc_repo):
         ignore_build_cache=True,
         remove_outs=True,
         no_commit=True,
+        always_changed=True,
         cmd="command",
     )
 
@@ -83,6 +85,7 @@ def test_run_args_from_cli(mocker, dvc_repo):
         ignore_build_cache=False,
         remove_outs=False,
         no_commit=False,
+        always_changed=False,
         cmd="echo foo",
     )
 
@@ -108,5 +111,6 @@ def test_run_args_with_spaces(mocker, dvc_repo):
         ignore_build_cache=False,
         remove_outs=False,
         no_commit=False,
+        always_changed=False,
         cmd='echo "foo bar"',
     )

tests/unit/test_stage.py

@@ -107,3 +107,10 @@ def test_stage_run_ignore_sigint(mocker):
     assert communicate.called_once_with()
     signal_mock.assert_any_call(signal.SIGINT, signal.SIG_IGN)
     assert signal.getsignal(signal.SIGINT) == signal.default_int_handler
+
+
+def test_always_changed():
+    stage = Stage(None, "path", always_changed=True)
+    stage.save()
+    assert stage.changed()
+    assert stage.status()["path"] == ["always changed"]