02Data model

this warehouse, as a star        the same domain, as a Data Vault
------------------------------   -------------------------------------
gold.dim_customer   (SCD2)       hub_customer + sat_customer_sap
                                              + sat_customer_shop
gold.dim_product    (SCD2)       hub_product  + sat_product
gold.fct_order_line              hub_sales_order + lnk_order_line
                                              + sat_order_line
gold.agg_revenue_month           still to be built, on top of all of it

-- gold.dim_customer
sha2(concat_ws('||', customer_id), 256)              AS customer_sk

-- gold.fct_order_line
sha2(concat_ws('||', order_id, line_number), 256)    AS order_line_sk

-- Shorthand: both lines omit the normalisation. Nothing in this warehouse
-- builds a key from them. make_sk() below is the executable form, and a key
-- hashed without trim/upper/coalesce does not equal the one already stored
-- in the fact — the join returns nothing and raises nothing.

SELECT sha2(concat('4711', '2'), 256)
     = sha2(concat('471', '12'), 256)  AS collides;
-- true — both hash the string '47112'.
--
-- Order 4711 line 2 and order 471 line 12 now share one order_line_sk.
-- Nothing errors. The dimension join fans out and one line's revenue
-- is attributed to the other.

from pyspark.sql import Column, functions as F

SEPARATOR = "||"
NULL_SENTINEL = "~"


def make_sk(*columns: str) -> Column:
    """Deterministic surrogate key: trim, upper, coalesce, join, sha2-256."""
    parts = [
        F.coalesce(F.upper(F.trim(F.col(c).cast("string"))), F.lit(NULL_SENTINEL))
        for c in columns
    ]
    return F.sha2(F.concat_ws(SEPARATOR, *parts), 256)

-- every column of a primary key must ALREADY be NOT NULL,
-- or the ADD CONSTRAINT is rejected before it declares anything
ALTER TABLE gold.dim_customer ALTER COLUMN customer_sk      SET NOT NULL;
ALTER TABLE gold.dim_customer ALTER COLUMN _tech_valid_from SET NOT NULL;

-- dim_customer is SCD2, so the key is the VERSION, not the customer
ALTER TABLE gold.dim_customer
  ADD CONSTRAINT pk_dim_customer
  PRIMARY KEY (customer_sk, _tech_valid_from);

-- now insert the same version twice
INSERT INTO gold.dim_customer
  (customer_sk, customer_id, customer_name, customer_segment, country,
   _tech_valid_from, _tech_valid_to, _tech_is_current, _tech_hash)
VALUES
  ('a1b2', 'C-4711', 'Muster GmbH', 'A', 'DE',
   TIMESTAMP '2026-07-01 00:00:00', NULL, true, 'h1'),
  ('a1b2', 'C-4711', 'Muster GmbH', 'B', 'DE',
   TIMESTAMP '2026-07-01 00:00:00', NULL, true, 'h2');

-- 2 rows inserted. No error. The primary key is still declared.

ALTER TABLE gold.dim_customer
  ADD CONSTRAINT country_domain
  CHECK (country IN ('DE', 'AT', 'CH', 'NL'));

ALTER TABLE gold.fct_order_line
  ADD CONSTRAINT amount_positive
  CHECK (status = 'CANCELLED' OR amount_eur > 0);

ALTER TABLE gold.fct_order_line
  ALTER COLUMN order_line_sk SET NOT NULL;

-- not a business rule: a NULL _tech_hash makes the SCD2 MERGE's
-- tgt._tech_hash <> src._tech_hash comparison NULL, so that customer
-- silently stops versioning. NOT NULL is the cheapest place to stop it.
ALTER TABLE gold.dim_customer
  ALTER COLUMN _tech_hash SET NOT NULL;

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.

UPDATE gold.dim_customer d
   SET _tech_valid_to   = :as_of,
       _tech_is_current = false
 WHERE d._tech_is_current
   -- the seeded fallback member is in no extract, ever. Without this line it
   -- is the first row every deletion pass closes.
   AND d.customer_sk <> 'UNKNOWN'
   -- NOT EXISTS, not NOT IN: one NULL customer_sk in the staging table makes
   -- NOT IN evaluate to NULL for every row, and the pass closes nothing.
   AND NOT EXISTS (SELECT 1
                   FROM   stg_customer s
                   WHERE  s.customer_sk = d.customer_sk);

SELECT raise_error(
         'sap_kna1 shrank by more than 10 % — refusing to apply deletions')
FROM   (SELECT count(*) AS staged    FROM stg_customer)   s,
       (SELECT count(*) AS live_rows FROM gold.dim_customer
         WHERE _tech_is_current AND customer_sk <> 'UNKNOWN')  t
WHERE  s.staged < t.live_rows * 0.9;
-- the alias is live_rows, not current: CURRENT_DATE / CURRENT_USER and
-- friends make a bare 'current' a word you do not want to be arguing with.

SELECT l.order_id,
       l.line_number,
       coalesce(d.customer_sk, 'UNKNOWN') AS customer_sk,
       o.order_ts
FROM       silver.sales_order_line l
JOIN       silver.sales_order      o ON o.order_id = l.order_id
LEFT JOIN  gold.dim_customer       d
       -- same recipe as make_sk(), or the join silently matches nothing
       ON  d.customer_sk = sha2(
             concat_ws('||', coalesce(upper(trim(o.customer_id)), '~')), 256)
      AND  o.order_ts >= d._tech_valid_from
      AND (o.order_ts <  d._tech_valid_to OR d._tech_valid_to IS NULL);

SELECT date_sk,
       count(*)        AS unresolved_lines,
       sum(amount_eur) AS revenue_at_risk
FROM   gold.fct_order_line
WHERE  customer_sk = 'UNKNOWN'
  AND  date_sk >= current_date() - INTERVAL 7 DAYS
GROUP BY date_sk
ORDER BY date_sk;