This exercise builds a complete Snowflake-style data pipeline for the IPL analytics platform using pure Python and DuckDB to simulate Snowflake's key patterns: COPY INTO batch loading with load history tracking, a Stream-style CDC mechanism that captures only new rows for incremental processing, a Task-style scheduled MERGE into the fact table, Time Travel recovery simulation, and Zero-Copy Clone simulation for a staging environment. All patterns are implemented with the same logic that applies to production Snowflake, enabling full validation without any Snowflake account or credits.
The exercise covers three stages. Stage 1 simulates COPY INTO: write three delivery batches, verify each file loads exactly once (idempotency), and assert the total row count. Stage 2 simulates Streams and Tasks: attach a Stream to the staging table, run two Task cycles consuming only new rows via MERGE, and verify the fact table equals the staging count. Stage 3 simulates Time Travel recovery and Zero-Copy Clone: recover from an accidental delete using the snapshot, and create a clone for migration testing.
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.
Step 1 — COPY INTO with Load History and Streams + Tasks
Implement the COPY INTO simulation: generate three batches of delivery files, write each to the staging table with load history tracking, verify that re-running the load for the same file is blocked by the load history check (idempotency), and assert the total staging row count equals 360 (3 files × 120 rows). Then implement the Stream + Task CDC cycle: consume the stream in two task runs, merge into the fact table, and verify the fact table equals the staging table after both runs.
Analogy🏏Cricket
🏏 Think of it like cricket: COPY INTO with load history is the ground's official scorecard-receipt ledger. Just as a match clerk logs each submitted scorecard file and refuses to process the same file twice — so re-sending match_10001's card changes nothing — the pipeline records every loaded file and blocks a re-run, which is why loading three 120-delivery files lands exactly 360 rows and a second attempt is SKIPPED. The Stream is then like the ledger's 'new since last read' marker, and the Task is the overnight statistician who reads only the deliveries logged since yesterday and merges them into the official season fact table. Just as running that overnight job a second time with nothing new added produces zero updates, the second Task run processes zero rows because the stream offset has already advanced past every staged delivery. Confirming the fact table count equals the staging count is the same reconciliation a scorer does matching the official record against the raw submissions. The payoff: exactly-once loading with no duplicated or missed deliveries.
python
# exercise_snowflake_pipeline.py — Step 1: COPY INTO and Streams + Tasks
import duckdb
import pandas as pd
import numpy as np
from datetime import datetime, timezone
np.random.seed(42)
con = duckdb.connect(":memory:")
# ── Tables ────────────────────────────────────────────────────────────────────
con.execute("""
CREATE TABLE staging_deliveries (
delivery_id BIGINT, match_id INTEGER, over_number INTEGER,
runs_scored INTEGER, is_wicket BOOLEAN, bowler VARCHAR, batter VARCHAR,
_source_file VARCHAR, _loaded_at TIMESTAMPTZ
);
CREATE TABLE fact_delivery (
delivery_id BIGINT PRIMARY KEY,
match_id INTEGER, over_number INTEGER, phase VARCHAR,
runs_scored INTEGER, is_wicket BOOLEAN, bowler VARCHAR, batter VARCHAR,
is_boundary BOOLEAN
);
CREATE TABLE copy_load_history (
file_path VARCHAR PRIMARY KEY,
rows_loaded INTEGER, loaded_at TIMESTAMPTZ, status VARCHAR
);
""")
# ── COPY INTO simulation ──────────────────────────────────────────────────────
def copy_into(
con,
file_path: str,
df: pd.DataFrame,
force: bool = False,
) -> dict:
# Check load history (idempotency)
existing = con.execute(
"SELECT file_path FROM copy_load_history WHERE file_path = ?", [file_path]
).fetchone()
if existing and not force:
return {"status": "SKIPPED", "reason": "already_loaded", "rows": 0}
df2 = df.copy()
df2["_source_file"] = file_path
df2["_loaded_at"] = datetime.now(timezone.utc).isoformat()
con.register("_batch", df2)
con.execute("INSERT INTO staging_deliveries SELECT * FROM _batch")
con.execute(
"INSERT OR REPLACE INTO copy_load_history VALUES (?, ?, ?, ?)",
[file_path, len(df2), datetime.now(timezone.utc).isoformat(), "LOADED"]
)
return {"status": "LOADED", "rows": len(df2)}
# Generate 3 batches
batches = {}
for match_id in [10001, 10002, 10003]:
np.random.seed(match_id)
batches[f"scorecards/match_{match_id}.parquet"] = pd.DataFrame({
"delivery_id": [match_id*1000+i for i in range(120)],
"match_id": match_id, "over_number": [(i//6)+1 for i in range(120)],
"runs_scored": [int(np.random.choice([0,1,2,4,6])) for _ in range(120)],
"is_wicket": [bool(np.random.random()<0.05) for _ in range(120)],
"bowler": list(np.random.choice(["Bumrah","Shami","Hardik"], 120)),
"batter": list(np.random.choice(["Rohit","Kohli","Gill"], 120)),
})
for path, df in batches.items():
r1 = copy_into(con, path, df)
r2 = copy_into(con, path, df) # idempotency test
print(f" {path}: load1={r1['status']}({r1['rows']}), load2={r2['status']}")
assert r1["status"] == "LOADED"
assert r2["status"] == "SKIPPED"
staging_count = con.execute("SELECT COUNT(*) FROM staging_deliveries").fetchone()[0]
assert staging_count == 360, f"Expected 360, got {staging_count}"
print(f" Staging total: {staging_count} rows (3 files × 120) ✓")
# ── Stream + Task CDC ─────────────────────────────────────────────────────────
stream_offset = [0] # tracks how many staging rows have been processed
def stream_has_data() -> bool:
total = con.execute("SELECT COUNT(*) FROM staging_deliveries").fetchone()[0]
return stream_offset[0] < total
def task_merge_from_stream() -> int:
if not stream_has_data():
return 0
total = con.execute("SELECT COUNT(*) FROM staging_deliveries").fetchone()[0]
# Get unprocessed rows
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_delivery
SELECT delivery_id, match_id, over_number,
CASE WHEN over_number<=6 THEN 'powerplay'
WHEN over_number<=15 THEN 'middle' ELSE 'death' END,
runs_scored, is_wicket, bowler, batter,
runs_scored >= 4
FROM _stream_batch
""")
rows_processed = len(batch)
stream_offset[0] = total
return rows_processed
# Task run 1: processes all 360 rows
processed1 = task_merge_from_stream()
assert processed1 == 360
# Task run 2: no new data
assert not stream_has_data()
processed2 = task_merge_from_stream()
assert processed2 == 0
fact_count = con.execute("SELECT COUNT(*) FROM fact_delivery").fetchone()[0]
assert fact_count == staging_count
print(f" Task run 1: processed {processed1} rows ✓")
print(f" Task run 2: {processed2} rows (no new data) ✓")
print(f" Fact table: {fact_count} rows = staging ({staging_count}) ✓")
print("Step 1 ✓: COPY INTO idempotency and Streams + Tasks CDC complete")Step 2 — Time Travel Recovery and Zero-Copy Clone
Simulate Time Travel recovery: record the current fact table state as a snapshot, execute an accidental DELETE of all match 10001 rows, verify the deletion, recover using the snapshot, and assert the fact table returns to its original row count. Then simulate Zero-Copy Clone: create a cloned version of the fact table, apply a schema migration (add a `wicket_type` column) to the clone, verify the migration does not affect the production table, and assert delivery total conservation between the clone and production table.
Analogy🏏Cricket
🏏 Think of it like cricket: Time Travel recovery is the third umpire's DRS replay for your data. Just as a wrongly given wicket is reversed by rolling back to the recorded footage and restoring the batsman, an accidental DELETE of all match 10001 deliveries is undone by re-inserting those exact rows from a snapshot taken before the error — and asserting the fact table returns to its original row count is like confirming the scoreboard reads exactly as it did before the mistaken decision. Zero-Copy Clone, meanwhile, is the practice-ground replica of the official record: just as a franchise tests a radical new field-placement plan in the nets without risking the live match, you clone the fact table and add a wicket_type column to the clone alone, verifying the production table stays untouched. Checking that total runs match between clone and production is the same as confirming the practice replica started from an identical scorecard. The payoff: safe recovery from mistakes and risk-free schema experiments that never endanger live data.
python
# exercise_snowflake_pipeline.py — Step 2: Time Travel and Zero-Copy Clone
import duckdb
import pandas as pd
# ── Time Travel: accidental delete and recovery ───────────────────────────────
# Snapshot (Snowflake Time Travel equivalent)
time_travel_snapshot = con.execute("SELECT * FROM fact_delivery").df()
original_count = len(time_travel_snapshot)
# Accidental DELETE
con.execute("DELETE FROM fact_delivery WHERE match_id = 10001")
after_delete = con.execute("SELECT COUNT(*) FROM fact_delivery").fetchone()[0]
assert after_delete < original_count
print(f" Accidental delete: {original_count - after_delete} rows removed")
# Time Travel recovery: re-insert deleted rows from snapshot
deleted_rows = time_travel_snapshot[
~time_travel_snapshot["delivery_id"].isin(
con.execute("SELECT delivery_id FROM fact_delivery").df()["delivery_id"]
)
]
con.register("_recovered", deleted_rows)
con.execute("INSERT INTO fact_delivery SELECT * FROM _recovered")
after_recovery = con.execute("SELECT COUNT(*) FROM fact_delivery").fetchone()[0]
assert after_recovery == original_count, \
f"Recovery failed: {after_recovery} != {original_count}"
print(f" Time Travel recovery: {len(deleted_rows)} rows restored ✓")
print(f" Fact table: {after_recovery} rows (back to original) ✓")
# ── Zero-Copy Clone: staging environment for migration testing ────────────────
# Clone production table (shares underlying data — zero initial storage)
clone_df = con.execute("SELECT * FROM fact_delivery").df().copy()
con.execute("""
CREATE TABLE fact_delivery_staging AS
SELECT * FROM fact_delivery; -- simulates Clone (initially identical)
""")
# Apply schema migration to CLONE only
con.execute("ALTER TABLE fact_delivery_staging ADD COLUMN wicket_type VARCHAR")
con.execute("""
UPDATE fact_delivery_staging
SET wicket_type = CASE WHEN is_wicket THEN 'bowled' ELSE NULL END
""")
# Verify migration applied to clone but NOT to production
clone_cols = set(con.execute("DESCRIBE fact_delivery_staging").df()["column_name"])
production_cols = set(con.execute("DESCRIBE fact_delivery").df()["column_name"])
assert "wicket_type" in clone_cols
assert "wicket_type" not in production_cols
print(f" Clone migration: wicket_type column added to staging only ✓")
# Conservation: total runs must match between clone and production
runs_production = con.execute("SELECT SUM(runs_scored) FROM fact_delivery").fetchone()[0]
runs_clone = con.execute("SELECT SUM(runs_scored) FROM fact_delivery_staging").fetchone()[0]
assert runs_production == runs_clone
print(f" Conservation: production={runs_production} = clone={runs_clone} ✓")
print("Step 2 ✓: Time Travel recovery and Zero-Copy Clone complete")