Homework 10 — Solutions

Lesson 10 · Data Engineering with Python

Solution to Task 1 — Force a write conflict and observe the retry

The key is to call table.append(…) in two processes within a short window so that the catalog’s version pointer is still the same when both processes read it. Here is a minimal sketch for the two scripts:

# proc1.py  (run from data-engineering/sample-pipeline/)
import pyarrow as pa
from datetime import date
from iceberg.tables import create_or_load_events

t = create_or_load_events()
batch = pa.table({
    "event_id":   pa.array([1], type=pa.int64()),
    "type":       pa.array(["PushEvent"]),
    "actor":      pa.array(["alice"]),
    "repo":       pa.array(["alice/repo-a"]),
    "created_at": pa.array([1748300000000000], type=pa.timestamp("us")),
    "event_date": pa.array([date(2026, 5, 26)], type=pa.date32()),
})
t.append(batch)
print("proc1: committed")

# proc2.py  (run from data-engineering/sample-pipeline/)
import pyarrow as pa
from datetime import date
from iceberg.tables import create_or_load_events

t = create_or_load_events()
batch = pa.table({
    "event_id":   pa.array([2], type=pa.int64()),
    "type":       pa.array(["IssuesEvent"]),
    "actor":      pa.array(["bob"]),
    "repo":       pa.array(["bob/repo-b"]),
    "created_at": pa.array([1748300001000000], type=pa.timestamp("us")),
    "event_date": pa.array([date(2026, 5, 26)], type=pa.date32()),
})
t.append(batch)
print("proc2: committed")

Run both scripts within about one second of each other in two terminals:

python proc1.py & python proc2.py

Expected log: one process prints “committed” immediately; the other may log an internal CommitFailedException once before pyiceberg’s retry loop rebuilds the manifest against the new metadata head and commits cleanly. Both processes always print “committed” — the failure is handled internally unless retries are exhausted.

Common mistake: assuming the retry requires application code. pyiceberg handles the retry internally up to commit_max_retries attempts (default: 4). The user-facing experience is silent unless retries exhaust. If you want to observe the retry in your logs, set PYICEBERG_LOG_LEVEL=DEBUG before running the script.

Solution to Task 2 — Snapshot-based regression triage

Complete script for the per-type delta between the last two snapshots:

from iceberg.tables import create_or_load_events
from collections import Counter

t = create_or_load_events()
snaps = t.snapshots()

if len(snaps) < 2:
    print("Need at least 2 snapshots; run write_from_parquet twice first.")
    raise SystemExit(1)

snap_n_minus_1 = snaps[-2]
snap_n         = snaps[-1]


def type_counts(snapshot_id: int) -> Counter:
    arrow = t.scan(snapshot_id=snapshot_id).to_arrow()
    return Counter(arrow["type"].to_pylist())


before = type_counts(snap_n_minus_1.snapshot_id)
after  = type_counts(snap_n.snapshot_id)

all_types = sorted(set(before) | set(after))
print(f"Delta from snapshot {snap_n_minus_1.snapshot_id} to {snap_n.snapshot_id}:")
for event_type in all_types:
    delta = after.get(event_type, 0) - before.get(event_type, 0)
    if delta != 0:
        print(f"  {event_type}: {delta:+d}")

Common mistake: comparing the current snapshot to itself by using snaps[0] and snaps[-1] when the table has been expired and only one snapshot remains. Always use snaps[-2] and snaps[-1] (checking len(snaps) >= 2 first) rather than assuming position 0 is an old snapshot. After expire_snapshots(--retention-days 0) in the lab, you may need to run two more appends before this script has two snapshots to compare.

Solution to Task 3 — Add an `event_hour` partition column

The critical constraint: schema evolution must come before partition-spec evolution. The new partition field must exist as a schema column before you can reference it in the spec.

from pyiceberg.types import IntegerType
from pyiceberg.transforms import IdentityTransform
from iceberg.tables import create_or_load_events

t = create_or_load_events()

# Step 1: evolve the schema (add the column first)
with t.update_schema() as upd:
    upd.add_column("event_hour", IntegerType())

# Step 2: evolve the partition spec (reference the new column)
with t.update_spec() as spec:
    spec.add_field("event_hour", IdentityTransform(), "event_hour")

print("Schema and partition spec updated.")
print("New schema fields:", [f.name for f in t.schema().fields])
print("Active partition spec:", t.spec())

Write one small batch with event_hour populated, then query both old and new partitions in DuckDB:

import pyarrow as pa
from datetime import date

batch = pa.table({
    "event_id":   pa.array([9001], type=pa.int64()),
    "type":       pa.array(["PushEvent"]),
    "actor":      pa.array(["carol"]),
    "repo":       pa.array(["carol/repo-c"]),
    "created_at": pa.array([1748310000000000], type=pa.timestamp("us")),
    "event_date": pa.array([date(2026, 5, 27)], type=pa.date32()),
    "event_hour": pa.array([14], type=pa.int32()),
})
t.append(batch)
print("New partition written.")

from iceberg.query_duckdb import open_with_iceberg

# Reopen after schema change so DuckDB picks up the new metadata:
with open_with_iceberg() as con:
    rows = con.execute(
        "SELECT event_date, event_hour, type, count(*) AS n "
        "FROM events "
        "GROUP BY event_date, event_hour, type "
        "ORDER BY event_date DESC, event_hour NULLS LAST "
        "LIMIT 10"
    ).fetchall()
    for row in rows:
        print(row)

Expected: old rows appear with event_hour = NULL; the new row appears with event_hour = 14. The DuckDB query spans both partition specs transparently.

Common mistake: calling update_spec() before update_schema(). The partition spec references the new field by name; if the field does not exist in the schema yet, pyiceberg raises a validation error. Always evolve schema first, partition spec second.

Homework 10 — Solutions

Solution to Task 1 — Force a write conflict and observe the retry

Solution to Task 2 — Snapshot-based regression triage

Solution to Task 3 — Add an event_hour partition column

Solution to Task 3 — Add an `event_hour` partition column