repro: accept multiple targets (#5111)

Currently, we iterate over targets manually in CLI, which results in a
complicated code and a DAG collection/check overhead. With this PR we
collect the DAG once, figure out all the needed targets within it and
just reproduce them in one pass.

Author: efiop

dvc/command/base.py

@@ -45,14 +45,6 @@ def __init__(self, args):
             updater = Updater(self.repo.tmp_dir, hardlink_lock=hardlink_lock)
             updater.check()
 
-    @property
-    def default_targets(self):
-        """Default targets for `dvc repro`."""
-        from dvc.dvcfile import PIPELINE_FILE
-
-        logger.trace(f"assuming default target '{PIPELINE_FILE}'.")
-        return [PIPELINE_FILE]
-
     @abstractmethod
     def run(self):
         pass

dvc/command/experiments.py

@@ -441,31 +441,17 @@ def run(self):
 
 class CmdExperimentsRun(CmdRepro):
     def run(self):
-        # Dirty hack so the for loop below can at least enter once
-        if self.args.all_pipelines:
-            self.args.targets = [None]
-        elif not self.args.targets:
-            self.args.targets = self.default_targets
-
-        ret = 0
-        for target in self.args.targets:
-            try:
-                self.repo.experiments.run(
-                    target,
-                    name=self.args.name,
-                    queue=self.args.queue,
-                    run_all=self.args.run_all,
-                    jobs=self.args.jobs,
-                    params=self.args.params,
-                    checkpoint_resume=self.args.checkpoint_resume,
-                    **self._repro_kwargs,
-                )
-            except DvcException:
-                logger.exception("")
-                ret = 1
-                break
+        self.repo.experiments.run(
+            name=self.args.name,
+            queue=self.args.queue,
+            run_all=self.args.run_all,
+            jobs=self.args.jobs,
+            params=self.args.params,
+            checkpoint_resume=self.args.checkpoint_resume,
+            **self._repro_kwargs,
+        )
 
-        return ret
+        return 0
 
 
 class CmdExperimentsGC(CmdRepro):

dvc/command/repro.py

@@ -6,46 +6,30 @@
 from dvc.command.metrics import _show_metrics
 from dvc.command.status import CmdDataStatus
 from dvc.dvcfile import PIPELINE_FILE
-from dvc.exceptions import DvcException
 
 logger = logging.getLogger(__name__)
 
 
 class CmdRepro(CmdBase):
     def run(self):
-        # Dirty hack so the for loop below can at least enter once
-        if self.args.all_pipelines:
-            self.args.targets = [None]
-        elif not self.args.targets:
-            self.args.targets = self.default_targets
+        stages = self.repo.reproduce(**self._repro_kwargs)
+        if len(stages) == 0:
+            logger.info(CmdDataStatus.UP_TO_DATE_MSG)
+        else:
+            logger.info(
+                "Use `dvc push` to send your updates to " "remote storage."
+            )
 
-        ret = 0
-        for target in self.args.targets:
-            try:
-                stages = self.repo.reproduce(target, **self._repro_kwargs)
+        if self.args.metrics:
+            metrics = self.repo.metrics.show()
+            logger.info(_show_metrics(metrics))
 
-                if len(stages) == 0:
-                    logger.info(CmdDataStatus.UP_TO_DATE_MSG)
-                else:
-                    logger.info(
-                        "Use `dvc push` to send your updates to "
-                        "remote storage."
-                    )
-
-                if self.args.metrics:
-                    metrics = self.repo.metrics.show()
-                    logger.info(_show_metrics(metrics))
-
-            except DvcException:
-                logger.exception("")
-                ret = 1
-                break
-
-        return ret
+        return 0
 
     @property
     def _repro_kwargs(self):
         return {
+            "targets": self.args.targets,
             "single_item": self.args.single_item,
             "force": self.args.force,
             "dry": self.args.dry,

dvc/repo/experiments/run.py

@@ -37,13 +37,13 @@ def _parse_params(path_params: Iterable):
 @locked
 def run(
     repo,
-    target: Optional[str] = None,
+    targets: Optional[Iterable] = None,
     params: Optional[Iterable] = None,
     run_all: Optional[bool] = False,
     jobs: Optional[int] = 1,
     **kwargs,
 ) -> dict:
-    """Reproduce the specified target as an experiment.
+    """Reproduce the specified targets as an experiment.
 
     Accepts the same additional kwargs as Repo.reproduce.
 
@@ -59,10 +59,10 @@ def run(
         params = []
     try:
         return repo.experiments.reproduce_one(
-            target=target, params=params, **kwargs
+            targets=targets, params=params, **kwargs
         )
     except UnchangedExperimentError:
         # If experiment contains no changes, just run regular repro
         kwargs.pop("queue", None)
         kwargs.pop("checkpoint_resume", None)
-        return {None: repo.reproduce(target=target, **kwargs)}
+        return {None: repo.reproduce(targets=targets, **kwargs)}

dvc/repo/reproduce.py

@@ -3,7 +3,7 @@
 import typing
 from functools import partial
 
-from dvc.exceptions import InvalidArgumentError, ReproductionError
+from dvc.exceptions import ReproductionError
 from dvc.repo.scm_context import scm_context
 from dvc.stage.run import CheckpointKilledError
 
@@ -92,19 +92,22 @@ def _get_active_graph(G):
 @scm_context
 def reproduce(
     self: "Repo",
-    target=None,
+    targets=None,
     recursive=False,
     pipeline=False,
     all_pipelines=False,
     **kwargs,
 ):
     glob = kwargs.pop("glob", False)
     accept_group = not glob
-    assert target is None or isinstance(target, str)
-    if not target and not all_pipelines:
-        raise InvalidArgumentError(
-            "Neither `target` nor `--all-pipelines` are specified."
-        )
+
+    if isinstance(targets, str):
+        targets = [targets]
+
+    if not all_pipelines and targets is None:
+        from dvc.dvcfile import PIPELINE_FILE
+
+        targets = [PIPELINE_FILE]
 
     interactive = kwargs.get("interactive", False)
     if not interactive:
@@ -113,28 +116,33 @@ def reproduce(
     active_graph = _get_active_graph(self.graph)
     active_pipelines = get_pipelines(active_graph)
 
+    stages = set()
     if pipeline or all_pipelines:
         if all_pipelines:
             pipelines = active_pipelines
         else:
-            stage = self.stage.get_target(target)
-            pipelines = [get_pipeline(active_pipelines, stage)]
+            pipelines = []
+            for target in targets:
+                stage = self.stage.get_target(target)
+                pipelines.append(get_pipeline(active_pipelines, stage))
 
-        targets = []
         for pipeline in pipelines:
             for stage in pipeline:
                 if pipeline.in_degree(stage) == 0:
-                    targets.append(stage)
+                    stages.add(stage)
     else:
-        targets = self.stage.collect(
-            target,
-            recursive=recursive,
-            graph=active_graph,
-            accept_group=accept_group,
-            glob=glob,
-        )
+        for target in targets:
+            stages.update(
+                self.stage.collect(
+                    target,
+                    recursive=recursive,
+                    graph=active_graph,
+                    accept_group=accept_group,
+                    glob=glob,
+                )
+            )
 
-    return _reproduce_stages(active_graph, targets, **kwargs)
+    return _reproduce_stages(active_graph, list(stages), **kwargs)
 
 
 def _reproduce_stages(
@@ -220,30 +228,28 @@ def _reproduce_stages(
 def _get_pipeline(G, stages, downstream, single_item):
     import networkx as nx
 
-    if single_item:
-        all_pipelines = stages
-    else:
-        all_pipelines = []
-        for stage in stages:
-            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.
-                nodes = nx.dfs_postorder_nodes(
-                    G.copy().reverse(copy=False), stage
-                )
-                all_pipelines += reversed(list(nodes))
-            else:
-                all_pipelines += nx.dfs_postorder_nodes(G, stage)
+    all_pipelines = []
+    for stage in stages:
+        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.
+            nodes = nx.dfs_postorder_nodes(G.copy().reverse(copy=False), stage)
+            all_pipelines += reversed(list(nodes))
+        else:
+            all_pipelines += nx.dfs_postorder_nodes(G, stage)
 
     pipeline = []
     for stage in all_pipelines:
         if stage not in pipeline:
+            if single_item and stage not in stages:
+                continue
+
             pipeline.append(stage)
 
     return pipeline

tests/func/test_repro_multistage.py

@@ -528,7 +528,7 @@ def test_repro_list_of_commands_in_order(tmp_dir, dvc):
         """
         )
     )
-    dvc.reproduce(target="multi")
+    dvc.reproduce(targets=["multi"])
     assert (tmp_dir / "foo").read_text() == "foo\n"
     assert (tmp_dir / "bar").read_text() == "bar\n"
 
@@ -547,6 +547,6 @@ def test_repro_list_of_commands_raise_and_stops_after_failure(tmp_dir, dvc):
         )
     )
     with pytest.raises(ReproductionError):
-        dvc.reproduce(target="multi")
+        dvc.reproduce(targets=["multi"])
     assert (tmp_dir / "foo").read_text() == "foo\n"
     assert not (tmp_dir / "bar").exists()

tests/unit/command/test_experiments.py

@@ -12,7 +12,6 @@
     CmdExperimentsRun,
     CmdExperimentsShow,
 )
-from dvc.dvcfile import PIPELINE_FILE
 from dvc.exceptions import InvalidArgumentError
 
 from .test_repro import default_arguments as repro_arguments
@@ -103,9 +102,7 @@ def test_experiments_run(dvc, scm, mocker, args, resume):
     mocker.patch.object(cmd.repo, "reproduce")
     mocker.patch.object(cmd.repo.experiments, "run")
     cmd.run()
-    cmd.repo.experiments.run.assert_called_with(
-        PIPELINE_FILE, **default_arguments
-    )
+    cmd.repo.experiments.run.assert_called_with(**default_arguments)
 
 
 def test_experiments_gc(dvc, scm, mocker):

tests/unit/command/test_repro.py

@@ -1,6 +1,5 @@
 from dvc.cli import parse_args
 from dvc.command.repro import CmdRepro
-from dvc.dvcfile import PIPELINE_FILE
 
 default_arguments = {
     "all_pipelines": False,
@@ -16,14 +15,15 @@
     "force_downstream": False,
     "pull": False,
     "glob": False,
+    "targets": [],
 }
 
 
 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(PIPELINE_FILE, **default_arguments)
+    cmd.repo.reproduce.assert_called_with(**default_arguments)
 
 
 def test_downstream(dvc, mocker):
@@ -32,4 +32,4 @@ def test_downstream(dvc, mocker):
     cmd.run()
     arguments = default_arguments.copy()
     arguments.update({"downstream": True})
-    cmd.repo.reproduce.assert_called_with(PIPELINE_FILE, **arguments)
+    cmd.repo.reproduce.assert_called_with(**arguments)