Hi, Iam a newbie to DBT here and considering the use of macros to customise the snapshot handling of DBT.
Specifically ,Iam trying to replace the names of the default meta columns and then I end up getting errors when i run the snapshot command on the model for the second time.
The error message i get is : Snapshot target has (“x”, “y”, “z”) but not (“dbt_scd_id”, “dbt_valid_from”, “dbt_valid_to”) - is it an unmigrated previous version archive?
Could you help me figure out what the mistake iam doing here ?
Thanks in advance.
Here are the codes that I use
– Person Snapshot
{% snapshot person_snapshot %}
{{
config(
unique_key='ID',
strategy='check' ,
target_schema='myschema',
check_cols='all',
)
}}
– Overriding Macros
{% macro build_snapshot_table(strategy, sql) %}
select *,
{{ strategy.scd_id }} as scd_id,
{{ strategy.updated_at }} as updated_at,
{{ strategy.updated_at }} as valid_from,
nullif({{ strategy.updated_at }}, {{ strategy.updated_at }}) as valid_to,
Case When ({{ strategy.updated_at }}= {{ strategy.updated_at }})
Then 1
Else 0
End
as aktiv
from (
{{ sql }}
) sbq
{% endmacro %}
– Overriding Macro build staging tables snapshot
{% macro snapshot_staging_table(strategy, source_sql, target_relation) %}
with snapshot_query as (
{{ source_sql }}
),
snapshotted_data as (
select *,
{{ strategy.unique_key }} as dbt_unique_key
from {{ target_relation }}
where valid_to is null
),
insertions_source_data as (
select
*,
{{ strategy.unique_key }} as dbt_unique_key,
{{ strategy.updated_at }} as dbt_updated_at,
{{ strategy.updated_at }} as dbt_valid_from,
nullif({{ strategy.updated_at }}, {{ strategy.updated_at }}) as dbt_valid_to,
{{ strategy.scd_id }} as dbt_scd_id
from snapshot_query
),
updates_source_data as (
select
*,
{{ strategy.unique_key }} as dbt_unique_key,
{{ strategy.updated_at }} as dbt_updated_at,
{{ strategy.updated_at }} as dbt_valid_from,
{{ strategy.updated_at }} as dbt_valid_to
from snapshot_query
),
{%- if strategy.invalidate_hard_deletes %}
deletes_source_data as (
select
*,
{{ strategy.unique_key }} as dbt_unique_key
from snapshot_query
),
{% endif %}
insertions as (
select
'insert' as dbt_change_type,
source_data.*
from insertions_source_data as source_data
left outer join snapshotted_data on snapshotted_data.scd_id = source_data.dbt_unique_key
where snapshotted_data.scd_id is null
or (
snapshotted_data.scd_id is not null
and (
{{ strategy.row_changed }}
)
)
),
updates as (
select
'update' as dbt_change_type,
source_data.*,
snapshotted_data.dbt_scd_id
from updates_source_data as source_data
join snapshotted_data on snapshotted_data.scd_id = source_data.dbt_unique_key
where (
{{ strategy.row_changed }}
)
)
{%- if strategy.invalidate_hard_deletes -%}
,
deletes as (
select
'delete' as dbt_change_type,
source_data.*,
{{ snapshot_get_time() }} as dbt_valid_from,
{{ snapshot_get_time() }} as dbt_updated_at,
{{ snapshot_get_time() }} as dbt_valid_to,
snapshotted_data.dbt_scd_id
from snapshotted_data
left join deletes_source_data as source_data on snapshotted_data.scd_id = source_data.dbt_unique_key
where source_data.dbt_unique_key is null
)
{%- endif %}
select * from insertions
union all
select * from updates
{%- if strategy.invalidate_hard_deletes %}
union all
select * from deletes
{%- endif %}
{% endmacro %}
select ID,name,CityID
from {{ ref(‘vw_person’) }}
{% endsnapshot %}