dvc: bump to 1.11.8

* [WIP] repro: simplify the logic for frozen stages (#5114)

Fixes #5082

(cherry picked from commit 64dfaffd7c)

# Conflicts:
#	dvc/repo/reproduce.py

* resolve conflicts

Co-authored-by: Ruslan Kuprieiev <[email protected]>
Co-authored-by: Ruslan Kuprieiev <[email protected]>

.github/workflows/packages.yaml

@@ -1,8 +1,6 @@
 name: Build packages
 on:
-  push:
-    branches:
-    - master
+  push: {}
   release:
     types: [released]
 jobs:
@@ -121,11 +119,11 @@ jobs:
       with:
         store_login: ${{ secrets.STORE_LOGIN }}
         snap: ${{ steps.snapcraft.outputs.snap }}
-        release: edge
+        release: edge,v1/edge
     - name: Publish to the stable channel
       uses: snapcore/action-publish@v1
       if: github.event_name == 'release'
       with:
         store_login: ${{ secrets.STORE_LOGIN }}
         snap: ${{ steps.snapcraft.outputs.snap }}
-        release: stable
+        release: stable,v1/stable

.github/workflows/tests.yaml

@@ -1,12 +1,8 @@
 name: Tests
 
 on:
-  pull_request:
-    branches:
-      - master
-  push:
-    branches:
-      - master
+  - pull_request
+  - push
 
 env:
   DVC_TEST: "true"

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

dvc/command/config.py

@@ -6,6 +6,25 @@ from dvc.config import Config, ConfigError
 
 logger = logging.getLogger(__name__)
 
+NAME_REGEX = r"^(?P<remote>remote\.)?(?P<section>[^\.]*)\.(?P<option>[^\.]*)$"
+
+
+def _name_type(value):
+    import re
+
+    match = re.match(NAME_REGEX, value)
+    if not match:
+        raise argparse.ArgumentTypeError(
+            "name argument should look like "
+            "remote.name.option or "
+            "section.option"
+        )
+    return (
+        bool(match.group("remote")),
+        match.group("section").lower(),
+        match.group("option").lower(),
+    )
+
 
 class CmdConfig(CmdBaseNoRepo):
     def __init__(self, args):
@@ -14,20 +33,24 @@ class CmdConfig(CmdBaseNoRepo):
         self.config = Config(validate=False)
 
     def run(self):
-        section, opt = self.args.name.lower().strip().split(".", 1)
+        remote, section, opt = self.args.name
 
         if self.args.value is None and not self.args.unset:
             conf = self.config.load_one(self.args.level)
-            self._check(conf, section, opt)
+            if remote:
+                conf = conf["remote"]
+            self._check(conf, remote, section, opt)
             logger.info(conf[section][opt])
             return 0
 
         with self.config.edit(self.args.level) as conf:
+            if remote:
+                conf = conf["remote"]
             if self.args.unset:
-                self._check(conf, section, opt)
+                self._check(conf, remote, section, opt)
                 del conf[section][opt]
             else:
-                self._check(conf, section)
+                self._check(conf, remote, section)
                 conf[section][opt] = self.args.value
 
         if self.args.name == "cache.type":
@@ -39,14 +62,15 @@ class CmdConfig(CmdBaseNoRepo):
 
         return 0
 
-    def _check(self, conf, section, opt=None):
+    def _check(self, conf, remote, section, opt=None):
+        name = "remote" if remote else "section"
         if section not in conf:
-            msg = "section {} doesn't exist"
-            raise ConfigError(msg.format(self.args.name))
+            raise ConfigError(f"{name} '{section}' doesn't exist")
 
         if opt and opt not in conf[section]:
-            msg = "option {} doesn't exist"
-            raise ConfigError(msg.format(self.args.name))
+            raise ConfigError(
+                f"option '{opt}' doesn't exist in {name} '{section}'"
+            )
 
 
 parent_config_parser = argparse.ArgumentParser(add_help=False)
@@ -92,6 +116,10 @@ def add_parser(subparsers, parent_parser):
         action="store_true",
         help="Unset option.",
     )
-    config_parser.add_argument("name", help="Option name.")
+    config_parser.add_argument(
+        "name",
+        type=_name_type,
+        help="Option name (remote.name.option or section.option).",
+    )
     config_parser.add_argument("value", nargs="?", help="Option value.")
     config_parser.set_defaults(func=CmdConfig)

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):

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,

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

@@ -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
 
@@ -54,33 +54,11 @@ def _dump_stage(stage):
     dvcfile.dump(stage, update_pipeline=False)
 
 
-def _get_active_graph(G):
-    import networkx as nx
-
-    active = G.copy()
-    for stage in G:
-        if not stage.frozen:
-            continue
-        active.remove_edges_from(G.out_edges(stage))
-        for edge in G.out_edges(stage):
-            _, to_stage = edge
-            for node in nx.dfs_preorder_nodes(G, to_stage):
-                # NOTE: `in_degree` will return InDegreeView({}) if stage
-                # no longer exists in the `active` DAG.
-                if not active.in_degree(node):
-                    # NOTE: if some edge no longer exists `remove_edges_from`
-                    # will ignore it without error.
-                    active.remove_edges_from(G.out_edges(node))
-                    active.remove_node(node)
-
-    return active
-
-
 @locked
 @scm_context
 def reproduce(
     self: "Repo",
-    target=None,
+    targets=None,
     recursive=False,
     pipeline=False,
     all_pipelines=False,
@@ -88,41 +66,46 @@ 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 not targets:
+        from dvc.dvcfile import PIPELINE_FILE
+
+        targets = [PIPELINE_FILE]
 
     interactive = kwargs.get("interactive", False)
     if not interactive:
         kwargs["interactive"] = self.config["core"].get("interactive", False)
 
-    active_graph = _get_active_graph(self.graph)
-    active_pipelines = get_pipelines(active_graph)
-
+    stages = set()
     if pipeline or all_pipelines:
+        pipelines = get_pipelines(self.graph)
         if all_pipelines:
-            pipelines = active_pipelines
+            used_pipelines = pipelines
         else:
-            stage = self.stage.get_target(target)
-            pipelines = [get_pipeline(active_pipelines, stage)]
+            used_pipelines = []
+            for target in targets:
+                stage = self.stage.get_target(target)
+                used_pipelines.append(get_pipeline(pipelines, stage))
 
-        targets = []
-        for pipeline in pipelines:
-            for stage in pipeline:
-                if pipeline.in_degree(stage) == 0:
-                    targets.append(stage)
+        for pline in used_pipelines:
+            for stage in pline:
+                if pline.in_degree(stage) == 0:
+                    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,
+                    accept_group=accept_group,
+                    glob=glob,
+                )
+            )
 
-    return _reproduce_stages(active_graph, targets, **kwargs)
+    return _reproduce_stages(self.graph, list(stages), **kwargs)
 
 
 def _reproduce_stages(
@@ -163,7 +146,7 @@ def _reproduce_stages(
 
     The derived evaluation of _downstream_ B would be: [B, D, E]
     """
-    pipeline = _get_pipeline(G, stages, downstream, single_item)
+    steps = _get_steps(G, stages, downstream, single_item)
 
     force_downstream = kwargs.pop("force_downstream", False)
     result = []
@@ -171,7 +154,7 @@ def _reproduce_stages(
     # `ret` is used to add a cosmetic newline.
     ret = []
     checkpoint_func = kwargs.pop("checkpoint_func", None)
-    for stage in pipeline:
+    for stage in steps:
         if ret:
             logger.info("")
 
@@ -205,36 +188,43 @@ def _reproduce_stages(
     return result
 
 
-def _get_pipeline(G, stages, downstream, single_item):
+def _get_steps(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)
+    active = G.copy()
+    if not single_item:
+        # NOTE: frozen stages don't matter for single_item
+        for stage in G:
+            if stage.frozen:
+                # NOTE: disconnect frozen stage from its dependencies
+                active.remove_edges_from(G.out_edges(stage))
 
-    pipeline = []
+    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(active.reverse(copy=False), stage)
+            all_pipelines += reversed(list(nodes))
+        else:
+            all_pipelines += nx.dfs_postorder_nodes(active, stage)
+
+    steps = []
     for stage in all_pipelines:
-        if stage not in pipeline:
-            pipeline.append(stage)
+        if stage not in steps:
+            # NOTE: order of steps still matters for single_item
+            if single_item and stage not in stages:
+                continue
 
-    return pipeline
+            steps.append(stage)
+
+    return steps
 
 
 def _repro_callback(experiments_callback, unchanged, stages):

dvc/system.py

@@ -71,8 +71,8 @@ class System:
         clonefile.restype = ctypes.c_int
 
         return clonefile(
-            ctypes.c_char_p(src.encode("utf-8")),
-            ctypes.c_char_p(dst.encode("utf-8")),
+            ctypes.c_char_p(os.fsencode(src)),
+            ctypes.c_char_p(os.fsencode(dst)),
             ctypes.c_int(0),
         )
 

dvc/tree/pool.py

@@ -11,7 +11,7 @@ def get_connection(conn_func, *args, **kwargs):
     conn = pool.get_connection()
     try:
         yield conn
-    except BaseException:
+    except Exception:
         conn.close()
         raise
     else:

dvc/version.py

@@ -6,7 +6,7 @@
 import os
 import subprocess
 
-_BASE_VERSION = "1.11.0"
+_BASE_VERSION = "1.11.8"
 
 
 def _generate_version(base_version):

scripts/build-requirements.txt

@@ -1,2 +1,3 @@
 PyInstaller==3.5
+numpy==1.19.3
 psutil

setup.py

@@ -71,7 +71,7 @@ install_requires = [
     "flufl.lock>=3.2,<4",
     "win-unicode-console>=0.5; sys_platform == 'win32'",
     "pywin32>=225; sys_platform == 'win32'",
-    "networkx>=2.1,<2.5",
+    "networkx>=2.1",
     "pydot>=1.2.4",
     "speedcopy>=2.0.1; python_version < '3.8' and sys_platform == 'win32'",
     "dataclasses==0.7; python_version < '3.7'",

tests/func/test_cli.py

@@ -141,14 +141,14 @@ class TestGCMultipleDvcRepos(TestDvc):
 
 class TestConfig(TestDvc):
     def test(self):
-        name = "param"
+        name = "section.option"
         value = "1"
 
         args = parse_args(["config", "-u", "--unset", name, value])
 
         self.assertIsInstance(args.func(args), CmdConfig)
         self.assertEqual(args.unset, True)
-        self.assertEqual(args.name, name)
+        self.assertEqual(args.name, (False, "section", "option"))
         self.assertEqual(args.value, value)
 
 

tests/func/test_config.py

@@ -145,3 +145,19 @@ def test_load_relative_paths(dvc, field, remote_url):
     assert cfg["remote"]["test"][field] == os.path.join(
         dvc_dir, "..", "file.txt"
     )
+
+
+def test_config_remote(tmp_dir, dvc, caplog):
+    (tmp_dir / ".dvc" / "config").write_text(
+        "['remote \"myremote\"']\n"
+        "  url = s3://bucket/path\n"
+        "  region = myregion\n"
+    )
+
+    caplog.clear()
+    assert main(["config", "remote.myremote.url"]) == 0
+    assert "s3://bucket/path" in caplog.text
+
+    caplog.clear()
+    assert main(["config", "remote.myremote.region"]) == 0
+    assert "myregion" in caplog.text

tests/func/test_repro_multistage.py

@@ -170,6 +170,19 @@ def test_non_existing_stage_name(tmp_dir, dvc, run_copy):
     assert main(["freeze", ":copy-file1-file3"]) != 0
 
 
+def test_repro_frozen(tmp_dir, dvc, run_copy):
+    (data_stage,) = tmp_dir.dvc_gen("data", "foo")
+    stage0 = run_copy("data", "stage0", name="copy-data-stage0")
+    run_copy("stage0", "stage1", name="copy-data-stage1")
+    run_copy("stage1", "stage2", name="copy-data-stage2")
+
+    dvc.freeze("copy-data-stage1")
+
+    tmp_dir.gen("data", "bar")
+    stages = dvc.reproduce()
+    assert stages == [data_stage, stage0]
+
+
 def test_downstream(tmp_dir, dvc):
     # The dependency graph should look like this:
     #

tests/unit/command/test_config.py

@@ -0,0 +1,11 @@
+import pytest
+
+from dvc.cli import DvcParserError, parse_args
+
+
+@pytest.mark.parametrize(
+    "name", ["way.too.long", "no_option", "remote.way.too.long"],
+)
+def test_config_bad_name(name):
+    with pytest.raises(DvcParserError):
+        parse_args(["config", name])

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):

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)

tests/unit/repo/test_reproduce.py

@@ -1,37 +1,5 @@
 import mock
 
-from dvc.repo.reproduce import _get_active_graph
-
-
-def test_get_active_graph(tmp_dir, dvc):
-    (pre_foo_stage,) = tmp_dir.dvc_gen({"pre-foo": "pre-foo"})
-    foo_stage = dvc.run(
-        single_stage=True, deps=["pre-foo"], outs=["foo"], cmd="echo foo > foo"
-    )
-    bar_stage = dvc.run(
-        single_stage=True, deps=["foo"], outs=["bar"], cmd="echo bar > bar"
-    )
-    baz_stage = dvc.run(
-        single_stage=True, deps=["foo"], outs=["baz"], cmd="echo baz > baz"
-    )
-
-    dvc.freeze("bar.dvc")
-
-    graph = dvc.graph
-    active_graph = _get_active_graph(graph)
-    assert active_graph.nodes == graph.nodes
-    assert set(active_graph.edges) == {
-        (foo_stage, pre_foo_stage),
-        (baz_stage, foo_stage),
-    }
-
-    dvc.freeze("baz.dvc")
-
-    graph = dvc.graph
-    active_graph = _get_active_graph(graph)
-    assert set(active_graph.nodes) == {bar_stage, baz_stage}
-    assert not active_graph.edges
-
 
 @mock.patch("dvc.repo.reproduce._reproduce_stage", returns=[])
 def test_number_reproduces(reproduce_stage_mock, tmp_dir, dvc):