Module 3: Snapshots And Time Travel

Project: Kiosko's time-traveled dim_product

Description

This project closes module 3. You have kiosko.dim_product created with no history columns (lesson 2), you know how to trigger the P002 change with table.overwrite() (lesson 3), you know why you never hardcode a snapshot_id and how to correctly capture or recover it (lesson 4), you know how to travel through time with table.scan(snapshot_id=...) (lesson 5), you know how to recover P002's correct margin by joining that historical read against fact_orders (lesson 6), and you know, with executed evidence, where this technique ends (lesson 7). One step is left: bringing the six pieces together in a single script, run end to end, with automatic assert statements confirming every number.

Connection to the module. This project doesn't introduce any new concept — it's the final integration of the seven previous lessons. It literally revisits the promise that opened this module in lesson 1: recovering the same correct result data-modeling-for-analytics-guide and dbt-analytics-engineering-guide already confirmed, with a table that has zero history columns.

An analogy: the whole archive, walked in one pass

Lessons 2 through 7 of this module built, one piece at a time, dim_product's complete photo archive: the shelf installed and the first photo taken (lesson 2), the restock that changed P002 (lesson 3), the discipline of writing down the right roll number (lesson 4), the request for the old photo to the archive keeper (lesson 5), the accountant's balance done with the right photo (lesson 6), and the honest warning about when that archive isn't enough (lesson 7). This project is the moment to repeat the whole process, end to end, in a single continuous gesture — the same kind of integration you already did closing module 1.

The material: everything this module built, in one place

You need, in a new working directory:

kiosko_time_travel/
├── raw_orders.py                              (module 1, lesson 6: the fixed week of 40 orders)
└── kiosko_time_traveled_dim_product.py         (this project: brings the 7 pieces together)

With PyIceberg installed in your environment (pip install "pyiceberg[sql-sqlite,pyarrow]", module 1, lesson 4).

The reference solution, verified

# kiosko_time_traveled_dim_product.py -- module 3 closing project
# dim_product with no history column at all + time travel recovers P002 V1
import os
from collections import defaultdict
from datetime import datetime

import pyarrow as pa
from pyiceberg.catalog import load_catalog
from pyiceberg.schema import Schema
from pyiceberg.types import DoubleType, IntegerType, NestedField, StringType, TimestampType

from raw_orders import RAW_ORDERS

DIM_STORE = [
    {"store_id": "S01", "store_name": "Kiosko Centro", "city": "Bogota"},
    {"store_id": "S02", "store_name": "Kiosko Norte", "city": "Lima"},
    {"store_id": "S03", "store_name": "Kiosko Sur", "city": "Santiago"},
]
DIM_PRODUCT_V1 = [
    {"product_id": "P001", "product_name": "Bottled Water 600ml", "category": "beverages", "unit_cost": 0.40},
    {"product_id": "P002", "product_name": "Energy Bar", "category": "snacks", "unit_cost": 0.60},
    {"product_id": "P003", "product_name": "Instant Coffee Sachet", "category": "beverages", "unit_cost": 0.35},
    {"product_id": "P004", "product_name": "Phone Charger Cable", "category": "electronics", "unit_cost": 2.10},
]
DIM_PRODUCT_V2 = [
    {"product_id": "P001", "product_name": "Bottled Water 600ml", "category": "beverages", "unit_cost": 0.40},
    {"product_id": "P002", "product_name": "Energy Bar", "category": "health-snacks", "unit_cost": 0.68},
    {"product_id": "P003", "product_name": "Instant Coffee Sachet", "category": "beverages", "unit_cost": 0.35},
    {"product_id": "P004", "product_name": "Phone Charger Cable", "category": "electronics", "unit_cost": 2.10},
]

FACT_ORDERS_SCHEMA = Schema(
    NestedField(field_id=1, name="order_id", field_type=StringType(), required=True),
    NestedField(field_id=2, name="store_id", field_type=StringType(), required=True),
    NestedField(field_id=3, name="product_id", field_type=StringType(), required=True),
    NestedField(field_id=4, name="quantity", field_type=IntegerType(), required=True),
    NestedField(field_id=5, name="unit_price", field_type=DoubleType(), required=True),
    NestedField(field_id=6, name="revenue", field_type=DoubleType(), required=True),
    NestedField(field_id=7, name="order_ts", field_type=TimestampType(), required=True),
)
DIM_PRODUCT_SCHEMA = Schema(
    NestedField(field_id=1, name="product_id", field_type=StringType(), required=True),
    NestedField(field_id=2, name="product_name", field_type=StringType(), required=True),
    NestedField(field_id=3, name="category", field_type=StringType(), required=True),
    NestedField(field_id=4, name="unit_cost", field_type=DoubleType(), required=True),
)


def build_fact_orders_table() -> pa.Table:
    store_ids = {s["store_id"] for s in DIM_STORE}
    product_ids = {p["product_id"] for p in DIM_PRODUCT_V1}
    cols = {"order_id": [], "store_id": [], "product_id": [], "quantity": [],
            "unit_price": [], "revenue": [], "order_ts": []}
    for order_id, store_id, product_id, quantity, unit_price, ts in RAW_ORDERS:
        if store_id not in store_ids:
            raise ValueError(f"unknown store_id: {store_id}")
        if product_id not in product_ids:
            raise ValueError(f"unknown product_id: {product_id}")
        cols["order_id"].append(order_id)
        cols["store_id"].append(store_id)
        cols["product_id"].append(product_id)
        cols["quantity"].append(quantity)
        cols["unit_price"].append(unit_price)
        cols["revenue"].append(round(quantity * unit_price, 10))
        cols["order_ts"].append(datetime.fromisoformat(ts))
    schema = pa.schema([
        pa.field("order_id", pa.string(), nullable=False),
        pa.field("store_id", pa.string(), nullable=False),
        pa.field("product_id", pa.string(), nullable=False),
        pa.field("quantity", pa.int32(), nullable=False),
        pa.field("unit_price", pa.float64(), nullable=False),
        pa.field("revenue", pa.float64(), nullable=False),
        pa.field("order_ts", pa.timestamp("us"), nullable=False),
    ])
    return pa.Table.from_arrays(
        [
            pa.array(cols["order_id"], type=pa.string()),
            pa.array(cols["store_id"], type=pa.string()),
            pa.array(cols["product_id"], type=pa.string()),
            pa.array(cols["quantity"], type=pa.int32()),
            pa.array(cols["unit_price"], type=pa.float64()),
            pa.array(cols["revenue"], type=pa.float64()),
            pa.array(cols["order_ts"], type=pa.timestamp("us")),
        ],
        schema=schema,
    )


def dim_product_pa_table(rows: list[dict]) -> pa.Table:
    schema = pa.schema([
        pa.field("product_id", pa.string(), nullable=False),
        pa.field("product_name", pa.string(), nullable=False),
        pa.field("category", pa.string(), nullable=False),
        pa.field("unit_cost", pa.float64(), nullable=False),
    ])
    return pa.Table.from_pylist(rows, schema=schema)


def margin_by_category(dim_rows: list[dict], fact_rows: list[dict]) -> tuple[dict, dict]:
    dim_by_id = {r["product_id"]: r for r in dim_rows}
    revenue, margin = defaultdict(float), defaultdict(float)
    for f in fact_rows:
        d = dim_by_id[f["product_id"]]
        revenue[d["category"]] += f["revenue"]
        margin[d["category"]] += f["revenue"] - f["quantity"] * d["unit_cost"]
    return revenue, margin


def main() -> None:
    print("=== Kiosko: dim_product with no history columns + time travel ===\n")

    warehouse_path = os.path.abspath("kiosko_warehouse")
    catalog_db_path = os.path.abspath("kiosko_catalog.db")
    os.makedirs(warehouse_path, exist_ok=True)
    catalog = load_catalog(
        "kiosko", type="sql",
        uri=f"sqlite:///{catalog_db_path}", warehouse=f"file://{warehouse_path}",
    )
    catalog.create_namespace("kiosko")
    print(f"Step 1/7 -- catalog '{catalog.name}' and namespace 'kiosko' ready")

    fact_orders = catalog.create_table("kiosko.fact_orders", schema=FACT_ORDERS_SCHEMA)
    fact_orders.append(build_fact_orders_table())
    print(f"Step 2/7 -- kiosko.fact_orders loaded: {fact_orders.scan().to_arrow().num_rows} rows")

    dim_product = catalog.create_table("kiosko.dim_product", schema=DIM_PRODUCT_SCHEMA)
    print("Step 3/7 -- kiosko.dim_product created, columns:",
          [f.name for f in dim_product.schema().fields], "(no valid_from/valid_to/is_current)")

    dim_product.append(dim_product_pa_table(DIM_PRODUCT_V1))
    snap_v1 = dim_product.current_snapshot().snapshot_id
    print("Step 4/7 -- V1 loaded with table.append(), snap_v1 captured (P002=snacks/0.60)")

    dim_product.overwrite(dim_product_pa_table(DIM_PRODUCT_V2))
    print("Step 5/7 -- V2 loaded with table.overwrite() (P002=health-snacks/0.68)")

    current_rows = dim_product.scan().to_arrow().to_pylist()
    v1_rows = dim_product.scan(snapshot_id=snap_v1).to_arrow().to_pylist()
    current_p002 = next(r for r in current_rows if r["product_id"] == "P002")
    v1_p002 = next(r for r in v1_rows if r["product_id"] == "P002")
    print(f"Step 6/7 -- current P002 state: {current_p002['category']}/{current_p002['unit_cost']}"
          f" | AS OF snap_v1 P002: {v1_p002['category']}/{v1_p002['unit_cost']}")

    fact_rows = fact_orders.scan().to_arrow().to_pylist()
    revenue_broken, margin_broken = margin_by_category(current_rows, fact_rows)
    revenue_correct, margin_correct = margin_by_category(v1_rows, fact_rows)
    print("Step 7/7 -- margin by category calculated, broken vs. correct\n")

    print("=== Final verification ===\n")
    print("BROKEN (no time travel, category='health-snacks'):")
    for cat in sorted(revenue_broken):
        print(f"  {cat:14} revenue={round(revenue_broken[cat], 2):>6}  margin={round(margin_broken[cat], 2):>6}")
    print("\nCORRECT (AS OF snap_v1, category='snacks'):")
    for cat in sorted(revenue_correct):
        print(f"  {cat:14} revenue={round(revenue_correct[cat], 2):>6}  margin={round(margin_correct[cat], 2):>6}")

    total_revenue = round(sum(f["revenue"] for f in fact_rows), 2)
    print(f"\nTotal revenue (identical in both cases): {total_revenue}")

    assert dim_product.scan().to_arrow().num_rows == 4, "dim_product should have 4 rows (one per product)"
    assert current_p002["category"] == "health-snacks" and current_p002["unit_cost"] == 0.68
    assert v1_p002["category"] == "snacks" and v1_p002["unit_cost"] == 0.60
    assert round(margin_broken["health-snacks"], 2) == 9.36
    assert round(margin_correct["snacks"], 2) == 10.8
    assert total_revenue == 106.15
    assert len(dim_product.schema().fields) == 4, "dim_product must have exactly 4 columns, none for history"
    print("\nAll checks passed: 4 columns in dim_product (zero for history), "
          "broken margin=9.36, correct margin=10.8 via time travel, total revenue=106.15.")


if __name__ == "__main__":
    main()

(raw_orders.py is exactly the same file with the forty fixed orders from module 1, lesson 6 — not repeated here for space.)

What to expect (verified by running the actual python3 kiosko_time_traveled_dim_product.py, end to end, in a new directory; no snapshot_id is printed as a literal — this module's lesson 4 explains why):

=== Kiosko: dim_product with no history columns + time travel ===

Step 1/7 -- catalog 'kiosko' and namespace 'kiosko' ready
Step 2/7 -- kiosko.fact_orders loaded: 40 rows
Step 3/7 -- kiosko.dim_product created, columns: ['product_id', 'product_name', 'category', 'unit_cost'] (no valid_from/valid_to/is_current)
Step 4/7 -- V1 loaded with table.append(), snap_v1 captured (P002=snacks/0.60)
Step 5/7 -- V2 loaded with table.overwrite() (P002=health-snacks/0.68)
Step 6/7 -- current P002 state: health-snacks/0.68 | AS OF snap_v1 P002: snacks/0.6
Step 7/7 -- margin by category calculated, broken vs. correct

=== Final verification ===

BROKEN (no time travel, category='health-snacks'):
  beverages      revenue=  44.05  margin=  14.75
  electronics    revenue=   40.5  margin=   21.6
  health-snacks  revenue=   21.6  margin=   9.36

CORRECT (AS OF snap_v1, category='snacks'):
  beverages      revenue=  44.05  margin=  14.75
  electronics    revenue=   40.5  margin=   21.6
  snacks         revenue=   21.6  margin=    10.8

Total revenue (identical in both cases): 106.15

All checks passed: 4 columns in dim_product (zero for history), broken margin=9.36, correct margin=10.8 via time travel, total revenue=106.15.

Notice the seven assert statements at the end: they're not decorative. They confirm, in a single automatic pass, each of this module's central claims — that dim_product has the correct grain (four rows), that the current and historical states are the ones they should be, that both margins exactly match the numbers already verified by data-modeling and dbt, that the total revenue didn't move a cent, and that the table's schema never had more than four columns.

Diagram: where you came from, where you landed

flowchart LR
    A["Modules 1-2:\nfact_orders loaded,\nanatomy inspected"] --> B["Lesson 2:\ndim_product created,\nV1 loaded, snap_v1"]
    B --> C["Lesson 3:\ntable.overwrite(V2)\nP002 changes"]
    C --> D["Lesson 4-5:\nsnap_v1 correctly captured,\ntime travel executed"]
    D --> E["Lesson 6:\ncorrect margin 10.8\nrecovered, 0 columns"]
    E --> F["Lesson 7:\nthe honest limit,\nproven with evidence"]
    F --> G["This project:\nthe 7 pieces, one script,\nautomatic asserts"]
    G --> H["Module 4:\nschema evolution\n(dim_store + country)"]

Closing the module's promise, point by point

What lesson 1 promised	Evidence this module delivered it
Every write is a new snapshot	Lesson 2 (`append`), lesson 3 (`overwrite`, which revealed two internal snapshots: `delete` + `append`)
Recover `P002` V1 with no history column at all	`kiosko.dim_product`, four columns, verified in this project's final `assert`
The `snapshot-id` is never hardcoded	Lesson 4: evidence from two runs with completely different values, and the content-based recovery technique
Time travel `AS OF` a `snapshot-id`	Lesson 5: `table.scan(snapshot_id=snap_v1)`, executed, read-only
`P002`'s correct margin (`10.8`), matching `data-modeling` and `dbt`	Lesson 6 and this project: `margin_correct["snacks"] == 10.8`, verified with `assert`
The honest limit: what time travel does NOT solve	Lesson 7: the `toy_price` experiment, with evidence that no single `snapshot_id` serves two sales with different costs at once

This module didn't solve the general case of a dimension with multiple changes and facts spread across them — that boundary was stated, with evidence, in lesson 7. What this module delivers is exactly what it promised: Kiosko's favorable case, solved with pure time travel, with no history column at all, and verified number for number against the two previous guides in the ecosystem that already solved the same problem with different techniques.

Common mistakes

Running this project against a catalog that already has kiosko.fact_orders or kiosko.dim_product from an earlier lesson. What happens: someone runs this project in the same directory where they already completed lessons 2 through 7, and catalog.create_table(...) fails because the tables are already registered. Why it happens: this project deliberately repeats the whole creation from scratch, so it's self-contained and reproducible without depending on the exact state the earlier lessons left. How to spot it: if you see TableAlreadyExistsError when running kiosko_time_traveled_dim_product.py, you already have a catalog with those tables registered in the same directory. How to fix it: run this project in a new working directory, separate from where you did lessons 2 through 7 — as this lesson's "The material" section suggests.

Interpreting assert round(margin_broken["health-snacks"], 2) == 9.36 as a bug in the script. What happens: someone, reviewing the code, sees the script asserts a number its own section calls "BROKEN," and wonders whether that's a mistake in the project. Why it happens: it's easy to assume an assert always verifies "the correct thing," and get confused seeing this one deliberately verify the result the guide itself calls incorrect for the business. How to spot it: if you're unsure why the script verifies a "broken" number, reread the purpose: the assert doesn't say 9.36 is the correct business margin — it says the calculation without time travel consistently, reproducibly produces that specific number, exactly as data-modeling and dbt predict. How to fix it: nothing to fix — verifying the broken number is just as important as verifying the correct one, because it confirms the whole experiment — with and without time travel — behaves exactly as this guide predicted, not just the "nice" half of the result.

Adapting this project for a real case without having read lesson 7 first. What happens: someone takes this project's pattern — create a table with no history columns, rely on time travel to recover earlier versions — and applies it directly to a production problem, without checking whether that problem satisfies lesson 7's condition (every relevant fact on the same side of each dimension change). Why it happens: this project's result is clean and convincing, and it's tempting to copy the pattern without reviewing its applicability conditions. How to spot it: if your real case has a dimension that changes more than once, or facts with dates spread across both sides of a change, and you plan to use only table.scan(snapshot_id=...) to reconstruct history, you're at risk of this mistake. How to fix it: before applying this pattern outside this guide, review all of lesson 7 and explicitly confirm your case satisfies the favorable condition — if it doesn't, you need row-level valid_from/valid_to, not pure time travel.

Exercises

Exercise 1 — Run the whole project yourself, from scratch. In a new directory, with only raw_orders.py and kiosko_time_traveled_dim_product.py, run python3 kiosko_time_traveled_dim_product.py. Confirm you see the seven steps complete and the final "All checks passed" message.

See solution

If raw_orders.py is in the same directory and PyIceberg is installed, the output should exactly reproduce this lesson's structure: seven numbered steps, followed by the final verification with the four correct margins, 106.15 total revenue, and the success message with all seven conditions confirmed.

Exercise 2 — Break an assert on purpose, and watch it fail. Temporarily change DIM_PRODUCT_V2 so P002 has unit_cost=0.70 instead of 0.68, run the script again, and observe which assert fails first. Then revert the change.

See solution

The first assert to fail should be assert current_p002["category"] == "health-snacks" and current_p002["unit_cost"] == 0.68 — because you changed V2's value, the current state no longer matches what the script expects. If you only fixed that assert to accept 0.70, the next one to fail would be assert round(margin_broken["health-snacks"], 2) == 9.36, because the broken margin would change with the new cost. This exercise demonstrates that this project's assert statements are chained to Kiosko's exact canonical values — any deviation gets caught immediately, at the first point it stops holding.

Exercise 3 — Explain, in your own words, why this project verifies seven different conditions instead of just the final margin. In 3-4 sentences, justify why this script's final assert block doesn't just check margin_correct["snacks"] == 10.8, but also checks the grain, the schema, and the broken result.

See solution

Verifying only the final number — 10.8 — would confirm the result is correct, but wouldn't confirm why it's correct: it could reach that number by coincidence, with a table of the wrong grain, or with a schema that does have hidden history columns. Verifying all seven conditions — the grain (4 rows), the schema (exactly 4 columns), the broken state (9.36) and the correct one (10.8), plus the total revenue — confirms the correct result arrives by the correct path: a table with exactly the design this module set out to build, not just a number that happens to match at the end. It's the same "verify, don't trust" discipline you already saw in module 1's project, applied here to an experiment with more moving pieces.

Summary and next step: closing this module

With this project you close module 3. You integrated the seven previous lessons — creating dim_product with no history columns, two writes, the snapshot_id capture discipline, the trip through time, the correct margin recovered, and the technique's honest limit — into a single script, run end to end, with automatic assert statements confirming every number against data-modeling-for-analytics-guide and dbt-analytics-engineering-guide.

For the first time in this ecosystem, Kiosko has a dimension with recoverable history without having designed a single column for it. kiosko.dim_product still has exactly four business columns — the mechanism that makes recovering P002 V1 possible lives entirely in the snapshot, not in the schema.

Where you go next. Module 4 — Schema evolution without rewriting — takes kiosko.dim_store, Kiosko's three-store table, and adds it a new column — country, derived from city — without rewriting a single existing data file. You're going to see why that guarantee — adding, renaming, or dropping a column without touching Parquet files already written — depends on exactly the same field_id you already learned in module 1, and what "ACID" guarantees in this precise context.

Resources

PyIceberg — official documentation (quickstart), the complete catalog, table, append(), and overwrite() flow this project integrates. py.iceberg.apache.org. In English.
PyIceberg — API reference, table.scan(snapshot_id=...), table.current_snapshot(), table.history(). py.iceberg.apache.org/api. In English.
data-modeling-for-analytics-guide DESIGN doc — source of the canonical 9.36/10.8 numbers this project verifies with assert. src/guides/data-modeling-for-analytics-guide/DISENO.md. In Spanish.
dbt-analytics-engineering-guide DESIGN doc — source of dbt snapshot, the second independent confirmation of the same numbers. src/guides/dbt-analytics-engineering-guide/DISENO.md. In Spanish.
This guide's DESIGN doc — the full map of the eight modules, including module 4 which follows. src/guides/lakehouse-and-iceberg-guide/DISENO.md. In Spanish.

← PreviousWhat time travel does not replace Next →Module overview: schema evolution without rewriting