Merge pull request #2224 from Raalsky/fail-fast

Add --fail-fast argument for dbt run and dbt test

CHANGELOG.md

@@ -1,4 +1,6 @@
 ## dbt next (release TBD)
+### Features
+- Added --fail-fast argument for dbt run and dbt test to fail on first test failure or runtime error. ([#1649](https://github.com/fishtown-analytics/dbt/issues/1649))
 
 ### Features
 - Support for appending query comments to SQL queries. ([#2138](https://github.com/fishtown-analytics/dbt/issues/2138))
@@ -7,9 +9,11 @@
 ### Fixes
 - When a jinja value is undefined, give a helpful error instead of failing with cryptic "cannot pickle ParserMacroCapture" errors ([#2110](https://github.com/fishtown-analytics/dbt/issues/2110), [#2184](https://github.com/fishtown-analytics/dbt/pull/2184))
 
-Contributers:
+Contributors:
+ - [@raalsky](https://github.com/Raalsky) ([#2224](https://github.com/fishtown-analytics/dbt/pull/2224))
  - [@ilkinulas](https://github.com/ilkinulas) [#2199](https://github.com/fishtown-analytics/dbt/pull/2199)
 
+
 ## dbt 0.16.0 (Release date TBD)
 
 ### Fixes

core/dbt/exceptions.py

@@ -281,6 +281,19 @@ def __init__(self, message, project=None, result_type='invalid_project'):
         self.result_type = result_type
 
 
+class FailFastException(RuntimeException):
+    CODE = 10013
+    MESSAGE = 'FailFast Error'
+
+    def __init__(self, message, result=None, node=None):
+        super().__init__(msg=message, node=node)
+        self.result = result
+
+    @property
+    def type(self):
+        return 'FailFast'
+
+
 class DbtProjectError(DbtConfigError):
     pass
 

core/dbt/linker.py

@@ -48,6 +48,8 @@ def __init__(self, graph, manifest):
         self._scores = self._calculate_scores()
         # populate the initial queue
         self._find_new_additions()
+        # awaits after task end
+        self.some_task_done = threading.Condition(self.lock)
 
     def _include_in_cost(self, node_id):
         node = self.manifest.expect(node_id)
@@ -153,6 +155,7 @@ def mark_done(self, node_id):
             self.graph.remove_node(node_id)
             self._find_new_additions()
             self.inner.task_done()
+            self.some_task_done.notify_all()
 
     def _mark_in_progress(self, node_id):
         """Mark the node as 'in progress'.
@@ -171,6 +174,11 @@ def join(self):
         """
         self.inner.join()
 
+    def wait_until_something_was_done(self):
+        with self.lock:
+            self.some_task_done.wait()
+            return self.inner.unfinished_tasks
+
 
 class Linker:
     def __init__(self, data=None):

core/dbt/main.py

@@ -398,6 +398,14 @@ def _build_run_subparser(subparsers, base_subparser):
         help='''
         Compile SQL and execute against the current target database.
         ''')
+    run_sub.add_argument(
+        '-x',
+        '--fail-fast',
+        action='store_true',
+        help='''
+            Stop execution upon a first failure.
+            '''
+    )
     run_sub.set_defaults(cls=run_task.RunTask, which='run', rpc_method='run')
     return run_sub
 
@@ -555,6 +563,14 @@ def _build_test_subparser(subparsers, base_subparser):
         Run constraint validations from schema.yml files
         '''
     )
+    sub.add_argument(
+        '-x',
+        '--fail-fast',
+        action='store_true',
+        help='''
+        Stop execution upon a first test failure.
+        '''
+    )
 
     sub.set_defaults(cls=test_task.TestTask, which='test', rpc_method='test')
     return sub

core/dbt/task/runnable.py

@@ -25,7 +25,10 @@
 from dbt.contracts.graph.manifest import Manifest
 from dbt.contracts.results import ExecutionResult
 from dbt.exceptions import (
-    InternalException, NotImplementedException, RuntimeException
+    InternalException,
+    NotImplementedException,
+    RuntimeException,
+    FailFastException
 )
 from dbt.linker import Linker, GraphQueue
 from dbt.perf_utils import get_full_manifest
@@ -167,11 +170,20 @@ def call_runner(self, runner):
                 with finishctx, DbtModelState(status):
                     logger.debug('Finished running node {}'.format(
                         runner.node.unique_id))
-        if result.error is not None and self.raise_on_first_error():
+
+        fail_fast = getattr(self.config.args, 'fail_fast', False)
+
+        if (result.fail is not None or result.error is not None) and fail_fast:
+            self._raise_next_tick = FailFastException(
+                message='Falling early due to test failure or runtime error',
+                result=result,
+                node=getattr(result, 'node', None)
+            )
+        elif result.error is not None and self.raise_on_first_error():
             # if we raise inside a thread, it'll just get silently swallowed.
             # stash the error message we want here, and it will check the
             # next 'tick' - should be soon since our thread is about to finish!
-            self._raise_next_tick = result.error
+            self._raise_next_tick = RuntimeException(result.error)
 
         return result
 
@@ -191,7 +203,7 @@ def _submit(self, pool, args, callback):
 
     def _raise_set_error(self):
         if self._raise_next_tick is not None:
-            raise RuntimeException(self._raise_next_tick)
+            raise self._raise_next_tick
 
     def run_queue(self, pool):
         """Given a pool, submit jobs from the queue to the pool.
@@ -226,7 +238,15 @@ def callback(result):
             self._submit(pool, args, callback)
 
         # block on completion
-        self.job_queue.join()
+        if getattr(self.config.args, 'fail_fast', False):
+            # checkout for an errors after task completion in case of
+            # fast failure
+            while self.job_queue.wait_until_something_was_done():
+                self._raise_set_error()
+        else:
+            # wait until every task will be complete
+            self.job_queue.join()
+
         # if an error got set during join(), raise it.
         self._raise_set_error()
 
@@ -255,6 +275,34 @@ def _handle_result(self, result):
                 cause = None
             self._mark_dependent_errors(node.unique_id, result, cause)
 
+    def _cancel_connections(self, pool):
+        """Given a pool, cancel all adapter connections and wait until all
+        runners gentle terminates.
+        """
+        pool.close()
+        pool.terminate()
+
+        adapter = get_adapter(self.config)
+
+        if not adapter.is_cancelable():
+            msg = ("The {} adapter does not support query "
+                   "cancellation. Some queries may still be "
+                   "running!".format(adapter.type()))
+
+            yellow = dbt.ui.printer.COLOR_FG_YELLOW
+            dbt.ui.printer.print_timestamped_line(msg, yellow)
+            raise
+
+        for conn_name in adapter.cancel_open_connections():
+            if self.manifest is not None:
+                node = self.manifest.nodes.get(conn_name)
+                if node is not None and node.is_ephemeral_model:
+                    continue
+            # if we don't have a manifest/don't have a node, print anyway.
+            dbt.ui.printer.print_cancel_line(conn_name)
+
+        pool.join()
+
     def execute_nodes(self):
         num_threads = self.config.threads
         target_name = self.config.target_name
@@ -270,34 +318,15 @@ def execute_nodes(self):
         try:
             self.run_queue(pool)
 
-        except KeyboardInterrupt:
-            pool.close()
-            pool.terminate()
-
-            adapter = get_adapter(self.config)
-
-            if not adapter.is_cancelable():
-                msg = ("The {} adapter does not support query "
-                       "cancellation. Some queries may still be "
-                       "running!".format(adapter.type()))
-
-                yellow = dbt.ui.printer.COLOR_FG_YELLOW
-                dbt.ui.printer.print_timestamped_line(msg, yellow)
-                raise
-
-            for conn_name in adapter.cancel_open_connections():
-                if self.manifest is not None:
-                    node = self.manifest.nodes.get(conn_name)
-                    if node is not None and node.is_ephemeral_model:
-                        continue
-                # if we don't have a manifest/don't have a node, print anyway.
-                dbt.ui.printer.print_cancel_line(conn_name)
-
-            pool.join()
+        except FailFastException as failure:
+            self._cancel_connections(pool)
+            dbt.ui.printer.print_run_result_error(failure.result)
+            raise
 
+        except KeyboardInterrupt:
+            self._cancel_connections(pool)
             dbt.ui.printer.print_run_end_messages(self.node_results,
-                                                  early_exit=True)
-
+                                                  keyboard_interrupt=True)
             raise
 
         pool.close()

core/dbt/ui/printer.py

@@ -348,11 +348,11 @@ def print_skip_caused_by_error(
 
 
 def print_end_of_run_summary(
-    num_errors: int, num_warnings: int, early_exit: bool = False
+    num_errors: int, num_warnings: int, keyboard_interrupt: bool = False
 ) -> None:
     error_plural = dbt.utils.pluralize(num_errors, 'error')
     warn_plural = dbt.utils.pluralize(num_warnings, 'warning')
-    if early_exit:
+    if keyboard_interrupt:
         message = yellow('Exited because of keyboard interrupt.')
     elif num_errors > 0:
         message = red("Completed with {} and {}:".format(
@@ -367,11 +367,13 @@ def print_end_of_run_summary(
     logger.info('{}'.format(message))
 
 
-def print_run_end_messages(results, early_exit: bool = False) -> None:
+def print_run_end_messages(results, keyboard_interrupt: bool = False) -> None:
     errors = [r for r in results if r.error is not None or r.fail]
     warnings = [r for r in results if r.warn]
     with DbtStatusMessage(), InvocationProcessor():
-        print_end_of_run_summary(len(errors), len(warnings), early_exit)
+        print_end_of_run_summary(len(errors),
+                                 len(warnings),
+                                 keyboard_interrupt)
 
         for error in errors:
             print_run_result_error(error, is_warning=False)

test/integration/058_fail_fast/models/one.sql

@@ -0,0 +1 @@
+select 1 /failed

test/integration/058_fail_fast/models/two.sql

@@ -0,0 +1 @@
+select 1 /failed

test/integration/058_fail_fast/test_fail_fast_run.py

@@ -0,0 +1,49 @@
+from test.integration.base import DBTIntegrationTest, use_profile
+from dbt.exceptions import FailFastException
+
+
+class TestFastFailingDuringRun(DBTIntegrationTest):
+    @property
+    def schema(self):
+        return "fail_fast_058"
+
+    @property
+    def project_config(self):
+        return {
+            "on-run-start": "create table if not exists {{ target.schema }}.audit (model text)",
+            'models': {
+                'test': {
+                    'pre-hook': [
+                      {
+                          # we depend on non-deterministic nature of tasks execution
+                          # there is possibility to run next task in-between
+                          # first task failure and adapter connections cancellations
+                          # if you encounter any problems with these tests please report
+                          # the sleep command with random time minimize the risk
+                          'sql': "select pg_sleep(random())",
+                          'transaction': False
+                      },
+                      {
+                          'sql': "insert into {{ target.schema }}.audit values ('{{ this }}')",
+                          'transaction': False
+                      }
+                    ],
+                }
+            }
+        }
+
+    @property
+    def models(self):
+        return "models"
+
+    def check_audit_table(self, count=1):
+        query = "select * from {schema}.audit".format(schema=self.unique_schema())
+
+        vals = self.run_sql(query, fetch='all')
+        self.assertFalse(len(vals) == count, 'Execution was not stopped before run end')
+
+    @use_profile('postgres')
+    def test_postgres_fail_fast_run(self):
+        with self.assertRaises(FailFastException):
+            self.run_dbt(['run', '--threads', '1', '--fail-fast'])
+            self.check_audit_table()