Modernize legacy insert by period materialization

CHANGELOG.md

@@ -5,7 +5,7 @@
 - Declare compatibility with dbt v0.21.0, which has no breaking changes for this package ([#398](https://github.com/fishtown-analytics/dbt-utils/pull/398))
 
 ## Features
-
+- Modernize the `insert_by_period` materialization and make it Snowflake-compatible. 
 - Allow user to provide any case type when defining the `exclude` argument in `dbt_utils.star()` ([#403](https://github.com/dbt-labs/dbt-utils/pull/403))
 
 

README.md

@@ -1079,19 +1079,9 @@ with events as (
 * `stop_date`: literal date or timestamp (default=current_timestamp)
 
 **Caveats:**
-* This materialization is compatible with dbt 0.10.1.
-* This materialization has been written for Redshift.
+* This materialization is compatible with dbt 0.21.0.
+* This materialization has been written primarily for Snowflake.
 * This materialization can only be used for a model where records are not expected to change after they are created.
-* Any model post-hooks that use `{{ this }}` will fail using this materialization. For example:
-```yaml
-models:
-    project-name:
-        post-hook: "grant select on {{ this }} to db_reader"
-```
-A useful workaround is to change the above post-hook to:
-```yaml
-        post-hook: "grant select on {{ this.schema }}.{{ this.name }} to db_reader"
-```
 
 ----
 

integration_tests/models/materializations/expected_insert_by_period.sql

@@ -1,7 +1,6 @@
 {{
 	config(
-		materialized = 'view',
-		enabled=(target.type == 'redshift')
+		materialized = 'view'
 	)
 }}
 

integration_tests/models/materializations/test_insert_by_period.sql

@@ -4,8 +4,7 @@
 		period = 'month',
 		timestamp_field = 'created_at',
 		start_date = '2018-01-01',
-		stop_date = '2018-06-01',
-		enabled=(target.type == 'redshift')
+		stop_date = '2018-06-01'
 	)
 }}
 

macros/materializations/helpers.sql

@@ -0,0 +1,113 @@
+{% macro check_for_period_filter(model_unique_id, sql) %}
+    {{ return(adapter.dispatch('check_for_period_filter', 'dbt_utils')(model_unique_id, sql)) }}
+{% endmacro %}
+
+{% macro default__check_for_period_filter(model_unique_id, sql) %}
+    {%- if sql.find('__PERIOD_FILTER__') == -1 -%}
+        {%- set error_message -%}
+        Model '{{ model_unique_id }}' does not include the required string '__PERIOD_FILTER__' in its sql
+        {%- endset -%}
+        {{ exceptions.raise_compiler_error(error_message) }}
+    {%- endif -%}
+{% endmacro %}
+
+{% macro get_period_boundaries(target_schema, target_table, timestamp_field, start_date, stop_date, period) -%}
+    {{ return(adapter.dispatch('get_period_boundaries', 'dbt_utils')(target_schema, target_table, timestamp_field, start_date, stop_date, period)) }}
+{% endmacro %}
+
+{% macro default__get_period_boundaries(target_schema, target_table, timestamp_field, start_date, stop_date, period) -%}
+
+  {% call statement('period_boundaries', fetch_result=True) -%}
+    {%- set model_name = model['name'] -%}
+
+    with data as (
+      select
+          coalesce(max("{{timestamp_field}}"), '{{start_date}}')::timestamp as start_timestamp,
+          coalesce(
+            {{dbt_utils.dateadd('millisecond',
+                                -1,
+                                "nullif('" ~ stop_date ~ "','')::timestamp")}},
+            {{dbt_utils.current_timestamp()}}
+          ) as stop_timestamp
+      from "{{target_schema}}"."{{model_name}}"
+    )
+
+    select
+      start_timestamp,
+      stop_timestamp,
+      {{dbt_utils.datediff('start_timestamp',
+                           'stop_timestamp',
+                           period)}}  + 1 as num_periods
+    from data
+  {%- endcall %}
+
+{%- endmacro %}
+
+{% macro snowflake__get_period_boundaries(target_schema, target_table, timestamp_field, start_date, stop_date, period) -%}
+
+  {% call statement('period_boundaries', fetch_result=True) -%}
+    {%- set model_name = model['name'] -%}
+
+    with data as (
+      select
+          coalesce(max({{timestamp_field}}), '{{start_date}}')::timestamp as start_timestamp,
+          coalesce(
+            {{dbt_utils.dateadd('millisecond',
+                                -1,
+                                "nullif('" ~ stop_date ~ "','')::timestamp")}},
+            {{dbt_utils.current_timestamp()}}
+          ) as stop_timestamp
+      from {{target.schema}}.{{model_name}}
+    )
+
+    select
+      start_timestamp,
+      stop_timestamp,
+      {{dbt_utils.datediff('start_timestamp',
+                           'stop_timestamp',
+                           period)}}  + 1 as num_periods
+    from data
+  {%- endcall %}
+
+{%- endmacro %}
+
+
+{% macro get_period_sql(target_cols_csv, sql, timestamp_field, period, start_timestamp, stop_timestamp, offset) -%}
+    {{ return(adapter.dispatch('get_period_sql', 'dbt_utils')(target_cols_csv, sql, timestamp_field, period, start_timestamp, stop_timestamp, offset)) }}
+{% endmacro %}
+
+{% macro default__get_period_sql(target_cols_csv, sql, timestamp_field, period, start_timestamp, stop_timestamp, offset) -%}
+
+  {%- set period_filter -%}
+    ("{{timestamp_field}}" >  '{{start_timestamp}}'::timestamp + interval '{{offset}} {{period}}' and
+     "{{timestamp_field}}" <= '{{start_timestamp}}'::timestamp + interval '{{offset}} {{period}}' + interval '1 {{period}}' and
+     "{{timestamp_field}}" <  '{{stop_timestamp}}'::timestamp)
+  {%- endset -%}
+
+  {%- set filtered_sql = sql | replace("__PERIOD_FILTER__", period_filter) -%}
+
+  select
+    {{target_cols_csv}}
+  from (
+    {{filtered_sql}}
+  ) as t -- has to have an alias
+
+{%- endmacro %}
+
+{% macro snowflake__get_period_sql(target_cols_csv, sql, timestamp_field, period, start_timestamp, stop_timestamp, offset) -%}
+
+  {%- set period_filter -%}
+    ({{timestamp_field}} >  '{{start_timestamp}}'::timestamp + interval '{{offset}} {{period}}' and
+     {{timestamp_field}} <= '{{start_timestamp}}'::timestamp + interval '{{offset}} {{period}}' + interval '1 {{period}}' and
+     {{timestamp_field}} <  '{{stop_timestamp}}'::timestamp)
+  {%- endset -%}
+
+  {%- set filtered_sql = sql | replace("__PERIOD_FILTER__", period_filter) -%}
+
+  select
+    {{target_cols_csv}}
+  from (
+    {{filtered_sql}}
+  )
+
+{%- endmacro %}