I am new to DBT and trying to write a DBT model with following behaviour (expected behaviour)
- incremental model
- update existing rows when conflict by 'id' happens
- source and destination are both datalake tables
I am using the following model config for the same destination_table.sql
{{
config(
tags=['platform','model_test'],
materialized = 'incremental',
partition_by='created_date',
unique_key='id',
strategy='merge',
merge_behavior='upsert',
on_conflict='update'
)
}}
{%- call statement('max_updated_at', fetch_result=True) -%}
select max(updated_at) from {{ this }}
{%- endcall -%}
{%- set max_updated_at = load_result('max_updated_at')['data'][0][0] -%}
{{ log("Getting maximum updated at, for model_test incremental run", info=True) }}
{{ log("Maximum updated at, for model_test incremental run: " ~ max_updated_at, info=True) }}
select * from source_table where updated_at > CAST('{{ max_updated_at }}' AS INTEGER)
But, the behaviour i see is - then incremental records are fetched from source but in case of conflict a new row with same id is being created in my destination table
Have tried various model configs including the 1 shared above, but not seeing the expected behaviour