Fix snapshot check all with an added column (#1797)

core/dbt/include/global_project/macros/materializations/snapshot/strategies.sql

@@ -82,29 +82,80 @@
 {% endmacro %}
 
 
+{% macro snapshot_string_as_time(timestamp) -%}
+ {{ adapter_macro('snapshot_string_as_time', timestamp) }}
+{%- endmacro %}
+
+
+{% macro default__snapshot_string_as_time(timestamp) %}
+ {% do exceptions.raise_not_implemented(
+ 'snapshot_string_as_time macro not implemented for adapter '+adapter.type()
+ ) %}
+{% endmacro %}
+
+
+{% macro snapshot_check_all_get_existing_columns(node, target_exists) -%}
+ {%- set query_columns = get_columns_in_query(node['injected_sql']) -%}
+ {%- if not target_exists -%}
+ {# no table yet -> return whatever the query does #}
+ {{ return([false, query_columns]) }}
+ {%- endif -%}
+ {# handle any schema changes #}
+ {%- set target_table = node.get('alias', node.get('name')) -%}
+ {%- set target_relation = adapter.get_relation(database=node.database, schema=node.schema, identifier=target_table) -%}
+ {%- set existing_cols = get_columns_in_query('select * from ' ~ target_relation) -%}
+ {%- set ns = namespace() -%} {# handle for-loop scoping with a namespace #}
+ {%- set ns.column_added = false -%}
+
+ {%- set intersection = [] -%}
+ {%- for col in query_columns -%}
+ {%- if col in existing_cols -%}
+ {%- do intersection.append(col) -%}
+ {%- else -%}
+ {% set ns.column_added = true %}
+ {%- endif -%}
+ {%- endfor -%}
+ {{ return([ns.column_added, intersection]) }}
+{%- endmacro %}
+
+
 {% macro snapshot_check_strategy(node, snapshotted_rel, current_rel, config, target_exists) %}
  {% set check_cols_config = config['check_cols'] %}
  {% set primary_key = config['unique_key'] %}
- {% set updated_at = snapshot_get_time() %}
+ {% set select_current_time -%}
+ select {{ snapshot_get_time() }} as snapshot_start
+ {%- endset %}
+
+ {# don't access the column by name, to avoid dealing with casing issues on snowflake #}
+ {%- set now = run_query(select_current_time)[0][0] -%}
+ {% if now is none or now is undefined -%}
+ {%- do exceptions.raise_compiler_error('Could not get a snapshot start time from the database') -%}
+ {%- endif %}
+ {% set updated_at = snapshot_string_as_time(now) %}
+
+ {% set column_added = false %}
 
  {% if check_cols_config == 'all' %}
- {% set check_cols = get_columns_in_query(node['injected_sql']) %}
+ {% set column_added, check_cols = snapshot_check_all_get_existing_columns(node, target_exists) %}
  {% elif check_cols_config is iterable and (check_cols_config | length) > 0 %}
  {% set check_cols = check_cols_config %}
  {% else %}
  {% do exceptions.raise_compiler_error("Invalid value for 'check_cols': " ~ check_cols_config) %}
  {% endif %}
 
- {% set row_changed_expr -%}
- (
- {% for col in check_cols %}
- {{ snapshotted_rel }}.{{ col }} != {{ current_rel }}.{{ col }}
- or
- ({{ snapshotted_rel }}.{{ col }} is null) != ({{ current_rel }}.{{ col }} is null)
- {%- if not loop.last %} or {% endif %}
-
- {% endfor %}
- )
+ {%- set row_changed_expr -%}
+ (
+ {%- if column_added -%}
+ TRUE
+ {%- else -%}
+ {%- for col in check_cols -%}
+ {{ snapshotted_rel }}.{{ col }} != {{ current_rel }}.{{ col }}
+ or
+ ({{ snapshotted_rel }}.{{ col }} is null) != ({{ current_rel }}.{{ col }} is null)
+ {%- if not loop.last %} or {% endif -%}
+ {%- endfor -%}
+ {%- endif -%}
+ )
  {%- endset %}
 
  {% set scd_id_expr = snapshot_hash_arguments([primary_key, updated_at]) %}

plugins/bigquery/dbt/include/bigquery/macros/adapters.sql

@@ -103,6 +103,12 @@
 {%- endmacro %}
 
 
+{% macro bigquery__snapshot_string_as_time(timestamp) -%}
+ {%- set result = 'TIMESTAMP("' ~ timestamp ~ '")' -%}
+ {{ return(result) }}
+{%- endmacro %}
+
+
 {% macro bigquery__list_schemas(database) -%}
  {{ return(adapter.list_schemas()) }}
 {% endmacro %}

plugins/postgres/dbt/include/postgres/macros/adapters.sql

@@ -104,6 +104,12 @@
  now()
 {%- endmacro %}
 
+{% macro postgres__snapshot_string_as_time(timestamp) -%}
+ {%- set result = "'" ~ timestamp ~ "'::timestamp without time zone" -%}
+ {{ return(result) }}
+{%- endmacro %}
+
+
 {% macro postgres__snapshot_get_time() -%}
  {{ current_timestamp() }}::timestamp without time zone
 {%- endmacro %}

plugins/redshift/dbt/include/redshift/macros/adapters.sql

@@ -172,6 +172,12 @@
  {{ current_timestamp() }}::timestamp
 {%- endmacro %}
 
+
+{% macro redshift__snapshot_string_as_time(timestamp) -%}
+ {%- set result = "'" ~ timestamp ~ "'::timestamp" -%}
+ {{ return(result) }}
+{%- endmacro %}
+
 {% macro redshift__make_temp_relation(base_relation, suffix) %}
  {% do return(postgres__make_temp_relation(base_relation, suffix)) %}
 {% endmacro %}

plugins/snowflake/dbt/include/snowflake/macros/adapters.sql

@@ -108,6 +108,13 @@
  convert_timezone('UTC', current_timestamp())
 {%- endmacro %}
 
+
+{% macro snowflake__snapshot_string_as_time(timestamp) -%}
+ {%- set result = "to_timestamp_ntz('" ~ timestamp ~ "')" -%}
+ {{ return(result) }}
+{%- endmacro %}
+
+
 {% macro snowflake__snapshot_get_time() -%}
  to_timestamp_ntz({{ current_timestamp() }})
 {%- endmacro %}

test/integration/004_simple_snapshot_test/data/seed.csv

@@ -0,0 +1,4 @@
+id,first_name
+1,Judith
+2,Arthur
+3,Rachel

test/integration/004_simple_snapshot_test/data/seed_newcol.csv

@@ -0,0 +1,4 @@
+id,first_name,last_name
+1,Judith,Kennedy
+2,Arthur,Kelly
+3,Rachel,Moreno

test/integration/004_simple_snapshot_test/macros/test_no_overlaps.sql

@@ -0,0 +1,85 @@
+{% macro get_snapshot_unique_id() -%}
+ {{ return(adapter_macro('get_snapshot_unique_id', )) }}
+{%- endmacro %}
+
+{% macro default__get_snapshot_unique_id() -%}
+ {% do return("id || '-' || first_name") %}
+{%- endmacro %}
+
+
+{% macro bigquery__get_snapshot_unique_id() -%}
+ {%- do return('concat(cast(id as string), "-", first_name)') -%}
+{%- endmacro %}
+
+{#
+ mostly copy+pasted from dbt_utils, but I removed some parameters and added
+ a query that calls get_snapshot_unique_id
+#}
+{% macro test_mutually_exclusive_ranges(model) %}
+
+with base as (
+ select {{ get_snapshot_unique_id() }} as dbt_unique_id,
+ *
+ from {{ model }}
+),
+window_functions as (
+
+ select
+ dbt_valid_from as lower_bound,
+ coalesce(dbt_valid_to, '2099-1-1T00:00:01') as upper_bound,
+
+ lead(dbt_valid_from) over (
+ partition by dbt_unique_id
+ order by dbt_valid_from
+ ) as next_lower_bound,
+
+ row_number() over (
+ partition by dbt_unique_id
+ order by dbt_valid_from desc
+ ) = 1 as is_last_record
+
+ from base
+
+),
+
+calc as (
+ -- We want to return records where one of our assumptions fails, so we'll use
+ -- the `not` function with `and` statements so we can write our assumptions nore cleanly
+ select
+ *,
+
+ -- For each record: lower_bound should be < upper_bound.
+ -- Coalesce it to return an error on the null case (implicit assumption
+ -- these columns are not_null)
+ coalesce(
+ lower_bound < upper_bound,
+ is_last_record
+ ) as lower_bound_less_than_upper_bound,
+
+ -- For each record: upper_bound {{ allow_gaps_operator }} the next lower_bound.
+ -- Coalesce it to handle null cases for the last record.
+ coalesce(
+ upper_bound = next_lower_bound,
+ is_last_record,
+ false
+ ) as upper_bound_equal_to_next_lower_bound
+
+ from window_functions
+
+),
+
+validation_errors as (
+
+ select
+ *
+ from calc
+
+ where not(
+ -- THE FOLLOWING SHOULD BE TRUE --
+ lower_bound_less_than_upper_bound
+ and upper_bound_equal_to_next_lower_bound
+ )
+)
+
+select count(*) from validation_errors
+{% endmacro %}

test/integration/004_simple_snapshot_test/models/schema.yml

@@ -0,0 +1,5 @@
+version: 2
+models:
+ - name: snapshot_actual
+ tests:
+ - mutually_exclusive_ranges

test/integration/004_simple_snapshot_test/test-snapshots-checkall/snapshot.sql

@@ -0,0 +1,4 @@
+{% snapshot my_snapshot %}
+ {{ config(check_cols='all', unique_key='id', strategy='check', target_database=database, target_schema=schema) }}
+ select * from {{ ref(var('seed_name', 'seed')) }}
+{% endsnapshot %}