wrap models using macros

CHANGELOG.md

@@ -8,6 +8,8 @@
 - Ignore commented-out schema tests ([#330](https://github.com/fishtown-analytics/dbt/pull/330), [#328](https://github.com/fishtown-analytics/dbt/issues/328))
 - Fix run levels ([#343](https://github.com/fishtown-analytics/dbt/pull/343), [#340](https://github.com/fishtown-analytics/dbt/issues/340), [#338](https://github.com/fishtown-analytics/dbt/issues/338))
 - Fix concurrency, open a unique transaction per model ([#345](https://github.com/fishtown-analytics/dbt/pull/345), [#336](https://github.com/fishtown-analytics/dbt/issues/336))
+- Handle concurrent `DROP ... CASCADE`s in Redshift ([#349](https://github.com/fishtown-analytics/dbt/pull/349))
+- Always release connections (use `try .. finally`) ([#354](https://github.com/fishtown-analytics/dbt/pull/354))
 
 ### Changes
 

dbt/adapters/redshift.py

@@ -1,5 +1,11 @@
+import multiprocessing
+
 from dbt.adapters.postgres import PostgresAdapter
 from dbt.logger import GLOBAL_LOGGER as logger  # noqa
+from dbt.compat import basestring
+
+
+drop_lock = multiprocessing.Lock()
 
 
 class RedshiftAdapter(PostgresAdapter):
@@ -30,7 +36,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
@@ -42,3 +48,28 @@ def sort_qualifier(cls, sort_type, sort):
         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
+
+        to_return = None
+
+        try:
+            drop_lock.acquire()
+
+            connection = cls.get_connection(profile, model_name)
+
+            cls.commit(connection)
+            cls.begin(profile, connection.get('name'))
+
+            to_return = super(PostgresAdapter, cls).drop(
+                profile, relation, relation_type, model_name)
+
+            cls.commit(connection)
+            cls.begin(profile, connection.get('name'))
+
+            return to_return
+
+        finally:
+            drop_lock.release()

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,14 @@ def get_compiler_context(self, linker, model, flat_graph):
         context = self.project.context()
         adapter = get_adapter(self.project.run_environment())
 
+        this_table = 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'))),
+            this_table,
             model.get('name')
         )
         context['var'] = Var(model, context=context)
@@ -291,33 +293,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')))
 
@@ -402,22 +377,18 @@ def compile_graph(self, linker, flat_graph):
                 # the SQL at the parser level.
                 pass
 
+            elif(is_type(injected_node, NodeType.Model) and
+                 get_materialization(injected_node) == 'ephemeral'):
+                pass
+
             else:
-                model = Model(
+                context = self.get_compiler_context(
+                    linker, injected_node, injected_graph)
+                wrapped_stmt = dbt.wrapper.wrap(
+                    injected_node,
                     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,
+                    flat_graph)
 
                 injected_node['wrapped_sql'] = wrapped_stmt
                 wrapped_graph['nodes'][name] = injected_node

dbt/flags.py

@@ -1,2 +1,3 @@
 STRICT_MODE = False
 NON_DESTRUCTIVE = False
+FULL_REFRESH = False

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

@@ -0,0 +1,56 @@
+
+
+{# this shouldnt be duplicated - use the existing macro! #}
+{% macro dbt__create_table_for_incremental(schema, identifier, dist, sort, sql, flags, funcs) -%}
+    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, model, identifier, dist, sort, sql, sql_where, funcs, flags, unique_key=None) -%}
+
+    {% if not funcs['already_exists'](schema, identifier) -%}
+
+        {{ dbt__create_table_for_incremental(schema, identifier, dist, sort, sql, flags, funcs) }}
+
+    {%- else -%}
+
+        create temporary table "{{ identifier }}__dbt_incremental_tmp" as (
+            with dbt_incr_sbq as (
+                {{ sql }}
+            )
+            select * from dbt_incr_sbq
+            where ({{ model.get('config', {}).get('sql_where', 'null') }})
+              or ({{ model.get('config', {}).get('sql_where', 'null') }}) is null
+        );
+
+        -- 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 model.get('config', {}).get('unique_key') is not none -%}
+
+            {{ 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,24 @@
+
+{% macro dbt__create_table(schema, identifier, dist, sort, sql, flags, funcs) -%}
+
+    {% if not flags.NON_DESTRUCTIVE or
+          not funcs['already_exists'](schema, identifier) -%}
+        create table {{ schema }}.{{ identifier }} {{ dist }} {{ sort }} as (
+            {{ sql }}
+        );
+    {%- else -%}
+        create temporary table {{ identifier }}__dbt_tmp {{ dist }} {{ sort }} as (
+            {{ sql }}
+        );
+
+        {% set dest_columns = funcs['get_columns_in_table'](schema, identifier) %}
+        {% set dest_cols_csv = dest_columns | map(attribute='quoted') | join(', ') %}
+
+        insert into {{ schema }}.{{ identifier }} ({{ dest_cols_csv }})
+        (
+            select {{ dest_cols_csv }}
+            from "{{ identifier }}__dbt_tmp"
+        );
+    {%- endif %}
+
+{%- 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)
@@ -184,13 +185,8 @@ def invoke_dbt(parsed):
 
             return None
 
-    log_dir = proj.get('log-path', 'logs')
-
-    if hasattr(proj.args, 'non_destructive') and \
-       proj.args.non_destructive is True:
-        flags.NON_DESTRUCTIVE = True
-    else:
-        flags.NON_DESTRUCTIVE = False
+    flags.NON_DESTRUCTIVE = getattr(proj.args, 'non_destructive', False)
+    flags.FULL_REFRESH = getattr(proj.args, 'full_refresh', False)
 
     logger.debug("running dbt with arguments %s", parsed)
 

dbt/model.py

@@ -92,11 +92,7 @@ def config(self):
         return cfg
 
     def is_full_refresh(self):
-        if hasattr(self.active_project, 'args') and \
-           hasattr(self.active_project.args, 'full_refresh'):
-            return self.active_project.args.full_refresh
-        else:
-            return False
+        return dbt.flags.FULL_REFRESH
 
     def update_in_model_config(self, config):
         config = config.copy()
@@ -345,47 +341,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]

dbt/runner.py

@@ -478,23 +478,26 @@ def execute_node(self, node, existing):
         adapter = get_adapter(profile)
         connection = adapter.begin(profile, node.get('name'))
 
-        logger.debug("executing node %s", node.get('unique_id'))
+        try:
+            logger.debug("executing node %s", node.get('unique_id'))
+
+            if node.get('skip') is True:
+                return "SKIP"
 
-        if node.get('skip') is True:
-            return "SKIP"
+            node = self.inject_runtime_config(node)
 
-        node = self.inject_runtime_config(node)
+            if is_type(node, NodeType.Model):
+                result = execute_model(profile, node, existing)
+            elif is_type(node, NodeType.Test):
+                result = execute_test(profile, node)
+            elif is_type(node, NodeType.Archive):
+                result = execute_archive(
+                    profile, node, self.node_context(node))
 
-        if is_type(node, NodeType.Model):
-            result = execute_model(profile, node, existing)
-        elif is_type(node, NodeType.Test):
-            result = execute_test(profile, node)
-        elif is_type(node, NodeType.Archive):
-            result = execute_archive(profile, node, self.node_context(node))
+            adapter.commit(connection)
 
-        adapter.commit(connection)
-        adapter.close(connection)
-        adapter.release_connection(profile, node.get('name'))
+        finally:
+            adapter.release_connection(profile, node.get('name'))
 
         return result