Implement the base macro (where possible)

.github/workflows/integration_tests.yml

@@ -115,7 +115,7 @@ jobs:
           dbt run-operation post_ci_cleanup --target ${{ matrix.warehouse }}
 
       - name: Run tests
-        run: ./.scripts/integration_test.sh -d ${{ matrix.warehouse }}
+        run: ./.scripts/integration_tests.sh -d ${{ matrix.warehouse }}
 
       - name: "Post-test: Drop ci schemas"
         run: |

dbt_project.yml

@@ -67,9 +67,4 @@ models:
       scratch:
         +schema: "scratch"
         +tags: "scratch"
-        bigquery:
-          enabled: "{{ target.type == 'bigquery' | as_bool() }}"
-        databricks:
-          enabled: "{{ target.type in ['databricks', 'spark'] | as_bool() }}"
-        snowflake:
-          enabled: "{{ target.type == 'snowflake' | as_bool() }}"
+        +enabled: "{{ target.type in ['bigquery', 'databricks', 'spark', 'snowflake'] | as_bool() }}"

integration_tests/dbt_project.yml

@@ -31,6 +31,7 @@ vars:
     snowplow__backfill_limit_days: 2
     snowplow__derived_tstamp_partitioned: false
     snowplow__atomic_schema: "{{ target.schema ~ 'sp_normalize_int_test' }}"
+    snowplow__session_timestamp: 'collector_tstamp'
 
 models:
   snowplow_normalize_integration_tests:

models/base/manifest/snowplow_normalize_incremental_manifest.sql

@@ -8,13 +8,4 @@
 -- Boilerplate to generate table.
 -- Table updated as part of end-run hook
 
-with prep as (
-  select
-    cast(null as {{ snowplow_utils.type_max_string() }}) model,
-    cast('1970-01-01' as {{ type_timestamp() }}) as last_success
-)
-
-select *
-
-from prep
-where false
+{{ snowplow_utils.base_create_snowplow_incremental_manifest() }}

models/base/scratch/snowplow_normalize_base_events_this_run.sql

@@ -0,0 +1,30 @@
+{{
+  config(
+    tags=["this_run"],
+    sql_header=snowplow_utils.set_query_tag(var('snowplow__query_tag', 'snowplow_dbt'))
+  )
+}}
+
+{%- set lower_limit, upper_limit, session_start_limit = snowplow_utils.return_base_new_event_limits(ref('snowplow_normalize_base_new_event_limits')) %}
+
+select
+    a.*
+
+from {{ var('snowplow__events') }} as a
+
+where
+  {# dvce_sent_tstamp is an optional field and not all trackers/webhooks populate it, this means this filter needs to be optional #}
+  {% if var("snowplow__days_late_allowed") == -1 %}
+    1 = 1
+  {% else %}
+    a.dvce_sent_tstamp <= {{ snowplow_utils.timestamp_add('day', var("snowplow__days_late_allowed", 3), 'a.dvce_created_tstamp') }}
+  {% endif %}
+  and a.{{ var('snowplow__session_timestamp', 'collector_tstamp') }} >= {{ lower_limit }}
+  and a.{{ var('snowplow__session_timestamp', 'collector_tstamp') }} <= {{ upper_limit }}
+  {% if var('snowplow__derived_tstamp_partitioned', true) and target.type == 'bigquery' | as_bool() %}
+    and a.derived_tstamp >= {{ snowplow_utils.timestamp_add('hour', -1, lower_limit) }}
+    and a.derived_tstamp <= {{ upper_limit }}
+  {% endif %}
+  and {{ snowplow_utils.app_id_filter(var("snowplow__app_id",[])) }}
+
+qualify row_number() over (partition by a.event_id order by a.collector_tstamp{% if target.type in ['databricks', 'spark'] -%}, a.etl_tstamp {%- endif %}) = 1

models/base/scratch/snowplow_normalize_base_new_event_limits.sql

@@ -10,14 +10,15 @@
 {% set min_last_success,
          max_last_success,
          models_matched_from_manifest,
-         has_matched_all_models = snowplow_utils.get_incremental_manifest_status(ref('snowplow_normalize_incremental_manifest'), models_in_run) -%}
+         has_matched_all_models = snowplow_utils.get_incremental_manifest_status(ref('snowplow_normalize_incremental_manifest'),
+                                                                                 models_in_run) -%}
 
 
 {% set run_limits_query = snowplow_utils.get_run_limits(min_last_success,
-                                                          max_last_success,
-                                                          models_matched_from_manifest,
-                                                          has_matched_all_models,
-                                                          var("snowplow__start_date","2020-01-01")) -%}
+                                                         max_last_success,
+                                                         models_matched_from_manifest,
+                                                         has_matched_all_models,
+                                                         var("snowplow__start_date","2020-01-01")) -%}
 
 
 {{ run_limits_query }}

packages.yml

@@ -1,3 +1,3 @@
 packages:
   - package: snowplow/snowplow_utils
-    version: [">=0.14.0", "<0.16.0"]
+    version: [">=0.15.0", "<0.16.0"]