From 29db8a1a9534e57153f4e981dac5289e5a38831e Mon Sep 17 00:00:00 2001 From: "mergify[bot]" <37929162+mergify[bot]@users.noreply.github.com> Date: Wed, 16 Dec 2020 05:17:55 +0200 Subject: [PATCH] repro: accept multiple targets (bp #5111) (#5115) * 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. (cherry picked from commit aec8d657912b7fd2b4a0a13afc220d9496a1b999) # Conflicts: # dvc/command/base.py # dvc/command/experiments.py # dvc/command/repro.py # tests/func/test_repro_multistage.py * resolve conflicts * resolve test conflicts Co-authored-by: Ruslan Kuprieiev Co-authored-by: Ruslan Kuprieiev --- dvc/command/base.py | 8 --- dvc/command/experiments.py | 34 +++-------- dvc/command/repro.py | 40 ++++-------- dvc/repo/experiments/run.py | 8 +-- dvc/repo/reproduce.py | 84 ++++++++++++++------------ tests/unit/command/test_experiments.py | 5 +- tests/unit/command/test_repro.py | 6 +- 7 files changed, 75 insertions(+), 110 deletions(-) diff --git a/dvc/command/base.py b/dvc/command/base.py index e55471b4d..c70e99ad9 100644 --- a/dvc/command/base.py +++ b/dvc/command/base.py @@ -45,14 +45,6 @@ class CmdBase(ABC): 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.debug(f"assuming default target '{PIPELINE_FILE}'.") - return [PIPELINE_FILE] - @abstractmethod def run(self): pass diff --git a/dvc/command/experiments.py b/dvc/command/experiments.py index c3d2b5747..75c62580b 100644 --- a/dvc/command/experiments.py +++ b/dvc/command/experiments.py @@ -451,32 +451,18 @@ class CmdExperimentsRun(CmdRepro): saved_dir = os.path.realpath(os.curdir) os.chdir(self.args.cwd) - # 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, + ) os.chdir(saved_dir) - return ret + return 0 class CmdExperimentsGC(CmdRepro): diff --git a/dvc/command/repro.py b/dvc/command/repro.py index 20ee20f89..b941b25b3 100644 --- a/dvc/command/repro.py +++ b/dvc/command/repro.py @@ -7,7 +7,6 @@ from dvc.command.base import CmdBase, append_doc_link 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__) @@ -17,40 +16,25 @@ class CmdRepro(CmdBase): saved_dir = os.path.realpath(os.curdir) os.chdir(self.args.cwd) - # 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 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 + if self.args.metrics: + metrics = self.repo.metrics.show() + logger.info(_show_metrics(metrics)) os.chdir(saved_dir) - 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, diff --git a/dvc/repo/experiments/run.py b/dvc/repo/experiments/run.py index 8a9e21f81..54b43bc1f 100644 --- a/dvc/repo/experiments/run.py +++ b/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)} diff --git a/dvc/repo/reproduce.py b/dvc/repo/reproduce.py index 99cdd4058..6d0846b87 100644 --- a/dvc/repo/reproduce.py +++ b/dvc/repo/reproduce.py @@ -2,7 +2,7 @@ import logging 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 @@ -80,7 +80,7 @@ def _get_active_graph(G): @scm_context def reproduce( self: "Repo", - target=None, + targets=None, recursive=False, pipeline=False, all_pipelines=False, @@ -88,11 +88,14 @@ def reproduce( ): 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: @@ -101,28 +104,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( @@ -208,30 +216,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 diff --git a/tests/unit/command/test_experiments.py b/tests/unit/command/test_experiments.py index 0f3408830..690dbe484 100644 --- a/tests/unit/command/test_experiments.py +++ b/tests/unit/command/test_experiments.py @@ -9,7 +9,6 @@ from dvc.command.experiments import ( CmdExperimentsRun, CmdExperimentsShow, ) -from dvc.dvcfile import PIPELINE_FILE from dvc.exceptions import InvalidArgumentError from .test_repro import default_arguments as repro_arguments @@ -100,9 +99,7 @@ def test_experiments_run(dvc, 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, mocker): diff --git a/tests/unit/command/test_repro.py b/tests/unit/command/test_repro.py index 172893b37..75fabbfe6 100644 --- a/tests/unit/command/test_repro.py +++ b/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,6 +15,7 @@ default_arguments = { "force_downstream": False, "pull": False, "glob": False, + "targets": [], } @@ -23,7 +23,7 @@ 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)