Merge branch 'develop'

Datavault-UK · Mar 27, 2021 · d15aeab · d15aeab
2 parents ad08bca + 695ec99
commit d15aeab
Show file tree

Hide file tree

Showing 10 changed files with 76 additions and 16 deletions.
diff --git a/dbt_project.yml b/dbt_project.yml
@@ -1,5 +1,5 @@
 name: 'dbtvault'
-version: '0.7.3'
+version: '0.7.4'
 require-dbt-version: [">=0.18.0", "<0.20.0"]
 config-version: 2
 

diff --git a/macros/internal/check_required_parameters.sql b/macros/internal/check_required_parameters.sql
@@ -0,0 +1,16 @@
+{%- macro check_required_parameters() -%}
+
+    {%- set ns = namespace(missing_parameters=[]) -%}
+
+    {%- if kwargs is not none -%}
+
+        {%- for k, v in kwargs.items() %}
+            {%- do ns.missing_parameters.append(k) if v is none -%}
+        {%- endfor -%}
+
+        {%- if ns.missing_parameters -%}
+            {{- exceptions.raise_compiler_error("Required parameter(s) missing or none in '{}': {}".format(this, ns.missing_parameters | join(", "))) -}}
+        {%- endif -%}
+    {%- endif -%}
+
+{%- endmacro -%}
diff --git a/macros/internal/expand_column_list.sql b/macros/internal/expand_column_list.sql
@@ -2,7 +2,7 @@
 
 {%- if not columns -%}
     {%- if execute -%}
-         {{ exceptions.raise_compiler_error("Expected a list of columns, got: " ~ columns) }}
+         {{- exceptions.raise_compiler_error("Expected a list of columns, got: " ~ columns) -}}
     {%- endif -%}
 {%- endif -%}
 
@@ -31,7 +31,7 @@
         {%- else -%}
 
             {%- if execute -%}
-                {{ exceptions.raise_compiler_error("Invalid columns object provided. Must be a list of lists, dictionaries or strings.") }}
+                {{- exceptions.raise_compiler_error("Invalid columns object provided. Must be a list of lists, dictionaries or strings.") -}}
             {%- endif %}
 
         {%- endif -%}
@@ -40,11 +40,11 @@
 {%- else -%}
 
     {%- if execute -%}
-        {{ exceptions.raise_compiler_error("Invalid columns object provided. Must be a list.") }}
+        {{- exceptions.raise_compiler_error("Invalid columns object provided. Must be a list.") -}}
     {%- endif %}
 
 {%- endif -%}
 
-{% do return(col_list) %}
+{%- do return(col_list) -%}
 
 {%- endmacro -%}
diff --git a/macros/internal/multikey.sql b/macros/internal/multikey.sql
@@ -16,13 +16,13 @@
 
     {%- if condition in ['<>', '!=', '='] -%}
         {%- for col in columns -%}
-            {{ prefix[0] ~ '.' if prefix }}{{ col }} {{ condition }} {{ prefix[1] ~ '.' if prefix }}{{ col }}
+            {{ (prefix[0] ~ '.') if prefix }}{{ col }} {{ condition }} {{ (prefix[1] ~ '.') if prefix }}{{ col }}
             {%- if not loop.last %} {{ operator }} {% endif %}
         {% endfor -%}
     {%- else -%}
         {%- if dbtvault.is_list(columns) -%}
             {%- for col in columns -%}
-                {{ prefix[0] ~ '.' if prefix }}{{ col }} {{ condition if condition else '' }}
+                {{ (prefix[0] ~ '.') if prefix }}{{ col }} {{ condition if condition else '' }}
                 {%- if not loop.last -%} {{ "\n    " ~ operator }} {% endif -%}
             {%- endfor -%}
         {%- else -%}

diff --git a/macros/staging/stage.sql b/macros/staging/stage.sql
@@ -18,7 +18,7 @@
 {% if (source_model is none) and execute %}
 
     {%- set error_message -%}
-    "Staging error: Missing source_model configuration. A source model name must be provided.
+    Staging error: Missing source_model configuration. A source model name must be provided.
     e.g. 
     [REF STYLE]
     source_model: model_name

diff --git a/macros/tables/eff_sat.sql b/macros/tables/eff_sat.sql
@@ -8,8 +8,14 @@
 
 {%- macro default__eff_sat(src_pk, src_dfk, src_sfk, src_start_date, src_end_date, src_eff, src_ldts, src_source, source_model) -%}
 
+{{- dbtvault.check_required_parameters(src_pk=src_pk, src_dfk=src_dfk, src_sfk=src_sfk,
+                                       src_start_date=src_start_date, src_end_date=src_end_date,
+                                       src_eff=src_eff, src_ldts=src_ldts, src_source=src_source,
+                                       source_model=source_model) -}}
+
 {%- set source_cols = dbtvault.expand_column_list(columns=[src_pk, src_dfk, src_sfk, src_start_date, src_end_date, src_eff, src_ldts, src_source]) -%}
 {%- set fk_cols = dbtvault.expand_column_list(columns=[src_dfk, src_sfk]) -%}
+{%- set dfk_cols = dbtvault.expand_column_list(columns=[src_dfk]) -%}
 {%- set is_auto_end_dating = config.get('is_auto_end_dating', default=false) %}
 
 {{- dbtvault.prepend_generated_by() }}
@@ -43,7 +49,10 @@ latest_open_eff AS
 (
     SELECT {{ dbtvault.alias_all(source_cols, 'b') }},
            ROW_NUMBER() OVER (
-                PARTITION BY b.{{ src_pk }}
+                PARTITION BY
+                {%- for driving_key in dfk_cols %}
+                    {{ driving_key }}{{ ", " if not loop.last }}
+                {%- endfor %}
                 ORDER BY b.{{ src_ldts }} DESC
            ) AS row_number
     FROM {{ this }} AS b

diff --git a/macros/tables/hub.sql b/macros/tables/hub.sql
@@ -8,6 +8,10 @@
 
 {%- macro default__hub(src_pk, src_nk, src_ldts, src_source, source_model) -%}
 
+{{- dbtvault.check_required_parameters(src_pk=src_pk, src_nk=src_nk,
+                                       src_ldts=src_ldts, src_source=src_source,
+                                       source_model=source_model) -}}
+
 {%- set source_cols = dbtvault.expand_column_list(columns=[src_pk, src_nk, src_ldts, src_source]) -%}
 
 {%- if model.config.materialized == 'vault_insert_by_rank' %}
@@ -36,9 +40,10 @@ row_rank_{{ source_number }} AS (
     {%- endif %}
            ROW_NUMBER() OVER(
                PARTITION BY {{ src_pk }}
-               ORDER BY {{ src_ldts }} ASC
+               ORDER BY {{ src_ldts }}
            ) AS row_number
     FROM {{ ref(src) }}
+    WHERE {{ dbtvault.multikey(src_pk, condition='IS NOT NULL') }}
     QUALIFY row_number = 1
     {%- set ns.last_cte = "row_rank_{}".format(source_number) %}
 ),{{ "\n" if not loop.last }}
@@ -78,7 +83,7 @@ row_rank_union AS (
                ORDER BY {{ src_ldts }}, {{ src_source }} ASC
            ) AS row_rank_number
     FROM {{ ns.last_cte }}
-    WHERE {{ src_pk }} IS NOT NULL
+    WHERE {{ dbtvault.multikey(src_pk, condition='IS NOT NULL') }}
     QUALIFY row_rank_number = 1
     {%- set ns.last_cte = "row_rank_union" %}
 ),

diff --git a/macros/tables/link.sql b/macros/tables/link.sql
@@ -8,6 +8,10 @@
 
 {%- macro default__link(src_pk, src_fk, src_ldts, src_source, source_model) -%}
 
+{{- dbtvault.check_required_parameters(src_pk=src_pk, src_fk=src_fk,
+                                       src_ldts=src_ldts, src_source=src_source,
+                                       source_model=source_model) -}}
+
 {%- set source_cols = dbtvault.expand_column_list(columns=[src_pk, src_fk, src_ldts, src_source]) -%}
 {%- set fk_cols = dbtvault.expand_column_list([src_fk]) -%}
 
@@ -40,6 +44,10 @@ row_rank_{{ source_number }} AS (
                ORDER BY {{ src_ldts }} ASC
            ) AS row_number
     FROM {{ ref(src) }}
+    {%- if source_model | length == 1 %}
+    WHERE {{ dbtvault.multikey(src_pk, condition='IS NOT NULL') }}
+    AND {{ dbtvault.multikey(fk_cols, condition='IS NOT NULL') }}
+    {%- endif %}
     QUALIFY row_number = 1
     {%- set ns.last_cte = "row_rank_{}".format(source_number) %}
 ),{{ "\n" if not loop.last }}
@@ -79,7 +87,8 @@ row_rank_union AS (
                ORDER BY {{ src_ldts }}, {{ src_source }} ASC
            ) AS row_rank_number
     FROM {{ ns.last_cte }}
-    WHERE {{ dbtvault.multikey(fk_cols, condition='IS NOT NULL') }}
+    WHERE {{ src_pk }} IS NOT NULL
+    AND {{ dbtvault.multikey(fk_cols, condition='IS NOT NULL') }}
     QUALIFY row_rank_number = 1
     {%- set ns.last_cte = "row_rank_union" %}
 ),

diff --git a/macros/tables/sat.sql b/macros/tables/sat.sql
@@ -8,8 +8,13 @@
 
 {%- macro default__sat(src_pk, src_hashdiff, src_payload, src_eff, src_ldts, src_source, source_model) -%}
 
+{{- dbtvault.check_required_parameters(src_pk=src_pk, src_hashdiff=src_hashdiff, src_payload=src_payload,
+                                       src_eff=src_eff, src_ldts=src_ldts, src_source=src_source,
+                                       source_model=source_model) -}}
+
 {%- set source_cols = dbtvault.expand_column_list(columns=[src_pk, src_hashdiff, src_payload, src_eff, src_ldts, src_source]) -%}
 {%- set rank_cols = dbtvault.expand_column_list(columns=[src_pk, src_hashdiff, src_ldts]) -%}
+{%- set pk_cols = dbtvault.expand_column_list(columns=[src_pk]) -%}
 
 {%- if model.config.materialized == 'vault_insert_by_rank' %}
     {%- set source_cols_with_rank = source_cols + [config.get('rank_column')] -%}
@@ -26,6 +31,9 @@ WITH source_data AS (
     FROM {{ ref(source_model) }} AS a
     {%- if model.config.materialized == 'vault_insert_by_period' %}
     WHERE __PERIOD_FILTER__
+    AND {{ dbtvault.multikey(src_pk, condition='IS NOT NULL') }}
+    {% elif model.config.materialized != 'vault_insert_by_rank' and model.config.materialized != 'vault_insert_by_period' %}
+    WHERE {{ dbtvault.multikey(src_pk, condition='IS NOT NULL') }}
     {% endif %}
     {%- set source_cte = "source_data" %}
 ),
@@ -34,6 +42,7 @@ WITH source_data AS (
 rank_col AS (
     SELECT * FROM source_data
     WHERE __RANK_FILTER__
+    AND {{ dbtvault.multikey(src_pk, condition='IS NOT NULL') }}
     {%- set source_cte = "rank_col" %}
 ),
 {% endif -%}
@@ -63,8 +72,9 @@ records_to_insert AS (
     FROM {{ source_cte }} AS e
     {%- if dbtvault.is_vault_insert_by_period() or dbtvault.is_vault_insert_by_rank() or is_incremental() %}
     LEFT JOIN latest_records
-    ON {{ dbtvault.prefix([src_hashdiff], 'latest_records', alias_target='target') }} = {{ dbtvault.prefix([src_hashdiff], 'e') }}
-    WHERE {{ dbtvault.prefix([src_hashdiff], 'latest_records', alias_target='target') }} IS NULL
+    ON {{ dbtvault.prefix([src_pk], 'latest_records', alias_target='target') }} = {{ dbtvault.prefix([src_pk], 'e') }}
+    WHERE {{ dbtvault.prefix([src_hashdiff], 'latest_records', alias_target='target') }} != {{ dbtvault.prefix([src_hashdiff], 'e') }}
+        OR {{ dbtvault.prefix([src_hashdiff], 'latest_records', alias_target='target') }} IS NULL
     {%- endif %}
 )
 

diff --git a/macros/tables/t_link.sql b/macros/tables/t_link.sql
@@ -8,7 +8,12 @@
 
 {%- macro default__t_link(src_pk, src_fk, src_payload, src_eff, src_ldts, src_source, source_model) -%}
 
+{{- dbtvault.check_required_parameters(src_pk=src_pk, src_fk=src_fk, src_eff=src_eff,
+                                       src_ldts=src_ldts, src_source=src_source,
+                                       source_model=source_model) -}}
+
 {%- set source_cols = dbtvault.expand_column_list(columns=[src_pk, src_fk, src_payload, src_eff, src_ldts, src_source]) -%}
+{%- set fk_cols = dbtvault.expand_column_list([src_fk]) -%}
 
 {{ dbtvault.prepend_generated_by() }}
 
@@ -17,9 +22,15 @@ WITH stage AS (
     FROM {{ ref(source_model) }}
     {%- if model.config.materialized == 'vault_insert_by_period' %}
     WHERE __PERIOD_FILTER__
-    {%- endif %}
-    {%- if model.config.materialized == 'vault_insert_by_rank' %}
+    AND {{ dbtvault.multikey(src_pk, condition='IS NOT NULL') }}
+    AND {{ dbtvault.multikey(fk_cols, condition='IS NOT NULL') }}
+    {%- elif model.config.materialized == 'vault_insert_by_rank' %}
     WHERE __RANK_FILTER__
+    AND {{ dbtvault.multikey(src_pk, condition='IS NOT NULL') }}
+    AND {{ dbtvault.multikey(fk_cols, condition='IS NOT NULL') }}
+    {%- else %}
+    WHERE {{ dbtvault.multikey(src_pk, condition='IS NOT NULL') }}
+    AND {{ dbtvault.multikey(fk_cols, condition='IS NOT NULL') }}
     {%- endif %}
 ),
 records_to_insert AS (