From 36ec25ffd78a23bb718036419eb28062edee6177 Mon Sep 17 00:00:00 2001 From: skshetry <18718008+skshetry@users.noreply.github.com> Date: Fri, 30 Jun 2023 10:05:44 +0545 Subject: [PATCH] repro: remove on_unchanged kwarg and _repro_callback (#9674) repro: remove on_unchaged kwarg and _repro_callback --- dvc/repo/experiments/executor/base.py | 16 +--------------- dvc/repo/reproduce.py | 19 +++++-------------- 2 files changed, 6 insertions(+), 29 deletions(-) diff --git a/dvc/repo/experiments/executor/base.py b/dvc/repo/experiments/executor/base.py index 4e6d8615b..ab0f17e2b 100644 --- a/dvc/repo/experiments/executor/base.py +++ b/dvc/repo/experiments/executor/base.py @@ -459,25 +459,16 @@ class BaseExecutor(ABC): should force overwrite any existing duplicates. """ from dvc.repo.checkout import checkout as dvc_checkout - from dvc.repo.reproduce import reproduce as dvc_reproduce - from dvc.stage import PipelineStage from dvc.ui import ui auto_push = env2bool(DVC_EXP_AUTO_PUSH) git_remote = os.getenv(DVC_EXP_GIT_REMOTE, None) - unchanged = [] - if queue is not None: queue.put((rev, os.getpid())) if log_errors and log_level is not None: cls._set_log_level(log_level) - def filter_pipeline(stages): - unchanged.extend( - [stage for stage in stages if isinstance(stage, PipelineStage)] - ) - exp_hash: Optional[str] = None exp_ref: Optional["ExpRefInfo"] = None repro_force: bool = False @@ -519,12 +510,7 @@ class BaseExecutor(ABC): recursive=kwargs.get("recursive", False), ) - stages = dvc_reproduce( - dvc, - *args, - on_unchanged=filter_pipeline, - **kwargs, - ) + stages = dvc.reproduce(*args, **kwargs) if paths := cls._get_top_level_paths(dvc): logger.debug("Staging top-level files: %s", paths) dvc.scm_context.add(paths) diff --git a/dvc/repo/reproduce.py b/dvc/repo/reproduce.py index 66c303893..527adcc99 100644 --- a/dvc/repo/reproduce.py +++ b/dvc/repo/reproduce.py @@ -122,7 +122,6 @@ def _reproduce_stages( # noqa: C901 stages, downstream=False, single_item=False, - on_unchanged=None, **kwargs, ): r"""Derive the evaluation of the given node for the given graph. @@ -175,24 +174,20 @@ def _reproduce_stages( # noqa: C901 try: kwargs["upstream"] = result + unchanged ret = _reproduce_stage(stage, **kwargs) - - if len(ret) == 0: + if not ret: unchanged.extend([stage]) - elif force_downstream: + continue + + result.extend(ret) + if force_downstream: # NOTE: we are walking our pipeline from the top to the # bottom. If one stage is changed, it will be reproduced, # which tells us that we should force reproducing all of # the other stages down below, even if their direct # dependencies didn't change. kwargs["force"] = True - - if ret: - result.extend(ret) except Exception as exc: # noqa: BLE001 raise ReproductionError(stage.addressing) from exc - - if on_unchanged is not None: - on_unchanged(unchanged) return result @@ -233,7 +228,3 @@ def _get_steps(graph, stages, downstream, single_item): steps.append(stage) return steps - - -def _repro_callback(experiments_callback, unchanged, stages): - experiments_callback(unchanged, stages)