wrap models using macros

dbt/adapters/redshift.py

@@ -1,5 +1,6 @@
 from dbt.adapters.postgres import PostgresAdapter
 from dbt.logger import GLOBAL_LOGGER as logger  # noqa
+from dbt.compat import basestring
 
 
 class RedshiftAdapter(PostgresAdapter):
@@ -30,7 +31,7 @@ def sort_qualifier(cls, sort_type, sort):
                 .format(sort_type, valid_sort_types)
             )
 
-        if type(sort) == str:
+        if isinstance(sort, basestring):
             sort_keys = [sort]
         else:
             sort_keys = sort

dbt/compilation.py

@@ -5,6 +5,7 @@
 import dbt.project
 import dbt.utils
 import dbt.include
+import dbt.wrapper
 
 from dbt.model import Model
 from dbt.utils import This, Var, is_enabled, get_materialization, NodeType, \
@@ -269,13 +270,18 @@ def get_compiler_context(self, linker, model, flat_graph):
         context = self.project.context()
         adapter = get_adapter(self.project.run_environment())
 
+        if dbt.flags.NON_DESTRUCTIVE or \
+                get_materialization(model) == 'incremental':
+            table_name = model.get('name')
+        else:
+            table_name = '{}__dbt_tmp'.format(model.get('name'))
+
         # built-ins
         context['ref'] = self.__ref(context, model, flat_graph)
         context['config'] = self.__model_config(model, linker)
         context['this'] = This(
             context['env']['schema'],
-            (model.get('name') if dbt.flags.NON_DESTRUCTIVE
-             else '{}__dbt_tmp'.format(model.get('name'))),
+            table_name,
             model.get('name')
         )
         context['var'] = Var(model, context=context)
@@ -291,33 +297,6 @@ def get_compiler_context(self, linker, model, flat_graph):
 
         return context
 
-    def get_context(self, linker, model, models):
-        # THIS IS STILL USED FOR WRAPPING, BUT SHOULD GO AWAY
-        #    - Connor
-        runtime = RuntimeContext(model=model)
-
-        context = self.project.context()
-
-        # built-ins
-        context['ref'] = self.__ref(context, model, models)
-        context['config'] = self.__model_config(model, linker)
-        context['this'] = This(
-            context['env']['schema'], model.immediate_name, model.name
-        )
-        context['var'] = Var(model, context=context)
-        context['target'] = self.project.get_target()
-
-        # these get re-interpolated at runtime!
-        context['run_started_at'] = '{{ run_started_at }}'
-        context['invocation_id'] = '{{ invocation_id }}'
-
-        adapter = get_adapter(self.project.run_environment())
-        context['sql_now'] = adapter.date_function()
-
-        runtime.update_global(context)
-
-        return runtime
-
     def compile_node(self, linker, node, flat_graph):
         logger.debug("Compiling {}".format(node.get('unique_id')))
 
@@ -403,21 +382,13 @@ def compile_graph(self, linker, flat_graph):
                 pass
 
             else:
-                model = Model(
-                    self.project,
-                    injected_node.get('root_path'),
-                    injected_node.get('path'),
-                    all_projects.get(injected_node.get('package_name')))
 
-                cfg = injected_node.get('config', {})
-                model._config = cfg
-
-                context = self.get_context(linker,
-                                           model,
-                                           injected_graph.get('nodes'))
-
-                wrapped_stmt = model.compile(
-                    injected_node.get('injected_sql'), self.project, context)
+                context = self.get_compiler_context(linker, injected_node, injected_graph)
+                wrapped_stmt = dbt.wrapper.wrap(
+                    injected_node,
+                    self.project,
+                    context,
+                    flat_graph)
 
                 injected_node['wrapped_sql'] = wrapped_stmt
                 wrapped_graph['nodes'][name] = injected_node

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

@@ -0,0 +1,57 @@
+
+
+{# this shouldnt be duplicated - use the existing macro! #}
+{% macro dbt__create_table(schema, identifier, dist, sort, sql) -%}
+
+    create table {{ schema }}.{{ identifier }} {{ dist }} {{ sort }} as (
+        {{ sql }}
+    );
+
+{%- endmacro %}
+
+{% macro dbt__incremental_delete(schema, identifier, unique_key) -%}
+
+    delete
+    from "{{ schema }}"."{{ identifier }}"
+    where ({{ unique_key }}) in (
+        select ({{ unique_key }}) from "{{ identifier }}__dbt_incremental_tmp"
+    );
+
+{%- endmacro %}
+
+{% macro dbt__create_incremental(schema, identifier, dist, sort, sql, sql_where, funcs, unique_key=None) -%}
+
+    {% if not funcs['already_exists'](schema, identifier) -%}
+
+        {{ dbt__create_table(schema, identifier, dist, sort, sql) }}
+
+    {%- else -%}
+
+        create temporary table "{{ identifier }}__dbt_incremental_tmp" as (
+            with dbt_incr_sbq as (
+                {{ sql }}
+            )
+            select * from dbt_incr_sbq
+            where coalesce({{ sql_where }}, TRUE)
+        );
+
+        -- DBT_OPERATION { function: expand_column_types_if_needed, args: { temp_table: "{{ identifier }}__dbt_incremental_tmp", to_schema: "{{ schema }}", to_table: "{{ identifier }}"} }
+
+        {% set dest_columns = funcs['get_columns_in_table'](schema, identifier) %}
+        {% set dest_cols_csv = dest_columns | map(attribute='quoted') | join(', ') %}
+
+        {% if unique_key is defined -%}
+
+            {{ dbt__incremental_delete(schema, identifier, unique_key) }}
+
+        {%- endif %}
+
+        insert into "{{ schema }}"."{{ identifier }}" ({{ dest_cols_csv }})
+        (
+            select {{ dest_cols_csv }}
+            from "{{ identifier }}__dbt_incremental_tmp"
+        );
+
+    {%- endif %}
+
+{%- endmacro %}

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

@@ -0,0 +1,8 @@
+
+{% macro dbt__create_table(schema, identifier, dist, sort, sql) -%}
+
+    create table {{ schema }}.{{ identifier }} {{ dist }} {{ sort }} as (
+        {{ sql }}
+    );
+
+{%- endmacro %}

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

@@ -0,0 +1,8 @@
+
+{% macro dbt__create_view(schema, identifier, sql) -%}
+
+    create view {{ schema }}.{{ identifier }} as (
+        {{ sql }}
+    );
+
+{%- endmacro %}

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

@@ -0,0 +1,20 @@
+
+{% macro dbt__wrap(sql, pre_hooks, post_hooks) -%}
+
+    -- Compiled by DBT
+
+    {% for hook in pre_hooks %}
+
+        {{ hook }};
+
+    {% endfor %}
+
+    {{ sql }}
+
+    {% for hook in post_hooks -%}
+
+        {{ hook }};
+
+    {% endfor %}
+
+{% endmacro %}

dbt/main.py

@@ -31,6 +31,7 @@ def main(args=None):
         handle(args)
 
     except RuntimeError as e:
+        raise
         logger.info("Encountered an error:")
         logger.info(str(e))
         sys.exit(1)

dbt/model.py

@@ -345,47 +345,6 @@ def get_prologue_string(self):
         blob = "\n".join("-- {}".format(s) for s in self.prologue)
         return "-- Compiled by DBT\n{}".format(blob)
 
-    def sort_qualifier(self, model_config):
-        if 'sort' not in model_config:
-            return ''
-
-        if (self.is_view or self.is_ephemeral) and 'sort' in model_config:
-            return ''
-
-        sort_keys = model_config['sort']
-        sort_type = model_config.get('sort_type', 'compound')
-
-        if type(sort_type) != str:
-            compiler_error(
-                self,
-                "The provided sort_type '{}' is not valid!".format(sort_type)
-            )
-
-        sort_type = sort_type.strip().lower()
-
-        adapter = get_adapter(self.project.run_environment())
-        return adapter.sort_qualifier(sort_type, sort_keys)
-
-    def dist_qualifier(self, model_config):
-        if 'dist' not in model_config:
-            return ''
-
-        if (self.is_view or self.is_ephemeral) and 'dist' in model_config:
-            return ''
-
-        dist_key = model_config['dist']
-
-        if type(dist_key) != str:
-            compiler_error(
-                self,
-                "The provided distkey '{}' is not valid!".format(dist_key)
-            )
-
-        dist_key = dist_key.strip().lower()
-
-        adapter = get_adapter(self.project.run_environment())
-        return adapter.dist_qualifier(dist_key)
-
     def build_path(self):
         filename = "{}.sql".format(self.name)
         path_parts = [self.build_dir] + self.fqn[:-1] + [filename]