#!/usr/bin/env python3 """Capture harness for the DDIA Ch. 8 update-bloat / MVCC exercise. Demonstrates that in PostgreSQL an UPDATE is internally a DELETE + INSERT. Updating ONE row 100,000 times leaves exactly one LIVE tuple but writes 100,000 row versions; whether the dead ones pile up visibly depends on Heap-Only Tuple (HOT) pruning: Scenario A -- t(id, v), v NOT indexed. Updates are HOT-eligible, so Postgres prunes dead versions within a page as it goes. n_dead_tup stays small, yet the heap STILL bloats many-fold and the row's ctid marches across dozens of pages. Scenario B -- t2(id, v) with an INDEX on v. Each update must add an index entry, so it is NOT a HOT update and cannot be pruned in place. Now the ~100,000 dead tuples the book's sentence predicts actually pile up, plus a bloated index -- until VACUUM. Then VACUUM reclaims the dead tuples (n_dead_tup -> 0, space marked reusable). This is the CAPTURE / VERIFY path, not the reader path. A reader can do the whole thing by hand in plain `psql` (the .md/.html show the exact statements); this script drives the same statements programmatically so the printed numbers are real, verbatim measurements the write-up quotes. It is idempotent: it drops and recreates its own database `ex_ch08_bloat` on the shared, already-running server, disables autovacuum on both tables so the version pile-up is observable, runs the updates, measures, VACUUMs, measures again. Exact dead-tuple counts and sizes depend on HOT pruning and page fill, so the script reports whatever it really observes. Run: uv run --python 3.12 --with "psycopg[binary]" python3 code/update_bloat_mvcc.py Substrate (already running; this script does NOT manage containers): PostgreSQL 16 container `ddia-ch08-pg` on localhost:5432, user=postgres, trust auth, maintenance db `study`. """ import psycopg ADMIN_DSN = "host=localhost port=5432 user=postgres dbname=study" DBNAME = "ex_ch08_bloat" EX_DSN = f"host=localhost port=5432 user=postgres dbname={DBNAME}" N_UPDATES = 100_000 def recreate_database() -> None: """Drop and recreate our own database via an autocommit admin connection.""" with psycopg.connect(ADMIN_DSN, autocommit=True) as admin: admin.execute(f"DROP DATABASE IF EXISTS {DBNAME} WITH (FORCE)") admin.execute(f"CREATE DATABASE {DBNAME}") def seed() -> None: """Create both tables (t: v unindexed / HOT; t2: v indexed / non-HOT).""" with psycopg.connect(EX_DSN, autocommit=True) as conn: conn.execute("CREATE EXTENSION IF NOT EXISTS pgstattuple") conn.execute("DROP TABLE IF EXISTS t") conn.execute("CREATE TABLE t (id int PRIMARY KEY, v int)") conn.execute("ALTER TABLE t SET (autovacuum_enabled = false)") conn.execute("INSERT INTO t VALUES (1, 0)") conn.execute("DROP TABLE IF EXISTS t2") conn.execute("CREATE TABLE t2 (id int PRIMARY KEY, v int)") conn.execute("ALTER TABLE t2 SET (autovacuum_enabled = false)") # Indexing v makes every UPDATE touch an index -> NOT a HOT update. conn.execute("CREATE INDEX t2_v ON t2 (v)") conn.execute("INSERT INTO t2 VALUES (1, 0)") def ctid_of(conn, table: str) -> str: return conn.execute(f"SELECT ctid FROM {table} WHERE id = 1").fetchone()[0] def measure(conn, table: str) -> dict: """Live/dead tuple counts and heap size, from the stats system + catalog.""" conn.execute(f"ANALYZE {table}") # refresh n_live_tup / n_dead_tup live, dead = conn.execute( "SELECT n_live_tup, n_dead_tup FROM pg_stat_user_tables WHERE relname = %s", (table,), ).fetchone() size_bytes, size_pretty = conn.execute( f"SELECT pg_relation_size('{table}'), pg_size_pretty(pg_relation_size('{table}'))" ).fetchone() pst = conn.execute( f"SELECT tuple_count, dead_tuple_count, dead_tuple_percent, free_percent " f"FROM pgstattuple('{table}')" ).fetchone() return { "n_live_tup": live, "n_dead_tup": dead, "size_bytes": size_bytes, "size_pretty": size_pretty, "pst_tuple_count": pst[0], "pst_dead_tuple_count": pst[1], "pst_dead_tuple_percent": pst[2], "pst_free_percent": pst[3], } def print_measure(tag: str, m: dict) -> None: print(f" {tag}") print(f" live tuples (n_live_tup) = {m['n_live_tup']}") print(f" dead tuples (n_dead_tup) = {m['n_dead_tup']}") print(f" heap size (pg_relation_size) = {m['size_bytes']} bytes ({m['size_pretty']})") print(f" pgstattuple: tuple_count={m['pst_tuple_count']} " f"dead_tuple_count={m['pst_dead_tuple_count']} " f"dead%={m['pst_dead_tuple_percent']} free%={m['pst_free_percent']}") def run_updates(conn, table: str, n: int) -> None: for _ in range(n): conn.execute(f"UPDATE {table} SET v = v + 1 WHERE id = 1") def main() -> None: recreate_database() seed() with psycopg.connect(EX_DSN, autocommit=True) as conn: print("Two one-row tables (id, v) seeded (1, 0), autovacuum disabled on both.") print(" t -- v NOT indexed -> UPDATEs are HOT-eligible") print(" t2 -- v indexed -> UPDATEs are NOT HOT (each adds an index entry)\n") # --- Step 1: initial state --- print("[1] Initial state of t (one live row):") first_ctid = ctid_of(conn, "t") print(f" ctid of row id=1 = {first_ctid}") print_measure("before any updates:", measure(conn, "t")) print() # --- Step 2: the ctid moves on every update --- print("[2] ctid of t's row id=1 over 6 successive UPDATE t SET v=v+1 WHERE id=1:") ctids = [first_ctid] for _ in range(6): conn.execute("UPDATE t SET v = v + 1 WHERE id = 1") ctids.append(ctid_of(conn, "t")) for i, c in enumerate(ctids): note = "" if i == 0 else (" <- moved" if ctids[i] != ctids[i - 1] else " (same slot)") print(f" after update {i}: ctid = {c}{note}") row = conn.execute("SELECT ctid, xmin, xmax, v FROM t WHERE id = 1").fetchone() print(f" live row now: ctid={row[0]} xmin={row[1]} xmax={row[2]} v={row[3]}") print(" (xmin = txid that created this version; each UPDATE writes a new one)") print() # --- Step 3: 100k updates, HOT-eligible table --- remaining = N_UPDATES - 6 print(f"[3] Scenario A -- {remaining:,} more updates of t " f"({N_UPDATES:,} total; v unindexed, HOT-eligible)...") run_updates(conn, "t", remaining) a_after = measure(conn, "t") a_ctid = ctid_of(conn, "t") v_final = conn.execute("SELECT v FROM t WHERE id = 1").fetchone()[0] print(f" done. v = {v_final}; ctid of row id=1 = {a_ctid} (was {first_ctid})") print_measure("t after 100k HOT-eligible updates:", a_after) print() # --- Step 4: 100k updates, non-HOT table --- print(f"[4] Scenario B -- {N_UPDATES:,} updates of t2 " f"(v INDEXED, so each UPDATE is NON-HOT)...") run_updates(conn, "t2", N_UPDATES) b_after = measure(conn, "t2") b_ctid = ctid_of(conn, "t2") idx_pretty = conn.execute("SELECT pg_size_pretty(pg_relation_size('t2_v'))").fetchone()[0] v2_final = conn.execute("SELECT v FROM t2 WHERE id = 1").fetchone()[0] print(f" done. v = {v2_final}; ctid of row id=1 = {b_ctid}") print_measure("t2 after 100k NON-HOT updates:", b_after) print(f" index t2_v size = {idx_pretty}") print() # --- Step 5: VACUUM both --- print("[5] VACUUM both tables (the garbage collector removes dead tuples):") conn.execute("VACUUM t") conn.execute("VACUUM t2") a_vac = measure(conn, "t") b_vac = measure(conn, "t2") print_measure("t after VACUUM:", a_vac) print_measure("t2 after VACUUM:", b_vac) print() # --- summary --- print("=== summary ===") print(f" Both tables kept exactly 1 live tuple the whole time.") print(f" Scenario A (HOT): n_dead_tup={a_after['n_dead_tup']:>6} " f"heap {measure_delta(8192, a_after['size_bytes'])} " f"ctid {first_ctid} -> {a_ctid}") print(f" Scenario B (non-HOT): n_dead_tup={b_after['n_dead_tup']:>6} " f"heap {measure_delta(8192, b_after['size_bytes'])} " f"+ index bloated to {idx_pretty}") print(f" After VACUUM: dead tuples {a_after['n_dead_tup']}->{a_vac['n_dead_tup']} (t), " f"{b_after['n_dead_tup']}->{b_vac['n_dead_tup']} (t2); " f"space marked reusable (files not shrunk).") def measure_delta(before: int, after: int) -> str: factor = after / before if before else 0 return f"{before}B -> {after}B ({factor:.0f}x)" if __name__ == "__main__": main()