diff --git a/core/dbt/task/run_operation.py b/core/dbt/task/run_operation.py
index 6ea5d91bd98..fe90649d1e0 100644
--- a/core/dbt/task/run_operation.py
+++ b/core/dbt/task/run_operation.py
@@ -1,6 +1,5 @@
 from dbt.logger import GLOBAL_LOGGER as logger
-
-from dbt.task.base import BaseTask
+from dbt.task.base import ConfiguredTask
 from dbt.adapters.factory import get_adapter
 from dbt.loader import GraphLoader
 
@@ -9,7 +8,7 @@
 import dbt.exceptions
 
 
-class RunOperationTask(BaseTask):
+class RunOperationTask(ConfiguredTask):
     def _get_macro_parts(self):
         macro_name = self.args.macro
         if '.' in macro_name:
@@ -22,19 +21,44 @@ def _get_macro_parts(self):
     def _get_kwargs(self):
         return dbt.utils.parse_cli_vars(self.args.args)
 
-    def run(self):
+    def _run_unsafe(self):
         manifest = GraphLoader.load_all(self.config)
         adapter = get_adapter(self.config)
 
         package_name, macro_name = self._get_macro_parts()
         macro_kwargs = self._get_kwargs()
 
-        res = adapter.execute_macro(
-            macro_name,
-            project=package_name,
-            kwargs=macro_kwargs,
-            manifest=manifest,
-            connection_name="macro_{}".format(macro_name)
-        )
+        with adapter.connection_named('macro_{}'.format(macro_name)):
+            adapter.clear_transaction()
+            res = adapter.execute_macro(
+                macro_name,
+                project=package_name,
+                kwargs=macro_kwargs,
+                manifest=manifest
+            )
 
         return res
+
+    def run(self):
+        try:
+            result = self._run_unsafe()
+        except dbt.exceptions.Exception as exc:
+            logger.error(
+                'Encountered an error while running operation: {}'
+                .format(exc)
+            )
+            logger.debug('', exc_info=True)
+            return False, None
+        except Exception as exc:
+            logger.error(
+                'Encountered an uncaught exception while running operation: {}'
+                .format(exc)
+            )
+            logger.debug('', exc_info=True)
+            return False, None
+        else:
+            return True, result
+
+    def interpret_results(self, results):
+        success, _ = results
+        return success
diff --git a/test/integration/044_run_operations_test/macros/happy_macros.sql b/test/integration/044_run_operations_test/macros/happy_macros.sql
new file mode 100644
index 00000000000..6170ebc7657
--- /dev/null
+++ b/test/integration/044_run_operations_test/macros/happy_macros.sql
@@ -0,0 +1,24 @@
+{% macro no_args() %}
+  {% if execute %}
+    {% call statement(auto_begin=True) %}
+      create table "{{ schema }}"."no_args" (id int);
+      commit;
+    {% endcall %}
+  {% endif %}
+{% endmacro %}
+
+
+{% macro table_name_args(table_name) %}
+  {% if execute %}
+    {% call statement(auto_begin=True) %}
+      create table "{{ schema }}"."{{ table_name }}" (id int);
+      commit;
+    {% endcall %}
+  {% endif %}
+{% endmacro %}
+
+{% macro vacuum(table_name) %}
+  {% call statement(auto_begin=false) %}
+    vacuum "{{ schema }}"."{{ table_name }}"
+  {% endcall %}
+{% endmacro %}
diff --git a/test/integration/044_run_operations_test/macros/sad_macros.sql b/test/integration/044_run_operations_test/macros/sad_macros.sql
new file mode 100644
index 00000000000..4f2c80bc40f
--- /dev/null
+++ b/test/integration/044_run_operations_test/macros/sad_macros.sql
@@ -0,0 +1,7 @@
+{% macro syntax_error() %}
+  {% if execute %}
+    {% call statement() %}
+        select NOPE NOT A VALID QUERY
+    {% endcall %}
+  {% endif %}
+{% endmacro %}
diff --git a/test/integration/044_run_operations_test/models/model.sql b/test/integration/044_run_operations_test/models/model.sql
new file mode 100644
index 00000000000..43258a71464
--- /dev/null
+++ b/test/integration/044_run_operations_test/models/model.sql
@@ -0,0 +1 @@
+select 1 as id
diff --git a/test/integration/044_run_operations_test/test_run_operations.py b/test/integration/044_run_operations_test/test_run_operations.py
new file mode 100644
index 00000000000..c66de6d8af5
--- /dev/null
+++ b/test/integration/044_run_operations_test/test_run_operations.py
@@ -0,0 +1,58 @@
+from test.integration.base import DBTIntegrationTest, use_profile
+import yaml
+
+
+class TestOperations(DBTIntegrationTest):
+    @property
+    def schema(self):
+        return "run_operations_044"
+
+    @property
+    def models(self):
+        return "test/integration/044_run_operations_test/models"
+
+    @property
+    def project_config(self):
+        return {
+            "macro-paths": ['test/integration/044_run_operations_test/macros'],
+        }
+
+    def run_operation(self, macro, expect_pass=True, extra_args=None, **kwargs):
+        args = ['run-operation']
+        if macro:
+            args.extend(('--macro', macro))
+        if kwargs:
+            args.extend(('--args', yaml.safe_dump(kwargs)))
+        if extra_args:
+            args.extend(extra_args)
+        return self.run_dbt(args, expect_pass=expect_pass)
+
+    @use_profile('postgres')
+    def test__postgres_macro_noargs(self):
+        self.run_operation('no_args')
+        self.assertTableDoesExist('no_args')
+
+    @use_profile('postgres')
+    def test__postgres_macro_args(self):
+        self.run_operation('table_name_args', table_name='my_fancy_table')
+        self.assertTableDoesExist('my_fancy_table')
+
+    @use_profile('postgres')
+    def test__postgres_macro_exception(self):
+        self.run_operation('syntax_error', False)
+
+    @use_profile('postgres')
+    def test__postgres_macro_missing(self):
+        self.run_operation('this_macro_does_not_exist', False)
+
+    @use_profile('postgres')
+    def test__postgres_cannot_connect(self):
+        self.run_operation('no_args',
+                           extra_args=['--target', 'noaccess'],
+                           expect_pass=False)
+
+    @use_profile('postgres')
+    def test__postgres_vacuum(self):
+        self.run_dbt(['run'])
+        # this should succeed
+        self.run_operation('vacuum', table_name='model')