Fix incremental filtering (#155)

* Fix incremental filtering

* Bump version

* Add unique key

* Cut new version

.changes/5.1.2.md

@@ -0,0 +1,7 @@
+## dbt-snowflake-monitoring 5.1.2 - May 23, 2024
+
+### Fixes
+
+- Fix incremental filtering ([#254](https://github.com/get-select/dbt-snowflake-monitoring/pull/254))
+
+

CHANGELOG.md

@@ -5,6 +5,14 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html),
 and is generated by [Changie](https://github.com/miniscruff/changie).
 
+## dbt-snowflake-monitoring 5.1.2 - May 23, 2024
+
+### Fixes
+
+- Fix incremental filtering ([#254](https://github.com/get-select/dbt-snowflake-monitoring/pull/254))
+
+
+
 ## dbt-snowflake-monitoring 5.1.1 - May 14, 2024
 
 ### Fixes

dbt_project.yml

@@ -1,5 +1,5 @@
 name: 'dbt_snowflake_monitoring'
-version: '5.1.1'
+version: '5.1.2'
 config-version: 2
 
 profile: dbt_snowflake_monitoring

models/dbt_queries.sql

@@ -53,5 +53,5 @@ where dbt_metadata is not null
     {% if is_incremental() %}
         -- Conservatively re-process the last 3 days to account for late arriving rates data which changes the cost per query. 
         -- Allow an override from project variable
-        and end_time > (select dateadd(day, -{{ var('dbt_snowflake_monitoring_incremental_days', '3') }}, max(end_time)) from {{ this }})
+        and end_time > (select coalesce(dateadd(day, -{{ var('dbt_snowflake_monitoring_incremental_days', '3') }}, max(end_time)), '1970-01-01') from {{ this }})
     {% endif %}

models/staging/stg_access_history.sql

@@ -10,7 +10,7 @@ select
 from {{ source('snowflake_account_usage', 'access_history') }}
 
 {% if is_incremental() %}
-    where query_start_time > (select coalesce(max(query_start_time), date '1970-01-01') from {{ this }})
+    where query_start_time > (select coalesce(max(query_start_time), '1970-01-01') from {{ this }})
 {% endif %}
 
 order by query_start_time asc

models/staging/stg_query_history.sql

@@ -1,4 +1,7 @@
-{{ config(materialized='incremental') }}
+{{ config(
+    materialized='incremental',
+    unique_key=['query_id', 'start_time'],
+) }}
 
 select
     query_id,
@@ -72,7 +75,8 @@ from {{ source('snowflake_account_usage', 'query_history') }}
 
 {% if is_incremental() %}
     -- must use end time in case query hasn't completed
-    where end_time > (select max(end_time) from {{ this }})
+    -- add lookback window of 2 days to account for late arriving queries
+    where end_time > (select dateadd(day, -2, coalesce(max(end_time), '1970-01-01') ) from {{ this }})
 {% endif %}
 
 order by start_time

models/staging/stg_serverless_task_history.sql

@@ -1,4 +1,7 @@
-{{ config(materialized='incremental') }}
+{{ config(
+    materialized='incremental',
+    unique_key=['start_time', 'task_id'],
+)  }}
 
 select
     start_time,
@@ -10,7 +13,7 @@ select
 from {{ source('snowflake_account_usage', 'serverless_task_history') }}
 
 {% if is_incremental() %}
-    where end_time > (select max(end_time) from {{ this }})
+    where end_time > (select dateadd(day, -3, coalesce(max(end_time), '1970-01-01') ) from {{ this }})
 {% endif %}
 
 order by start_time

models/staging/stg_warehouse_events_history.sql

@@ -14,7 +14,7 @@ select
 from {{ source('snowflake_account_usage', 'warehouse_events_history') }}
 
 {% if is_incremental() %}
-    where timestamp > (select max(timestamp) from {{ this }})
+    where timestamp > (select coalesce(max(timestamp), '1970-01-01') from {{ this }})
 {% endif %}
 
 order by timestamp asc