SCD Type 2: as-was against as-is, and a late arrival

# /// script
# requires-python = ">=3.10"
# ///
# Copyright (c) 2026 Aayush Sharma · SPDX-License-Identifier: MIT
# https://thedataengineering.com/lab-code/LICENSE.txt
"""SCD type 2: as-was against as-is, and the cost of a late correction.

A customer dimension that keeps history (a new row per version, with a
validity range), facts that point to the version current at the time, a late
fact that must find the right version, and a correction of when the change
happened.
"""
import argparse, json, random, sqlite3

p = argparse.ArgumentParser(description=__doc__.splitlines()[0])
p.add_argument("--facts", type=int, default=10_000, help="Extra daily facts for Lena across the year, for the re-keying count.")
p.add_argument("--seed", type=int, default=0, help="Random seed.")
p.add_argument("--json", help="Also write the results to this JSON file.")
args = p.parse_args()
random.seed(args.seed)

db = sqlite3.connect(":memory:")
db.executescript("""
CREATE TABLE dim_customer (sk INTEGER PRIMARY KEY, customer_id TEXT, name TEXT, region TEXT, valid_from TEXT, valid_to TEXT, is_current INT);
INSERT INTO dim_customer VALUES (1, 'C42', 'Lena', 'West',  '2025-01-01', '2025-07-01', 0),
                                (2, 'C42', 'Lena', 'North', '2025-07-01', '9999-12-31', 1);
CREATE TABLE fact_sales (sale_date TEXT, customer_sk INT, amount INT);
INSERT INTO fact_sales VALUES ('2025-03-10', 1, 100), ('2025-09-15', 2, 200);
""")
q = lambda sql, *a: db.execute(sql, a).fetchall()
as_was = q("SELECT d.region, SUM(f.amount) FROM fact_sales f JOIN dim_customer d ON d.sk = f.customer_sk GROUP BY 1 ORDER BY 1")
as_is = q("""SELECT cur.region, SUM(f.amount) FROM fact_sales f JOIN dim_customer d ON d.sk = f.customer_sk
             JOIN dim_customer cur ON cur.customer_id = d.customer_id AND cur.is_current = 1 GROUP BY 1""")
print("as-was:", as_was, "   as-is:", as_is)

# A late fact dated May, arriving today: find the version valid on its date, not the current one
late_sk = q("SELECT sk FROM dim_customer WHERE customer_id = 'C42' AND ? >= valid_from AND ? < valid_to", "2025-05-20", "2025-05-20")[0][0]
db.execute("INSERT INTO fact_sales VALUES ('2025-05-20', ?, 50)", (late_sk,))
print(f"late fact dated 2025-05-20 gets surrogate key {late_sk} (region {q('SELECT region FROM dim_customer WHERE sk = ?', late_sk)[0][0]})")

# A year of daily facts, keyed correctly, then the correction "moved in April, not July"
for _ in range(args.facts):
    d = f"2025-{random.randint(1, 12):02d}-{random.randint(1, 28):02d}"
    sk = q("SELECT sk FROM dim_customer WHERE customer_id = 'C42' AND ? >= valid_from AND ? < valid_to", d, d)[0][0]
    db.execute("INSERT INTO fact_sales VALUES (?, ?, ?)", (d, sk, random.randint(10, 500)))
rekey = q("SELECT COUNT(*) FROM fact_sales WHERE customer_sk = 1 AND sale_date >= '2025-04-01'")[0][0]
total = q("SELECT COUNT(*) FROM fact_sales")[0][0]
db.execute("UPDATE dim_customer SET valid_to = '2025-04-01' WHERE sk = 1")
db.execute("UPDATE dim_customer SET valid_from = '2025-04-01' WHERE sk = 2")
db.execute("UPDATE fact_sales SET customer_sk = 2 WHERE customer_sk = 1 AND sale_date >= '2025-04-01'")
print(f"correction 'moved in April, not July': {rekey:,} of {total:,} fact rows re-keyed")

if args.json:
    with open(args.json, "w") as f:
        json.dump({"tables": {"queries": [{"view": "as-was", "result": ", ".join(f"{r} {a}" for r, a in as_was)},
                                          {"view": "as-is", "result": ", ".join(f"{r} {a}" for r, a in as_is)}],
                              "summary": [{"late_sk": late_sk, "rekey": rekey, "total": total}]}}, f)