From 1c99bd08281d2d6f747278cf119970969cd67558 Mon Sep 17 00:00:00 2001
From: Ian Whitestone <ianjameswhitestone@gmail.com>
Date: Fri, 24 May 2024 11:28:51 -0400
Subject: [PATCH] Fix incremental filtering (#155)

* Fix incremental filtering

* Bump version

* Add unique key

* Cut new version
---
 .changes/5.1.2.md                               | 7 +++++++
 CHANGELOG.md                                    | 8 ++++++++
 dbt_project.yml                                 | 2 +-
 models/dbt_queries.sql                          | 2 +-
 models/staging/stg_access_history.sql           | 2 +-
 models/staging/stg_query_history.sql            | 8 ++++++--
 models/staging/stg_serverless_task_history.sql  | 7 +++++--
 models/staging/stg_warehouse_events_history.sql | 2 +-
 8 files changed, 30 insertions(+), 8 deletions(-)
 create mode 100644 .changes/5.1.2.md

diff --git a/.changes/5.1.2.md b/.changes/5.1.2.md
new file mode 100644
index 0000000..b1a1a2a
--- /dev/null
+++ b/.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))
+
+
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 339eb96..28b513b 100644
--- a/CHANGELOG.md
+++ b/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
diff --git a/dbt_project.yml b/dbt_project.yml
index f051fad..d609f51 100644
--- a/dbt_project.yml
+++ b/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
diff --git a/models/dbt_queries.sql b/models/dbt_queries.sql
index a21b25a..799c3bb 100644
--- a/models/dbt_queries.sql
+++ b/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 %}
diff --git a/models/staging/stg_access_history.sql b/models/staging/stg_access_history.sql
index b073fd7..aad63d3 100644
--- a/models/staging/stg_access_history.sql
+++ b/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
diff --git a/models/staging/stg_query_history.sql b/models/staging/stg_query_history.sql
index c3f95f8..e7da17c 100644
--- a/models/staging/stg_query_history.sql
+++ b/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
diff --git a/models/staging/stg_serverless_task_history.sql b/models/staging/stg_serverless_task_history.sql
index 9de688e..8c83982 100644
--- a/models/staging/stg_serverless_task_history.sql
+++ b/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
diff --git a/models/staging/stg_warehouse_events_history.sql b/models/staging/stg_warehouse_events_history.sql
index 0f97502..70ac2d7 100644
--- a/models/staging/stg_warehouse_events_history.sql
+++ b/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