Release 0.10.1

Datavault-UK · Aug 28, 2023 · f0cf634 · f0cf634
2 parents 838522a + db10154
commit f0cf634
Show file tree

Hide file tree

Showing 3 changed files with 60 additions and 9 deletions.
diff --git a/dbt_project.yml b/dbt_project.yml
@@ -1,5 +1,5 @@
 name: automate_dv
-version: 0.10.0
+version: 0.10.1
 require-dbt-version: [">=1.0.0", "<2.0.0"]
 config-version: 2
 

diff --git a/macros/tables/postgres/sat.sql b/macros/tables/postgres/sat.sql
@@ -5,10 +5,12 @@
 
 {%- macro postgres__sat(src_pk, src_hashdiff, src_payload, src_extra_columns, src_eff, src_ldts, src_source, source_model) -%}
 
+{%- set apply_source_filter = config.get('apply_source_filter', false) -%}
+{%- set enable_ghost_record = var('enable_ghost_records', false) %}
+
 {%- set source_cols = automate_dv.expand_column_list(columns=[src_pk, src_hashdiff, src_payload, src_extra_columns, src_eff, src_ldts, src_source]) -%}
 {%- set window_cols = automate_dv.expand_column_list(columns=[src_pk, src_hashdiff, src_ldts]) -%}
 {%- set pk_cols = automate_dv.expand_column_list(columns=[src_pk]) -%}
-{%- set enable_ghost_record = var('enable_ghost_records', false) -%}
 
 {%- if model.config.materialized == 'vault_insert_by_rank' %}
     {%- set source_cols_with_rank = source_cols + [config.get('rank_column')] -%}
@@ -49,6 +51,21 @@ latest_records AS (
     WHERE b.rank = 1
 ),
 
+{%- if apply_source_filter %}
+
+valid_stg AS (
+    SELECT {{ automate_dv.prefix(source_cols, 's', alias_target='source') }}
+    FROM source_data AS s
+    LEFT JOIN latest_records AS sat
+    ON {{ automate_dv.multikey(src_pk, prefix=['s', 'sat'], condition='=') }}
+    WHERE {{ automate_dv.multikey(src_pk, prefix='sat', condition='IS NULL') }}
+    OR {{ automate_dv.prefix([src_ldts], 's') }} > (
+        SELECT MAX({{ src_ldts }}) FROM latest_records AS sat
+        WHERE {{ automate_dv.multikey(src_pk, prefix=['sat','s'], condition='=') }}
+    )
+),
+{%- endif %}
+
 {%- endif %}
 
 first_record_in_set AS (
@@ -59,7 +76,12 @@ first_record_in_set AS (
                 PARTITION BY {{ automate_dv.prefix([src_pk], 'sd', alias_target='source') }}
                 ORDER BY {{ automate_dv.prefix([src_ldts], 'sd', alias_target='source') }} ASC
             ) as asc_rank
-        FROM source_data as sd ) rin
+        {%- if automate_dv.is_any_incremental() and apply_source_filter %}
+        FROM valid_stg as sd
+        {%- else %}
+        FROM source_data as sd
+        {%- endif %}
+    ) AS rin
     WHERE rin.asc_rank = 1
 ),
 
@@ -72,8 +94,12 @@ unique_source_records AS (
             LAG({{ automate_dv.prefix([src_hashdiff], 'sd', alias_target='source') }}) OVER (
                 PARTITION BY {{ automate_dv.prefix([src_pk], 'sd', alias_target='source') }}
                 ORDER BY {{ automate_dv.prefix([src_ldts], 'sd', alias_target='source') }} ASC) as prev_hashdiff
+        {%- if automate_dv.is_any_incremental() and apply_source_filter %}
+        FROM valid_stg as sd
+        {%- else %}
         FROM source_data as sd
-        ) b
+        {%- endif %}
+        ) AS b
     WHERE {{ automate_dv.prefix([src_hashdiff], 'b', alias_target='source') }} != b.prev_hashdiff
 ),
 

diff --git a/macros/tables/snowflake/sat.sql b/macros/tables/snowflake/sat.sql
@@ -25,10 +25,12 @@
 
 {%- macro default__sat(src_pk, src_hashdiff, src_payload, src_extra_columns, src_eff, src_ldts, src_source, source_model) -%}
 
+{%- set apply_source_filter = config.get('apply_source_filter', false) -%}
+{%- set enable_ghost_record = var('enable_ghost_records', false) %}
+
 {%- set source_cols = automate_dv.expand_column_list(columns=[src_pk, src_hashdiff, src_payload, src_extra_columns, src_eff, src_ldts, src_source]) -%}
 {%- set window_cols = automate_dv.expand_column_list(columns=[src_pk, src_hashdiff, src_ldts]) -%}
 {%- set pk_cols = automate_dv.expand_column_list(columns=[src_pk]) -%}
-{%- set enable_ghost_record = var('enable_ghost_records', false) %}
 
 {%- if model.config.materialized == 'vault_insert_by_rank' %}
     {%- set source_cols_with_rank = source_cols + [config.get('rank_column')] -%}
@@ -66,6 +68,21 @@ latest_records AS (
     QUALIFY rank_num = 1
 ),
 
+{%- if apply_source_filter %}
+
+valid_stg AS (
+    SELECT {{ automate_dv.prefix(source_cols, 's', alias_target='source') }}
+    FROM source_data AS s
+    LEFT JOIN latest_records AS sat
+    ON {{ automate_dv.multikey(src_pk, prefix=['s', 'sat'], condition='=') }}
+    WHERE {{ automate_dv.multikey(src_pk, prefix='sat', condition='IS NULL') }}
+    OR {{ automate_dv.prefix([src_ldts], 's') }} > (
+        SELECT MAX({{ src_ldts }}) FROM latest_records AS sat
+        WHERE {{ automate_dv.multikey(src_pk, prefix=['sat','s'], condition='=') }}
+    )
+),
+{%- endif %}
+
 {%- endif %}
 
 first_record_in_set AS (
@@ -75,14 +92,22 @@ first_record_in_set AS (
             PARTITION BY {{ automate_dv.prefix([src_pk], 'sd', alias_target='source') }}
             ORDER BY {{ automate_dv.prefix([src_ldts], 'sd', alias_target='source') }} ASC
         ) as asc_rank
+    {%- if automate_dv.is_any_incremental() and apply_source_filter %}
+    FROM valid_stg as sd
+    {%- else %}
     FROM source_data as sd
+    {%- endif %}
     QUALIFY asc_rank = 1
 ),
 
 unique_source_records AS (
     SELECT DISTINCT
         {{ automate_dv.prefix(source_cols, 'sd', alias_target='source') }}
+    {%- if automate_dv.is_any_incremental() and apply_source_filter %}
+    FROM valid_stg as sd
+    {%- else %}
     FROM source_data as sd
+    {%- endif %}
     QUALIFY {{ automate_dv.prefix([src_hashdiff], 'sd', alias_target='source') }} != LAG({{ automate_dv.prefix([src_hashdiff], 'sd', alias_target='source') }}) OVER (
         PARTITION BY {{ automate_dv.prefix([src_pk], 'sd', alias_target='source') }}
         ORDER BY {{ automate_dv.prefix([src_ldts], 'sd', alias_target='source') }} ASC)
@@ -93,9 +118,9 @@ unique_source_records AS (
 
 ghost AS (
     {{ automate_dv.create_ghost_record(src_pk=src_pk, src_hashdiff=src_hashdiff,
-                                    src_payload=src_payload, src_extra_columns=src_extra_columns,
-                                    src_eff=src_eff, src_ldts=src_ldts,
-                                    src_source=src_source, source_model=source_model) }}
+                                       src_payload=src_payload, src_extra_columns=src_extra_columns,
+                                       src_eff=src_eff, src_ldts=src_ldts,
+                                       src_source=src_source, source_model=source_model) }}
 ),
 
 {%- endif %}
@@ -113,7 +138,7 @@ records_to_insert AS (
     SELECT {{ automate_dv.alias_all(source_cols, 'frin') }}
     FROM first_record_in_set AS frin
     {%- if automate_dv.is_any_incremental() %}
-    LEFT JOIN LATEST_RECORDS lr
+    LEFT JOIN latest_records lr
         ON {{ automate_dv.multikey(src_pk, prefix=['lr','frin'], condition='=') }}
         AND {{ automate_dv.prefix([src_hashdiff], 'lr', alias_target='target') }} = {{ automate_dv.prefix([src_hashdiff], 'frin') }}
         WHERE {{ automate_dv.prefix([src_hashdiff], 'lr', alias_target='target') }} IS NULL