Homework 10 — Iceberg Extensions

Task 1 — Force a write conflict and observe the retry

Open two Python processes (two terminal tabs, or two notebooks). In each process, call create_or_load_events() to get a handle on the same table. Prepare a minimal Arrow table in each process (two small rows of distinct data), then call table.append(…) in both processes as close in time as possible — ideally within 100 ms of each other.

Observe what happens:

• One process commits on the first attempt.
• The other process may raise CommitFailedException once, after which pyiceberg retries internally against the updated metadata head, and the second commit lands cleanly.

After both processes finish, inspect the snapshot list:

from iceberg.tables import create_or_load_events
t = create_or_load_events()
for s in t.snapshots():
    print(s.snapshot_id, s.timestamp_ms)

Self-check: the snapshot list shows two new snapshots (or more, if you ran the lab steps earlier). Each snapshot contains exactly the rows from one of the two appends — no data is lost and no rows from one process bleed into the other’s snapshot.

Task 2 — Snapshot-based regression triage

Simulate a “broken DAG” scenario by writing two batches to the table:

1. Batch N (bad): append a small Arrow table where every row has actor = None (all-null actor column). This represents a broken pipeline that dropped actor data.
2. Batch N+1 (corrected): append a second Arrow table with proper actor values.

Write a short Python script that loads the table, takes the last two snapshots, and prints the delta in type distribution between them — i.e., how did the event-type counts change between snapshot N and snapshot N+1?

Self-check: the script prints per-type deltas for every type that changed between the two snapshots. Running it on your two batches should show the corrected batch’s type distribution added on top of the bad batch’s distribution.

Task 3 — Add an event_hour partition column

Extend the table schema and partition spec to add hourly partitioning alongside the existing daily partitioning. Do this in two phases, in order:

1. Evolve the schema — add a new column event_hour of type IntegerType:

   from pyiceberg.types import IntegerType
   from iceberg.tables import create_or_load_events
   
   t = create_or_load_events()
   with t.update_schema() as upd:
       upd.add_column("event_hour", IntegerType())

2. Evolve the partition spec — add event_hour as an identity partition field:

   from pyiceberg.transforms import IdentityTransform
   
   with t.update_spec() as spec:
       spec.add_field("event_hour", IdentityTransform(), "event_hour")

After both updates, write one small new batch that populates event_hour, then query all data (old partitions + new partition) in a single DuckDB query via query_duckdb.py.

Self-check: iceberg/tables.py still imports cleanly with python -m py_compile iceberg/tables.py. The DuckDB query returns a unified result across both partition specs (old event_date-only partitions and new event_date + event_hour partitions) without any manual SQL gymnastics. Old rows have NULL for event_hour.