Data Modeling Guide — Kimball, Inmon, Star Schema, and Slowly Changing Dimensions

-- Star schema for sales analytics
CREATE TABLE dim_customer (
    customer_sk INTEGER PRIMARY KEY,
    customer_id TEXT NOT NULL,
    full_name TEXT,
    email TEXT,
    city TEXT,
    state TEXT,
    segment TEXT,
    effective_date DATE,
    end_date DATE,
    is_current BOOLEAN DEFAULT TRUE
);

CREATE TABLE dim_product (
    product_sk INTEGER PRIMARY KEY,
    product_id TEXT,
    product_name TEXT,
    category TEXT,
    subcategory TEXT,
    unit_price DECIMAL(10,2)
);

CREATE TABLE dim_date (
    date_sk INTEGER PRIMARY KEY,
    full_date DATE,
    year INTEGER,
    quarter INTEGER,
    month INTEGER,
    day_of_week INTEGER,
    is_holiday BOOLEAN
);

CREATE TABLE fact_sales (
    sales_sk INTEGER PRIMARY KEY,
    customer_sk INTEGER REFERENCES dim_customer(customer_sk),
    product_sk INTEGER REFERENCES dim_product(product_sk),
    date_sk INTEGER REFERENCES dim_date(date_sk),
    quantity INTEGER,
    unit_price DECIMAL(10,2),
    discount DECIMAL(5,2),
    total_amount DECIMAL(12,2)
);

-- SCD Type 1: Update in place
UPDATE dim_customer
SET city = 'New York', state = 'NY'
WHERE customer_id = 'CUST001';

-- SCD Type 2: Expire old, insert new
UPDATE dim_customer
SET end_date = '2026-06-15',
    is_current = FALSE
WHERE customer_id = 'CUST001'
  AND is_current = TRUE;

INSERT INTO dim_customer
(customer_sk, customer_id, full_name, city, state,
 effective_date, end_date, is_current)
VALUES
(1002, 'CUST001', 'Alice', 'New York', 'NY',
 '2026-06-15', NULL, TRUE);

-- SCD Type 3: Add previous value columns
ALTER TABLE dim_customer
ADD COLUMN previous_city TEXT,
ADD COLUMN previous_state TEXT;

UPDATE dim_customer
SET previous_city = city,
    previous_state = state,
    city = 'New York',
    state = 'NY'
WHERE customer_id = 'CUST001'
  AND is_current = TRUE;

-- Data Vault example
CREATE TABLE hub_customer (
    customer_hk CHAR(32) PRIMARY KEY,
    customer_id TEXT NOT NULL,
    load_date TIMESTAMP,
    record_source TEXT
);

CREATE TABLE sat_customer_detail (
    customer_hk CHAR(32) REFERENCES hub_customer(customer_hk),
    hash_diff CHAR(32),
    full_name TEXT,
    email TEXT,
    city TEXT,
    state TEXT,
    effective_date TIMESTAMP,
    load_date TIMESTAMP,
    PRIMARY KEY (customer_hk, effective_date)
);

CREATE TABLE link_customer_order (
    link_hk CHAR(32) PRIMARY KEY,
    customer_hk CHAR(32) REFERENCES hub_customer(customer_hk),
    order_hk CHAR(32) REFERENCES hub_order(order_hk),
    load_date TIMESTAMP
);

# scd_type2_demo.py
# Implement SCD Type 2 for a customer dimension
from datetime import date, datetime
import json

class CustomerDim:
    def __init__(self):
        self.records = []
        self.next_sk = 1

    def get_current(self, customer_id):
        for r in self.records:
            if r["customer_id"] == customer_id and r["is_current"]:
                return r
        return None

    def upsert(self, customer_id, name, city, state, change_date):
        current = self.get_current(customer_id)

        if current and current["city"] == city and current["state"] == state:
            return

        # Expire current record
        if current:
            current["end_date"] = change_date
            current["is_current"] = False

        # Insert new record
        new_record = {
            "customer_sk": self.next_sk,
            "customer_id": customer_id,
            "full_name": name,
            "city": city,
            "state": state,
            "effective_date": change_date,
            "end_date": None,
            "is_current": True,
        }
        self.records.append(new_record)
        self.next_sk += 1
        print(f"[SCD2] {customer_id}: {'updated' if current else 'created'} — {city}, {state}")

    def as_of(self, query_date):
        """Return records as they were on a specific date."""
        result = []
        for r in self.records:
            eff = r["effective_date"]
            end = r["end_date"] or date(9999, 12, 31)
            if eff <= query_date <= end:
                result.append(r)
        return result

dim = CustomerDim()

# Initial load
dim.upsert("C001", "Alice", "Boston", "MA", date(2020, 1, 1))
dim.upsert("C002", "Bob", "Chicago", "IL", date(2020, 1, 1))
dim.upsert("C003", "Charlie", "Seattle", "WA", date(2020, 6, 1))

# Alice moves to New York
dim.upsert("C001", "Alice", "New York", "NY", date(2023, 3, 15))

# Bob moves to San Francisco
dim.upsert("C002", "Bob", "San Francisco", "CA", date(2024, 7, 1))

# Query history
print("\n=== Alice's History ===")
for r in dim.records:
    if r["customer_id"] == "C001":
        print(f"  {r['effective_date']} to {r['end_date'] or 'now'}: {r['city']}, {r['state']}")

print(f"\n=== Customers as of 2022-01-01 ===")
for r in dim.as_of(date(2022, 1, 1)):
    print(f"  {r['full_name']:<10} {r['city']:<15} {r['state']}")

print(f"\n=== Current Customers ===")
for r in dim.records:
    if r["is_current"]:
        print(f"  {r['full_name']:<10} {r['city']:<15} {r['state']}")

# scd_loader.py
# Generic SCD Type 2 loader for any dimension
from datetime import date
import hashlib

class SCD2Loader:
    def __init__(self, dim_name, business_key):
        self.dim_name = dim_name
        self.business_key = business_key
        self.current_state = {}

    def hash_record(self, record):
        """Hash all attributes except business key for change detection."""
        relevant = {k: v for k, v in sorted(record.items()) if k != self.business_key}
        return hashlib.md5(str(relevant).encode()).hexdigest()[:16]

    def process(self, records, change_date):
        changes = 0
        for record in records:
            bk = record[self.business_key]
            new_hash = self.hash_record(record)

            if bk in self.current_state:
                if self.current_state[bk]["hash"] == new_hash:
                    continue

            old_record = self.current_state.get(bk)
            if old_record:
                print(f"  [{self.dim_name}] {bk}: changed {old_record['record']} → {record}")

            self.current_state[bk] = {
                "hash": new_hash,
                "record": record,
                "effective_date": change_date,
            }
            changes += 1

        return changes

loader = SCD2Loader("dim_customer", "customer_id")

print("=== Day 1: Initial Load ===")
day1 = [
    {"customer_id": "C001", "name": "Alice", "tier": "basic"},
    {"customer_id": "C002", "name": "Bob", "tier": "premium"},
]
changes = loader.process(day1, date(2026, 1, 1))
print(f"Changes: {changes}")

print("\n=== Day 90: Alice upgrades ===")
day2 = [
    {"customer_id": "C001", "name": "Alice", "tier": "premium"},
    {"customer_id": "C002", "name": "Bob", "tier": "premium"},
]
changes = loader.process(day2, date(2026, 4, 1))
print(f"Changes: {changes}")

print("\n=== Day 180: Bob downgrades ===")
day3 = [
    {"customer_id": "C001", "name": "Alice", "tier": "premium"},
    {"customer_id": "C002", "name": "Bob", "tier": "basic"},
]
changes = loader.process(day3, date(2026, 7, 1))
print(f"Changes: {changes}")

print(f"\nFinal state: {len(loader.current_state)} customers")