Merge pull request #74 from simon-stepper/73-duplicated-months

fix: aggregate by month in fivetran_log__mar_table_history

CHANGELOG.md

@@ -1,3 +1,13 @@
+# dbt_fivetran_log v0.7.2
+## Bug Fixes
+- Fixed duplicated rows in `fivetran_log__mar_table_history` and set the model back to a monthly granularity for each source, destination, and table. ([#74](https://github.com/fivetran/dbt_fivetran_log/pull/74))
+
+## Under the Hood
+- Adjusted the uniqueness test within the `fivetran_log__mar_table_history` to also include the `schema_name` as the same table may exist in multiple schemas within a connector/destination. ([#74](https://github.com/fivetran/dbt_fivetran_log/pull/74))
+
+## Contributors
+- [@simon-stepper](https://github.com/simon-stepper) ([#73](https://github.com/fivetran/dbt_fivetran_log/issues/73))
+
 # dbt_fivetran_log v0.7.1
 ## Bug Fixes
 - Modified the logic within the `fivetran_log__mar_table_history` model to no longer filter out previous historical MAR records. Previously, these fields were filtered out as the `active_volume` source (since deprecated and replaced with `incremental_mar`) produced a cumulative daily MAR total. However, the `incremental_mar` source is not cumulative and will need to include all historical records. ([#72](https://github.com/fivetran/dbt_fivetran_log/pull/72))
@@ -62,9 +72,9 @@
 
 ```
       - dbt_utils.unique_combination_of_columns:
-          combination_of_columns: 
+          combination_of_columns:
             - destination_id
-            - measured_month 
+            - measured_month
 ```
 
 ## Under the Hood

dbt_project.yml

@@ -1,6 +1,6 @@
 config-version: 2
 name: 'fivetran_log'
-version: '0.7.1'
+version: '0.7.2'
 require-dbt-version: [">=1.3.0", "<2.0.0"]
 
 models:
@@ -10,7 +10,7 @@ models:
     staging:
       +schema: stg_fivetran_log
       tmp:
-        +materialized: view 
+        +materialized: view
 
 vars:
   fivetran_log:

integration_tests/dbt_project.yml

@@ -1,5 +1,5 @@
 name: 'fivetran_log_integration_tests'
-version: '0.7.1'
+version: '0.7.2'
 
 config-version: 2
 profile: 'integration_tests'
@@ -77,4 +77,4 @@ seeds:
       +enabled: "{{ true if target.type == 'snowflake' else false }}"
     user:
       +column_types:
-        created_at: timestamp
+        created_at: timestamp

models/fivetran_log.yml

@@ -66,14 +66,15 @@ models:
     
   - name: fivetran_log__mar_table_history
     description: > 
-      Table of each table's monthly active rows (MAR) for all connectors and destinations, per month.
+      Table of each table's monthly active rows (MAR) for all connectors, destinations, and schemas, per month.
       Read more about how monthly active rows are defined and calculated 
       [here](https://fivetran.com/docs/getting-started/consumption-based-pricing).
     tests:
       - dbt_utils.unique_combination_of_columns:
           combination_of_columns: 
             - connector_id
-            - destination_id  
+            - destination_id
+            - schema_name
             - table_name
             - measured_month
     columns:

models/fivetran_log__mar_table_history.sql

@@ -1,16 +1,15 @@
 with incremental_mar as (
 
-    select 
-        *, 
+    select
+        *,
         {{ dbt.date_trunc('month', 'measured_date') }} as measured_month
 
-    from {{ ref('stg_fivetran_log__incremental_mar') }} 
-
+    from {{ ref('stg_fivetran_log__incremental_mar') }}
 ),
 
 connector as (
 
-    select * 
+    select *
     from {{ ref('stg_fivetran_log__connector') }}
 ),
 
@@ -27,18 +26,18 @@ ordered_mar as (
         schema_name,
         table_name,
         destination_id,
-        measured_date,
         measured_month,
-        incremental_rows,
-        coalesce(case when lower(free_type) = 'paid'
+        max(measured_date) as last_measured_at,
+        sum(incremental_rows) as incremental_rows,
+        sum(coalesce(case when lower(free_type) = 'paid'
             then incremental_rows
-        end, 0) as paid_monthly_active_rows,
-        coalesce(case when lower(free_type) != 'paid'
+        end, 0)) as paid_monthly_active_rows,
+        sum(coalesce(case when lower(free_type) != 'paid'
             then incremental_rows
-        end, 0) as free_monthly_active_rows
+        end, 0)) as free_monthly_active_rows
 
     from incremental_mar
-
+    {{dbt_utils.group_by(5)}}
 ),
 
 latest_mar as (
@@ -48,29 +47,29 @@ latest_mar as (
         table_name,
         destination_id,
         measured_month,
-        date(measured_date) as last_measured_at,
+        date(last_measured_at) as last_measured_at,
         free_monthly_active_rows,
         paid_monthly_active_rows,
         (free_monthly_active_rows + paid_monthly_active_rows) as total_monthly_active_rows
-
-    from ordered_mar
 
+    from ordered_mar
 ),
 
 mar_join as (
 
-    select 
+    select
         latest_mar.*,
         connector.connector_type,
         connector.connector_id,
         destination.destination_name
 
     from latest_mar
-    join connector 
+    join connector
         on latest_mar.connector_name = connector.connector_name
         and latest_mar.destination_id = connector.destination_id
     join destination on latest_mar.destination_id = destination.destination_id
 )
 
 select * from mar_join
 order by measured_month desc, destination_id, connector_name
+