Fix for "table dropped by concurrent query" on Redshift

Makefile

@@ -15,7 +15,7 @@ test-unit:
 
 test-integration:
 	@echo "Integration test run starting..."
-	@time docker-compose run test tox -e integration-postgres-py27,integration-postgres-py36,integration-snowflake-py27,integration-snowflake-py36,integration-bigquery-py27,integration-bigquery-py36
+	@time docker-compose run test tox -e integration-postgres-py27,integration-postgres-py36,integration-redshift-py27,integration-redshift-py36,integration-snowflake-py27,integration-snowflake-py36,integration-bigquery-py27,integration-bigquery-py36
 
 test-quick:
 	@echo "Integration test run starting..."

dbt/adapters/default/impl.py

@@ -162,7 +162,8 @@ def drop_relation(cls, profile, project_cfg, relation, model_name=None):
 
         sql = 'drop {} if exists {} cascade'.format(relation.type, relation)
 
-        connection, cursor = cls.add_query(profile, sql, model_name)
+        connection, cursor = cls.add_query(profile, sql, model_name,
+                                           auto_begin=False)
 
     @classmethod
     def truncate(cls, profile, project_cfg, schema, table, model_name=None):

dbt/include/global_project/macros/materializations/table/table.sql

@@ -1,6 +1,7 @@
 {% materialization table, default %}
   {%- set identifier = model['alias'] -%}
   {%- set tmp_identifier = identifier + '__dbt_tmp' -%}
+  {%- set backup_identifier = identifier + '__dbt_backup' -%}
   {%- set non_destructive_mode = (flags.NON_DESTRUCTIVE == True) -%}
 
   {%- set existing_relations = adapter.list_relations(schema=schema) -%}
@@ -10,15 +11,18 @@
                                                 schema=schema, type='table') -%}
   {%- set intermediate_relation = api.Relation.create(identifier=tmp_identifier,
                                                       schema=schema, type='table') -%}
+  {%- set backup_relation = api.Relation.create(identifier=backup_identifier,
+                                                schema=schema, type='table') -%}
   {%- set exists_as_table = (old_relation is not none and old_relation.is_table) -%}
   {%- set exists_as_view = (old_relation is not none and old_relation.is_view) -%}
   {%- set create_as_temporary = (exists_as_table and non_destructive_mode) -%}
 
 
-  -- drop the temp relation if it exists for some reason
+  -- drop the temp relations if they exists for some reason
   {{ adapter.drop_relation(intermediate_relation) }}
+  {{ adapter.drop_relation(backup_relation) }}
 
-  -- setup: if the target relation already exists, truncate or drop it
+  -- setup: if the target relation already exists, truncate or drop it (if it's a view)
   {% if non_destructive_mode -%}
     {% if exists_as_table -%}
       {{ adapter.truncate_relation(old_relation) }}
@@ -60,12 +64,17 @@
   {% if non_destructive_mode -%}
     -- noop
   {%- else -%}
-    {{ drop_relation_if_exists(old_relation) }}
+    {% if exists_as_table %}
+      -- move the existing table out of the way
+      {{ adapter.rename_relation(target_relation, backup_relation) }}
+    {% endif %}
+
     {{ adapter.rename_relation(intermediate_relation, target_relation) }}
   {%- endif %}
 
   -- `COMMIT` happens here
   {{ adapter.commit() }}
+  {{ drop_relation_if_exists(backup_relation) }}
 
   {{ run_hooks(post_hooks, inside_transaction=False) }}
 {% endmaterialization %}

dbt/include/global_project/macros/materializations/view/view.sql

@@ -2,6 +2,7 @@
 
   {%- set identifier = model['alias'] -%}
   {%- set tmp_identifier = identifier + '__dbt_tmp' -%}
+  {%- set backup_identifier = identifier + '__dbt_backup' -%}
   {%- set non_destructive_mode = (flags.NON_DESTRUCTIVE == True) -%}
 
   {%- set existing_relations = adapter.list_relations(schema=schema) -%}
@@ -11,14 +12,19 @@
                                                 type='view') -%}
   {%- set intermediate_relation = api.Relation.create(identifier=tmp_identifier,
                                                       schema=schema, type='view') -%}
+  {%- set backup_relation = api.Relation.create(identifier=backup_identifier,
+                                                schema=schema, type='view') -%}
 
   {%- set exists_as_view = (old_relation is not none and old_relation.is_view) -%}
 
   {%- set has_transactional_hooks = (hooks | selectattr('transaction', 'equalto', True) | list | length) > 0 %}
   {%- set should_ignore = non_destructive_mode and exists_as_view %}
 
   {{ run_hooks(pre_hooks, inside_transaction=False) }}
+
+  -- drop the temp relations if they exists for some reason
   {{ adapter.drop_relation(intermediate_relation) }}
+  {{ adapter.drop_relation(backup_relation) }}
 
   -- `BEGIN` happens here:
   {{ run_hooks(pre_hooks, inside_transaction=True) }}
@@ -45,7 +51,10 @@
 
   -- cleanup
   {% if not should_ignore -%}
-    {{ drop_relation_if_exists(old_relation) }}
+    -- move the existing view out of the way
+    {% if exists_as_view %}
+      {{ adapter.rename_relation(target_relation, backup_relation) }}
+    {% endif %}
     {{ adapter.rename_relation(intermediate_relation, target_relation) }}
   {%- endif %}
 
@@ -55,6 +64,7 @@
   #}
   {% if has_transactional_hooks or not should_ignore %}
       {{ adapter.commit() }}
+      {{ drop_relation_if_exists(backup_relation) }}
   {% endif %}
 
   {{ run_hooks(post_hooks, inside_transaction=False) }}

test.env.sample

@@ -15,3 +15,9 @@ BIGQUERY_AUTH_URI=
 BIGQUERY_TOKEN_URI=
 BIGQUERY_AUTH_PROVIDER_X509_CERT_URL=
 BIGQUERY_CLIENT_X509_CERT_URL=
+
+REDSHIFT_TEST_HOST=
+REDSHIFT_TEST_USER=
+REDSHIFT_TEST_PASSWORD=
+REDSHIFT_TEST_PORT=
+REDSHIFT_TEST_DBNAME=

test/integration/032_concurrent_transaction_test/README.md

@@ -0,0 +1,33 @@
+
+This test warrants some explanation. In dbt <=0.10.1, Redshift table and view materializations suffered from issues around concurrent transactions. In order to reliably reproduce this error, a query needs to select from a dbt model as the table is being rebuilt. Critically, this concurrent select needs to query the table during the drop/swap portition of the materialization. This looks like:
+
+```sql
+begin;
+create table as (...);
+drop table old_table cascade;
+// <---- The concurrent query needs to be running here!
+alter table new_table rename to old_table;
+commit;
+```
+
+In order to reliably reproduce this failure, the model shown above needs to block for a long time between the `drop` and `alter` statements. We can't just stick a sleep() call in there, as this code is defined in the materialization. Instead, we can reliably reproduce the failure by:
+
+1) creating a view that depends on this model
+2) issuing a long-running query on the view before `dbt run` is invoked
+3) issuing _another_ long-running query against the original model
+
+Since long-running query (step 2) is selecting from the view, Redshift blocks on the `drop ... cascade`, of the materialization, which causes the query from step 3 time to overlap with the critical section of the materialization between the `drop` and `alter` statements.
+
+In dbt v0.10.1, this integration test results in:
+
+```
+======================================================================
+FAIL: test__redshift__concurrent_transaction (test_concurrent_transaction.TestConcurrentTransaction)
+----------------------------------------------------------------------
+Traceback (most recent call last):
+  File "/usr/src/app/test/integration/032_concurrent_transaction_test/test_concurrent_transaction.py", line 84, in test__redshift__concurrent_transaction
+    self.assertEqual(self.query_state['model_1'], 'good')
+AssertionError: 'error: table 3379442 dropped by concurrent transaction\n' != 'good'
+- error: table 3379442 dropped by concurrent transaction
++ good
+```

test/integration/032_concurrent_transaction_test/macros/udfs.sql

@@ -0,0 +1,13 @@
+
+{% macro create_udfs() %}
+
+CREATE OR REPLACE FUNCTION {{ target.schema }}.f_sleep (x float)
+RETURNS bool IMMUTABLE
+AS
+$$
+  from time import sleep
+  sleep(x)
+  return True
+$$ LANGUAGE plpythonu;
+
+{% endmacro %}

test/integration/032_concurrent_transaction_test/models-incremental/model_1.sql

@@ -0,0 +1,5 @@
+
+{{ config(materialized='incremental', sql_where=True, unique_key='id') }}
+
+-- incremental model
+select 1 as id

test/integration/032_concurrent_transaction_test/models-incremental/view_model.sql

@@ -0,0 +1,3 @@
+
+
+select * from {{ ref('model_1') }}

test/integration/032_concurrent_transaction_test/models-table/model_1.sql

@@ -0,0 +1,5 @@
+
+{{ config(materialized='table') }}
+
+-- table model
+select 1 as id

test/integration/032_concurrent_transaction_test/models-table/view_model.sql

@@ -0,0 +1,3 @@
+
+
+select * from {{ ref('model_1') }}

test/integration/032_concurrent_transaction_test/models-view/model_1.sql

@@ -0,0 +1,5 @@
+
+{{ config(materialized='view') }}
+
+-- view model
+select 1 as id

test/integration/032_concurrent_transaction_test/models-view/view_model.sql

@@ -0,0 +1,3 @@
+
+
+select * from {{ ref('model_1') }}

test/integration/032_concurrent_transaction_test/test_concurrent_transaction.py

@@ -0,0 +1,111 @@
+from nose.plugins.attrib import attr
+from test.integration.base import DBTIntegrationTest
+import threading
+
+class BaseTestConcurrentTransaction(DBTIntegrationTest):
+
+    def reset(self):
+        self.query_state = {
+            'view_model': 'wait',
+            'model_1': 'wait',
+        }
+
+    @property
+    def schema(self):
+        return "concurrent_transaction_032"
+
+    @property
+    def project_config(self):
+        return {
+            "macro-paths": ["test/integration/032_concurrent_transaction_test/macros"],
+            "on-run-start": [
+                "{{ create_udfs() }}",
+            ],
+        }
+
+    def run_select_and_check(self, rel, sql):
+        try:
+            res = self.run_sql(sql, fetch='one')
+
+            # The result is the output of f_sleep(), which is True
+            if res[0] == True:
+                self.query_state[rel] = 'good'
+            else:
+                self.query_state[rel] = 'bad'
+
+        except Exception as e:
+            if 'concurrent transaction' in str(e):
+                self.query_state[rel] = 'error: {}'.format(e)
+            else:
+                self.query_state[rel] = 'error: {}'.format(e)
+
+    def async_select(self, rel, sleep=10):
+        # Run the select statement in a thread. When the query returns, the global
+        # query_state will be update with a state of good/bad/error, and the associated
+        # error will be reported if one was raised.
+
+        schema = self.unique_schema()
+        query = '''
+        -- async_select: {rel}
+        select {schema}.f_sleep({sleep}) from {schema}.{rel}
+        '''.format(
+                schema=schema,
+                sleep=sleep,
+                rel=rel)
+
+        thread = threading.Thread(target=lambda: self.run_select_and_check(rel, query))
+        thread.start()
+        return thread
+
+    def run_test(self):
+        self.use_profile("redshift")
+
+        # First run the project to make sure the models exist
+        self.run_dbt(args=['run'])
+
+        # Execute long-running queries in threads
+        t1 = self.async_select('view_model', 10)
+        t2 = self.async_select('model_1', 5)
+
+        # While the queries are executing, re-run the project
+        res = self.run_dbt(args=['run', '--threads', '8'])
+
+        # Finally, wait for these threads to finish
+        t1.join()
+        t2.join()
+
+        self.assertTrue(len(res) > 0)
+
+        # If the query succeeded, the global query_state should be 'good'
+        self.assertEqual(self.query_state['view_model'], 'good')
+        self.assertEqual(self.query_state['model_1'], 'good')
+
+class TableTestConcurrentTransaction(BaseTestConcurrentTransaction):
+    @property
+    def models(self):
+        return "test/integration/032_concurrent_transaction_test/models-table"
+
+    @attr(type="redshift")
+    def test__redshift__concurrent_transaction_table(self):
+        self.reset()
+        self.run_test()
+
+class ViewTestConcurrentTransaction(BaseTestConcurrentTransaction):
+    @property
+    def models(self):
+        return "test/integration/032_concurrent_transaction_test/models-view"
+
+    @attr(type="redshift")
+    def test__redshift__concurrent_transaction_view(self):
+        self.reset()
+        self.run_test()
+
+class IncrementalTestConcurrentTransaction(BaseTestConcurrentTransaction):
+    @property
+    def models(self):
+        return "test/integration/032_concurrent_transaction_test/models-incremental"
+
+    @attr(type="redshift")
+    def test__redshift__concurrent_transaction_incremental(self):
+        self.reset()
+        self.run_test()

test/integration/base.py

@@ -64,6 +64,28 @@ def postgres_profile(self):
             }
         }
 
+    def redshift_profile(self):
+        return {
+            'config': {
+                'send_anonymous_usage_stats': False
+            },
+            'test': {
+                'outputs': {
+                    'default2': {
+                        'type': 'redshift',
+                        'threads': 1,
+                        'host': os.getenv('REDSHIFT_TEST_HOST'),
+                        'port': os.getenv('REDSHIFT_TEST_PORT'),
+                        'user': os.getenv('REDSHIFT_TEST_USER'),
+                        'pass': os.getenv('REDSHIFT_TEST_PASS'),
+                        'dbname': os.getenv('REDSHIFT_TEST_DBNAME'),
+                        'schema': self.unique_schema()
+                    }
+                },
+                'target': 'default2'
+            }
+        }
+
     def snowflake_profile(self):
         return {
             'config': {
@@ -137,6 +159,8 @@ def get_profile(self, adapter_type):
             return self.snowflake_profile()
         elif adapter_type == 'bigquery':
             return self.bigquery_profile()
+        elif adapter_type == 'redshift':
+            return self.redshift_profile()
 
     def setUp(self):
         self.adapter_type = 'postgres'