Mini-project: Kiosko's first analytics warehouse

Description

This project closes the module — and the whole guide — by integrating the seven previous pieces: bronze and silver rebuilt from foundations (lesson 3), the star with historized dim_product_scd and the point-in-time join (lesson 4), fact_sessions and fact_store_activity (lesson 5), mart_daily_sales_obt published for BI (lesson 6), and the map of sibling guides (lesson 7). What's left is to bring it all together in a single script, start to finish, in a single DuckDB connection, verified with assert at every step, closing with the final report Kiosko's management asked for back in lesson 2.

The project has eight parts. First, bronze: raw landing of orders and events. Second, silver: the quality gate and fact_orders modeled. Third, the star: dim_store, dim_date, dim_product_scd historized with two MERGE INTO runs. Fourth, correct historical revenue: the point-in-time join contrasted against the broken one. Fifth, the funnel and activity: fact_sessions and fact_store_activity. Sixth, the OBT: mart_daily_sales_obt, with the correct join already resolved. Seventh, the contract: validate_gold_schema() over the guide's four gold tables. Eighth, the formal declaration: KIOSKO_WAREHOUSE, the structure documenting the whole project, and the final report for management.

Connection to the module. This project introduces no new concept — it's the final integration of this guide's eight complete modules, packaged as KIOSKO_WAREHOUSE, Kiosko's complete analytics warehouse, verified end to end.

An analogy: the grand opening of the complete building, with all seven inspections already passed

Every module in this guide was a separate inspection, approved on its own: the foundation (grain), the complete structure (star), the electrical and plumbing systems compared against alternatives (snowflake/OBT), the building's change log (SCD), the certification that every question gets answered with the correct date (point-in-time join), two additional systems no traditional blueprint covers (funnel, activity), and the complete inventory with its quality contract (messy domain, Medallion). This project is the grand opening: the complete building, with all seven inspections already passed, opened for the first time so someone — Kiosko's management — can walk in and really use it.

The material you need

You need, in the same folder: kiosko.py (with DIM_STORE, DIM_PRODUCT, Order, transform_fact_orders), raw_orders.py (the 40 fixed orders), and events.py (the 32 canonical events) — the same three files you used in every mini-project across this guide's eight modules. You don't need any additional file: generate_date_dim(), validate_orders(), validate_gold_schema(), and all the MERGE INTO and cumulative-design logic are defined directly in this project's script.

The reference solution, verified

Part 1 — BRONZE: raw landing of orders and events

# kiosko_analytics_warehouse.py -- Kiosko's first analytics warehouse, the guide's closing project
from datetime import date, datetime, timedelta

import duckdb

from kiosko import DIM_PRODUCT, DIM_STORE, Order, transform_fact_orders
from raw_orders import RAW_ORDERS
from events import RAW_EVENTS

DAY_NAMES = ["Monday", "Tuesday", "Wednesday", "Thursday", "Friday", "Saturday", "Sunday"]
REQUIRED_COLUMNS = ["order_id", "store_id", "product_id", "quantity", "unit_price", "order_ts"]


def generate_date_dim(start_date: str, end_date: str) -> list[dict]:
    start = date.fromisoformat(start_date)
    end = date.fromisoformat(end_date)
    rows = []
    current = start
    while current <= end:
        weekday_index = current.weekday()
        rows.append({
            "date_key": int(current.strftime("%Y%m%d")), "calendar_date": current,
            "day_of_week": DAY_NAMES[weekday_index], "month": current.month,
            "quarter": (current.month - 1) // 3 + 1, "year": current.year,
            "is_weekend": weekday_index >= 5,
        })
        current += timedelta(days=1)
    return rows


def validate_orders(rows: list[dict]) -> tuple[list[dict], list[dict]]:
    valid: list[dict] = []
    rejected: list[dict] = []
    seen_order_ids: set[str] = set()
    for row in rows:
        reasons: list[str] = []
        missing = [c for c in REQUIRED_COLUMNS if not row.get(c)]
        if missing:
            reasons.append(f"missing or empty fields: {missing}")
            rejected.append({"row": row, "reasons": reasons})
            continue
        try:
            quantity = int(row["quantity"])
        except ValueError:
            reasons.append(f"quantity is not a valid integer: {row['quantity']!r}")
            quantity = None
        try:
            unit_price = float(row["unit_price"])
        except ValueError:
            reasons.append(f"unit_price is not a valid decimal: {row['unit_price']!r}")
            unit_price = None
        if quantity is not None and quantity <= 0:
            reasons.append(f"quantity must be > 0, got {quantity}")
        if unit_price is not None and unit_price < 0:
            reasons.append(f"unit_price must be >= 0, got {unit_price}")
        if row["order_id"] in seen_order_ids:
            reasons.append(f"duplicate order_id: {row['order_id']}")
        if reasons:
            rejected.append({"row": row, "reasons": reasons})
        else:
            seen_order_ids.add(row["order_id"])
            valid.append(row)
    return valid, rejected


def validate_gold_schema(con, table, expected_columns):
    actual_rows = con.sql(f"DESCRIBE {table}").fetchall()
    actual_columns = {row[0]: row[1] for row in actual_rows}
    discrepancies = []
    for column_name, expected_type in expected_columns.items():
        if column_name not in actual_columns:
            discrepancies.append(f"{table}: missing column '{column_name}' (expected type {expected_type})")
        elif actual_columns[column_name] != expected_type:
            discrepancies.append(
                f"{table}: '{column_name}' has type {actual_columns[column_name]}, expected {expected_type}"
            )
    for column_name in actual_columns:
        if column_name not in expected_columns:
            discrepancies.append(f"{table}: unexpected column '{column_name}', not declared in the contract")
    return discrepancies


print("=== Kiosko: the first complete analytics warehouse, end to end ===")
print(f"DuckDB version: {duckdb.__version__}\n")

con = duckdb.connect()

bronze_rows = [
    {"order_id": r[0], "store_id": r[1], "product_id": r[2],
     "quantity": str(r[3]), "unit_price": str(r[4]), "order_ts": r[5]}
    for r in RAW_ORDERS
]
con.execute("""
    CREATE TABLE bronze_orders (
        order_id VARCHAR, store_id VARCHAR, product_id VARCHAR,
        quantity VARCHAR, unit_price VARCHAR, order_ts VARCHAR
    )
""")
con.executemany("INSERT INTO bronze_orders VALUES (?, ?, ?, ?, ?, ?)",
    [(r["order_id"], r["store_id"], r["product_id"], r["quantity"], r["unit_price"], r["order_ts"]) for r in bronze_rows])
con.execute("CREATE TABLE bronze_events (event_id VARCHAR, event_type VARCHAR, session_id VARCHAR, event_ts VARCHAR)")
con.executemany("INSERT INTO bronze_events VALUES (?, ?, ?, ?)", RAW_EVENTS)

bronze_orders_count = con.sql("SELECT COUNT(*) FROM bronze_orders").fetchone()[0]
bronze_events_count = con.sql("SELECT COUNT(*) FROM bronze_events").fetchone()[0]
print("Part 1 -- BRONZE: raw landing, untransformed")
print(f"  bronze_orders   {bronze_orders_count:3} rows")
print(f"  bronze_events   {bronze_events_count:3} rows")

Part 2 — SILVER: quality gate + fact_orders modeled

valid_rows, rejected_rows = validate_orders(bronze_rows)

orders = [
    Order(order_id=r["order_id"], store_id=r["store_id"], product_id=r["product_id"],
          quantity=int(r["quantity"]), unit_price=float(r["unit_price"]),
          order_ts=datetime.fromisoformat(r["order_ts"]))
    for r in valid_rows
]
fact_orders_rows = transform_fact_orders(orders, DIM_STORE, DIM_PRODUCT)

con.execute("""
    CREATE TABLE fact_orders (
        order_id VARCHAR, store_id VARCHAR, product_id VARCHAR,
        quantity INTEGER, unit_price DOUBLE, revenue DOUBLE, order_ts TIMESTAMP
    )
""")
con.executemany("INSERT INTO fact_orders VALUES (?, ?, ?, ?, ?, ?, ?)",
    [(r["order_id"], r["store_id"], r["product_id"], r["quantity"],
      r["unit_price"], r["revenue"], r["order_ts"]) for r in fact_orders_rows])
con.execute("CREATE TABLE events (event_id VARCHAR, event_type VARCHAR, session_id VARCHAR, event_ts TIMESTAMP)")
con.executemany("INSERT INTO events VALUES (?, ?, ?, ?)",
    [(r[0], r[1], r[2], datetime.fromisoformat(r[3])) for r in RAW_EVENTS])

total_orders = con.sql("SELECT COUNT(*) FROM fact_orders").fetchone()[0]
total_revenue = con.sql("SELECT ROUND(SUM(revenue), 2) FROM fact_orders").fetchone()[0]
grain_check = con.sql("SELECT COUNT(*), COUNT(DISTINCT order_id || '-' || product_id) FROM fact_orders").fetchone()

print("\nPart 2 -- SILVER: quality gate + fact_orders modeled")
print(f"  validate_orders(): {len(valid_rows)} valid, {len(rejected_rows)} rejected")
print(f"  fact_orders        {total_orders:3} rows, total revenue = {total_revenue}")
print(f"  grain verified: COUNT(*)={grain_check[0]} == COUNT(DISTINCT order_id-product_id)={grain_check[1]}")
assert len(valid_rows) == 40 and len(rejected_rows) == 0
assert total_orders == 40 and total_revenue == 106.15
assert grain_check[0] == grain_check[1] == 40
print("  Verification OK: bronze -> silver -> gold reproduces fact_orders identical to foundations")

What to expect (Parts 1 and 2).

=== Kiosko: the first complete analytics warehouse, end to end ===
DuckDB version: 1.5.5

Part 1 -- BRONZE: raw landing, untransformed
  bronze_orders    40 rows
  bronze_events    32 rows

Part 2 -- SILVER: quality gate + fact_orders modeled
  validate_orders(): 40 valid, 0 rejected
  fact_orders         40 rows, total revenue = 106.15
  grain verified: COUNT(*)=40 == COUNT(DISTINCT order_id-product_id)=40
  Verification OK: bronze -> silver -> gold reproduces fact_orders identical to foundations

Part 3 — THE STAR: dim_store, dim_date, historized dim_product_scd

con.execute("CREATE TABLE dim_store_natural (store_id VARCHAR, store_name VARCHAR, city VARCHAR)")
con.executemany("INSERT INTO dim_store_natural VALUES (?, ?, ?)",
                 [(s["store_id"], s["store_name"], s["city"]) for s in DIM_STORE])
con.execute("CREATE TABLE dim_store AS SELECT ROW_NUMBER() OVER (ORDER BY store_id) AS store_key, store_id, store_name, city FROM dim_store_natural")

dim_date_rows = generate_date_dim("2026-08-01", "2026-08-31")
con.execute("""
    CREATE TABLE dim_date (
        date_key INTEGER, calendar_date DATE, day_of_week VARCHAR,
        month INTEGER, quarter INTEGER, year INTEGER, is_weekend BOOLEAN
    )
""")
con.executemany("INSERT INTO dim_date VALUES (?, ?, ?, ?, ?, ?, ?)",
    [(r["date_key"], r["calendar_date"], r["day_of_week"], r["month"],
      r["quarter"], r["year"], r["is_weekend"]) for r in dim_date_rows])

con.execute("CREATE SEQUENCE product_key_seq START 1")
con.execute("""
    CREATE TABLE dim_product_scd (
        product_key INTEGER PRIMARY KEY, product_id VARCHAR NOT NULL, product_name VARCHAR,
        category VARCHAR, unit_cost DOUBLE, valid_from DATE NOT NULL, valid_to DATE,
        is_current BOOLEAN NOT NULL DEFAULT true
    )
""")
con.executemany(
    "INSERT INTO dim_product_scd VALUES (nextval('product_key_seq'), ?, ?, ?, ?, DATE '2026-08-01', NULL, true)",
    [(p["product_id"], p["product_name"], p["category"], p["unit_cost"]) for p in DIM_PRODUCT])

PRODUCTS_V1 = [("P001", "Bottled Water 600ml", "beverages", 0.40), ("P002", "Energy Bar", "snacks", 0.60),
               ("P003", "Instant Coffee Sachet", "beverages", 0.35), ("P004", "Phone Charger Cable", "electronics", 2.10)]
PRODUCTS_V2 = [("P001", "Bottled Water 600ml", "beverages", 0.40), ("P002", "Energy Bar", "health-snacks", 0.68),
               ("P003", "Instant Coffee Sachet", "beverages", 0.35), ("P004", "Phone Charger Cable", "electronics", 2.10)]


def load_staging(products):
    con.execute("DROP TABLE IF EXISTS staging_product")
    con.execute("CREATE TABLE staging_product (product_id VARCHAR, product_name VARCHAR, category VARCHAR, unit_cost DOUBLE)")
    con.executemany("INSERT INTO staging_product VALUES (?, ?, ?, ?)", products)


def merge_scd(change_date):
    result = con.sql(f"""
        MERGE INTO dim_product_scd AS target USING staging_product AS source
        ON target.product_id = source.product_id AND target.is_current = true
        WHEN MATCHED AND (target.unit_cost <> source.unit_cost OR target.category <> source.category)
        THEN UPDATE SET valid_to = DATE '{change_date}' - INTERVAL 1 DAY, is_current = false
        RETURNING merge_action, product_id
    """)
    changed_ids = [row[1] for row in result.fetchall()]
    if changed_ids:
        placeholders = ", ".join("?" for _ in changed_ids)
        con.execute(f"""
            INSERT INTO dim_product_scd (product_key, product_id, product_name, category, unit_cost, valid_from, valid_to, is_current)
            SELECT nextval('product_key_seq'), source.product_id, source.product_name, source.category, source.unit_cost,
                   DATE '{change_date}', NULL, true
            FROM staging_product AS source WHERE source.product_id IN ({placeholders})
        """, changed_ids)
    return len(changed_ids)


load_staging(PRODUCTS_V1)
rows_changed_1 = merge_scd("2026-08-15")
load_staging(PRODUCTS_V2)
rows_changed_2 = merge_scd("2026-08-15")

dim_product_scd_count = con.sql("SELECT COUNT(*) FROM dim_product_scd").fetchone()[0]
p002_versions = con.sql("""
    SELECT COUNT(*), SUM(CASE WHEN is_current THEN 1 ELSE 0 END) FROM dim_product_scd WHERE product_id = 'P002'
""").fetchone()

con.execute("""
    CREATE TABLE fact_orders_star AS
    SELECT f.order_id, f.store_id, f.product_id, s.store_key, d.date_key, f.quantity, f.unit_price, f.revenue
    FROM fact_orders f
    JOIN dim_store s ON f.store_id = s.store_id
    JOIN dim_date  d ON CAST(strftime(f.order_ts, '%Y%m%d') AS INTEGER) = d.date_key
""")
star_rows = con.sql("SELECT COUNT(*) FROM fact_orders_star").fetchone()[0]

print("\nPart 3 -- THE STAR: dim_store, dim_date, historized dim_product_scd")
print(f"  dim_store         {con.sql('SELECT COUNT(*) FROM dim_store').fetchone()[0]:3} rows")
print(f"  dim_date          {con.sql('SELECT COUNT(*) FROM dim_date').fetchone()[0]:3} rows")
print(f"  dim_product_scd   {dim_product_scd_count:3} rows (MERGE #1 closed {rows_changed_1}, MERGE #2 closed {rows_changed_2})")
print(f"  P002: {p002_versions[0]} versions, {p002_versions[1]} current")
print(f"  fact_orders_star  {star_rows:3} rows")
assert dim_product_scd_count == 5 and p002_versions == (2, 1) and star_rows == 40

What to expect (Part 3).

Part 3 -- THE STAR: dim_store, dim_date, historized dim_product_scd
  dim_store           3 rows
  dim_date           31 rows
  dim_product_scd     5 rows (MERGE #1 closed 0, MERGE #2 closed 1)
  P002: 2 versions, 1 current
  fact_orders_star   40 rows

Part 4 — CORRECT HISTORICAL REVENUE: point-in-time join vs is_current

broken = con.sql("""
    SELECT d.category, COUNT(*) AS orders, ROUND(SUM(f.revenue), 2) AS revenue,
           ROUND(SUM(f.revenue - f.quantity * d.unit_cost), 2) AS margin
    FROM fact_orders f JOIN dim_product_scd d ON f.product_id = d.product_id AND d.is_current = true
    GROUP BY d.category ORDER BY d.category
""").fetchall()
correct = con.sql("""
    SELECT d.category, COUNT(*) AS orders, ROUND(SUM(f.revenue), 2) AS revenue,
           ROUND(SUM(f.revenue - f.quantity * d.unit_cost), 2) AS margin
    FROM fact_orders f
    JOIN dim_product_scd d ON f.product_id = d.product_id
       AND f.order_ts BETWEEN d.valid_from AND COALESCE(d.valid_to, DATE '9999-12-31')
    GROUP BY d.category ORDER BY d.category
""").fetchall()
total_broken = con.sql("""
    SELECT ROUND(SUM(f.revenue), 2) FROM fact_orders f
    JOIN dim_product_scd d ON f.product_id = d.product_id AND d.is_current = true
""").fetchone()[0]
total_correct = con.sql("""
    SELECT ROUND(SUM(f.revenue), 2) FROM fact_orders f
    JOIN dim_product_scd d ON f.product_id = d.product_id
       AND f.order_ts BETWEEN d.valid_from AND COALESCE(d.valid_to, DATE '9999-12-31')
""").fetchone()[0]

print("\nPart 4 -- historical revenue: BROKEN JOIN vs correct JOIN")
print("  BROKEN (is_current = true):")
for category, orders_count, revenue, margin in broken:
    print(f"    {category:14} {orders_count:2} orders  revenue={revenue:7}  margin={margin:6}")
print("  CORRECT (BETWEEN valid_from AND valid_to):")
for category, orders_count, revenue, margin in correct:
    print(f"    {category:14} {orders_count:2} orders  revenue={revenue:7}  margin={margin:6}")
print(f"  Total revenue: identical in both cases -> {total_broken} == {total_correct}")

broken_categories = {c for c, *_ in broken}
correct_categories = {c for c, *_ in correct}
assert "health-snacks" in broken_categories and "health-snacks" not in correct_categories
assert "snacks" in correct_categories and "snacks" not in broken_categories
assert total_broken == total_correct == 106.15
print("  Verification OK: 'snacks' is the correct category, not 'health-snacks'")

What to expect (Part 4).

Part 4 -- historical revenue: BROKEN JOIN vs correct JOIN
  BROKEN (is_current = true):
    beverages      23 orders  revenue=  44.05  margin= 14.75
    electronics     7 orders  revenue=   40.5  margin=  21.6
    health-snacks  10 orders  revenue=   21.6  margin=  9.36
  CORRECT (BETWEEN valid_from AND valid_to):
    beverages      23 orders  revenue=  44.05  margin= 14.75
    electronics     7 orders  revenue=   40.5  margin=  21.6
    snacks         10 orders  revenue=   21.6  margin=  10.8
  Total revenue: identical in both cases -> 106.15 == 106.15
  Verification OK: 'snacks' is the correct category, not 'health-snacks'

Part 5 — fact_sessions and fact_store_activity

STORE_ROTATION = ["S01", "S02", "S03"]


def store_for_session(session_id):
    session_number = int(session_id.split("-")[1])
    return STORE_ROTATION[(session_number - 1) % 3]


ALL_SESSIONS = [f"SESS-{n:02d}" for n in range(1, 18)]
con.execute("CREATE TABLE session_store_map (session_id VARCHAR, store_id VARCHAR)")
con.executemany("INSERT INTO session_store_map VALUES (?, ?)", [(sid, store_for_session(sid)) for sid in ALL_SESSIONS])

con.execute("""
    CREATE TABLE fact_sessions AS
    SELECT e.session_id, m.store_id, MIN(CAST(e.event_ts AS DATE)) AS session_date,
        MAX(CASE WHEN e.event_type = 'page_view'   THEN e.event_ts END) AS view_ts,
        MAX(CASE WHEN e.event_type = 'add_to_cart' THEN e.event_ts END) AS add_to_cart_ts,
        MAX(CASE WHEN e.event_type = 'purchase'    THEN e.event_ts END) AS purchase_ts,
        MAX(CASE WHEN e.event_type = 'purchase' THEN true ELSE false END) AS is_converted
    FROM events e JOIN session_store_map m ON e.session_id = m.session_id
    GROUP BY e.session_id, m.store_id
""")
funnel = con.sql("SELECT COUNT(*), COUNT(view_ts), COUNT(add_to_cart_ts), COUNT(purchase_ts) FROM fact_sessions").fetchone()
conversion_pct = con.sql("SELECT ROUND(100.0 * COUNT(purchase_ts) / COUNT(*), 1) FROM fact_sessions").fetchone()[0]

con.execute("""
    CREATE TABLE fact_store_activity (
        store_id VARCHAR, activity_date DATE, daily_revenue DOUBLE,
        revenue_array_7d DOUBLE[], active_days_7d INTEGER,
        revenue_array_30d DOUBLE[], active_days_30d INTEGER
    )
""")
DAYS = ["2026-08-03", "2026-08-04", "2026-08-05", "2026-08-06", "2026-08-07", "2026-08-08", "2026-08-09"]
for day in DAYS:
    for store_id in STORE_ROTATION:
        daily_revenue = con.sql(f"""
            SELECT COALESCE(ROUND(SUM(revenue), 2), 0.0) FROM fact_orders
            WHERE store_id = '{store_id}' AND CAST(order_ts AS DATE) = DATE '{day}'
        """).fetchone()[0]
        prev = con.sql(f"""
            SELECT revenue_array_7d, revenue_array_30d FROM fact_store_activity
            WHERE store_id = '{store_id}' ORDER BY activity_date DESC LIMIT 1
        """).fetchone()
        if prev is None:
            new_7d, new_30d = [daily_revenue], [daily_revenue]
        else:
            new_7d = ([daily_revenue] + list(prev[0]))[:7]
            new_30d = ([daily_revenue] + list(prev[1]))[:30]
        active_7d = sum(1 for v in new_7d if v > 0)
        active_30d = sum(1 for v in new_30d if v > 0)
        con.execute("INSERT INTO fact_store_activity VALUES (?, ?, ?, ?, ?, ?, ?)",
            (store_id, day, daily_revenue, new_7d, active_7d, new_30d, active_30d))

activity_rows = con.sql("SELECT COUNT(*) FROM fact_store_activity").fetchone()[0]
snapshot = con.sql("""
    SELECT store_id, ROUND(list_sum(revenue_array_7d), 2) AS revenue_7d, active_days_7d, active_days_30d
    FROM fact_store_activity WHERE activity_date = DATE '2026-08-09' ORDER BY store_id
""").fetchall()

print("\nPart 5 -- fact_sessions (accumulating snapshot) + fact_store_activity (cumulative)")
print(f"  fact_sessions        {con.sql('SELECT COUNT(*) FROM fact_sessions').fetchone()[0]:3} rows")
print(f"  funnel: total={funnel[0]} viewed={funnel[1]} cart={funnel[2]} purchased={funnel[3]} conversion={conversion_pct}%")
print(f"  fact_store_activity  {activity_rows:3} rows")
for store_id, revenue_7d, active_7d, active_30d in snapshot:
    print(f"    {store_id}: revenue_7d={revenue_7d:6}  active_days_7d={active_7d}  active_days_30d={active_30d}")
assert funnel == (17, 17, 9, 6) and conversion_pct == 35.3
assert activity_rows == 21

What to expect (Part 5).

Part 5 -- fact_sessions (accumulating snapshot) + fact_store_activity (cumulative)
  fact_sessions         17 rows
  funnel: total=17 viewed=17 cart=9 purchased=6 conversion=35.3%
  fact_store_activity   21 rows
    S01: revenue_7d=  38.3  active_days_7d=7  active_days_30d=7
    S02: revenue_7d=  38.8  active_days_7d=7  active_days_30d=7
    S03: revenue_7d= 29.05  active_days_7d=6  active_days_30d=6

Part 6 — mart_daily_sales_obt: the OBT for the BI team

con.execute("""
    CREATE TABLE mart_daily_sales_obt AS
    SELECT CAST(f.order_ts AS DATE) AS sale_date, dt.day_of_week, dt.is_weekend,
        s.store_id, s.store_name, s.city, p.product_id, p.product_name, p.category, p.unit_cost,
        SUM(f.quantity) AS quantity, ROUND(SUM(f.revenue), 2) AS revenue,
        ROUND(SUM(f.revenue - f.quantity * p.unit_cost), 2) AS margin
    FROM fact_orders f
    JOIN dim_store s ON f.store_id = s.store_id
    JOIN dim_date  dt ON CAST(strftime(f.order_ts, '%Y%m%d') AS INTEGER) = dt.date_key
    JOIN dim_product_scd p ON f.product_id = p.product_id
       AND f.order_ts BETWEEN p.valid_from AND COALESCE(p.valid_to, DATE '9999-12-31')
    GROUP BY 1,2,3,4,5,6,7,8,9,10
""")
obt_rows = con.sql("SELECT COUNT(*) FROM mart_daily_sales_obt").fetchone()[0]
obt_revenue = con.sql("SELECT ROUND(SUM(revenue), 2) FROM mart_daily_sales_obt").fetchone()[0]
obt_categories = con.sql("SELECT DISTINCT category FROM mart_daily_sales_obt ORDER BY category").fetchall()

print("\nPart 6 -- mart_daily_sales_obt: the OBT for the BI team")
print(f"  mart_daily_sales_obt {obt_rows:3} rows, total revenue = {obt_revenue}")
print(f"  categories present: {[c[0] for c in obt_categories]}")
assert obt_revenue == 106.15
assert "snacks" in {c[0] for c in obt_categories} and "health-snacks" not in {c[0] for c in obt_categories}

What to expect (Part 6).

Part 6 -- mart_daily_sales_obt: the OBT for the BI team
  mart_daily_sales_obt  39 rows, total revenue = 106.15
  categories present: ['beverages', 'electronics', 'snacks']

Part 7 — validate_gold_schema() over the guide's 4 gold tables

GOLD_CONTRACTS = {
    "fact_orders": {
        "order_id": "VARCHAR", "store_id": "VARCHAR", "product_id": "VARCHAR",
        "quantity": "INTEGER", "unit_price": "DOUBLE", "revenue": "DOUBLE", "order_ts": "TIMESTAMP",
    },
    "fact_sessions": {
        "session_id": "VARCHAR", "store_id": "VARCHAR", "session_date": "DATE",
        "view_ts": "TIMESTAMP", "add_to_cart_ts": "TIMESTAMP", "purchase_ts": "TIMESTAMP", "is_converted": "BOOLEAN",
    },
    "fact_store_activity": {
        "store_id": "VARCHAR", "activity_date": "DATE", "daily_revenue": "DOUBLE",
        "revenue_array_7d": "DOUBLE[]", "active_days_7d": "INTEGER",
        "revenue_array_30d": "DOUBLE[]", "active_days_30d": "INTEGER",
    },
    "dim_date": {
        "date_key": "INTEGER", "calendar_date": "DATE", "day_of_week": "VARCHAR",
        "month": "INTEGER", "quarter": "INTEGER", "year": "INTEGER", "is_weekend": "BOOLEAN",
    },
}
print("\nPart 7 -- validate_gold_schema() over the guide's 4 gold tables")
total_discrepancies = 0
for table, expected_columns in GOLD_CONTRACTS.items():
    discrepancies = validate_gold_schema(con, table, expected_columns)
    total_discrepancies += len(discrepancies)
    status = "OK, 0 discrepancies" if not discrepancies else f"{len(discrepancies)} discrepancies"
    print(f"  {table:20} {status}")
assert total_discrepancies == 0
print(f"  TOTAL: {len(GOLD_CONTRACTS)} gold tables, {total_discrepancies} discrepancies -- OK")

What to expect (Part 7).

Part 7 -- validate_gold_schema() over the guide's 4 gold tables
  fact_orders          OK, 0 discrepancies
  fact_sessions        OK, 0 discrepancies
  fact_store_activity  OK, 0 discrepancies
  dim_date             OK, 0 discrepancies
  TOTAL: 4 gold tables, 0 discrepancies -- OK

Part 8 — KIOSKO_WAREHOUSE: the formal declaration, and the final report

KIOSKO_WAREHOUSE = {
    "bronze": {"bronze_orders": bronze_orders_count, "bronze_events": bronze_events_count},
    "silver": {"valid_rows": len(valid_rows), "rejected_rows": len(rejected_rows), "fact_orders_rows": total_orders},
    "gold_star": {"dim_store": 3, "dim_date": 31, "dim_product_scd": dim_product_scd_count, "fact_orders_star_rows": star_rows},
    "historized_product": {
        "product_id": "P002", "versions": p002_versions[0], "current_versions": p002_versions[1],
        "change_date": "2026-08-15", "category_before": "snacks", "category_after": "health-snacks",
    },
    "point_in_time_revenue": {
        "broken_category_for_p002": "health-snacks", "correct_category_for_p002": "snacks",
        "total_revenue_broken_vs_correct": [total_broken, total_correct],
    },
    "gold_funnel_and_activity": {
        "fact_sessions_rows": 17, "funnel_total_viewed_cart_purchased": list(funnel),
        "funnel_conversion_pct": conversion_pct, "fact_store_activity_rows": activity_rows,
    },
    "gold_obt": {"mart_daily_sales_obt_rows": obt_rows, "revenue": obt_revenue},
    "gold_contract": {"tables_validated": list(GOLD_CONTRACTS.keys()), "discrepancies": total_discrepancies},
}
print("\nPart 8 -- KIOSKO_WAREHOUSE: the formal declaration that closes the guide")
for key, value in KIOSKO_WAREHOUSE.items():
    print(f"  {key}: {value}")

print("\n=== Final report for Kiosko's management ===")
print(con.sql("""
    SELECT store_name, ROUND(SUM(revenue), 2) AS revenue, ROUND(SUM(margin), 2) AS margin
    FROM mart_daily_sales_obt GROUP BY store_name ORDER BY store_name
"""))
print(con.sql("""
    SELECT category, ROUND(SUM(revenue), 2) AS revenue, ROUND(SUM(margin), 2) AS margin
    FROM mart_daily_sales_obt GROUP BY category ORDER BY category
"""))

What to expect. Running python3 kiosko_analytics_warehouse.py in full (all eight parts together), the output ends exactly like this:

Part 8 -- KIOSKO_WAREHOUSE: the formal declaration that closes the guide
  bronze: {'bronze_orders': 40, 'bronze_events': 32}
  silver: {'valid_rows': 40, 'rejected_rows': 0, 'fact_orders_rows': 40}
  gold_star: {'dim_store': 3, 'dim_date': 31, 'dim_product_scd': 5, 'fact_orders_star_rows': 40}
  historized_product: {'product_id': 'P002', 'versions': 2, 'current_versions': 1, 'change_date': '2026-08-15', 'category_before': 'snacks', 'category_after': 'health-snacks'}
  point_in_time_revenue: {'broken_category_for_p002': 'health-snacks', 'correct_category_for_p002': 'snacks', 'total_revenue_broken_vs_correct': [106.15, 106.15]}
  gold_funnel_and_activity: {'fact_sessions_rows': 17, 'funnel_total_viewed_cart_purchased': [17, 17, 9, 6], 'funnel_conversion_pct': 35.3, 'fact_store_activity_rows': 21}
  gold_obt: {'mart_daily_sales_obt_rows': 39, 'revenue': 106.15}
  gold_contract: {'tables_validated': ['fact_orders', 'fact_sessions', 'fact_store_activity', 'dim_date'], 'discrepancies': 0}

=== Final report for Kiosko's management ===
┌───────────────┬─────────┬────────┐
│  store_name   │ revenue │ margin │
│    varchar    │ double  │ double │
├───────────────┼─────────┼────────┤
│ Kiosko Centro │    38.3 │   17.4 │
│ Kiosko Norte  │    38.8 │  17.65 │
│ Kiosko Sur    │   29.05 │   12.1 │
└───────────────┴─────────┴────────┘

┌─────────────┬─────────┬────────┐
│  category   │ revenue │ margin │
│   varchar   │ double  │ double │
├─────────────┼─────────┼────────┤
│ beverages   │   44.05 │  14.75 │
│ electronics │    40.5 │   21.6 │
│ snacks      │    21.6 │   10.8 │
└─────────────┴─────────┴────────┘

Stop on KIOSKO_WAREHOUSE and the final report together, because they summarize the whole guide in a single picture. Eight fields, each with the numeric evidence of a distinct lesson in this module — and, underneath, of a distinct module among the eight that make up the whole guide. And the two final tables are, literally, what Kiosko's management asked for in lesson 2's brief: revenue and margin by store, revenue and margin by category — with snacks instead of health-snacks, without anyone on the BI side having had to write a single BETWEEN valid_from AND valid_to.

Diagram: the complete warehouse, all eight parts closed with evidence

flowchart TD
    A["P1: BRONZE\nVERIFIED -- 40+32 rows"] --> B
    B["P2: SILVER\nVERIFIED -- 40 rows, 106.15, 0 rejected"] --> C
    C["P3: STAR + SCD\nVERIFIED -- P002 in 2 versions"] --> D
    D["P4: point-in-time JOIN\nVERIFIED -- snacks correct, 106.15"] --> E
    E["P5: fact_sessions + fact_store_activity\nVERIFIED -- 17->9->6, 35.3%, active 7/30d"] --> F
    F["P6: mart_daily_sales_obt\nVERIFIED -- 39 rows, correct join inside"] --> G
    G["P7: validate_gold_schema()\nVERIFIED -- 0 discrepancies"] --> H
    H["KIOSKO_WAREHOUSE\nthe formal contract that closes the guide"]
    H --> I["Final report for management\nCLOSES data-modeling-for-analytics-guide"]

Closing module 1's lesson 2 checklist, the whole guide

The seven rows of the checklist that opened this guide — in module 1's lesson 2 — are resolved, each verified twice: first in its own module, now again, integrated, in this final project. No piece of the Kimball dimensional modeling this guide set out to teach remains pending.

Common mistakes

Delivering KIOSKO_WAREHOUSE without the eight parts' asserts. What happens: someone, in a hurry to show the final structure as the result, builds it directly after running the queries, without having gone through the asserts that confirm each number. Why it happens: the summary structure looks more presentable as "the deliverable," and the asserts feel like discardable preliminary steps. How to spot it: if your final delivery doesn't include any executed evidence that fact_orders has 40 rows with revenue 106.15, that P002 ended up with two versions, that the correct JOIN gives snacks and not health-snacks, that the funnel gives 35.3%, and that the 4 gold tables pass with no discrepancies, you're documenting a process without having confirmed it worked. How to fix it: this project's asserts — eight, one per part — aren't optional, they're the guarantee that makes everything KIOSKO_WAREHOUSE documents trustworthy.

Thinking this project "ends" the guide in the sense that there's nothing left to learn about dimensional modeling. What happens: someone, seeing the complete checklist with all seven rows resolved, concludes they now fully master dimensional modeling, with no room left to grow. Why it happens: a complete checklist, with executed evidence in every row, naturally feels like a definitive ending. How to spot it: if you can't name, from memory, at least four of the sibling guides lesson 7 mapped, you lost sight of the fact that this project closes this guide, not the whole of data engineering learning. How to fix it: this project integrates, with excellence, everything Kimball and Zach Wilson teach about applied dimensional modeling — but, as lesson 7 warned, the production infrastructure around that model is still missing: real orchestration, versioning as code, native time travel, CDC, formal governance. That's the sibling guides' job, not this project's.

Confusing Kiosko's warehouse, as it stands here, with a system ready for real production. What happens: someone, impressed by how many pieces are integrated — bronze, silver, historized star, funnel, activity, OBT, schema contract — assumes this code, as is, could be deployed against a real retailer's production warehouse with millions of orders. Why it happens: the number of pieces and their internal coherence create a sense of completeness that can be mistaken for "production-ready." How to spot it: if your plan is to copy this script directly into a production environment without going through any of lesson 7's sibling guides, you're going to run into the same eight gaps that lesson named explicitly — no orchestration, no time travel, no CDC, no governance. How to fix it: this warehouse is the correct model, verified with real evidence — but "correct" and "production-ready at scale" are two different claims. The pattern you learned here is exactly the one you'd need at any scale; the infrastructure that supports it in production is the eight sibling guides' job.

Exercises

Exercise 1 — Verify that the OBT's revenue, grouped by day, adds up to exactly 106.15. Using mart_daily_sales_obt, write a query that groups by sale_date and confirms the sum of the seven days matches the known total revenue.

print(con.sql("""
    SELECT sale_date, ROUND(SUM(revenue), 2) AS revenue
    FROM mart_daily_sales_obt GROUP BY sale_date ORDER BY sale_date
"""))

┌────────────┬─────────┐
│ sale_date  │ revenue │
│    date    │ double  │
├────────────┼─────────┤
│ 2026-08-03 │   15.85 │
│ 2026-08-04 │   15.85 │
│ 2026-08-05 │    9.55 │
│ 2026-08-06 │   11.05 │
│ 2026-08-07 │   18.05 │
│ 2026-08-08 │   31.85 │
│ 2026-08-09 │    3.95 │
└────────────┴─────────┘

Exercise 2 — Extend KIOSKO_WAREHOUSE with a guide_complete field that confirms, with a boolean, that the seven pieces of the original checklist were resolved. Without using datetime.now(), add a field documenting this final confirmation.

CHECKLIST_ITEMS = [
    "grain_declared", "star_with_conformed_dimensions", "snowflake_vs_obt",
    "scd_historization", "point_in_time_join", "accumulating_and_cumulative",
    "junk_dimension_and_schema_contract",
]
KIOSKO_WAREHOUSE["guide_complete"] = {
    "checklist_items_resolved": len(CHECKLIST_ITEMS),
    "checklist_items_total": len(CHECKLIST_ITEMS),
    "all_resolved": True,
    "verified_on": "2026-08-09",
}
print(KIOSKO_WAREHOUSE["guide_complete"])

{'checklist_items_resolved': 7, 'checklist_items_total': 7, 'all_resolved': True, 'verified_on': '2026-08-09'}

Exercise 3 — Explain, from memory, which of this project's eight parts would be the first to break if Kiosko opened a fourth store tomorrow. Without writing code, describe in 4-6 sentences what would happen to each of the eight parts of this warehouse if S04 were added to the store catalog, and which one would require the deepest change.

Summary and next step: the end of the whole guide

With this mini-project you close module 8 — and data-modeling-for-analytics-guide in full. You built Kiosko's first real analytics warehouse: bronze and silver rebuilt from foundations (40 rows, revenue 106.15, 0 rejected), the star with historized dim_product_scd (P002 in two real versions, historized with MERGE INTO), the point-in-time join demonstrated against the broken one (snacks/10.8 correct, not health-snacks/9.36), fact_sessions and fact_store_activity (funnel 35.3%, active 7/30 days by store), mart_daily_sales_obt published with the correct join already resolved for the BI team, and validate_gold_schema() confirming zero discrepancies over the guide's four gold tables.

You started this guide with a flat fact_orders, with natural keys, with no history, inherited as-is from foundations. You're finishing it with a complete dimensional warehouse: grain declared and verified, star schema with surrogate keys and conformed dimensions, an evidence-backed comparison between star, snowflake, and wide table, a genuinely historized dimension with SCD type 2, the correct join against that history, two fact patterns no transactional model solves, and a domain with more than one fact and more than one type of dimension, with its schema contract verified. Every pattern — grain, star, SCD, point-in-time join, accumulating snapshot, cumulative design, messy domain, Medallion contract — is the same pattern a production warehouse uses at any scale, exactly as this guide warned since its first lesson.

Where you're headed next. Lesson 7 of this module already traced the complete map: seven sibling guides from the Data Engineering ecosystem — dbt-analytics-engineering-guide, airflow-and-declarative-orchestration-guide, spark-and-distributed-processing-guide, lakehouse-and-iceberg-guide, streaming-with-kafka-and-flink-guide, data-reliability-and-governance-guide, python-for-data-engineering-guide — plus advanced-sql-querying-guide, linked from the SQL ecosystem. Choose the one that solves the gap that matters most to you, and keep building on the dimensional warehouse you left behind, correct and verified, in this guide.

Resources

• Kimball Group — "Star Schema / OLAP Cube" — the complete dimensional vocabulary this project integrates end to end. kimballgroup.com/data-warehouse-business-intelligence-resources/kimball-techniques/dimensional-modeling-techniques/star-schema-olap-cube. In English.
• "The Data Warehouse Toolkit", 3rd edition (Kimball & Ross, Wiley) — the canonical reference that carried this guide's eight modules, start to finish. wiley.com/en-jp/The+Data+Warehouse+Toolkit. In English.
• Databricks — "What is the medallion lakehouse architecture?" — the bronze/silver/gold framework that organized this project's complete warehouse. docs.databricks.com/aws/en/lakehouse/medallion. In English.
• DataExpert-io — cumulative-table-design repository (Zach Wilson) — the pattern behind fact_store_activity, integrated in Part 5 of this project. github.com/DataExpert-io/cumulative-table-design. In English.
• DuckDB — official Python client documentation, the interface that ran every verification in this whole guide. duckdb.org/docs/current/clients/python/overview. In English.