Rebase + rework

dbt_project.yml

@@ -11,6 +11,8 @@ analysis-paths: ["analysis"]
 data-paths: ["data"]
 macro-paths: ["macros"]
 
+require-dbt-version: ">=0.16.0"
+
 models:
   snowplow:
     base:

integration_tests/dbt_project.yml

@@ -31,3 +31,17 @@ models:
       'snowplow:context:useragent': FALSE
       'snowplow:pass_through_columns': ['test_add_col']
 
+seeds:
+  sp_event_update:
+
+seeds:
+  snowplow_integration_tests:
+    snowplow:
+      sp_event_update:
+        column_types:
+          collector_tstamp: timestamp
+          derived_tstamp: timestamp
+      sp_event:
+        column_types:
+          collector_tstamp: timestamp
+          derived_tstamp: timestamp    

macros/bigquery/get_start_date.sql

@@ -0,0 +1,48 @@
+
+
+{%- macro get_most_recent_record(relation) -%}
+
+    {#-- do not run the query in parsing mode #}
+    {%- if not execute -%}
+        {{ return('') }}
+    {%- endif -%}
+
+    {# fix for tmp suffix #}
+    {%- set relation = api.Relation.create(identifier=relation.name, schema=relation.schema) -%}
+    {%- set partition_col = config.partition_by.render() -%}
+
+    {%- call statement('_', fetch_result=True) -%}
+
+        select max(partition_col) as start_date from {{ relation }}
+
+    {%- endcall -%}
+
+    {%- set data = load_result('_')['table'].rows -%}
+    {{ return(data[0]['start_date']) }}
+
+{%- endmacro -%}
+
+
+
+{%- macro get_start_date() -%}
+
+    {%- set start_date -%}
+
+    {%- if config.incremental_strategy == 'insert_overwrite' -%}
+
+        {%- if config.partitions -%} least({{partitions|join(',')}})
+        {%- elif config.partition_by.data_type == 'date' -%} _dbt_max_partition
+        {%- else -%} date(_dbt_max_partition)
+        {%- endif -%}
+
+    {%- else -%}
+
+        '{{ get_most_recent_record() }}'
+
+    {%- endif -%}
+
+    {%- endset -%}
+
+    {%- do return(start_date) -%}
+
+{%- endmacro -%}

models/identification/bigquery/snowplow_id_map.sql

@@ -5,19 +5,24 @@
 {{
     config(
         materialized='incremental',
-        partition_by='DATE(max_tstamp)',
+        partition_by={
+            'field': 'max_tstamp',
+            'data_type': 'timestamp'
+        },
         unique_key="domain_userid",
         enabled=is_adapter('bigquery')
     )
 }}
 
-{% set start_date = get_most_recent_record(this, "max_tstamp", "2001-01-01") %}
-
 with all_events as (
 
     select *
     from {{ ref('snowplow_base_events') }}
-    where DATE(collector_tstamp) >= date_sub('{{ start_date }}', interval 1 day)
+
+    {% if is_incremental() %}
+        {% set start_date = get_start_date(this) %}
+        where DATE(collector_tstamp) >= {{start_date}}
+    {% endif %}
 
 ),
 
@@ -27,7 +32,6 @@ new_sessions as (
         domain_sessionid
 
     from all_events
-    where DATE(collector_tstamp) >= '{{ start_date }}'
 
 ),
 
@@ -59,7 +63,7 @@ prep as (
             rows between unbounded preceding and unbounded following
         ) as user_id,
 
-        max(timestamp(collector_tstamp)) over (
+        max(collector_tstamp) over (
             partition by domain_userid
         ) as max_tstamp
 

models/page_views/bigquery/snowplow_page_views.sql

@@ -1,14 +1,16 @@
 {{
     config(
         materialized='incremental',
-        partition_by='DATE(page_view_start)',
+        partition_by={
+            'field': 'page_view_start',
+            'data_type': 'timestamp'
+        },
         unique_key="page_view_id",
         enabled=is_adapter('bigquery')
     )
 }}
 
 {% set timezone = var('snowplow:timezone', 'UTC') %}
-{% set start_date = get_most_recent_record(this, "page_view_start", "2001-01-01") %}
 
 /*
     General approach: find sessions that happened since the last time
@@ -25,9 +27,14 @@ with all_events as (
     select *
     from {{ ref('snowplow_base_events') }}
 
-    -- load up events from the start date, and the day before it, to ensure
-    -- that we capture pageviews that span midnight
-    where DATE(collector_tstamp) >= date_sub('{{ start_date }}', interval 1 day)
+    {% if is_incremental() %}
+        {% set start_date = get_start_date(this) %}
+        where DATE(collector_tstamp) >= 
+            date_sub(
+                {{start_date}},
+                interval {{var('snowplow:page_view_lookback_days')}} day
+            )
+    {% endif %}
 
 ),
 
@@ -38,9 +45,6 @@ new_sessions as (
 
     from all_events
 
-    -- only consider events for sessions that occurred on or after the start_date
-    where DATE(collector_tstamp) >= '{{ start_date }}'
-
 ),
 
 relevant_events as (

models/sessions/bigquery/snowplow_sessions.sql

@@ -1,7 +1,10 @@
 {{
     config(
         materialized='table',
-        partition_by='DATE(session_start)',
+        partition_by={
+            'field': 'session_start',
+            'data_type': 'timestamp'
+        },
         enabled=is_adapter('bigquery')
     )
 }}

models/sessions/bigquery/snowplow_sessions_tmp.sql

@@ -1,18 +1,23 @@
 {{
     config(
         materialized='incremental',
-        partition_by='DATE(session_start)',
+        partition_by={
+            'field': 'session_start',
+            'data_type': 'timestamp'
+        },
         unique_key="session_id",
         enabled=is_adapter('bigquery')
     )
 }}
 
-{% set start_date = get_most_recent_record(this, "session_start", "2001-01-01") %}
-
 with all_page_views as (
 
     select * from {{ ref('snowplow_page_views') }}
-    where DATE(page_view_start) >= date_sub('{{ start_date }}', interval 1 day)
+
+    {% if is_incremental() %}
+        {% set start_date = get_start_date(this) %}
+        where DATE(collector_tstamp) >= {{start_date}}
+    {% endif %}
 
 ),
 
@@ -22,7 +27,6 @@ new_page_views as (
         session_id
 
     from all_page_views
-    where DATE(page_view_start) >= '{{ start_date }}'
 
 ),