Project: Kiosko's evolved dim_store

Description

This project closes module 4. You know why an Iceberg write is atomic, with evidence from 490 concurrent reads that never captured a halfway state (lesson 2). You know how to add a column without touching any data file (lesson 3), and how to rename or drop one without that same cost (lesson 4). You populated country deterministically from city (lesson 5), confirmed a snapshot before the evolution still reads its own schema (lesson 6), and saw, with a real conflict between two writers, exactly what "ACID" guarantees in the context of an Iceberg table (lesson 7). One step is left: bringing the seven pieces together in a single script, run end to end, with automatic assert statements confirming every claim.

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: evolving kiosko.dim_store's schema — adding, populating, renaming, dropping — without rewriting a single existing data file, and with no reader, at any moment, ever seeing the table halfway between one state and the next.

An analogy: the complete form, reviewed end to end

Lessons 2 through 7 of this module built, one piece at a time, the complete evidence that Iceberg's schema evolution is safe: the proof that no reader sees an intermediate state (lesson 2), the mechanism for adding a column without touching data (lesson 3), the same guarantee for renaming and dropping (lesson 4), the real payoff — country populated (lesson 5), the confirmation that the past still reads with its own schema (lesson 6), and the precise limit of the word "ACID" (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 modules 1 and 3.

The material: everything this module built, in one place

You need, in a new working directory:

kiosko_dim_store_evolution/
└── kiosko_evolved_dim_store.py         (this project: brings the 6 data pieces together)

With PyIceberg installed in your environment (pip install "pyiceberg[sql-sqlite,pyarrow]", module 1, lesson 4). This project is self-contained: it creates kiosko.dim_store from scratch, so it doesn't depend on any file from this module's earlier lessons — only on the kiosko catalog existing in the directory you run it from (if you already have kiosko.fact_orders or kiosko.dim_product from modules 1 through 3 in that same catalog, this project leaves them intact; if you don't have them, kiosko.dim_store still gets created on its own).

The reference solution, verified

# kiosko_evolved_dim_store.py -- module 4 closing project
# dim_store evolves (add country, rename+drop temp_notes) with no existing Parquet ever rewritten
import os

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

DIM_STORE_V1 = [
    {"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"},
]
CITY_TO_COUNTRY = {"Bogota": "Colombia", "Lima": "Peru", "Santiago": "Chile"}

DIM_STORE_SCHEMA_V1 = Schema(
    NestedField(field_id=1, name="store_id", field_type=StringType(), required=True),
    NestedField(field_id=2, name="store_name", field_type=StringType(), required=True),
    NestedField(field_id=3, name="city", field_type=StringType(), required=True),
)


def dim_store_v1_pa_table() -> pa.Table:
    schema = pa.schema([
        pa.field("store_id", pa.string(), nullable=False),
        pa.field("store_name", pa.string(), nullable=False),
        pa.field("city", pa.string(), nullable=False),
    ])
    return pa.Table.from_pylist(DIM_STORE_V1, schema=schema)


def dim_store_with_country_pa_table() -> pa.Table:
    schema = pa.schema([
        pa.field("store_id", pa.string(), nullable=False),
        pa.field("store_name", pa.string(), nullable=False),
        pa.field("city", pa.string(), nullable=False),
        pa.field("country", pa.string(), nullable=True),
    ])
    rows = [{**s, "country": CITY_TO_COUNTRY[s["city"]]} for s in DIM_STORE_V1]
    return pa.Table.from_pylist(rows, schema=schema)


def data_file_names(table) -> list:
    return sorted(f["file_path"].split("/")[-1] for f in table.inspect.files().to_pylist())


def main() -> None:
    print("=== Kiosko: dim_store evolved with no data rewritten ===\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/8 -- catalog '{catalog.name}' and namespace 'kiosko' ready")

    dim_store = catalog.create_table("kiosko.dim_store", schema=DIM_STORE_SCHEMA_V1)
    dim_store.append(dim_store_v1_pa_table())
    snap_before_evolution = dim_store.current_snapshot().snapshot_id
    files_before_evolution = data_file_names(dim_store)
    print(f"Step 2/8 -- dim_store loaded: {dim_store.scan().to_arrow().num_rows} rows, "
          f"snap_before_evolution captured, {len(files_before_evolution)} data file(s)")

    with dim_store.update_schema() as update:
        update.add_column("country", StringType())
    files_after_add_column = data_file_names(dim_store)
    rows_after_add_column = dim_store.scan().to_arrow().to_pylist()
    print(f"Step 3/8 -- add_column('country') applied. Data files unchanged: "
          f"{files_before_evolution == files_after_add_column}. "
          f"country in existing rows: {[r['country'] for r in rows_after_add_column]}")

    with dim_store.update_schema() as update:
        update.rename_column("store_name", "outlet_name")
    with dim_store.update_schema() as update:
        update.rename_column("outlet_name", "store_name")
    with dim_store.update_schema() as update:
        update.add_column("temp_notes", StringType())
    with dim_store.update_schema() as update:
        update.delete_column("temp_notes")
    files_after_schema_ops = data_file_names(dim_store)
    schema_names_after_ops = [f.name for f in dim_store.schema().fields]
    print(f"Step 4/8 -- rename store_name<->outlet_name + add/drop temp_notes. "
          f"Data files unchanged: {files_before_evolution == files_after_schema_ops}. "
          f"Final schema: {schema_names_after_ops}")

    dim_store.overwrite(dim_store_with_country_pa_table())
    print("Step 5/8 -- table.overwrite() populates 'country' from 'city' for the 3 existing rows")

    current_rows = dim_store.scan().to_arrow().to_pylist()
    print("Step 6/8 -- dim_store's current state:")
    for row in sorted(current_rows, key=lambda r: r["store_id"]):
        print(f"    {row['store_id']}  {row['store_name']:<15} {row['city']:<10} country={row['country']}")

    old_scan = dim_store.scan(snapshot_id=snap_before_evolution).to_arrow()
    old_rows = old_scan.to_pylist()
    print(f"Step 7/8 -- table.scan(snapshot_id=snap_before_evolution) -- "
          f"columns: {old_scan.schema.names}, rows: {len(old_rows)}")

    total_snapshots = len(dim_store.history())
    print(f"Step 8/8 -- table.history() has {total_snapshots} entries total\n")

    print("=== Final verification ===\n")

    assert files_before_evolution == files_after_add_column, "add_column must not touch the data files"
    assert files_before_evolution == files_after_schema_ops, "rename/add/drop must not touch the data files"
    assert all(r["country"] is None for r in rows_after_add_column), \
        "after add_column, existing rows must have country=None (not populated yet)"
    assert schema_names_after_ops == ["store_id", "store_name", "city", "country"], \
        "the final schema must not keep temp_notes or the rename"

    country_by_store = {r["store_id"]: r["country"] for r in current_rows}
    assert country_by_store == {"S01": "Colombia", "S02": "Peru", "S03": "Chile"}

    assert old_scan.schema.names == ["store_id", "store_name", "city"], \
        "the snapshot before the evolution must read with the 3-column schema, no country"
    assert len(old_rows) == 3
    assert all("country" not in r for r in old_rows)

    print("All checks passed:")
    print("  - add_column('country') didn't touch any existing data file")
    print("  - rename_column + add/delete_column('temp_notes') didn't touch data files either")
    print("  - country ended up deterministically populated from city (S01=Colombia, S02=Peru, S03=Chile)")
    print("  - table.scan(snapshot_id=snap_before_evolution) still reads the 3-column schema, no country")


if __name__ == "__main__":
    main()

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

=== Kiosko: dim_store evolved with no data rewritten ===

Step 1/8 -- catalog 'kiosko' and namespace 'kiosko' ready
Step 2/8 -- dim_store loaded: 3 rows, snap_before_evolution captured, 1 data file(s)
Step 3/8 -- add_column('country') applied. Data files unchanged: True. country in existing rows: [None, None, None]
Step 4/8 -- rename store_name<->outlet_name + add/drop temp_notes. Data files unchanged: True. Final schema: ['store_id', 'store_name', 'city', 'country']
Step 5/8 -- table.overwrite() populates 'country' from 'city' for the 3 existing rows
Step 6/8 -- dim_store's current state:
    S01  Kiosko Centro   Bogota     country=Colombia
    S02  Kiosko Norte    Lima       country=Peru
    S03  Kiosko Sur      Santiago   country=Chile
Step 7/8 -- table.scan(snapshot_id=snap_before_evolution) -- columns: ['store_id', 'store_name', 'city'], rows: 3
Step 8/8 -- table.history() has 3 entries total

=== Final verification ===

All checks passed:
  - add_column('country') didn't touch any existing data file
  - rename_column + add/delete_column('temp_notes') didn't touch data files either
  - country ended up deterministically populated from city (S01=Colombia, S02=Peru, S03=Chile)
  - table.scan(snapshot_id=snap_before_evolution) still reads the 3-column schema, no country

Notice step 8: table.history() reports three entries, not four or five, even though this script ran add_column, two rename_columns, add_column again, delete_column, and finally overwrite(). The three entries are, in order: the original append (step 2), and the overwrite()'s internal delete+append that populated country (step 5) — exactly the same pattern module 3 found for the P002 change. None of the five schema operations from step 3 and step 4 added a single entry to the snapshot history, because none of them is a data write — it's, precisely, this whole module's consolidated evidence, in a single number.

Diagram: where you came from, where you landed

flowchart LR
    A["Modules 1-3:\nfact_orders, dim_product,\ntime travel"] --> B["Lesson 2:\n490 reads, 0 intermediate\nstates -- atomicity"]
    B --> C["Lesson 3:\nadd_column('country')\n0 files touched"]
    C --> D["Lesson 4:\nrename + add/drop\ntemp_notes, same guarantee"]
    D --> E["Lesson 5:\ncountry populated:\nColombia/Peru/Chile"]
    E --> F["Lesson 6:\nsnap_before_evolution\nreads its own schema"]
    F --> G["Lesson 7:\nreal CommitFailedException,\nACID's precise limit"]
    G --> H["This project:\nthe 6 data pieces,\none script, automatic assert"]
    H --> I["Module 5:\nhidden partitioning and\npartition evolution"]

Closing the module's promise, point by point

What lesson 1 promised	Evidence this module delivered it
Why overwrite-partition was never atomic, with the exact quote	Lesson 2: data-engineering-foundations-guide's literal quote, and 490 concurrent reads that never saw an intermediate state in Iceberg
Adding a column without rewriting data	Lesson 3: add_column('country'), same file list before and after, table.history() unchanged
Renaming and dropping columns with the same guarantee	Lesson 4: rename_column back and forth, temp_notes added and dropped, field_id never reused
country populated from city, deterministically	Lesson 5 and this project: country_by_store == {"S01": "Colombia", "S02": "Peru", "S03": "Chile"}, verified with assert
A snapshot before the evolution still reads its own schema	Lesson 6 and this project: old_scan.schema.names == ["store_id", "store_name", "city"], no country
What "ACID" guarantees here, precisely	Lesson 7: real CommitFailedException, with the exact message from a writer conflict, and the explicit boundary against multi-table transactions

This module didn't evolve kiosko.fact_orders's or kiosko.dim_product's schema — those tables stay exactly as module 3 left them — nor did it touch any partitioning — that arrives in module 5. What this module delivers is exactly what it promised: a new table, kiosko.dim_store, evolved from three columns to four, with data populated deterministically, with not a single existing data file rewritten at any step, and with atomicity and isolation guarantees verified with real code, not merely quoted from documentation.

Common mistakes

Running this project against a catalog that already has kiosko.dim_store from an earlier lesson in this module. What happens: someone runs this project in the same directory where they already completed lessons 3 through 7, and catalog.create_table(...) fails because the table is 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 earlier lessons left. How to spot it: if you see TableAlreadyExistsError when running kiosko_evolved_dim_store.py, you already have a catalog with kiosko.dim_store registered in the same directory. How to fix it: run this project in a new working directory, separate from where you did lessons 3 through 7 — as this lesson's "The material" section suggests.

Expecting table.history() to report 8 entries, one per numbered step in the script. What happens: someone, seeing the "Step 1/8" through "Step 8/8" the script prints, assumes each one corresponds to a new snapshot, and is surprised to see only 3 in table.history(). Why it happens: the "Step N/8" numbers count the script's narrative stages — catalog creation, load, schema evolution, population, historical read, verification — not each individual update_schema() call or write operation. How to spot it: if your expected snapshot count matches the number of "steps" printed on screen, revisit the distinction between schema operations (which create no snapshots) and data operations (which do) this module's lessons 3 and 5 already established. How to fix it: count snapshots with table.history() or table.inspect.snapshots() directly — the correct answer is 3: the original append, and the internal delete+append from the single overwrite() this project runs.

Exercises

Exercise 1 — Run the whole project yourself, from scratch. In a new directory, run python3 kiosko_evolved_dim_store.py. Confirm you see the eight steps complete and the final message with the four checks.

See solution

If PyIceberg is installed in your environment, the output should exactly reproduce this lesson's structure: eight numbered steps, followed by the final verification with the three correct countries, the three-column schema confirmed for the historical snapshot, and the success message with all four conditions confirmed.

Exercise 2 — Break an assert on purpose, and watch it fail. Temporarily change CITY_TO_COUNTRY["Lima"] from "Peru" to "Perú" (with an accent), run the script again, and observe which assert fails first. Then revert the change.

See solution

The assert that fails is assert country_by_store == {"S01": "Colombia", "S02": "Peru", "S03": "Chile"} — because the dictionary now produces "Perú" with an accent for S02, which no longer matches, character for character, the expected "Peru" with no accent. This exercise demonstrates two things at once: that this project's assert statements are chained to CITY_TO_COUNTRY's exact canonical values, and that this guide, following the whole ecosystem's hard convention that code identifiers and data go in English (with no accents or special characters), uses "Peru" with no accent as the correct business value, even though this same lesson's Spanish-language prose does write "Perú" with an accent when referring to the country.

Exercise 3 — Explain, in your own words, why this project verifies files_before_evolution == files_after_schema_ops instead of only verifying country's final result. In 3-4 sentences, justify why the data-files assert is just as important as the assert on country's final value.

See solution

Verifying only the final result — that country has the three correct countries — would confirm the data is correct, but wouldn't confirm how it got there: a different implementation, one that rewrote every data file on every schema operation (the behavior this whole module argues Iceberg avoids), could arrive at exactly the same final result, with none of the advantages this module set out to demonstrate. Verifying files_before_evolution == files_after_schema_ops confirms the module's central claim — that adding, renaming, and dropping columns are metadata operations, not data ones — with direct evidence on the filesystem, not just a query's correct result. It's the same "verify the path, not just the destination" discipline you already saw in module 3's project, where the "broken" margin's assert was just as important as the "correct" one's.

Summary and next step: closing this module

With this project you close module 4. You integrated the seven previous lessons — data-engineering-foundations-guide's exact quote and its contrast with Iceberg's atomicity, the mechanism for adding/renaming/dropping columns without rewriting data, country's real population, correctly reading the past, and "ACID"'s precise limit — into a single script, run end to end, with automatic assert statements confirming every claim with evidence, not a promise.

For the first time in this ecosystem, Kiosko has a dimension table whose schema evolved after the data already existed, with none of those operations touching a single already-written Parquet file. kiosko.dim_store went from three columns to four, with country deterministically derived from city — and the mechanism that made it possible is the same one that guarantees, in any Iceberg table, that no reader ever sees a schema halfway through changing.

Where you go next. Module 5 — Hidden partitioning and partition evolution — takes kiosko.fact_orders_at_scale, the ten-million-row table inherited from spark-and-distributed-processing-guide, and contrasts Spark's folder-based partitioning (partitionBy("store_id"), visible, you have to know the structure) with Iceberg's hidden partitioning — the query filters by the business column, the engine decides the layout — and its evolution going forward, with no rewrite of the ten million rows already existing.

Resources

• PyIceberg — official documentation (quickstart), the complete catalog, table, append(), update_schema(), and overwrite() flow this project integrates. py.iceberg.apache.org. In English.
• PyIceberg — API reference, table.update_schema(), table.scan(snapshot_id=...), table.history(), table.inspect.files(). py.iceberg.apache.org/api. In English.
• Apache Iceberg — official documentation, "Evolution" and "Reliability," the formal source for the guarantees this project verifies with assert. iceberg.apache.org/docs/latest/evolution · iceberg.apache.org/docs/latest/reliability. In English.
• data-engineering-foundations-guide DESIGN doc — source of stores's original schema and the exact quote about overwrite-partition's risk this whole module contrasts against. src/guides/data-engineering-foundations-guide/DISENO.md. In Spanish.
• This guide's DESIGN doc — the full map of the eight modules, including module 5 which follows. src/guides/lakehouse-and-iceberg-guide/DISENO.md. In Spanish.