Support TTL for BigQuery tables

CHANGELOG.md

@@ -9,6 +9,7 @@
 - Upgraded snowflake-connector-python dependency to 2.2.10 and enabled the SSO token cache ([#2613](https://github.com/fishtown-analytics/dbt/issues/2613), [#2689](https://github.com/fishtown-analytics/dbt/issues/2689), [#2698](https://github.com/fishtown-analytics/dbt/pull/2698))
 
 ### Features
+- Support TTL for BigQuery tables([#2711](https://github.com/fishtown-analytics/dbt/pull/2711))
 - Add better retry support when using the BigQuery adapter ([#2694](https://github.com/fishtown-analytics/dbt/pull/2694), follow-up to [#1963](https://github.com/fishtown-analytics/dbt/pull/1963))
 - Added a `dispatch` method to the context adapter and deprecated `adapter_macro`. ([#2302](https://github.com/fishtown-analytics/dbt/issues/2302), [#2679](https://github.com/fishtown-analytics/dbt/pull/2679))
 - The built-in schema tests now use `adapter.dispatch`, so they can be overridden for adapter plugins ([#2415](https://github.com/fishtown-analytics/dbt/issues/2415), [#2684](https://github.com/fishtown-analytics/dbt/pull/2684))
@@ -18,7 +19,7 @@
 
 Contributors:
 - [@bbhoss](https://github.com/bbhoss) ([#2677](https://github.com/fishtown-analytics/dbt/pull/2677))
-- [@kconvey](https://github.com/kconvey) ([#2694](https://github.com/fishtown-analytics/dbt/pull/2694))
+- [@kconvey](https://github.com/kconvey) ([#2694](https://github.com/fishtown-analytics/dbt/pull/2694), [#2711], (https://github.com/fishtown-analytics/dbt/pull/2711))
 
 ## dbt 0.18.0b2 (July 30, 2020)
 

plugins/bigquery/dbt/adapters/bigquery/impl.py

@@ -104,6 +104,7 @@ class BigqueryConfig(AdapterConfig):
     labels: Optional[Dict[str, str]] = None
     partitions: Optional[List[str]] = None
     grant_access_to: Optional[List[Dict[str, str]]] = None
+    time_to_expiration: Optional[int] = None
 
 
 class BigQueryAdapter(BaseAdapter):
@@ -745,6 +746,12 @@ def get_table_options(
             expiration = 'TIMESTAMP_ADD(CURRENT_TIMESTAMP(), INTERVAL 12 hour)'
             opts['expiration_timestamp'] = expiration
 
+        if (config.get('time_to_expiration') is not None) and (not temporary):
+            expiration = (
+                'TIMESTAMP_ADD(CURRENT_TIMESTAMP(), INTERVAL '
+                '{} hour)').format(config.get('time_to_expiration'))
+            opts['expiration_timestamp'] = expiration
+
         if config.persist_relation_docs() and 'description' in node:
             description = sql_escape(node['description'])
             opts['description'] = '"""{}"""'.format(description)

test/integration/022_bigquery_test/adapter-specific-models/expiring_table.sql

@@ -0,0 +1 @@
+select 1 as id

test/integration/022_bigquery_test/test_bigquery_adapter_specific.py

@@ -0,0 +1,37 @@
+""""Test adapter specific config options."""
+from test.integration.base import DBTIntegrationTest, use_profile
+import textwrap
+import yaml
+
+
+class TestBigqueryAdapterSpecific(DBTIntegrationTest):
+
+    @property
+    def schema(self):
+        return "bigquery_test_022"
+
+    @property
+    def models(self):
+        return "adapter-specific-models"
+
+    @property
+    def profile_config(self):
+        return self.bigquery_profile()
+
+    @property
+    def project_config(self):
+        return yaml.safe_load(textwrap.dedent('''\
+        config-version: 2
+        models:
+            test:
+                materialized: table
+                expiring_table:
+                    time_to_expiration: 4    
+        '''))
+
+    @use_profile('bigquery')
+    def test_bigquery_time_to_expiration(self):
+        _, stdout = self.run_dbt_and_capture()
+        self.assertIn(
+            'expiration_timestamp: TIMESTAMP_ADD(CURRENT_TIMESTAMP(), INTERVAL '
+            '4 hour)', stdout)

test/unit/test_bigquery_adapter.py

@@ -4,7 +4,7 @@
 import unittest
 from contextlib import contextmanager
 from requests.exceptions import ConnectionError
-from unittest.mock import patch, MagicMock, Mock
+from unittest.mock import patch, MagicMock, Mock, create_autospec
 
 import hologram
 
@@ -571,6 +571,35 @@ def test_parse_partition_by(self):
             }
         )
 
+    def test_time_to_expiration(self):
+        adapter = self.get_adapter('oauth')
+        mock_config = create_autospec(
+            dbt.context.providers.RuntimeConfigObject)
+        config = {'time_to_expiration': 4}
+        mock_config.get.side_effect = lambda name: config.get(name)
+
+        expected = {
+            'expiration_timestamp': 'TIMESTAMP_ADD(CURRENT_TIMESTAMP(), INTERVAL 4 hour)',
+        }
+        actual = adapter.get_table_options(mock_config, node={}, temporary=False)
+        self.assertEqual(expected, actual)
+
+
+    def test_time_to_expiration_temporary(self):
+        adapter = self.get_adapter('oauth')
+        mock_config = create_autospec(
+            dbt.context.providers.RuntimeConfigObject)
+        config={'time_to_expiration': 4}
+        mock_config.get.side_effect = lambda name: config.get(name)
+
+        expected = {
+            'expiration_timestamp': (
+                'TIMESTAMP_ADD(CURRENT_TIMESTAMP(), INTERVAL 12 hour)'),
+        }
+        actual = adapter.get_table_options(mock_config, node={}, temporary=True)
+        self.assertEqual(expected, actual)
+
+
 
 class TestBigQueryFilterCatalog(unittest.TestCase):
     def test__catalog_filter_table(self):