Present State → Future State

changes.sql

{{
    config(
        materialized='incremental',
        unique_key='insert_id',
        on_schema_change='sync_all_columns',
        incremental_strategy='delete+insert' 
    )
}}

{% set present_table = source('time_flip', 'present') %}
{% set future_table = source('time_flip', 'future') %}

-- Fetch columns dynamically excluding insert_id
{% set columns_to_hash_query %}
    SELECT column_name 
    FROM information_schema.columns 
    WHERE table_schema = '{{ present_table.schema }}'
        AND table_name = '{{ present_table.name }}' -- corrected from present_table.table
        AND column_name != 'insert_id'
{% endset %}

{% set columns_to_hash = run_query(columns_to_hash_query).columns[0] %} -- Extract the first column

WITH base_data AS (
    SELECT 
        *,
        SHA2(256, ARRAY_TO_STRING(ARRAY_AGG(CAST({{ columns_to_hash | join(' || ') }} AS STRING)), '')) AS cdc_hash
    FROM {{ present_table }}
    GROUP BY insert_id
),

future_data AS (
    SELECT 
        *,
        SHA2(256, ARRAY_TO_STRING(ARRAY_AGG(CAST({{ columns_to_hash | join(' || ') }} AS STRING)), '')) AS cdc_hash
    FROM {{ future_table }}
    GROUP BY insert_id
),

staged_data AS (
    SELECT 
        future_data.*
    FROM 
        future_data
    LEFT JOIN 
        base_data 
    ON 
        future_data.insert_id = base_data.insert_id
    WHERE 
        base_data.insert_id IS NULL
        OR future_data.cdc_hash <> base_data.cdc_hash
)

SELECT * FROM staged_data

future.csv

future.sql

-- future.sql
{{ config(materialized='table') }}
SELECT * FROM {{ source('time_flip', 'future') }}	

present.csv

present.sql

-- present.sql
{{ config(
    materialized='table',
    pre_hook="CREATE OR REPLACE STREAM base_stream ON TABLE {{ source('time_flip', 'present') }}"
) }}

SELECT * FROM {{ source('time_flip', 'present') }}

sources.yml

# sources.yml
version: 2

sources:
  - name: time_flip
    database: DBT_STUFF
    schema: RAW
    tables:
      - name: present
      - name: future