Add extra graph edges for build only (dbt-labs#4143)

* Resolve extra graph edges for build only

* Fix flake8

* Change test to reflect functional change

* Rename method + args. Add changelog entry

CHANGELOG.md

@@ -96,6 +96,7 @@ Contributors:
 - Added support for tests on databases that lack real boolean types. ([#4084](https://github.com/dbt-labs/dbt-core/issues/4084))
 - Scrub secrets coming from `CommandError`s so they don't get exposed in logs. ([#4138](https://github.com/dbt-labs/dbt-core/pull/4138))
 - Syntax fix in `alter_relation_add_remove_columns` if only removing columns in `on_schema_change: sync_all_columns` ([#4147](https://github.com/dbt-labs/dbt-core/issues/4147))
+- Add downstream test edges for `build` task _only_. Restore previous graph construction, compilation performance, and node selection behavior (`test+`) for all other tasks ([#4135](https://github.com/dbt-labs/dbt-core/issues/4135), [#4143](https://github.com/dbt-labs/dbt-core/pull/4143))
 
 Contributors:
 - [@ljhopkins2](https://github.com/ljhopkins2) ([#4077](https://github.com/dbt-labs/dbt-core/pull/4077))

core/dbt/compilation.py

@@ -418,7 +418,7 @@ def link_node(
             else:
                 dependency_not_found(node, dependency)
 
-    def link_graph(self, linker: Linker, manifest: Manifest):
+    def link_graph(self, linker: Linker, manifest: Manifest, add_test_edges: bool = False):
         for source in manifest.sources.values():
             linker.add_node(source.unique_id)
         for node in manifest.nodes.values():
@@ -431,11 +431,11 @@ def link_graph(self, linker: Linker, manifest: Manifest):
         if cycle:
             raise RuntimeError("Found a cycle: {}".format(cycle))
 
-        manifest.build_parent_and_child_maps()
+        if add_test_edges:
+            manifest.build_parent_and_child_maps()
+            self.add_test_edges(linker, manifest)
 
-        self.resolve_graph(linker, manifest)
-
-    def resolve_graph(self, linker: Linker, manifest: Manifest) -> None:
+    def add_test_edges(self, linker: Linker, manifest: Manifest) -> None:
         """ This method adds additional edges to the DAG. For a given non-test
         executable node, add an edge from an upstream test to the given node if
         the set of nodes the test depends on is a proper/strict subset of the
@@ -501,11 +501,11 @@ def resolve_graph(self, linker: Linker, manifest: Manifest) -> None:
                             node_id
                         )
 
-    def compile(self, manifest: Manifest, write=True) -> Graph:
+    def compile(self, manifest: Manifest, write=True, add_test_edges=False) -> Graph:
         self.initialize()
         linker = Linker()
 
-        self.link_graph(linker, manifest)
+        self.link_graph(linker, manifest, add_test_edges)
 
         stats = _generate_stats(manifest)
 

core/dbt/task/build.py

@@ -3,6 +3,7 @@
 from .seed import SeedRunner as seed_runner
 from .test import TestRunner as test_runner
 
+from dbt.adapters.factory import get_adapter
 from dbt.contracts.results import NodeStatus
 from dbt.exceptions import InternalException
 from dbt.graph import ResourceTypeSelector
@@ -64,3 +65,12 @@ def get_node_selector(self) -> ResourceTypeSelector:
 
     def get_runner_type(self, node):
         return self.RUNNER_MAP.get(node.resource_type)
+
+    def compile_manifest(self):
+        if self.manifest is None:
+            raise InternalException(
+                'compile_manifest called before manifest was loaded'
+            )
+        adapter = get_adapter(self.config)
+        compiler = adapter.get_compiler()
+        self.graph = compiler.compile(self.manifest, add_test_edges=True)

test/integration/062_defer_state_test/test_run_results_state.py

@@ -184,8 +184,8 @@ def test_postgres_build_run_results_state(self):
         assert nodes == {'table_model', 'unique_view_model_id'}
 
         results = self.run_dbt(['ls', '--select', 'result:fail+', '--state', './state'])
-        assert len(results) == 2
-        assert set(results) == {'test.table_model', 'test.unique_view_model_id'}
+        assert len(results) == 1
+        assert set(results) == {'test.unique_view_model_id'}
 
         # change the unique test severity from error to warn and reuse the same view_model.sql changes above
         f = open('models/schema.yml', 'r')
@@ -212,8 +212,8 @@ def test_postgres_build_run_results_state(self):
         assert nodes == {'table_model', 'unique_view_model_id'}
 
         results = self.run_dbt(['ls', '--select', 'result:warn+', '--state', './state'])
-        assert len(results) == 2
-        assert set(results) == {'test.table_model', 'test.unique_view_model_id'}
+        assert len(results) == 1
+        assert set(results) == {'test.unique_view_model_id'}
 
     @use_profile('postgres')
     def test_postgres_run_run_results_state(self):