repro: implement downstream option (#1926)

* repro: implement downstream option

Close #1555

* Update scripts/completion/dvc.zsh

Co-Authored-By: mroutis <[email protected]>

* Update dvc/command/repro.py

Co-Authored-By: mroutis <[email protected]>

dvc/command/repro.py

@@ -39,6 +39,7 @@ class CmdRepro(CmdBase):
                     all_pipelines=self.args.all_pipelines,
                     ignore_build_cache=self.args.ignore_build_cache,
                     no_commit=self.args.no_commit,
+                    downstream=self.args.downstream,
                 )
 
                 if len(stages) == 0:
@@ -140,4 +141,10 @@ def add_parser(subparsers, parent_parser):
         default=False,
         help="Don't put files/directories into cache.",
     )
+    repro_parser.add_argument(
+        "--downstream",
+        action="store_true",
+        default=False,
+        help="Reproduce the pipeline starting from the specified stage.",
+    )
     repro_parser.set_defaults(func=CmdRepro)

dvc/repo/__init__.py

@@ -296,6 +296,47 @@ class Repo(object):
         return cache
 
     def graph(self, stages=None, from_directory=None):
+        """Generate a graph by using the given stages on the given directory
+
+        The nodes of the graph are the stage's path relative to the root.
+
+        Edges are created when the output of one stage is used as a
+        dependency in other stage.
+
+        The direction of the edges goes from the stage to its dependency:
+
+        For example, running the following:
+
+            $ dvc run -o A "echo A > A"
+            $ dvc run -d A -o B "echo B > B"
+            $ dvc run -d B -o C "echo C > C"
+
+        Will create the following graph:
+
+               ancestors <--
+                           |
+                C.dvc -> B.dvc -> A.dvc
+                |          |
+                |          --> descendants
+                |
+                ------- pipeline ------>
+                           |
+                           v
+              (weakly connected components)
+
+        Args:
+            stages (list): used to build a graph, if None given, use the ones
+                on the `from_directory`.
+
+            from_directory (str): directory where to look at for stages, if
+                None is given, use the current working directory
+
+        Raises:
+            OutputDuplicationError: two outputs with the same path
+            StagePathAsOutputError: stage inside an output directory
+            OverlappingOutputPathsError: output inside output directory
+            CyclicGraphError: resulting graph has cycles
+        """
         import networkx as nx
         from dvc.exceptions import (
             OutputDuplicationError,
@@ -333,7 +374,6 @@ class Repo(object):
                 if path_dir.startswith(out.path + os.sep):
                     raise StagePathAsOutputError(stage.wdir, stage.relpath)
 
-        # collect the whole DAG
         for stage in stages:
             node = os.path.relpath(stage.path, self.root_dir)
 

dvc/repo/reproduce.py

@@ -43,6 +43,7 @@ def reproduce(
     all_pipelines=False,
     ignore_build_cache=False,
     no_commit=False,
+    downstream=False,
 ):
     from dvc.stage import Stage
 
@@ -82,6 +83,7 @@ def reproduce(
                 interactive=interactive,
                 ignore_build_cache=ignore_build_cache,
                 no_commit=no_commit,
+                downstream=downstream,
             )
             ret.extend(stages)
 
@@ -97,6 +99,7 @@ def _reproduce(
     interactive=False,
     ignore_build_cache=False,
     no_commit=False,
+    downstream=False,
 ):
     import networkx as nx
     from dvc.stage import Stage
@@ -116,6 +119,7 @@ def _reproduce(
             interactive,
             ignore_build_cache,
             no_commit,
+            downstream,
         )
     else:
         ret = _reproduce_stage(
@@ -126,12 +130,67 @@ def _reproduce(
 
 
 def _reproduce_stages(
-    G, stages, node, force, dry, interactive, ignore_build_cache, no_commit
+    G,
+    stages,
+    node,
+    force,
+    dry,
+    interactive,
+    ignore_build_cache,
+    no_commit,
+    downstream,
 ):
+    r"""Derive the evaluation of the given node for the given graph.
+
+    When you _reproduce a stage_, you want to _evaluate the descendants_
+    to know if it make sense to _recompute_ it. A post-ordered search
+    will give us an order list of the nodes we want.
+
+    For example, let's say that we have the following pipeline:
+
+                               E
+                              / \
+                             D   F
+                            / \   \
+                           B   C   G
+                            \ /
+                             A
+
+    The derived evaluation of D would be: [A, B, C, D]
+
+    In case that `downstream` option is specifed, the desired effect
+    is to derive the evaluation starting from the given stage up to the
+    ancestors. However, the `networkx.ancestors` returns a set, without
+    any guarantee of any order, so we are going to reverse the graph and
+    use a pre-ordered search using the given stage as a starting point.
+
+                   E                                   A
+                  / \                                 / \
+                 D   F                               B   C   G
+                / \   \        --- reverse -->        \ /   /
+               B   C   G                               D   F
+                \ /                                     \ /
+                 A                                       E
+
+    The derived evaluation of _downstream_ B would be: [B, D, E]
+    """
+
     import networkx as nx
 
+    if downstream:
+        # NOTE (py3 only):
+        # Python's `deepcopy` defaults to pickle/unpickle the object.
+        # Stages are complex objects (with references to `repo`, `outs`,
+        # and `deps`) that cause struggles when you try to serialize them.
+        # We need to create a copy of the graph itself, and then reverse it,
+        # instead of using graph.reverse() directly because it calls
+        # `deepcopy` underneath -- unless copy=False is specified.
+        pipeline = nx.dfs_preorder_nodes(G.copy().reverse(copy=False), node)
+    else:
+        pipeline = nx.dfs_postorder_nodes(G, node)
+
     result = []
-    for n in nx.dfs_postorder_nodes(G, node):
+    for n in pipeline:
         try:
             ret = _reproduce_stage(
                 stages, n, force, dry, interactive, no_commit

scripts/completion/dvc.zsh

@@ -157,6 +157,7 @@ _dvc_repro=(
   {-i,--interactive}"[Ask for confirmation before reproducing each stage.]"
   {-p,--pipeline}"[Reproduce the whole pipeline that the specified stage file belongs to.]"
   {-P,--all-pipelines}"[Reproduce all pipelines in the project.]"
+  {-d,--downstream}"[Reproduce the pipeline starting from the specified stage.]"
 )
 
 _dvc_remove=(

tests/func/test_repro.py

@@ -1420,3 +1420,39 @@ def _format_dvc_line(line):
 
 def _hide_md5(text):
     return re.sub(r"\b[a-f0-9]{32}\b", "<md5>", text)
+
+
+class TestReproDownstream(TestDvc):
+    def test(self):
+        # The dependency graph should look like this:
+        #
+        #       E
+        #      / \
+        #     D   F
+        #    / \   \
+        #   B   C   G
+        #    \ /
+        #     A
+        #
+        assert main(["run", "-o", "A", "echo A>A"]) == 0
+        assert main(["run", "-d", "A", "-o", "B", "echo B>B"]) == 0
+        assert main(["run", "-d", "A", "-o", "C", "echo C>C"]) == 0
+        assert main(["run", "-d", "B", "-d", "C", "-o", "D", "echo D>D"]) == 0
+        assert main(["run", "-o", "G", "echo G>G"]) == 0
+        assert main(["run", "-d", "G", "-o", "F", "echo F>F"]) == 0
+        assert main(["run", "-d", "D", "-d", "F", "-o", "E", "echo E>E"]) == 0
+
+        # We want the evaluation to move from B to E
+        #
+        #       E
+        #      /
+        #     D
+        #    /
+        #   B
+        #
+        evaluation = self.dvc.reproduce("B.dvc", downstream=True, force=True)
+
+        assert len(evaluation) == 3
+        assert evaluation[0].relpath == "B.dvc"
+        assert evaluation[1].relpath == "D.dvc"
+        assert evaluation[2].relpath == "E.dvc"

tests/requirements.txt

@@ -2,6 +2,7 @@ pytest>=4.4.0
 pytest-timeout>=1.3.3
 pytest-cov>=2.6.1
 pytest-xdist>=1.26.1
+pytest-mock>=1.10.4
 flaky>=3.5.3
 mock>=2.0.0
 xmltodict>=0.11.0

tests/unit/command/test_repro.py

@@ -0,0 +1,30 @@
+from dvc.command.repro import CmdRepro
+from dvc.cli import parse_args
+
+default_arguments = {
+    "all_pipelines": False,
+    "downstream": False,
+    "dry": False,
+    "force": False,
+    "ignore_build_cache": False,
+    "interactive": False,
+    "no_commit": False,
+    "pipeline": False,
+    "recursive": True,
+}
+
+
+def test_default_arguments(dvc, mocker):
+    cmd = CmdRepro(parse_args(["repro"]))
+    mocker.patch.object(cmd.repo, "reproduce")
+    cmd.run()
+    cmd.repo.reproduce.assert_called_with("Dvcfile", **default_arguments)
+
+
+def test_downstream(dvc, mocker):
+    cmd = CmdRepro(parse_args(["repro", "--downstream"]))
+    mocker.patch.object(cmd.repo, "reproduce")
+    cmd.run()
+    arguments = default_arguments.copy()
+    arguments.update({"downstream": True})
+    cmd.repo.reproduce.assert_called_with("Dvcfile", **arguments)