[CT-2280] Tracking changes in the non-chronological insertions of data in dbt snapshots (check strategy)

### Is this your first time submitting a feature request?

- [X] I have read the [expectations for open source contributors](https://docs.getdbt.com/docs/contributing/oss-expectations)
- [X] I have searched the existing issues, and I could not find an existing issue for this feature
- [X] I am requesting a straightforward extension of existing dbt functionality, rather than a Big Idea better suited to a discussion

### Describe the feature

# Current Behaviour
Check strategy of snapshots is only able to correctly store the changes in the data if they come in chronological order. It is not able to handle non-chronological insertion of data. For instance, assume we have an entry for a unique_id in the snapshot table as shown in the table below.

**Current Snapshot Table**
| unique_id | check_cols | dbt_scd_id | dbt_updated_at | dbt_valid_from | dbt_valid_to |
| - | - | - | - | - | - |
| 1 | X | checksum1| 2022-07-16T03:58:00.000+0000 | 2022-07-16T03:58:00.000+0000 | |

When a new row with the same unique_key is introduced for a timestamp **2022-07-15T03:58:00.000+0000** and values for one of the check_cols is different from that of the currently valid row, then it will go ahead and change the snapshot table as follow.

**New row in the source table**
| unique_id | check_cols | as_of_datetime |
| - | - | - |
| 1 | Y | 2022-07-15T03:58:00.000+0000 |

**Updated snapshot table**
| unique_id | check_cols | dbt_scd_id | dbt_updated_at | dbt_valid_from | dbt_valid_to |
| - | - | - | - | - | - |
| 1 | X | checksum1 | 2022-07-16T03:58:00.000+0000 | 2022-07-16T03:58:00.000+0000 | 2022-07-15T03:58:00.000+0000 |
| 1 | Y | checksum2 | 2022-07-15T03:58:00.000+0000 | 2022-07-15T03:58:00.000+0000 | |

Apparent from the changes made to the snapshots table, check strategy is not able to handle the non-chronological changes made to the data.

# Expected Behaviour
For the above example, the expected output for the snapshot table should have been as shown in the table below.

**Expected snapshot table**
| unique_id | check_cols | dbt_scd_id | dbt_updated_at | dbt_valid_from | dbt_valid_to |
| - | - | - | - | - | - |
| 1 | Y | checksum2 | 2022-07-15T03:58:00.000+0000 | 2022-07-15T03:58:00.000+0000 | 2022-07-16T03:58:00.000+0000 |
| 1 | X | checksum1 | 2022-07-16T03:58:00.000+0000 | 2022-07-16T03:58:00.000+0000 | |

### Describe alternatives you've considered

# Updates in the source code
To solve this problem, I have made some changes to the source code of the snapshots, specifically in the following macros and the snapshot materialization strategy:
- **default__snapshot_merge_sql**
- **default__snapshot_staging_table**

I have introduced changes in the way inserts and updates are identified in the default__snapshot_staging_table macro, for tracking the changes in the non-chronologically incoming data.

The new code introduces the following dbt_change_types.

## Types of Inserts

### **`Insertion Type 1`**

**When a latest record arrives for an already existing unique_id that has some changes or a previously unseen unique_id is encountered.**

### Scenario 1

**Data already existing in snapshot table for unique_id 1**

| unique_id | some_columns… | dbt_valid_from | dbt_valid_to |
| --- | --- | --- | --- |
| 1 | X | 15-07-2022 | null |

**New Data (with some changes)**

| unique_id | some_columns… | as_of_date |
| --- | --- | --- |
| 1 | Y | 16-07-2022 |

**Final Snapshot (Insertion Type 1)**

| unique_id | some_columns… | dbt_valid_from | dbt_valid_to | dbt_change_type |
| --- | --- | --- | --- | --- |
| 1 | X | 15-07-2022 | 16-07-2022 | Update 1 |
| 1 | Y | 16-07-2022 | null | Insertion 1 |

### Scenario 2

**Data already existing in snapshot table**

| unique_id | some_columns… | dbt_valid_from | dbt_valid_to |
| --- | --- | --- | --- |
| 1 | X | 15-07-2022 | null |

**New Data (with some changes)**

| unique_id | some_columns… | as_of_date |
| --- | --- | --- |
| 2 | Z | 16-07-2022 |

**Final Snapshot (Insertion Type 1)**

| unique_id | some_columns… | dbt_valid_from | dbt_valid_to | dbt_change_type |
| --- | --- | --- | --- | --- |
| 1 | X | 15-07-2022 | null | - |
| 2 | Z | 16-07-2022 | null | Insertion 1 |

### **`Insertion Type 2`**

When an older record of the same unique_id arrives, which is different from the its nearest future version.

**Data already existing in snapshot table for unique_id 1**

| unique_id | some_columns… | dbt_valid_from | dbt_valid_to |
| --- | --- | --- | --- |
| 1 | X | 15-07-2022 | 17-07-2022 |
| 1 | Y | 17-07-2022 | null |

**New Data (with some changes)**

| unique_id | some_columns… | as_of_date |
| --- | --- | --- |
| 1 | Z | 14-07-2022 |

**Final Snapshot (Insertion Type 2)**

| unique_id | some_columns… | dbt_valid_from | dbt_valid_to | dbt_change_type |
| --- | --- | --- | --- | --- |
| 1 | Z | 14-07-2022 | 15-07-2022 | Insertion 2 |
| 1 | X | 15-07-2022 | 17-07-2022 | - |
| 1 | Y | 17-02-2022 | null | - |

## Types of Updates

### **`Update Type 1`**

When a later record of the same unique_id arrives, which is different from the its nearest past version and has a dbt_valid_from < dbt_valid_to (snapshot table record).

### **`Update Type 2`**

When an older record of the same unique_id arrives, which is not different from the its future version.

**Data already existing in snapshot table for unique_id 1**

| unique_id | some_columns… | dbt_valid_from | dbt_valid_to |
| --- | --- | --- | --- |
| 1 | X | 15-07-2022 | 17-07-2022 |
| 1 | Y | 17-07-2022 | null |

**New Data (with some changes)**

| unique_id | some_columns… | as_of_date |
| --- | --- | --- |
| 1 | Y | 16-07-2022 |

**Final Snapshot (Update Type 1 and 2 performed simultaneously)**

| unique_id | some_columns… | dbt_valid_from | dbt_valid_to | dbt_change_type |
| --- | --- | --- | --- | --- |
| 1 | X | 15-07-2022 | 16-07-2022 | Update 1 |
| 1 | Y | 16-02-2022 | null | Update 2 |

# Source Code Changes

### **`Snapshot Materialization`**
```
{% materialization snapshot, default %}
{%- set config = model['config'] -%}

{%- set target_table = model.get('alias', model.get('name')) -%}

{%- set strategy_name = config.get('strategy') -%}
{%- set unique_key = config.get('unique_key') %}
-- grab current tables grants config for comparision later on
{%- set grant_config = config.get('grants') -%}

{% set target_relation_exists, target_relation = get_or_create_relation(
database=model.database,
schema=model.schema,
identifier=target_table,
type='table') -%}

{%- if not target_relation.is_table -%}
{% do exceptions.relation_wrong_type(target_relation, 'table') %}
{%- endif -%}

{{ run_hooks(pre_hooks, inside_transaction=False) }}

{{ run_hooks(pre_hooks, inside_transaction=True) }}

{% set strategy_macro = strategy_dispatch(strategy_name) %}
{% set strategy = strategy_macro(model, "snapshotted_data", "source_data", config, target_relation_exists) %}

{% if not target_relation_exists %}

{% set build_sql = build_snapshot_table(strategy, model['compiled_code']) %}
{% set final_sql = create_table_as(False, target_relation, build_sql) %}

{% else %}

{{ adapter.valid_snapshot_target(target_relation) }}

{% set staging_table = build_snapshot_staging_table(strategy, sql, target_relation) %}

-- this may no-op if the database does not require column expansion
{% do adapter.expand_target_column_types(from_relation=staging_table,
to_relation=target_relation) %}

{% set missing_columns = adapter.get_missing_columns(staging_table, target_relation)
| rejectattr('name', 'equalto', 'dbt_change_type')
| rejectattr('name', 'equalto', 'DBT_CHANGE_TYPE')
| rejectattr('name', 'equalto', 'dbt_unique_key')
| rejectattr('name', 'equalto', 'DBT_UNIQUE_KEY')
| rejectattr('name', 'equalto', 'dbt_new_scd_id')
| rejectattr('name', 'equalto', 'DBT_NEW_SCD_ID')
| list %}

{% do create_columns(target_relation, missing_columns) %}

{% set source_columns = adapter.get_columns_in_relation(staging_table)
| rejectattr('name', 'equalto', 'dbt_change_type')
| rejectattr('name', 'equalto', 'DBT_CHANGE_TYPE')
| rejectattr('name', 'equalto', 'dbt_unique_key')
| rejectattr('name', 'equalto', 'DBT_UNIQUE_KEY')
| rejectattr('name', 'equalto', 'dbt_new_scd_id')
| rejectattr('name', 'equalto', 'DBT_NEW_SCD_ID')
| list %}

{% set quoted_source_columns = [] %}
{% for column in source_columns %}
{% do quoted_source_columns.append(adapter.quote(column.name)) %}
{% endfor %}

{% set final_sql = snapshot_merge_sql(
target = target_relation,
source = staging_table,
insert_cols = quoted_source_columns
)
%}

{% endif %}

{% call statement('main') %}
{{ final_sql }}
{% endcall %}

{% set should_revoke = should_revoke(target_relation_exists, full_refresh_mode=False) %}
{% do apply_grants(target_relation, grant_config, should_revoke=should_revoke) %}

{% do persist_docs(target_relation, model) %}

{% if not target_relation_exists %}
{% do create_indexes(target_relation) %}
{% endif %}

{{ run_hooks(post_hooks, inside_transaction=True) }}

{{ adapter.commit() }}

{% if staging_table is defined %}
{% do post_snapshot(staging_table) %}
{% endif %}

{{ run_hooks(post_hooks, inside_transaction=False) }}

{{ return({'relations': [target_relation]}) }}

{% endmaterialization %}
```

### **`Snapshot Staging Table`**

```
{% macro default__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 dbt_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.updated_at }} as dbt_valid_from,
{{ strategy.unique_key }} as dbt_unique_key
from snapshot_query
),
{% endif %}

insertions1 as (

select
'insert1' as dbt_change_type,
source_data.*

from insertions_source_data as source_data
left outer join snapshotted_data on snapshotted_data.dbt_unique_key = source_data.dbt_unique_key
where snapshotted_data.dbt_unique_key is null
or (
snapshotted_data.dbt_unique_key is not null
and (
{{ strategy.row_changed }}
)
and snapshotted_data.dbt_valid_from < source_data.dbt_valid_from
)

),

insertions2 as (

select
'insert2' as dbt_change_type,
source_data.* except(source_data.dbt_valid_to, source_data.dbt_scd_id),
snapshotted_data.dbt_valid_from as dbt_valid_to,
source_data.dbt_scd_id

from insertions_source_data as source_data
inner join (
select
t.* except(t.rn)
from (
select
snap_data.*,
row_number() over(partition by snap_data.dbt_unique_key order by snap_data.dbt_valid_from asc) as rn
from (
select
*,
{{ strategy.unique_key }} as dbt_unique_key
from {{ target_relation }}
) snap_data inner join insertions_source_data as source_data
on
snap_data.dbt_unique_key = source_data.dbt_unique_key
where snap_data.dbt_valid_from > source_data.dbt_valid_from
) t
where t.rn = 1
) snapshotted_data
on
source_data.dbt_unique_key = snapshotted_data.dbt_unique_key
where (
{{ strategy.row_changed }}
)

),

updates1 as (

select
'update1' as dbt_change_type,
snapshotted_data.* except(snapshotted_data.dbt_valid_to, snapshotted_data.dbt_scd_id, snapshotted_data.dbt_valid_from),
snapshotted_data.dbt_valid_from as dbt_valid_from,
source_data.dbt_valid_to as dbt_valid_to,
snapshotted_data.dbt_scd_id
from (
select
t.* except(t.rn)
from (
select
snap_data.*,
row_number() over(partition by snap_data.dbt_unique_key order by snap_data.dbt_valid_from desc) as rn
from (
select
*,
{{ strategy.unique_key }} as dbt_unique_key
from {{ target_relation }}
) snap_data inner join updates_source_data as source_data
on
snap_data.dbt_unique_key = source_data.dbt_unique_key
where snap_data.dbt_valid_from < source_data.dbt_valid_from
) t
where t.rn = 1
) snapshotted_data inner join updates_source_data as source_data
on
snapshotted_data.dbt_unique_key = source_data.dbt_unique_key
where snapshotted_data.dbt_valid_to > source_data.dbt_valid_from
and (
{{ strategy.row_changed }}
)
),

updates2 as (

select
'update2' as dbt_change_type,
snapshotted_data.* except(snapshotted_data.dbt_valid_from, snapshotted_data.dbt_scd_id, snapshotted_data.dbt_valid_to),
source_data.dbt_valid_from as dbt_valid_from,
snapshotted_data.dbt_valid_to as dbt_valid_to,
snapshotted_data.dbt_scd_id as dbt_scd_id,
source_data.dbt_scd_id as dbt_new_scd_id

from (
select
t.* except(t.rn)
from (
select
snap_data.*,
row_number() over(partition by snap_data.dbt_unique_key order by snap_data.dbt_valid_from asc) as rn
from (
select
*,
{{ strategy.unique_key }} as dbt_unique_key
from {{ target_relation }}
) snap_data inner join insertions_source_data as source_data
on
snap_data.dbt_unique_key = source_data.dbt_unique_key
where snap_data.dbt_valid_from > source_data.dbt_valid_from
) t
where t.rn = 1
) snapshotted_data inner join insertions_source_data as source_data
on
snapshotted_data.dbt_unique_key = source_data.dbt_unique_key
where not (
{{ 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.dbt_unique_key = source_data.dbt_unique_key
where source_data.dbt_unique_key is null
)

{%- endif %}

select *, null as dbt_new_scd_id from insertions1
union all
select *, null as dbt_new_scd_id from insertions2
union all
select *, null as dbt_new_scd_id from updates1
union all
select * from updates2
{%- if strategy.invalidate_hard_deletes %}
union all
select *, null as dbt_new_scd_id from deletes
{% endif %}

{%- endmacro %}
```

### **`Snapshot Merge SQL`**
```
{% macro default__snapshot_merge_sql(target, source, insert_cols) -%}
{% set insert_cols_csv = insert_cols | join(', ') %}

merge into {{ target }} as DBT_INTERNAL_DEST
using {{ source }} as DBT_INTERNAL_SOURCE
on DBT_INTERNAL_SOURCE.dbt_scd_id = DBT_INTERNAL_DEST.dbt_scd_id

when matched
and DBT_INTERNAL_DEST.dbt_valid_to is null
and DBT_INTERNAL_SOURCE.dbt_change_type in ('delete')
then update
set dbt_valid_to = DBT_INTERNAL_SOURCE.dbt_valid_to

when matched
and DBT_INTERNAL_SOURCE.dbt_change_type in ('update1')
then update
set
dbt_valid_from = DBT_INTERNAL_SOURCE.dbt_valid_from,
dbt_valid_to = DBT_INTERNAL_SOURCE.dbt_valid_to

when matched
and DBT_INTERNAL_SOURCE.dbt_change_type in ('update2')
then update
set
dbt_valid_from = DBT_INTERNAL_SOURCE.dbt_valid_from,
dbt_valid_to = DBT_INTERNAL_SOURCE.dbt_valid_to,
dbt_scd_id = DBT_INTERNAL_SOURCE.dbt_new_scd_id

when not matched
and DBT_INTERNAL_SOURCE.dbt_change_type in ('insert1', 'insert2')
then insert ({{ insert_cols_csv }})
values ({{ insert_cols_csv }})

{% endmacro %}
```

### Who will this benefit?

**This feature will help to snapshot historical data in any order, it does not matter in what order they feed the data to the dbt snapshots. There is always a possibility that the organizations get access to data that is older than the data with which they started to generate the snapshots. This feature will allow them to feed the historical data in any order and they would still be able to generate snapshots as if the changes in the data were captured in the chronological order.**

### Are you interested in contributing this feature?

Already made some changes in the source code as described above.

### Anything else?

_No response_