rewrite materializations with statement block

dbt/adapters/bigquery.py

@@ -4,7 +4,6 @@
 
 import dbt.exceptions
 import dbt.flags as flags
-import dbt.materializers
 import dbt.clients.gcloud
 
 from dbt.adapters.postgres import PostgresAdapter
@@ -19,15 +18,13 @@
 
 class BigQueryAdapter(PostgresAdapter):
 
-    QUERY_TIMEOUT = 60 * 1000
+    context_functions = [
+        "query_for_existing",
+        "drop_view",
+        "execute_model",
+    ]
 
-    @classmethod
-    def get_materializer(cls, node, existing):
-        materializer = dbt.materializers.NonDDLMaterializer
-        return dbt.materializers.make_materializer(materializer,
-                                                   cls,
-                                                   node,
-                                                   existing)
+    QUERY_TIMEOUT = 60 * 1000
 
     @classmethod
     def handle_error(cls, error, message, sql):
@@ -171,7 +168,7 @@ def format_sql_for_bigquery(cls, sql):
         return "#standardSQL\n{}".format(sql)
 
     @classmethod
-    def execute_model(cls, profile, model):
+    def execute_model(cls, profile, model, model_name=None):
         connection = cls.get_connection(profile, model.get('name'))
 
         if flags.STRICT_MODE:

dbt/adapters/default.py

@@ -1,14 +1,12 @@
 import copy
+import itertools
 import multiprocessing
-import re
 import time
-import yaml
 
 from contextlib import contextmanager
 
 import dbt.exceptions
 import dbt.flags
-import dbt.materializers
 
 from dbt.contracts.connection import validate_connection
 from dbt.logger import GLOBAL_LOGGER as logger
@@ -24,9 +22,22 @@ class DefaultAdapter(object):
 
     requires = {}
 
-    @classmethod
-    def get_materializer(cls, model, existing):
-        return dbt.materializers.get_materializer(cls, model, existing)
+    context_functions = [
+        "already_exists",
+        "get_columns_in_table",
+        "get_missing_columns",
+        "query_for_existing",
+        "rename",
+        "drop",
+        "truncate",
+        "add_query",
+        "expand_target_column_types",
+    ]
+
+    raw_functions = [
+        "get_status",
+        "get_result_from_cursor"
+    ]
 
     ###
     # ADAPTER-SPECIFIC FUNCTIONS -- each of these must be overridden in
@@ -49,16 +60,6 @@ def date_function(cls):
         raise dbt.exceptions.NotImplementedException(
             '`date_function` is not implemented for this adapter!')
 
-    @classmethod
-    def dist_qualifier(cls):
-        raise dbt.exceptions.NotImplementedException(
-            '`dist_qualifier` is not implemented for this adapter!')
-
-    @classmethod
-    def sort_qualifier(cls):
-        raise dbt.exceptions.NotImplementedException(
-            '`sort_qualifier` is not implemented for this adapter!')
-
     @classmethod
     def get_status(cls, cursor):
         raise dbt.exceptions.NotImplementedException(
@@ -88,6 +89,18 @@ def cancel_connection(cls, project, connection):
     ###
     # FUNCTIONS THAT SHOULD BE ABSTRACT
     ###
+    @classmethod
+    def get_result_from_cursor(cls, cursor):
+        data = []
+
+        if cursor.description is not None:
+            column_names = [col[0] for col in cursor.description]
+            raw_results = cursor.fetchall()
+            data = [dict(zip(column_names, row))
+                    for row in raw_results]
+
+        return data
+
     @classmethod
     def drop(cls, profile, relation, relation_type, model_name=None):
         if relation_type == 'view':
@@ -144,35 +157,6 @@ def rename(cls, profile, from_name, to_name, model_name=None):
     def is_cancelable(cls):
         return True
 
-    @classmethod
-    def execute_model(cls, profile, model):
-        parts = re.split(r'-- (DBT_OPERATION .*)', model.get('wrapped_sql'))
-
-        for i, part in enumerate(parts):
-            matches = re.match(r'^DBT_OPERATION ({.*})$', part)
-            if matches is not None:
-                instruction_string = matches.groups()[0]
-                instruction = yaml.safe_load(instruction_string)
-                function = instruction['function']
-                kwargs = instruction['args']
-
-                def call_expand_target_column_types(kwargs):
-                    kwargs.update({'profile': profile,
-                                   'model_name': model.get('name')})
-                    return cls.expand_target_column_types(**kwargs)
-
-                func_map = {
-                    'expand_column_types_if_needed':
-                    call_expand_target_column_types
-                }
-
-                func_map[function](kwargs)
-            else:
-                connection, cursor = cls.add_query(
-                    profile, part, model.get('name'))
-
-        return cls.get_status(cursor)
-
     @classmethod
     def get_missing_columns(cls, profile,
                             from_schema, from_table,
@@ -259,27 +243,6 @@ def get_drop_schema_sql(cls, schema):
         return ('drop schema if exists "{schema} cascade"'
                 .format(schema=schema))
 
-    @classmethod
-    def get_create_table_sql(cls, schema, table, columns, sort, dist):
-        fields = ['"{field}" {data_type}'.format(
-            field=column.name, data_type=column.data_type
-        ) for column in columns]
-        fields_csv = ",\n  ".join(fields)
-        dist = cls.dist_qualifier(dist)
-        sort = cls.sort_qualifier('compound', sort)
-        sql = """
-        create table if not exists "{schema}"."{table}" (
-        {fields}
-        )
-        {dist} {sort}
-        """.format(
-            schema=schema,
-            table=table,
-            fields=fields_csv,
-            sort=sort,
-            dist=dist).strip()
-        return sql
-
     ###
     # ODBC FUNCTIONS -- these should not need to change for every adapter,
     #                   although some adapters may override them
@@ -595,13 +558,6 @@ def drop_schema(cls, profile, schema, model_name=None):
         sql = cls.get_drop_schema_sql(schema)
         return cls.add_query(profile, sql, model_name)
 
-    @classmethod
-    def create_table(cls, profile, schema, table, columns, sort, dist,
-                     model_name=None):
-        logger.debug('Creating table "%s"."%s".', schema, table)
-        sql = cls.get_create_table_sql(schema, table, columns, sort, dist)
-        return cls.add_query(profile, sql, model_name)
-
     @classmethod
     def table_exists(cls, profile, schema, table, model_name=None):
         tables = cls.query_for_existing(profile, schema, model_name)

dbt/adapters/postgres.py

@@ -39,14 +39,6 @@ def type(cls):
     def date_function(cls):
         return 'datenow()'
 
-    @classmethod
-    def dist_qualifier(cls, dist):
-        return ''
-
-    @classmethod
-    def sort_qualifier(cls, sort_type, sort):
-        return ''
-
     @classmethod
     def get_status(cls, cursor):
         return cursor.statusmessage

dbt/adapters/redshift.py

@@ -18,37 +18,6 @@ def type(cls):
     def date_function(cls):
         return 'getdate()'
 
-    @classmethod
-    def dist_qualifier(cls, dist):
-        dist_key = dist.strip().lower()
-
-        if dist_key in ['all', 'even']:
-            return 'diststyle {}'.format(dist_key)
-        else:
-            return 'diststyle key distkey("{}")'.format(dist_key)
-
-    @classmethod
-    def sort_qualifier(cls, sort_type, sort):
-        valid_sort_types = ['compound', 'interleaved']
-        if sort_type not in valid_sort_types:
-            raise RuntimeError(
-                "Invalid sort_type given: {} -- must be one of {}"
-                .format(sort_type, valid_sort_types)
-            )
-
-        if isinstance(sort, basestring):
-            sort_keys = [sort]
-        else:
-            sort_keys = sort
-
-        formatted_sort_keys = ['"{}"'.format(sort_key)
-                               for sort_key in sort_keys]
-        keys_csv = ', '.join(formatted_sort_keys)
-
-        return "{sort_type} sortkey({keys_csv})".format(
-            sort_type=sort_type, keys_csv=keys_csv
-        )
-
     @classmethod
     def drop(cls, profile, relation, relation_type, model_name=None):
         global drop_lock

dbt/adapters/snowflake.py

@@ -132,16 +132,6 @@ def rename(cls, profile, from_name, to_name, model_name=None):
 
         connection, cursor = cls.add_query(profile, sql, model_name)
 
-    @classmethod
-    def execute_model(cls, profile, model):
-        connection = cls.get_connection(profile, model.get('name'))
-
-        if flags.STRICT_MODE:
-            validate_connection(connection)
-
-        return super(PostgresAdapter, cls).execute_model(
-            profile, model)
-
     @classmethod
     def add_begin_query(cls, profile, name):
         return cls.add_query(profile, 'BEGIN', name, auto_begin=False,