Extract, Transform, Load and Pipeline Implementation Logic Questions

{{ config(
    materialized='incremental',
    unique_key='order_id'
) }}

with source_data as (

  select
    order_id,
    customer_id,
    amount,
    status,
    updated_at
  from {{ source('raw', 'orders') }}
  {% if is_incremental() %}
    -- only fetch rows that may have changed since last run
    where updated_at > (
      select coalesce(max(updated_at), '1970-01-01'::timestamp)
      from {{ this }}
    )
  {% endif %}
),

-- dedupe incoming rows keeping latest updated_at per order_id
incoming_dedup as (
  select *
  from (
    select
      sd.*,
      row_number() over (partition by order_id order by updated_at desc, hashtext(order_id::text) desc) as rn
    from source_data sd
  ) t
  where rn = 1
)

select
  id.order_id,
  id.customer_id,
  id.amount,
  id.status,
  id.updated_at
from incoming_dedup id

-- For full-refresh dbt will rebuild table; for incremental, dbt will upsert/insert

WITH src AS (
  SELECT COUNT(*) AS source_count
  FROM source_table
  WHERE event_date = DATE '2025-01-01'
),
tgt AS (
  SELECT COUNT(*) AS target_count
  FROM target_table
  WHERE event_date = DATE '2025-01-01'
)
SELECT
  s.source_count,
  t.target_count,
  -- percent difference relative to source; NULL if source_count = 0
  CASE
    WHEN s.source_count = 0 THEN NULL
    ELSE (t.target_count - s.source_count) * 1.0 / NULLIF(s.source_count, 0)
  END AS pct_diff,
  -- alert if abs(pct_diff) > 0.01 OR business rule: source=0 and target>0
  CASE
    WHEN s.source_count = 0 AND t.target_count > 0 THEN TRUE
    WHEN s.source_count = 0 AND t.target_count = 0 THEN FALSE
    WHEN ABS((t.target_count - s.source_count) * 1.0 / NULLIF(s.source_count,0)) > 0.01 THEN TRUE
    ELSE FALSE
  END AS alert
FROM src s CROSS JOIN tgt t;

BEGIN TRANSACTION;

DECLARE @now datetime2 = SYSUTCDATETIME();

-- Temporary table to collect audit rows from OUTPUT
CREATE TABLE #audit_tmp (
  order_id bigint,
  change_type varchar(10),
  changed_at datetime2
);

MERGE INTO orders_dim AS target
USING (
  SELECT * FROM staging_orders -- deduped and latest per order_id
) AS src
ON target.order_id = src.order_id AND target.is_current = 1

WHEN MATCHED AND (
     ISNULL(target.customer_id,0) <> ISNULL(src.customer_id,0)
  OR ISNULL(target.status,'') <> ISNULL(src.status,'')
  OR ISNULL(target.amount,0) <> ISNULL(src.amount,0)
) THEN
  -- Close current version
  UPDATE SET
    target.is_current = 0,
    target.effective_to = @now
  OUTPUT src.order_id, 'UPDATE' , @now INTO #audit_tmp;

WHEN NOT MATCHED BY TARGET THEN
  -- New order: insert current row
  INSERT (order_id, customer_id, status, amount, effective_from, effective_to, is_current)
  VALUES (src.order_id, src.customer_id, src.status, src.amount, @now, NULL, 1)
  OUTPUT src.order_id, 'INSERT', @now INTO #audit_tmp;

-- After closing the old current row, insert the new current version for updates
-- Do this by inserting rows from staging for those that matched and changed.
;WITH changed AS (
  SELECT s.*
  FROM staging_orders s
  JOIN #audit_tmp a ON a.order_id = s.order_id AND a.change_type = 'UPDATE'
)
INSERT INTO orders_dim (order_id, customer_id, status, amount, effective_from, effective_to, is_current)
SELECT order_id, customer_id, status, amount, @now, NULL, 1
FROM changed;

-- Persist audit entries
INSERT INTO orders_audit (order_id, change_type, changed_at)
SELECT order_id, change_type, changed_at FROM #audit_tmp;

DROP TABLE #audit_tmp;

COMMIT TRANSACTION;