diff --git a/dvc/command/repro.py b/dvc/command/repro.py index beb0088f1..2d3dfba9f 100644 --- a/dvc/command/repro.py +++ b/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) diff --git a/dvc/repo/__init__.py b/dvc/repo/__init__.py index 394cd585e..4d599a71d 100644 --- a/dvc/repo/__init__.py +++ b/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) diff --git a/dvc/repo/reproduce.py b/dvc/repo/reproduce.py index 54c4d3361..672d03f76 100644 --- a/dvc/repo/reproduce.py +++ b/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 diff --git a/scripts/completion/dvc.zsh b/scripts/completion/dvc.zsh index 9588e0af8..f7d71f3f5 100644 --- a/scripts/completion/dvc.zsh +++ b/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=( diff --git a/tests/func/test_repro.py b/tests/func/test_repro.py index 3a6b7eed2..4c4a08287 100644 --- a/tests/func/test_repro.py +++ b/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", "", 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" diff --git a/tests/requirements.txt b/tests/requirements.txt index 2bd83a275..fc290e945 100644 --- a/tests/requirements.txt +++ b/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 diff --git a/tests/unit/command/test_repro.py b/tests/unit/command/test_repro.py new file mode 100644 index 000000000..2d1bf4882 --- /dev/null +++ b/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)