History

-- gold.dim_customer, after a customer moves house AND is re-graded.

-- TYPE 2 columns: customer_segment. A change closes the row and opens a new one.
-- TYPE 1 columns: email, phone, street, postal_code, city. Overwritten in place.

customer_sk  customer_id  city      segment  _tech_valid_from  _tech_valid_to  _tech_is_current
-----------  -----------  --------  -------  ----------------  --------------  ----------------
a1f3...      C-4711       Hamburg   B        2024-01-01        2026-03-01      false
a1f3...      C-4711       Hamburg   A        2026-03-01        NULL            true

-- Read what the address did: the customer moved from Bremen to Hamburg,
-- and BOTH rows now say Hamburg - including the row covering 2024, when
-- they lived in Bremen. That is type 1 behaving correctly, not a bug.
-- Historical address reporting was traded away deliberately, in rule 4.

customer_sk  segment  _tech_valid_from     _tech_valid_to       _tech_is_current
a1b2...      A        2024-01-15 00:00:00  2026-03-01 00:00:00  false
a1b2...      B        2026-03-01 00:00:00  NULL                 true

SELECT f.order_id, f.amount_eur, d.customer_segment
FROM   gold.fct_order_line f
JOIN   gold.dim_customer   d
  ON   d.customer_sk = f.customer_sk
 AND   f.order_ts >= d._tech_valid_from
 AND  (f.order_ts <  d._tech_valid_to OR d._tech_valid_to IS NULL)
WHERE  f.status <> 'CANCELLED';

from datetime import datetime

from pyspark.sql import DataFrame


def apply_scd2(current: DataFrame, incoming: DataFrame,
               keys: list[str], as_of: datetime) -> DataFrame:
    """Close changed versions at as_of; open their successors at as_of.

    as_of is injected so the same input yields the same output on every
    run. That property is what the idempotency test asserts.
    """

MERGE INTO gold.dim_customer AS tgt
USING (
  -- every staged row once with its key (can close an open version) ...
  SELECT s.customer_sk AS merge_key, s.* FROM stg_customer s
  UNION ALL
  -- ... and the CHANGED ones again with a NULL key, so the second copy
  -- can only ever take the NOT MATCHED branch
  SELECT NULL AS merge_key, s.*
  FROM   stg_customer s
  JOIN   gold.dim_customer d
    ON   d.customer_sk = s.customer_sk AND d._tech_is_current
  WHERE  d._tech_hash <> s._tech_hash
) AS src
ON tgt.customer_sk = src.merge_key AND tgt._tech_is_current

WHEN MATCHED AND tgt._tech_hash <> src._tech_hash THEN UPDATE SET
  tgt._tech_valid_to   = :as_of,
  tgt._tech_is_current = false

-- rule 4: an address change restates the CURRENT row, it does not version
WHEN MATCHED AND (tgt.street      IS DISTINCT FROM src.street
               OR tgt.postal_code IS DISTINCT FROM src.postal_code
               OR tgt.city        IS DISTINCT FROM src.city) THEN UPDATE SET
  tgt.street          = src.street,
  tgt.postal_code     = src.postal_code,
  tgt.city            = src.city,
  tgt._tech_loaded_at = :as_of

WHEN NOT MATCHED THEN INSERT (
  customer_sk, customer_id, customer_name, email, phone, street, postal_code,
  city, country, customer_segment, region_sk,
  _tech_valid_from, _tech_valid_to, _tech_is_current, _tech_hash, _tech_loaded_at
) VALUES (
  src.customer_sk, src.customer_id, src.customer_name, src.email, src.phone,
  src.street, src.postal_code, src.city, src.country, src.customer_segment,
  src.region_sk, :as_of, NULL, true, src._tech_hash, :as_of
);

sha2(concat_ws('||',
       coalesce(upper(trim(customer_name)),    '~'),
       coalesce(upper(trim(customer_segment)), '~'),
       coalesce(upper(trim(country)),          '~')
     ), 256) AS _tech_hash
-- street / postal_code / city are deliberately absent.

-- silver.customer is the AUTO CDC target from the ingestion route.
-- Gold reads this view, never the target directly, and every as-of join,
-- assertion and monitor on this page then works unchanged.
CREATE OR REPLACE VIEW silver.customer_scd2 AS
SELECT
  customer_id,
  customer_name,
  street,
  postal_code,
  city,
  country,
  customer_segment,
  -- __END_AT carries the SUCCESSOR's __START_AT, so the interval is
  -- already half-open with an exclusive end — the one thing that does
  -- line up. NULL still means open.
  CAST(__START_AT AS TIMESTAMP) AS _tech_valid_from,
  CAST(__END_AT   AS TIMESTAMP) AS _tech_valid_to,
  __END_AT IS NULL              AS _tech_is_current,
  -- AUTO CDC decides versioning from track_history_column_list, so it
  -- never materialises a hash. Recompute it here from the SAME rule-4
  -- attribute list, or the two paths disagree about what a change is.
  sha2(concat_ws('||',
         coalesce(upper(trim(customer_name)),    '~'),
         coalesce(upper(trim(customer_segment)), '~'),
         coalesce(upper(trim(country)),          '~')
       ), 256)                  AS _tech_hash
FROM silver.customer;