Merge pull request #1252 from fishtown-analytics/feature/snowflake-tr…

…ansient-tables

Support transient tables on Snowflake

core/dbt/adapters/base/impl.py

@@ -111,6 +111,10 @@ class BaseAdapter(object):
     # This should be an implementation of BaseConnectionManager
     ConnectionManager = None
 
+    # A set of clobber config fields accepted by this adapter
+    # for use in materializations
+    AdapterSpecificConfigs = frozenset()
+
     def __init__(self, config):
         self.config = config
         self.cache = RelationsCache()

core/dbt/config/project.py

@@ -21,7 +21,7 @@
 from dbt.ui import printer
 from dbt.utils import deep_map
 from dbt.utils import parse_cli_vars
-from dbt.utils import DBTConfigKeys
+from dbt.parser.source_config import SourceConfig
 
 from dbt.contracts.project import Project as ProjectContract
 from dbt.contracts.project import PackageConfig
@@ -84,7 +84,7 @@ def _get_config_paths(config, path=(), paths=None):
 
     for key, value in config.items():
         if isinstance(value, dict):
-            if key in DBTConfigKeys:
+            if key in SourceConfig.ConfigKeys:
                 if path not in paths:
                     paths.add(path)
             else:

core/dbt/parser/source_config.py

@@ -1,28 +1,25 @@
 import dbt.exceptions
 
-from dbt.utils import deep_merge, DBTConfigKeys
+from dbt.utils import deep_merge
 from dbt.node_types import NodeType
+from dbt.adapters.factory import get_adapter_class_by_name
 
 
 class SourceConfig(object):
-    ConfigKeys = DBTConfigKeys
-
     AppendListFields = {'pre-hook', 'post-hook', 'tags'}
     ExtendDictFields = {'vars', 'column_types', 'quoting'}
     ClobberFields = {
         'alias',
         'schema',
         'enabled',
         'materialized',
-        'dist',
-        'sort',
         'sql_where',
         'unique_key',
-        'sort_type',
-        'bind',
         'database',
     }
 
+    ConfigKeys = AppendListFields | ExtendDictFields | ClobberFields
+
     def __init__(self, active_project, own_project, fqn, node_type):
         self._config = None
         # active_project is a RuntimeConfig, not a Project
@@ -31,16 +28,13 @@ def __init__(self, active_project, own_project, fqn, node_type):
         self.fqn = fqn
         self.node_type = node_type
 
+        adapter_type = active_project.credentials.type
+        adapter_class = get_adapter_class_by_name(adapter_type)
+        self.AdapterSpecificConfigs = adapter_class.AdapterSpecificConfigs
+
         # the config options defined within the model
         self.in_model_config = {}
 
-        # make sure we categorize all configs
-        all_configs = self.AppendListFields | self.ExtendDictFields | \
-            self.ClobberFields
-
-        for config in self.ConfigKeys:
-            assert config in all_configs, config
-
     def _merge(self, *configs):
         merged_config = {}
         for config in configs:
@@ -107,7 +101,7 @@ def update_in_model_config(self, config):
                     )
                 current.update(value)
                 self.in_model_config[key] = current
-            else:  # key in self.ClobberFields
+            else:  # key in self.ClobberFields or self.AdapterSpecificConfigs
                 self.in_model_config[key] = value
 
     @staticmethod
@@ -121,24 +115,25 @@ def __get_as_list(relevant_configs, key):
 
         return items
 
-    @classmethod
-    def smart_update(cls, mutable_config, new_configs):
+    def smart_update(self, mutable_config, new_configs):
+        config_keys = self.ConfigKeys | self.AdapterSpecificConfigs
+
         relevant_configs = {
             key: new_configs[key] for key
-            in new_configs if key in cls.ConfigKeys
+            in new_configs if key in config_keys
         }
 
-        for key in cls.AppendListFields:
-            append_fields = cls.__get_as_list(relevant_configs, key)
+        for key in self.AppendListFields:
+            append_fields = self.__get_as_list(relevant_configs, key)
             mutable_config[key].extend([
                 f for f in append_fields if f not in mutable_config[key]
             ])
 
-        for key in cls.ExtendDictFields:
+        for key in self.ExtendDictFields:
             dict_val = relevant_configs.get(key, {})
             mutable_config[key].update(dict_val)
 
-        for key in cls.ClobberFields:
+        for key in (self.ClobberFields | self.AdapterSpecificConfigs):
             if key in relevant_configs:
                 mutable_config[key] = relevant_configs[key]
 

core/dbt/utils.py

@@ -19,27 +19,6 @@
 from dbt.clients import yaml_helper
 
 
-DBTConfigKeys = [
-    'alias',
-    'schema',
-    'enabled',
-    'materialized',
-    'dist',
-    'sort',
-    'sql_where',
-    'unique_key',
-    'sort_type',
-    'pre-hook',
-    'post-hook',
-    'vars',
-    'column_types',
-    'bind',
-    'quoting',
-    'tags',
-    'database',
-]
-
-
 class ExitCodes(object):
     Success = 0
     ModelError = 1

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

@@ -51,6 +51,8 @@ class BigQueryAdapter(BaseAdapter):
     Column = dbt.schema.BigQueryColumn
     ConnectionManager = BigQueryConnectionManager
 
+    AdapterSpecificConfigs = frozenset({"cluster_by", "partition_by"})
+
     ###
     # Implementations of abstract methods
     ###

plugins/redshift/dbt/adapters/redshift/impl.py

@@ -7,6 +7,8 @@
 class RedshiftAdapter(PostgresAdapter):
     ConnectionManager = RedshiftConnectionManager
 
+    AdapterSpecificConfigs = frozenset({"sort_type", "dist", "sort", "bind"})
+
     @classmethod
     def date_function(cls):
         return 'getdate()'

plugins/snowflake/dbt/adapters/snowflake/impl.py

@@ -14,6 +14,8 @@ class SnowflakeAdapter(SQLAdapter):
     Relation = SnowflakeRelation
     ConnectionManager = SnowflakeConnectionManager
 
+    AdapterSpecificConfigs = frozenset({"transient"})
+
     @classmethod
     def date_function(cls):
         return 'CURRENT_TIMESTAMP()'

plugins/snowflake/dbt/include/snowflake/macros/adapters.sql

@@ -3,7 +3,16 @@
     use schema {{ adapter.quote_as_configured(schema, 'schema') }};
   {% endif %}
 
-  {{ default__create_table_as(temporary, relation, sql) }}
+  {%- set transient = config.get('transient', default=true) -%}
+
+  create {% if temporary -%}
+    temporary
+  {%- elif transient -%}
+    transient
+  {%- endif %} table {{ relation.include(database=(not temporary), schema=(not temporary)) }}
+  as (
+    {{ sql }}
+  );
 {% endmacro %}
 
 {% macro snowflake__create_view_as(relation, sql) -%}

test/unit/test_parser.py

@@ -35,7 +35,7 @@ def setUp(self):
             'quoting': {},
             'outputs': {
                 'test': {
-                    'type': 'postgres',
+                    'type': 'redshift',
                     'host': 'localhost',
                     'schema': 'analytics',
                     'user': 'test',