Merge pull request #1792 from fishtown-analytics/fix/run-operation-fl…

…at-graph

Make run-operation a ManifestTask

core/dbt/task/rpc_server.py

@@ -18,6 +18,7 @@ def __init__(self, args, config, tasks=None):
         super(RPCServerTask, self).__init__(args, config)
         # compile locally
         self.manifest = self._compile_manifest()
+        self.manifest.build_flat_graph()
         self.task_manager = rpc.TaskManager()
         tasks = tasks or [RemoteCompileTask, RemoteRunTask]
         for cls in tasks:

core/dbt/task/run_operation.py

@@ -1,14 +1,13 @@
 from dbt.logger import GLOBAL_LOGGER as logger
-from dbt.task.base import ConfiguredTask
+from dbt.task.runnable import ManifestTask
 from dbt.adapters.factory import get_adapter
-from dbt.loader import GraphLoader
 
 import dbt
 import dbt.utils
 import dbt.exceptions
 
 
-class RunOperationTask(ConfiguredTask):
+class RunOperationTask(ManifestTask):
     def _get_macro_parts(self):
         macro_name = self.args.macro
         if '.' in macro_name:
@@ -21,8 +20,11 @@ def _get_macro_parts(self):
     def _get_kwargs(self):
         return dbt.utils.parse_cli_vars(self.args.args)
 
+    def compile_manifest(self):
+        # skip building a linker, but do make sure to build the flat graph
+        self.manifest.build_flat_graph()
+
     def _run_unsafe(self):
-        manifest = GraphLoader.load_all(self.config)
         adapter = get_adapter(self.config)
 
         package_name, macro_name = self._get_macro_parts()
@@ -34,12 +36,13 @@ def _run_unsafe(self):
                 macro_name,
                 project=package_name,
                 kwargs=macro_kwargs,
-                manifest=manifest
+                manifest=self.manifest
             )
 
         return res
 
     def run(self):
+        self._runtime_initialize()
         try:
             result = self._run_unsafe()
         except dbt.exceptions.Exception as exc:

test/integration/044_run_operations_test/macros/happy_macros.sql

@@ -42,3 +42,10 @@
   {% endset %}
   {% do run_query(query) %}
 {% endmacro %}
+
+
+{% macro log_graph() %}
+  {% for node in graph.nodes.values() %}
+    {{ log((node | string), info=True)}}
+  {% endfor %}
+{% endmacro %}

test/integration/044_run_operations_test/test_run_operations.py

@@ -64,3 +64,7 @@ def test__postgres_vacuum_ref(self):
     @use_profile('postgres')
     def test__postgres_select(self):
         self.run_operation('select_something', name='world')
+
+    @use_profile('postgres')
+    def test__postgres_access_graph(self):
+        self.run_operation('log_graph')