This exercise builds Stage 1 and Stage 2 of the capstone pipeline: star schema design with SCD Type 2 and the Snowflake-style load pipeline. You will create the fact and dimension tables at the correct grain, populate a player dimension with SCD Type 2 tracking for a mid-season team change, implement a COPY INTO simulation with load history idempotency, and implement a Stream/Task-style incremental MERGE that processes only new delivery records on each run.
Analogy🏏Cricket
🏏 Think of it like cricket: OLTP is the IPL's live ticketing counter — it handles thousands of simultaneous seat reservations, each requiring a precise single-seat record update with immediate confirmation. Speed per transaction and data consistency under concurrent updates are everything. OLAP is the IPL's season statistics department — it runs complex analytical queries across every ball bowled in every match of every season to produce the published rankings, economy rates, and historical comparisons. No one books a seat through the statistics department, and no broadcaster calls the ticketing counter for Bumrah's career economy rate. The two workloads demand completely different systems. Just as the ticketing counter is built for speed and correctness on one seat at a time and would buckle if asked to tally a decade of attendance mid-sale, an OLTP row-store excels at single-record writes but chokes on full-table aggregation; and just as the statistics department pores over millions of past deliveries but would be hopeless at booking a live seat under contention, the OLAP columnar engine sweeps billions of rows yet is the wrong tool for a fast single-row update. The physical design of each — row-oriented for the counter, columnar for the stats desk — is what makes it superb at its own job and unfit for the other's.
The exercise reuses patterns from Module 1 (star schema, SCD Type 2) and Module 2 (COPY INTO idempotency, Streams and Tasks) and integrates them into a single coherent pipeline. Completing all assertions in this exercise produces the fact and dimension tables that the Lesson 33 dbt model hierarchy will transform, and the SCD Type 2 player dimension that the Lesson 34 RLS dashboard will join against for franchise-level filtering.
Step 1 — Star Schema with SCD Type 2 Dimensions
Create the star schema: `fact_ipl_delivery` at delivery grain with surrogate foreign keys, `dim_player` with SCD Type 2 tracking for team changes, and `dim_match`, `dim_venue`, `dim_team` dimensions. Populate `dim_player` with an initial load, apply a mid-season team change for one player, and verify the SCD Type 2 history shows two versions with correct valid_from/valid_to boundaries. Assert that a point-in-time query correctly returns the team a player was on for a delivery date before the change.
Analogy🏏Cricket
🏏 Think of it like cricket: The star schema is the franchise's match-data model — fact_ipl_delivery at one-row-per-delivery grain sits at the centre, surrounded by dim_player, dim_match, dim_venue and dim_team like a scorecard linking each ball to its bowler, fixture, ground and side. dim_player uses SCD Type 2 so a team change is versioned, not overwritten: just as Hardik Pandya's GT spell is closed with an end date and a fresh MI registration is opened, the old player_key gets valid_to stamped and is_current cleared while a new surrogate key is minted. The point-in-time lookup is the crucial check: just as a statistician crediting a 2023 delivery must attribute it to the GT Hardik, not today's MI Hardik, get_player_key_at_date returns the GT version for a 2023 date and the MI version for 2024. Storing that version-specific bowler_key on each fact row freezes the attribution forever. The payoff: deliveries stay linked to the team a player actually bowled for, so historical stats never shift under later transfers.
python
# capstone_star_schema.py — Step 1: Star schema with SCD Type 2
import duckdb
import pandas as pd
import numpy as np
from datetime import date, datetime, timezone
np.random.seed(42)
con = duckdb.connect(":memory:")
# ── Dimension: dim_player (SCD Type 2) ────────────────────────────────────────
con.execute("""
CREATE TABLE dim_player (
player_key INTEGER PRIMARY KEY,
player_id VARCHAR(20) NOT NULL,
player_name VARCHAR(80) NOT NULL,
current_team VARCHAR(4) NOT NULL,
bowling_style VARCHAR(40),
valid_from DATE NOT NULL,
valid_to DATE,
is_current BOOLEAN NOT NULL DEFAULT TRUE
);
INSERT INTO dim_player VALUES
(1001,'IPL-HP-2015','Hardik Pandya','GT','Right-arm medium-fast','2022-01-01',NULL,TRUE),
(1002,'IPL-BUJ-2008','Jasprit Bumrah','MI','Right-arm fast','2016-01-01',NULL,TRUE),
(1003,'IPL-MSH-2009','Mohammed Shami','PBKS','Right-arm fast-medium','2013-01-01',NULL,TRUE);
""")
# Mid-season change: Hardik moves to MI on 2024-01-01
con.execute("UPDATE dim_player SET valid_to='2024-01-01', is_current=FALSE WHERE player_key=1001")
con.execute("INSERT INTO dim_player VALUES (1004,'IPL-HP-2015','Hardik Pandya','MI','Right-arm medium-fast','2024-01-01',NULL,TRUE)")
# ── Other dimensions ───────────────────────────────────────────────────────────
con.execute("""
CREATE TABLE dim_match (
match_key INTEGER PRIMARY KEY, match_id INTEGER, season SMALLINT, match_date DATE
);
INSERT INTO dim_match VALUES (2001,10001,2024,'2024-04-20'), (2002,10002,2023,'2023-04-15');
CREATE TABLE dim_venue (
venue_key INTEGER PRIMARY KEY, venue_name VARCHAR(80), city VARCHAR(40)
);
INSERT INTO dim_venue VALUES (3001,'Wankhede','Mumbai'), (3002,'Chinnaswamy','Bengaluru');
CREATE TABLE dim_team (
team_key INTEGER PRIMARY KEY, team_code VARCHAR(4), team_name VARCHAR(60)
);
INSERT INTO dim_team VALUES (4001,'MI','Mumbai Indians'), (4002,'GT','Gujarat Titans');
""")
# ── Fact table: grain = one delivery ─────────────────────────────────────────
con.execute("""
CREATE TABLE fact_ipl_delivery (
delivery_key BIGINT PRIMARY KEY,
match_key INTEGER NOT NULL,
venue_key INTEGER NOT NULL,
bowler_key INTEGER NOT NULL, -- SCD Type 2 surrogate (version-specific)
batting_team_key INTEGER NOT NULL,
over_number SMALLINT NOT NULL,
phase VARCHAR(12) NOT NULL,
runs_scored SMALLINT NOT NULL,
is_wicket BOOLEAN NOT NULL
);
""")
# ── Point-in-time SCD Type 2 lookup ───────────────────────────────────────────
def get_player_key_at_date(con, player_id: str, event_date: str) -> int:
result = con.execute("""
SELECT player_key FROM dim_player
WHERE player_id = ?
AND valid_from <= ?
AND (valid_to IS NULL OR valid_to > ?)
""", [player_id, event_date, event_date]).fetchone()
return result[0] if result else None
# Verify point-in-time correctness
key_2023 = get_player_key_at_date(con, "IPL-HP-2015", "2023-06-01") # before change: GT
key_2024 = get_player_key_at_date(con, "IPL-HP-2015", "2024-04-20") # after change: MI
assert key_2023 == 1001, f"Expected GT version (1001), got {key_2023}"
assert key_2024 == 1004, f"Expected MI version (1004), got {key_2024}"
team_2023 = con.execute("SELECT current_team FROM dim_player WHERE player_key=?", [key_2023]).fetchone()[0]
team_2024 = con.execute("SELECT current_team FROM dim_player WHERE player_key=?", [key_2024]).fetchone()[0]
print(f"Point-in-time SCD Type 2 lookup:")
print(f" Hardik's team on 2023-06-01: {team_2023} ✓")
print(f" Hardik's team on 2024-04-20: {team_2024} ✓")
# Populate fact table using correct point-in-time keys
N = 120
bowler_key_2024 = key_2024 # use the current (MI) version for 2024 matches
deliveries = pd.DataFrame({
"delivery_key": range(1, N+1),
"match_key": 2001,
"venue_key": 3001,
"bowler_key": bowler_key_2024,
"batting_team_key": 4001,
"over_number": [(i//6)+1 for i in range(N)],
"phase": ["powerplay" if (i//6)+1<=6 else "middle" if (i//6)+1<=15 else "death" for i in range(N)],
"runs_scored": np.random.choice([0,1,2,4,6], N),
"is_wicket": (np.random.random(N)<0.05),
})
con.register("_deliveries", deliveries)
con.execute("INSERT INTO fact_ipl_delivery SELECT * FROM _deliveries")
fact_count = con.execute("SELECT COUNT(*) FROM fact_ipl_delivery").fetchone()[0]
assert fact_count == N
print(f" Fact table populated: {fact_count} deliveries ✓")
print("Step 1 ✓: Star schema with SCD Type 2 dimensions complete")Step 2 — Snowflake-Style COPY INTO and Stream/Task MERGE
Implement a COPY INTO simulation with load history tracking for a second batch of deliveries, verify idempotency by attempting to re-load the same file, then implement a Stream/Task-style incremental MERGE that processes only the newly staged rows and writes them into the fact table. Run the Task twice — the second run should process zero new rows — and verify the final fact table row count equals the sum of both delivery batches with no duplicates.
Analogy🏏Cricket
🏏 Think of it like cricket: This step is the official scorecard-ingestion desk. COPY INTO with load history is the receipt ledger: just as a clerk logs each submitted scorecard file and rejects a duplicate submission of match_10002's card, the second load of the same file returns SKIPPED and adds zero rows, so the 120-delivery batch is never counted twice. The Stream/Task MERGE is the overnight statistician who updates the season record: just as they process only the deliveries logged since the last run and merge them into the master fact table, the Task reads only rows past the stream offset and inserts them, advancing the marker afterward. Running the Task a second time with nothing new is like the statistician arriving to find no fresh scorecards — zero rows processed. Asserting the final fact table holds exactly 240 deliveries with no duplicates is the reconciliation confirming both batches landed once and only once. The payoff: exactly-once incremental loading you can trust.
python
# capstone_star_schema.py — Step 2: COPY INTO and Stream/Task MERGE
import duckdb
import pandas as pd
import numpy as np
from datetime import datetime, timezone
np.random.seed(43)
# ── Staging table and load history ────────────────────────────────────────────
con.execute("""
CREATE TABLE staging_deliveries (
delivery_key BIGINT, match_key INTEGER, venue_key INTEGER, bowler_key INTEGER,
batting_team_key INTEGER, over_number SMALLINT, phase VARCHAR, runs_scored SMALLINT,
is_wicket BOOLEAN, _source_file VARCHAR, _loaded_at TIMESTAMPTZ
);
CREATE TABLE copy_load_history (
file_path VARCHAR PRIMARY KEY, rows_loaded INTEGER, loaded_at TIMESTAMPTZ
);
""")
def copy_into(con, file_path: str, df: pd.DataFrame) -> dict:
existing = con.execute("SELECT 1 FROM copy_load_history WHERE file_path=?", [file_path]).fetchone()
if existing:
return {"status": "SKIPPED", "rows": 0}
df2 = df.copy()
df2["_source_file"] = file_path
df2["_loaded_at"] = datetime.now(timezone.utc).isoformat()
con.register("_b", df2)
con.execute("INSERT INTO staging_deliveries SELECT * FROM _b")
con.execute("INSERT INTO copy_load_history VALUES (?,?,?)",
[file_path, len(df2), datetime.now(timezone.utc).isoformat()])
return {"status": "LOADED", "rows": len(df2)}
# Second batch: match_key 2002 (2023 match)
batch2 = pd.DataFrame({
"delivery_key": range(121, 241),
"match_key": 2002,
"venue_key": 3002,
"bowler_key": 1001, # Hardik's GT version (2023 match)
"batting_team_key": 4002,
"over_number": [(i//6)+1 for i in range(120)],
"phase": ["powerplay" if (i//6)+1<=6 else "middle" if (i//6)+1<=15 else "death" for i in range(120)],
"runs_scored": np.random.choice([0,1,2,4,6], 120),
"is_wicket": (np.random.random(120)<0.05),
})
r1 = copy_into(con, "scorecards/match_10002.parquet", batch2)
r2 = copy_into(con, "scorecards/match_10002.parquet", batch2) # idempotency
assert r1["status"] == "LOADED" and r1["rows"] == 120
assert r2["status"] == "SKIPPED"
print(f" COPY INTO: load1={r1['status']}({r1['rows']}), load2={r2['status']} ✓")
# ── Stream/Task incremental MERGE ─────────────────────────────────────────────
stream_offset = [0]
def stream_has_data() -> bool:
total = con.execute("SELECT COUNT(*) FROM staging_deliveries").fetchone()[0]
return stream_offset[0] < total
def task_merge() -> int:
if not stream_has_data():
return 0
total = con.execute("SELECT COUNT(*) FROM staging_deliveries").fetchone()[0]
batch = con.execute(f"SELECT * FROM staging_deliveries LIMIT -1 OFFSET {stream_offset[0]}").df()
con.register("_stream_batch", batch)
con.execute("""
INSERT OR IGNORE INTO fact_ipl_delivery
SELECT delivery_key, match_key, venue_key, bowler_key, batting_team_key,
over_number, phase, runs_scored, is_wicket
FROM _stream_batch
""")
rows = len(batch)
stream_offset[0] = total
return rows
processed1 = task_merge()
assert processed1 == 120
processed2 = task_merge() # no new data
assert processed2 == 0
final_count = con.execute("SELECT COUNT(*) FROM fact_ipl_delivery").fetchone()[0]
assert final_count == 240, f"Expected 240, got {final_count}"
print(f" Task run 1: {processed1} rows merged ✓")
print(f" Task run 2: {processed2} rows (no new data) ✓")
print(f" Final fact table: {final_count} deliveries (120+120) ✓")
print("Step 2 ✓: COPY INTO and Stream/Task MERGE complete")