Avoid overlapping validity timestamps (#1736)

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

@@ -82,10 +82,30 @@
 {% 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_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) %}
 
  {% if check_cols_config == 'all' %}
  {% set check_cols = get_columns_in_query(node['injected_sql']) %}

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/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_simple_snapshot.py

@@ -33,6 +33,7 @@ def assert_case_tables_equal(self, actual, expected):
  self.assertTablesEqual(actual, expected)
 
  def assert_expected(self):
+ self.run_dbt(['test'])
  self.assert_case_tables_equal('snapshot_actual', 'snapshot_expected')
 
 
@@ -42,6 +43,7 @@ def project_config(self):
  return {
  "data-paths": ['data'],
  "snapshot-paths": ['test-snapshots-pg'],
+ 'macro-paths': ['macros'],
  }
 
  @use_profile('postgres')
@@ -98,7 +100,7 @@ class TestCustomSnapshotFiles(BaseSimpleSnapshotTest):
  def project_config(self):
  return {
  'data-paths': ['data'],
- 'macro-paths': ['custom-snapshot-macros'],
+ 'macro-paths': ['custom-snapshot-macros', 'macros'],
  'snapshot-paths': ['test-snapshots-pg-custom'],
  }
 
@@ -128,7 +130,7 @@ class TestNamespacedCustomSnapshotFiles(BaseSimpleSnapshotTest):
  def project_config(self):
  return {
  'data-paths': ['data'],
- 'macro-paths': ['custom-snapshot-macros'],
+ 'macro-paths': ['custom-snapshot-macros', 'macros'],
  'snapshot-paths': ['test-snapshots-pg-custom-namespaced'],
  }
 
@@ -152,7 +154,7 @@ class TestInvalidNamespacedCustomSnapshotFiles(BaseSimpleSnapshotTest):
  def project_config(self):
  return {
  'data-paths': ['data'],
- 'macro-paths': ['custom-snapshot-macros'],
+ 'macro-paths': ['custom-snapshot-macros', 'macros'],
  'snapshot-paths': ['test-snapshots-pg-custom-invalid'],
  }
 
@@ -179,6 +181,7 @@ def project_config(self):
  "data-paths": ['data'],
  "snapshot-paths": ['test-snapshots-select',
  'test-snapshots-pg'],
+ 'macro-paths': ['macros'],
  }
 
  @use_profile('postgres')
@@ -236,7 +239,8 @@ def project_config(self):
  'strategy': 'timestamp',
  'updated_at': 'updated_at',
  }
- }
+ },
+ 'macro-paths': ['macros'],
  }
 
 
@@ -253,16 +257,15 @@ def models(self):
  def project_config(self):
  return {
  "snapshot-paths": ['test-snapshots-bq'],
+ 'macro-paths': ['macros'],
  }
 
  def assert_expected(self):
+ self.run_dbt(['test'])
  self.assertTablesEqual('snapshot_actual', 'snapshot_expected')
 
  @use_profile('bigquery')
  def test__bigquery__simple_snapshot(self):
- self.use_default_project()
- self.use_profile('bigquery')
-
  self.run_sql_file("seed_bq.sql")
 
  self.run_dbt(["snapshot"])
@@ -276,11 +279,8 @@ def test__bigquery__simple_snapshot(self):
 
  self.assert_expected()
 
-
  @use_profile('bigquery')
  def test__bigquery__snapshot_with_new_field(self):
- self.use_default_project()
- self.use_profile('bigquery')
 
  self.run_sql_file("seed_bq.sql")
 
@@ -341,6 +341,7 @@ def project_config(self):
  paths = ['test-snapshots-bq']
  return {
  'snapshot-paths': paths,
+ 'macro-paths': ['macros'],
  }
 
  def run_snapshot(self):
@@ -395,6 +396,7 @@ def project_config(self):
  paths = ['test-snapshots-pg']
  return {
  'snapshot-paths': paths,
+ 'macro-paths': ['macros'],
  }
 
  def target_schema(self):
@@ -427,6 +429,7 @@ def models(self):
  def project_config(self):
  return {
  "snapshot-paths": ['test-snapshots-invalid'],
+ 'macro-paths': ['macros'],
  }
 
  @use_profile('postgres')
@@ -456,6 +459,7 @@ def project_config(self):
  return {
  "data-paths": ['data'],
  "snapshot-paths": ['test-check-col-snapshots'],
+ 'macro-paths': ['macros'],
  }
 
 
@@ -472,7 +476,8 @@ def project_config(self):
  "strategy": "check",
  "check_cols": ["email"],
  }
- }
+ },
+ 'macro-paths': ['macros'],
  }
 
 
@@ -493,6 +498,7 @@ def project_config(self):
  return {
  "data-paths": ['data'],
  "snapshot-paths": ['test-check-col-snapshots-bq'],
+ 'macro-paths': ['macros'],
  }
 
  @use_profile('bigquery')
@@ -562,6 +568,7 @@ def run_snapshot(self):
  def project_config(self):
  return {
  "snapshot-paths": ['test-snapshots-longtext'],
+ 'macro-paths': ['macros'],
  }
 
  @use_profile('postgres')

test/integration/base.py

@@ -545,10 +545,6 @@ def run_dbt_and_check(self, args=None, strict=True, parser=False, profiles_dir=T
  final_args.append('--log-cache-events')
 
  logger.info("Invoking dbt with {}".format(final_args))
- if args is None:
- args = ["run"]
-
- logger.info("Invoking dbt with {}".format(args))
  return dbt.handle_and_check(final_args)
 
  def run_sql_file(self, path, kwargs=None):