Merge pull request #1355 from fishtown-analytics/fix/out-of-order-exe…

…cution-on-model-select

Fix out of order execution on model select (#1354)

core/dbt/linker.py

@@ -174,6 +174,28 @@ def join(self):
         self.inner.join()
 
 
+def _subset_graph(graph, include_nodes):
+    """Create and return a new graph that is a shallow copy of graph but with
+    only the nodes in include_nodes. Transitive edges across removed nodes are
+    preserved as explicit new edges.
+    """
+    new_graph = nx.algorithms.transitive_closure(graph)
+
+    include_nodes = set(include_nodes)
+
+    for node in graph.nodes():
+        if node not in include_nodes:
+            new_graph.remove_node(node)
+
+    for node in include_nodes:
+        if node not in new_graph:
+            raise RuntimeError(
+                "Couldn't find model '{}' -- does it exist or is "
+                "it disabled?".format(node)
+            )
+    return new_graph
+
+
 class Linker(object):
     def __init__(self, data=None):
         if data is None:
@@ -209,23 +231,7 @@ def as_graph_queue(self, manifest, limit_to=None):
         else:
             graph_nodes = limit_to
 
-        new_graph = nx.DiGraph(self.graph)
-
-        to_remove = []
-        graph_nodes_lookup = set(graph_nodes)
-        for node in new_graph.nodes():
-            if node not in graph_nodes_lookup:
-                to_remove.append(node)
-
-        for node in to_remove:
-            new_graph.remove_node(node)
-
-        for node in graph_nodes:
-            if node not in new_graph:
-                raise RuntimeError(
-                    "Couldn't find model '{}' -- does it exist or is "
-                    "it disabled?".format(node)
-                )
+        new_graph = _subset_graph(self.graph, graph_nodes)
         return GraphQueue(new_graph, manifest)
 
     def get_dependent_nodes(self, node):

test/integration/007_graph_selection_tests/models/users_rollup_dependency.sql

@@ -0,0 +1,5 @@
+{{
+	config(materialized='table')
+}}
+
+select * from {{ ref('users_rollup') }}

test/integration/007_graph_selection_tests/test_graph_selection.py

@@ -60,7 +60,7 @@ def test__postgres__tags_and_children(self):
         self.run_sql_file("test/integration/007_graph_selection_tests/seed.sql")
 
         results = self.run_dbt(['run', '--models', 'tag:base+'])
-        self.assertEqual(len(results), 3)
+        self.assertEqual(len(results), 4)
 
         created_models = self.get_models_in_schema()
         self.assertFalse('base_users' in created_models)
@@ -89,7 +89,7 @@ def test__postgres__specific_model_and_children(self):
         self.run_sql_file("test/integration/007_graph_selection_tests/seed.sql")
 
         results = self.run_dbt(['run', '--models', 'users+'])
-        self.assertEqual(len(results),  3)
+        self.assertEqual(len(results),  4)
 
         self.assertTablesEqual("seed", "users")
         self.assertTablesEqual("summary_expected", "users_rollup")
@@ -104,7 +104,7 @@ def test__snowflake__specific_model_and_children(self):
         self.run_sql_file("test/integration/007_graph_selection_tests/seed.sql")
 
         results = self.run_dbt(['run', '--models', 'users+'])
-        self.assertEqual(len(results),  3)
+        self.assertEqual(len(results),  4)
 
         self.assertManyTablesEqual(
             ["SEED", "USERS"],
@@ -194,7 +194,7 @@ def test__postgres__locally_qualified_name(self):
     def test__postgres__childrens_parents(self):
         self.run_sql_file("test/integration/007_graph_selection_tests/seed.sql")
         results = self.run_dbt(['run', '--models', '@base_users'])
-        self.assertEqual(len(results), 3)
+        self.assertEqual(len(results), 4)
 
         created_models = self.get_models_in_schema()
         self.assertIn('users_rollup', created_models)
@@ -207,12 +207,33 @@ def test__postgres__childrens_parents(self):
     def test__postgres__more_childrens_parents(self):
         self.run_sql_file("test/integration/007_graph_selection_tests/seed.sql")
         results = self.run_dbt(['run', '--models', '@users'])
-        # base_users, emails, users_rollup, but not users (ephemeral)
-        self.assertEqual(len(results), 3)
+        # base_users, emails, users_rollup, users_rollup_dependency, but not users (ephemeral)
+        self.assertEqual(len(results), 4)
 
         created_models = self.get_models_in_schema()
         self.assertIn('users_rollup', created_models)
         self.assertIn('users', created_models)
         self.assertIn('emails_alt', created_models)
         self.assertNotIn('subdir', created_models)
         self.assertNotIn('nested_users', created_models)
+
+    @attr(type='snowflake')
+    def test__snowflake__skip_intermediate(self):
+        self.run_sql_file("test/integration/007_graph_selection_tests/seed.sql")
+        results = self.run_dbt(['run', '--models', '@users'])
+        # base_users, emails, users_rollup, users_rollup_dependency
+        self.assertEqual(len(results), 4)
+
+        # now re-run, skipping users_rollup
+        results = self.run_dbt(['run', '--models', '@users', '--exclude', 'users_rollup'])
+        self.assertEqual(len(results), 3)
+
+        # make sure that users_rollup_dependency and users don't interleave
+        users = [r for r in results if r.node.name == 'users'][0]
+        dep = [r for r in results if r.node.name == 'users_rollup_dependency'][0]
+        user_last_end = users.timing[1]['completed_at']
+        dep_first_start = dep.timing[0]['started_at']
+        self.assertTrue(
+            user_last_end < dep_first_start,
+            'dependency started before its transitive parent ({} > {})'.format(user_last_end, dep_first_start)
+        )

test/integration/007_graph_selection_tests/test_schema_test_graph_selection.py

@@ -26,7 +26,7 @@ def run_schema_and_assert(self, include, exclude, expected_tests):
         self.run_sql_file("test/integration/007_graph_selection_tests/seed.sql")
         self.run_dbt(["deps"])
         results = self.run_dbt(['run', '--exclude', 'never_selected'])
-        self.assertEqual(len(results), 8)
+        self.assertEqual(len(results), 9)
 
         args = FakeArgs()
         args.models = include

test/integration/007_graph_selection_tests/test_tag_selection.py

@@ -42,7 +42,7 @@ def test__postgres__select_tag_and_children(self):
         self.run_sql_file("test/integration/007_graph_selection_tests/seed.sql")
 
         results = self.run_dbt(['run', '--models', '+tag:specified_in_project+'])
-        self.assertEqual(len(results), 2)
+        self.assertEqual(len(results), 3)
 
         models_run = [r.node['name'] for r in results]
         self.assertTrue('users' in models_run)
@@ -69,8 +69,10 @@ def test__postgres__select_tag_in_model_with_project_Config(self):
         self.run_sql_file("test/integration/007_graph_selection_tests/seed.sql")
 
         results = self.run_dbt(['run', '--models', '@tag:users'])
-        self.assertEqual(len(results), 3)
+        self.assertEqual(len(results), 4)
 
         models_run = set(r.node['name'] for r in results)
-        self.assertEqual({'users', 'users_rollup', 'emails_alt'}, models_run)
-
+        self.assertEqual(
+            {'users', 'users_rollup', 'emails_alt', 'users_rollup_dependency'},
+            models_run
+        )

test/integration/base.py

@@ -439,7 +439,7 @@ def run_dbt(self, args=None, expect_pass=True, strict=True):
             args = ["run"]
 
         if strict:
-            args = ["--single-threaded", "--strict"] + args
+            args = ["--strict"] + args
         args.append('--log-cache-events')
         logger.info("Invoking dbt with {}".format(args))