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

dvc/command/run.py

@@ -51,6 +51,7 @@ class CmdRun(CmdBase):
                 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 @@ class Stage(object):
         locked=False,
         tag=None,
         state=None,
+        always_changed=False,
     ):
         if deps is None:
             deps = []
@@ -179,6 +182,7 @@ class Stage(object):
         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 @@ class Stage(object):
             )
             return True
 
+        if self.always_changed:
+            return True
+
         for dep in self.deps:
             status = dep.status()
             if status:
@@ -457,6 +464,7 @@ class Stage(object):
             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 @@ class Stage(object):
                 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 @@ class Stage(object):
             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 @@ class Stage(object):
                 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 @@ class Stage(object):
                 if (
                     not force
                     and not self.is_callback
+                    and not self.always_changed
                     and self._already_cached()
                 ):
                     self.checkout()
@@ -885,7 +897,7 @@ class Stage(object):
         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"]