Implement and Test an SCD2 Load Including Repeat Runs, Backdated Changes, Deletes, and Unknown Members

PRAGMA foreign_keys = ON;DROP TABLE IF EXISTS fact_sales_history;DROP TABLE IF EXISTS fact_sales_stage;DROP TABLE IF EXISTS applied_change_event;DROP TABLE IF EXISTS dim_customer_history;CREATE TABLE dim_customer_history (  customer_sk INTEGER PRIMARY KEY AUTOINCREMENT,  durable_customer_id TEXT NOT NULL,  source_customer_id TEXT NOT NULL,  customer_name TEXT NOT NULL,  segment TEXT NOT NULL,  geography_code TEXT NOT NULL,  lifecycle_status TEXT NOT NULL,  original_signup_channel TEXT NOT NULL,  effective_from TEXT NOT NULL,  effective_to TEXT NOT NULL,  is_current INTEGER NOT NULL CHECK (is_current IN (0,1)),  change_hash TEXT NOT NULL,  CHECK (effective_from < effective_to),  UNIQUE (durable_customer_id, effective_from));CREATE UNIQUE INDEX ux_customer_one_current  ON dim_customer_history(durable_customer_id)  WHERE is_current = 1;CREATE TABLE applied_change_event (  event_id TEXT PRIMARY KEY,  durable_customer_id TEXT NOT NULL,  business_effective_from TEXT NOT NULL,  applied_at TEXT NOT NULL);CREATE TABLE fact_sales_stage (  order_id TEXT NOT NULL,  line_no INTEGER NOT NULL,  event_date TEXT NOT NULL,  durable_customer_id TEXT NOT NULL,  product_id TEXT NOT NULL,  quantity INTEGER NOT NULL,  extended_amount NUMERIC NOT NULL,  extended_cost NUMERIC NOT NULL,  PRIMARY KEY(order_id,line_no));CREATE TABLE fact_sales_history (  order_id TEXT NOT NULL,  line_no INTEGER NOT NULL,  event_date TEXT NOT NULL,  durable_customer_id TEXT NOT NULL,  customer_sk INTEGER NOT NULL REFERENCES dim_customer_history(customer_sk),  product_id TEXT NOT NULL,  quantity INTEGER NOT NULL,  extended_amount NUMERIC NOT NULL,  extended_cost NUMERIC NOT NULL,  PRIMARY KEY(order_id,line_no));INSERT INTO fact_sales_stage VALUES('O1001',1,'2026-09-18','D-CUST-001','P100',2,100,60),('O1001',2,'2026-09-18','D-CUST-001','P200',1,25,10),('O1002',1,'2026-09-18','D-CUST-002','P300',1,200,140),('O1003',1,'2026-09-19','D-CUST-001','P400',1,100,60),('O1003',2,'2026-09-19','D-CUST-001','P200',2,50,20),('O1005',1,'2026-09-20','D-CUST-004','P100',1,50,30),('O1005',2,'2026-09-20','D-CUST-004','P400',1,100,60);

import hashlib, json, sqlite3OPEN_END = "9999-12-31"def tracked_hash(customer_name, segment, geography_code, lifecycle_status):    payload = json.dumps({        "customer_name": customer_name,        "segment": segment,        "geography_code": geography_code,        "lifecycle_status": lifecycle_status,    }, sort_keys=True, separators=(",", ":"), ensure_ascii=False)    return hashlib.sha256(payload.encode("utf-8")).hexdigest()def apply_type2(conn, event_id, durable_id, effective_from, updates, applied_at):    if conn.execute("SELECT 1 FROM applied_change_event WHERE event_id=?", (event_id,)).fetchone():        return "replay_ignored"    row = conn.execute("""      SELECT customer_sk, source_customer_id, customer_name, segment,             geography_code, lifecycle_status, original_signup_channel,             effective_from, effective_to, is_current, change_hash      FROM dim_customer_history      WHERE durable_customer_id=?        AND effective_from <= ? AND ? < effective_to      ORDER BY effective_from DESC      LIMIT 1    """, (durable_id, effective_from, effective_from)).fetchone()    if not row:        raise ValueError(f"No covering row for {durable_id} at {effective_from}")    sk, source_id, name, segment, geo, status, signup, old_from, old_to, was_current, old_hash = row    new_name = updates.get("customer_name", name)    new_segment = updates.get("segment", segment)    new_geo = updates.get("geography_code", geo)    new_status = updates.get("lifecycle_status", status)    new_hash = tracked_hash(new_name, new_segment, new_geo, new_status)    if new_hash == old_hash:        conn.execute("INSERT INTO applied_change_event VALUES (?,?,?,?)",                     (event_id, durable_id, effective_from, applied_at))        return "no_semantic_change"    if effective_from == old_from:        # Course policy: same-boundary arrival is correction of this interval.        conn.execute("""UPDATE dim_customer_history                        SET customer_name=?, segment=?, geography_code=?,                            lifecycle_status=?, change_hash=?                        WHERE customer_sk=?""",                     (new_name, new_segment, new_geo, new_status, new_hash, sk))        conn.execute("INSERT INTO applied_change_event VALUES (?,?,?,?)",                     (event_id, durable_id, effective_from, applied_at))        return "corrected_existing_interval"    conn.execute("UPDATE dim_customer_history SET effective_to=?, is_current=0 WHERE customer_sk=?",                 (effective_from, sk))    conn.execute("""INSERT INTO dim_customer_history      (durable_customer_id,source_customer_id,customer_name,segment,geography_code,       lifecycle_status,original_signup_channel,effective_from,effective_to,is_current,change_hash)      VALUES (?,?,?,?,?,?,?,?,?,?,?)""",      (durable_id, source_id, new_name, new_segment, new_geo, new_status,       signup, effective_from, old_to, was_current, new_hash))    conn.execute("INSERT INTO applied_change_event VALUES (?,?,?,?)",                 (event_id, durable_id, effective_from, applied_at))    return "new_version"def resolve_facts(conn):    conn.execute("DELETE FROM fact_sales_history")    for row in conn.execute("""SELECT order_id,line_no,event_date,durable_customer_id,                                      product_id,quantity,extended_amount,extended_cost                               FROM fact_sales_stage ORDER BY order_id,line_no"""):        order_id,line_no,event_date,durable,product,qty,amount,cost=row        hit=conn.execute("""SELECT customer_sk FROM dim_customer_history                            WHERE durable_customer_id=?                              AND effective_from<=? AND ?<effective_to                            ORDER BY effective_from DESC LIMIT 1""",                         (durable,event_date,event_date)).fetchone()        if hit:            customer_sk=hit[0]        else:            customer_sk=conn.execute("""SELECT customer_sk FROM dim_customer_history                                         WHERE durable_customer_id='D-UNKNOWN'""").fetchone()[0]        conn.execute("INSERT INTO fact_sales_history VALUES (?,?,?,?,?,?,?,?,?)",                     (order_id,line_no,event_date,durable,customer_sk,product,qty,amount,cost))

import sqlite3# Assumes schema, seed rows, tracked_hash(), apply_type2(), and resolve_facts() are loaded.# 1. Apply changes and prove exact-event replay is idempotent.assert apply_type2(conn,'EVT-C001-SEG-20260919','D-CUST-001','2026-09-19',{'segment':'Mid-Market'},now) == 'new_version'assert apply_type2(conn,'EVT-C002-GEO-20260910','D-CUST-002','2026-09-10',{'geography_code':'G-EAST'},now) == 'new_version'assert apply_type2(conn,'EVT-C004-INACTIVE-20260921','D-CUST-004','2026-09-21',{'lifecycle_status':'inactive'},now) == 'new_version'rows_before = conn.execute('SELECT COUNT(*) FROM dim_customer_history').fetchone()[0]assert apply_type2(conn,'EVT-C001-SEG-20260919','D-CUST-001','2026-09-19',{'segment':'Mid-Market'},now) == 'replay_ignored'assert conn.execute('SELECT COUNT(*) FROM dim_customer_history').fetchone()[0] == rows_before# 2. Exactly one current row per durable identity.bad_current = conn.execute('''SELECT durable_customer_id, SUM(is_current)                              FROM dim_customer_history                              GROUP BY durable_customer_id                              HAVING SUM(is_current) <> 1''').fetchall()assert bad_current == []# 3. No overlapping Type 2 intervals.overlaps = conn.execute('''SELECT a.durable_customer_id,a.customer_sk,b.customer_sk                           FROM dim_customer_history a                           JOIN dim_customer_history b                             ON a.durable_customer_id=b.durable_customer_id                            AND a.customer_sk < b.customer_sk                            AND a.effective_from < b.effective_to                            AND b.effective_from < a.effective_to''').fetchall()assert overlaps == []# 4. Resolve accepted sales facts by business event date.resolve_facts(conn)control = conn.execute('''SELECT COUNT(*),COUNT(DISTINCT order_id),SUM(quantity),                                 SUM(extended_amount),SUM(extended_cost),                                 SUM(extended_amount-extended_cost)                          FROM fact_sales_history''').fetchone()assert control == (7,4,9,625,380,245), control# 5. Historical/as-was segment attribution.as_was = dict(conn.execute('''SELECT d.segment,SUM(f.extended_amount)                              FROM fact_sales_history f                              JOIN dim_customer_history d ON d.customer_sk=f.customer_sk                              GROUP BY d.segment''').fetchall())assert as_was == {'Consumer':350,'Mid-Market':150,'SMB':125}, as_was# 6. Current/as-is segment attribution uses durable key + current row.as_is = dict(conn.execute('''SELECT d.segment,SUM(f.extended_amount)                             FROM fact_sales_history f                             JOIN dim_customer_history d                               ON d.durable_customer_id=f.durable_customer_id                              AND d.is_current=1                             GROUP BY d.segment''').fetchall())assert as_is == {'Consumer':350,'Mid-Market':275}, as_is# 7. Backdated C002 geography applies to its 18 September sale.c002_geo = conn.execute('''SELECT d.geography_code                           FROM fact_sales_history f                           JOIN dim_customer_history d ON d.customer_sk=f.customer_sk                           WHERE f.order_id='O1002' AND f.line_no=1''').fetchone()[0]assert c002_geo == 'G-EAST'# 8. C004 was still active on its 20 September sale despite current inactivation on the 21st.c004_status = conn.execute('''SELECT d.lifecycle_status                              FROM fact_sales_history f                              JOIN dim_customer_history d ON d.customer_sk=f.customer_sk                              WHERE f.order_id='O1005' AND f.line_no=1''').fetchone()[0]assert c004_status == 'active'current_c004 = conn.execute('''SELECT lifecycle_status FROM dim_customer_history                              WHERE durable_customer_id='D-CUST-004' AND is_current=1''').fetchone()[0]assert current_c004 == 'inactive'# 9. Unknown-member fallback exists and is current.unknown = conn.execute("SELECT customer_sk FROM dim_customer_history WHERE durable_customer_id='D-UNKNOWN' AND is_current=1").fetchone()assert unknown is not Noneprint('control=',control)print('as_was=',as_was)print('as_is=',as_is)print('Chapter 07 acceptance: PASS')