Intermediate Examples

"""Example 29: Append-Only Log Replay."""
 
from dataclasses import dataclass  # => a plain, typed record for one log entry
 
 
@dataclass
class LogRecord:  # => the smallest unit a WAL ever stores -- just enough to replay one change
    seq: int  # => this record's position in append order
    action: str  # => a human-readable description of what happened
 
 
# A write-ahead log is APPEND-ONLY (co-16) -- records are never edited or
# reordered once written, so replaying them in the SAME order they were
# appended in is always well-defined, with no ambiguity about what happened
# first.
log: list[LogRecord] = []  # => models a WAL file as an in-memory list, in append order
 
 
def append(action: str) -> LogRecord:  # => the only way a record enters the log
    record = LogRecord(seq=len(log), action=action)  # => seq is just "how many records came before"
    log.append(record)  # => append-only: always added at the END, never inserted elsewhere
    return record
 
 
def replay() -> list[str]:  # => walks the log front-to-back, in append order
    return [record.action for record in log]  # => list order == log order, by construction
 
 
append("insert row-1")  # => record 0
append("insert row-2")  # => record 1
append("delete row-1")  # => record 2
actions = replay()
print(actions)  # => Output: ['insert row-1', 'insert row-2', 'delete row-1']
 
assert actions == ["insert row-1", "insert row-2", "delete row-1"]  # => replay order == append order
assert [r.seq for r in log] == [0, 1, 2]  # => sequence numbers are strictly increasing, gap-free
print("ex-29 OK")  # => Output: ex-29 OK

['insert row-1', 'insert row-2', 'delete row-1']
ex-29 OK

"""Example 29: pytest verification for Append-Only Log Replay."""
 
from example import append, log, replay
 
 
def test_replay_matches_append_order() -> None:
    log.clear()
    append("a")
    append("b")
    append("c")
    assert replay() == ["a", "b", "c"]
 
 
def test_sequence_numbers_are_gap_free() -> None:
    log.clear()
    append("x")
    append("y")
    assert [r.seq for r in log] == [0, 1]  # => no gaps, no reordering, ever
 
 
# => Run: pytest -- Output: 2 passed

2 passed

"""Example 30: WAL-Before-Page Ordering Guard."""
 
# The WAL rule: a log record must be DURABLE before the page it describes is
# flushed (co-16) -- reversing that order is exactly what makes a crash
# unrecoverable, since the page's change would exist with no log record to
# redo it from.
 
 
class WalOrderingError(Exception):  # => raised when a flush is attempted out of order
    """Raised when a page flush is attempted before its log record is durable."""  # => documents intent
 
 
durable_lsns: set[int] = set()  # => LSNs that have actually reached stable storage
 
 
def make_durable(lsn: int) -> None:  # => simulates fsync-ing the log up through this LSN
    durable_lsns.add(lsn)
 
 
def flush_page(page_lsn: int) -> None:  # => the guarded page-write path
    if page_lsn not in durable_lsns:  # => the guard: check BEFORE writing the page
        raise WalOrderingError(f"log record {page_lsn} is not yet durable -- refusing to flush the page")
    print(f"page flushed (log record {page_lsn} was already durable)")  # => Output: page flushed line
 
 
raised = False  # => flips to True only if the guard actually fires below
try:
    flush_page(page_lsn=5)  # => LSN 5's log record was never made durable
except WalOrderingError:  # => the exact exception the guard raises
    raised = True  # => confirms the guard fired instead of flushing out of order
assert raised  # => the page write was refused
 
make_durable(5)  # => NOW the log record is durable
flush_page(page_lsn=5)  # => this time the guard passes -- Output: page flushed (log record 5 was already durable)
 
assert 5 in durable_lsns  # => the log record really was made durable before this second flush
print("ex-30 OK")  # => Output: ex-30 OK

page flushed (log record 5 was already durable)
ex-30 OK

"""Example 30: pytest verification for the WAL-Before-Page Ordering Guard."""
 
import pytest
 
from example import WalOrderingError, durable_lsns, flush_page, make_durable
 
 
def test_flush_before_durable_raises() -> None:
    durable_lsns.clear()
    with pytest.raises(WalOrderingError):
        flush_page(page_lsn=1)
 
 
def test_flush_after_durable_succeeds() -> None:
    durable_lsns.clear()
    make_durable(2)
    flush_page(page_lsn=2)  # => no exception -- the guard passed
 
 
# => Run: pytest -- Output: 2 passed

2 passed

"""Example 31: Monotonic LSN Assignment."""
# LSNs (Log Sequence Numbers) totally order every WAL record (co-17).
 
 
class LsnGenerator:  # => a strictly increasing counter -- the WAL's own clock
    def __init__(self) -> None:  # => starts before any LSN has been issued
        self._next: int = 1  # => LSN 0 is reserved to mean "nothing written yet"
 
    def next_lsn(self) -> int:  # => issues one new LSN, guaranteed greater than every prior one
        lsn = self._next  # => capture the value about to be issued
        self._next += 1  # => strictly increasing: no reuse, no gaps introduced by this generator
        return lsn  # => hand back this call's freshly issued LSN
 
 
gen = LsnGenerator()  # => a fresh generator for this example
lsns = [gen.next_lsn() for _ in range(5)]  # => issue five LSNs in a row
print(lsns)  # => Output: [1, 2, 3, 4, 5]
 
for i in range(1, len(lsns)):  # => walk every consecutive pair
    assert lsns[i] > lsns[i - 1]  # => each new LSN is STRICTLY greater than the previous one
print("ex-31 OK")  # => Output: ex-31 OK

[1, 2, 3, 4, 5]
ex-31 OK

"""Example 31: pytest verification for Monotonic LSN Assignment."""
 
from example import LsnGenerator
 
 
def test_lsns_strictly_increase() -> None:
    gen = LsnGenerator()
    a = gen.next_lsn()
    b = gen.next_lsn()
    c = gen.next_lsn()
    assert a < b < c
 
 
def test_two_generators_are_independent() -> None:
    gen1 = LsnGenerator()
    gen2 = LsnGenerator()
    assert gen1.next_lsn() == gen2.next_lsn() == 1  # => each generator starts its own count at 1
 
 
# => Run: pytest -- Output: 2 passed

2 passed

"""Example 32: Skip Already-Applied Records via pageLSN."""
 
from dataclasses import dataclass  # => a plain, typed record for one log entry
 
 
@dataclass  # => a plain, typed record -- no custom __init__ needed
class LogRecord:  # => a WAL record that changed exactly one page
    lsn: int  # => this record's own log sequence number
    page_id: int  # => which page it changed
    new_value: str  # => the value it wrote
 
 
def redo(log: list[LogRecord], page_lsns: dict[int, int]) -> dict[int, str]:  # => returns applied values
    applied: dict[int, str] = {}  # => page_id -> value, only for records actually re-applied
    for record in log:  # => walk the log in LSN order -- ARIES calls this "repeating history"
        current_page_lsn = page_lsns.get(record.page_id, 0)  # => 0 means "page never touched before crash"
        if record.lsn > current_page_lsn:  # => the redo rule: only re-apply what is NEWER than the page
            applied[record.page_id] = record.new_value  # => genuinely re-applied
            page_lsns[record.page_id] = record.lsn  # => the page's pageLSN now reflects this record too
        # => record.lsn <= current_page_lsn means this change was ALREADY on the page before the crash
    return applied  # => the caller can see exactly which pages were genuinely re-applied
 
 
log = [  # => three records against the same page, only one of which still needs redo
    LogRecord(lsn=1, page_id=10, new_value="v1"),  # => already reflected on page 10 before the crash
    LogRecord(lsn=2, page_id=10, new_value="v2"),  # => also already reflected (page_lsns[10] starts at 2)
    LogRecord(lsn=3, page_id=10, new_value="v3"),  # => NOT yet reflected -- this one needs redo
]  # => end of the log fixture
page_lsns = {10: 2}  # => page 10's pageLSN says "everything up to LSN 2 is already applied"
applied = redo(log, page_lsns)  # => run redo over the whole log
print(applied)  # => Output: {10: 'v3'}
 
assert applied == {10: "v3"}  # => only the record whose LSN exceeded the page's pageLSN was re-applied
assert page_lsns[10] == 3  # => the page's pageLSN now reflects the redo that actually happened
print("ex-32 OK")  # => Output: ex-32 OK

{10: 'v3'}
ex-32 OK

"""Example 32: pytest verification for Skip Already-Applied Records via pageLSN."""
 
from example import LogRecord, redo
 
 
def test_records_at_or_below_pagelsn_are_skipped() -> None:
    log = [LogRecord(lsn=1, page_id=1, new_value="a"), LogRecord(lsn=2, page_id=1, new_value="b")]
    page_lsns = {1: 2}
    assert redo(log, page_lsns) == {}  # => both records were already reflected on the page
 
 
def test_records_above_pagelsn_are_reapplied() -> None:
    log = [LogRecord(lsn=1, page_id=1, new_value="a"), LogRecord(lsn=2, page_id=1, new_value="b")]
    page_lsns = {1: 0}
    assert redo(log, page_lsns) == {1: "b"}  # => the LAST record wins -- page ends at its final value
 
 
# => Run: pytest -- Output: 2 passed

2 passed

"""Example 33: WAL Redo Replays Committed Writes."""
# ARIES-style redo replays EVERY logged write (co-19) -- undo removes the losers next.
 
from dataclasses import dataclass  # => a plain, typed record for one log entry
 
 
@dataclass  # => a plain, typed record -- no custom __init__ needed
class LogRecord:  # => a WAL record for one key's write, tagged with its owning transaction
    txn_id: int  # => which transaction wrote this
    key: str  # => the key that was written
    value: str  # => the value it was set to
 
 
log: list[LogRecord] = [  # => one committed write, one that never committed
    LogRecord(txn_id=1, key="a", value="committed-a"),  # => txn 1 later commits
    LogRecord(txn_id=2, key="b", value="uncommitted-b"),  # => txn 2 never commits (crashes mid-flight)
]  # => end of the log fixture
committed_txns: set[int] = {1}  # => only txn 1 reached COMMIT before the simulated crash
 
 
def redo_all(log: list[LogRecord]) -> dict[str, str]:  # => replays EVERY record, commit status ignored
    state: dict[str, str] = {}  # => simulates the page state, wiped clean by the "crash"
    for record in log:  # => redo does not check committed_txns at all -- that is undo's job (ex-34)
        state[record.key] = record.value  # => re-apply, regardless of whether the writer ever committed
    return state  # => the fully redone state, including transactions that never actually committed
 
 
state = redo_all(log)  # => in-memory state was discarded ("crash") -- this rebuilds it from the log
print(state)  # => Output: {'a': 'committed-a', 'b': 'uncommitted-b'}
 
assert state["a"] == "committed-a"  # => the committed write survived the crash, via redo
assert 1 in committed_txns  # => confirms txn 1 really did commit before the crash
print("ex-33 OK")  # => Output: ex-33 OK

{'a': 'committed-a', 'b': 'uncommitted-b'}
ex-33 OK

"""Example 33: pytest verification for WAL Redo Replaying Committed Writes."""
 
from example import LogRecord, redo_all
 
 
def test_committed_writes_survive_redo() -> None:
    log = [LogRecord(txn_id=1, key="k", value="v")]
    state = redo_all(log)
    assert state["k"] == "v"
 
 
def test_redo_replays_every_record_regardless_of_commit_status() -> None:
    log = [LogRecord(txn_id=1, key="k", value="v1"), LogRecord(txn_id=2, key="k", value="v2")]
    state = redo_all(log)
    assert state["k"] == "v2"  # => the LAST write in the log wins, whoever wrote it
 
 
# => Run: pytest -- Output: 2 passed

2 passed

"""Example 34: WAL Undo Rolls Back Uncommitted Writes."""
# Undo removes exactly what redo just replayed for transactions that never committed (co-19).
 
from dataclasses import dataclass  # => a plain, typed record for one log entry
 
 
@dataclass  # => a plain, typed record -- no custom __init__ needed
class LogRecord:  # => carries BOTH the old and new value -- undo needs the old one
    txn_id: int  # => which transaction wrote this
    key: str  # => the key that was written
    before: str  # => the value that existed BEFORE this write -- undo's restoration target
    after: str  # => the value this write set -- what redo would apply
 
 
log: list[LogRecord] = [  # => one committed write, one that never committed
    LogRecord(txn_id=1, key="a", before="orig-a", after="committed-a"),  # => txn 1 commits
    LogRecord(txn_id=2, key="b", before="orig-b", after="uncommitted-b"),  # => txn 2 never commits
]  # => end of the log fixture
committed_txns: set[int] = {1}  # => only txn 1 reached COMMIT before the crash
 
state: dict[str, str] = {record.key: record.after for record in log}  # => redo already ran (ex-33's step)
print(state)  # => Output: {'a': 'committed-a', 'b': 'uncommitted-b'}
 
 
def undo_losers(log: list[LogRecord], state: dict[str, str], committed: set[int]) -> None:  # => rolls back losers
    for record in reversed(log):  # => undo walks the log BACKWARD -- most recent uncommitted change first
        if record.txn_id not in committed:  # => a "loser" transaction: never reached COMMIT
            state[record.key] = record.before  # => restore the value that existed before this write
 
 
undo_losers(log, state, committed_txns)  # => run the repair pass over the redone state
print(state)  # => Output: {'a': 'committed-a', 'b': 'orig-b'}
 
assert state["a"] == "committed-a"  # => the committed write was untouched by undo
assert state["b"] == "orig-b"  # => the uncommitted write was rolled back to its before-image
print("ex-34 OK")  # => Output: ex-34 OK

{'a': 'committed-a', 'b': 'uncommitted-b'}
{'a': 'committed-a', 'b': 'orig-b'}
ex-34 OK

"""Example 34: pytest verification for WAL Undo Rolling Back Uncommitted Writes."""
 
from example import LogRecord, undo_losers
 
 
def test_uncommitted_write_is_rolled_back() -> None:
    log = [LogRecord(txn_id=1, key="k", before="old", after="new")]
    state = {"k": "new"}
    undo_losers(log, state, committed=set())
    assert state["k"] == "old"
 
 
def test_committed_write_is_left_alone() -> None:
    log = [LogRecord(txn_id=1, key="k", before="old", after="new")]
    state = {"k": "new"}
    undo_losers(log, state, committed={1})
    assert state["k"] == "new"  # => committed -- undo must never touch it
 
 
# => Run: pytest -- Output: 2 passed

2 passed

"""Example 35: A Checkpoint Bounds Redo Replay."""
# A checkpoint (co-20) is the only thing that bounds how far back redo must scan.
 
from dataclasses import dataclass  # => a plain, typed record for one log entry
 
 
@dataclass  # => a plain, typed record -- no custom __init__ needed
class LogRecord:  # => either a normal write or a checkpoint marker
    lsn: int  # => this record's own log sequence number
    kind: str  # => "write" or "checkpoint"
 
 
log: list[LogRecord] = [  # => two old records, a checkpoint, then two fresh records
    LogRecord(lsn=1, kind="write"),  # => old history -- already durable before the checkpoint
    LogRecord(lsn=2, kind="write"),  # => old history -- already durable before the checkpoint
    LogRecord(lsn=3, kind="checkpoint"),  # => the most recent checkpoint marker
    LogRecord(lsn=4, kind="write"),  # => needs redo -- happened AFTER the checkpoint
    LogRecord(lsn=5, kind="write"),  # => needs redo -- happened AFTER the checkpoint
]  # => end of the log fixture
 
 
def last_checkpoint_lsn(log: list[LogRecord]) -> int:  # => finds the MOST RECENT checkpoint's LSN
    checkpoints = [r.lsn for r in log if r.kind == "checkpoint"]  # => every checkpoint marker's LSN
    return checkpoints[-1] if checkpoints else 0  # => 0 means "no checkpoint yet -- scan from the start"
 
 
def redo_scan_range(log: list[LogRecord]) -> list[int]:  # => the LSNs redo actually has to look at
    start = last_checkpoint_lsn(log)  # => everything up to and including this LSN is already durable
    return [r.lsn for r in log if r.lsn > start]  # => only records strictly AFTER the checkpoint
 
 
scanned = redo_scan_range(log)  # => the bounded set of LSNs redo must actually replay
print(scanned)  # => Output: [4, 5]
 
assert scanned == [4, 5]  # => redo skips records 1, 2, and the checkpoint itself -- only 4 and 5 remain
print("ex-35 OK")  # => Output: ex-35 OK

[4, 5]
ex-35 OK

"""Example 35: pytest verification for a Checkpoint Bounding Redo Replay."""
 
from example import LogRecord, redo_scan_range
 
 
def test_scan_starts_strictly_after_the_checkpoint() -> None:
    log = [LogRecord(1, "write"), LogRecord(2, "checkpoint"), LogRecord(3, "write")]
    assert redo_scan_range(log) == [3]
 
 
def test_no_checkpoint_means_scan_the_whole_log() -> None:
    log = [LogRecord(1, "write"), LogRecord(2, "write")]
    assert redo_scan_range(log) == [1, 2]  # => nothing to skip -- no checkpoint exists yet
 
 
# => Run: pytest -- Output: 2 passed

2 passed

"""Example 36: Row Versions Tagged with xmin/xmax."""
# xmin/xmax (co-21) are the two columns MVCC tags every row version with.
 
from dataclasses import dataclass  # => a plain, typed record for one MVCC row version
 
 
@dataclass  # => a plain, typed record -- no custom __init__ needed
class RowVersion:  # => one version of one row -- MVCC never edits a version in place
    value: str  # => this version's actual data
    xmin: int  # => the transaction id that CREATED this version
    xmax: int | None = None  # => the transaction id that DELETED (superseded) it, or None if still live
 
 
def update(versions: list[RowVersion], new_value: str, txn_id: int) -> RowVersion:  # => co-21: append, never edit
    current = versions[-1]  # => the most recent (currently live) version
    current.xmax = txn_id  # => the OLD version is now superseded, tagged with the updating txn's id
    new_version = RowVersion(value=new_value, xmin=txn_id)  # => a BRAND NEW version, not an edit in place
    versions.append(new_version)  # => the row's version chain grows -- nothing was overwritten
    return new_version  # => hand back the freshly created version
 
 
versions: list[RowVersion] = [RowVersion(value="original", xmin=1)]  # => created by txn 1
update(versions, new_value="updated", txn_id=2)  # => txn 2 updates the row
print(versions[0])  # => Output: RowVersion(value='original', xmin=1, xmax=2)
print(versions[1])  # => Output: RowVersion(value='updated', xmin=2, xmax=None)
 
assert versions[0].xmax == 2  # => the OLD version is tagged as superseded by txn 2
assert versions[1].xmin == 2  # => the NEW version is tagged as created by txn 2
assert versions[1].xmax is None  # => the new version is still live -- nothing has superseded IT yet
print("ex-36 OK")  # => Output: ex-36 OK

RowVersion(value='original', xmin=1, xmax=2)
RowVersion(value='updated', xmin=2, xmax=None)
ex-36 OK

"""Example 36: pytest verification for Row Versions Tagged with xmin/xmax."""
 
from example import RowVersion, update
 
 
def test_update_sets_xmax_on_the_old_version() -> None:
    versions = [RowVersion(value="v1", xmin=1)]
    update(versions, new_value="v2", txn_id=5)
    assert versions[0].xmax == 5
 
 
def test_update_sets_xmin_on_the_new_version() -> None:
    versions = [RowVersion(value="v1", xmin=1)]
    update(versions, new_value="v2", txn_id=5)
    assert versions[1].xmin == 5
    assert versions[1].xmax is None  # => the newest version has no successor yet
 
 
# => Run: pytest -- Output: 2 passed

2 passed

"""Example 37: Snapshot Visibility Rule."""
# Snapshot visibility (co-22) is decided entirely from xmin/xmax and commit order.
 
from dataclasses import dataclass  # => a plain, typed record for one MVCC row version
 
 
@dataclass  # => a plain, typed record -- no custom __init__ needed
class RowVersion:  # => one version of one row, tagged with its creating/deleting transactions
    value: str  # => this version's actual data
    xmin: int  # => transaction id that created this version
    xmax: int | None = None  # => transaction id that deleted it, or None if still live
 
 
commit_order: dict[int, int] = {}  # => txn_id -> the position it committed at (lower = earlier)
 
 
def commit(txn_id: int) -> None:  # => records WHEN (in commit order) a transaction committed
    commit_order[txn_id] = len(commit_order)  # => the Nth commit gets position N
 
 
def is_visible(version: RowVersion, snapshot_at: int) -> bool:  # => snapshot_at is a commit-order position
    creator_pos = commit_order.get(version.xmin)  # => None if the creator hasn't committed at all yet
    if creator_pos is None or creator_pos >= snapshot_at:  # => created too late (or not committed) -- hide it
        return False  # => this snapshot cannot possibly see a version created after (or not before) it
    if version.xmax is not None:  # => this version WAS superseded by some later transaction
        deleter_pos = commit_order.get(version.xmax)  # => None means the deleter hasn't committed yet either
        if deleter_pos is not None and deleter_pos < snapshot_at:  # => the delete was ALREADY visible too
            return False  # => superseded before this snapshot was taken -- hidden
    return True  # => created before the snapshot, and not (visibly) deleted before it either
 
 
commit(txn_id=1)  # => commit position 0
version = RowVersion(value="v1", xmin=1)  # => a version created by txn 1
snapshot_before = 0  # => taken BEFORE txn 1's commit is visible (at position 0, nothing has committed yet)
snapshot_after = 1  # => taken AFTER txn 1's commit
print(is_visible(version, snapshot_before))  # => Output: False
print(is_visible(version, snapshot_after))  # => Output: True
 
assert is_visible(version, snapshot_before) is False  # => a snapshot before the commit cannot see it
assert is_visible(version, snapshot_after) is True  # => a snapshot after the commit CAN see it
print("ex-37 OK")  # => Output: ex-37 OK

False
True
ex-37 OK

"""Example 37: pytest verification for the Snapshot Visibility Rule."""
 
from example import RowVersion, commit, commit_order, is_visible
 
 
def test_snapshot_before_commit_cannot_see_it() -> None:
    commit_order.clear()
    commit(txn_id=9)
    version = RowVersion(value="v", xmin=9)
    assert is_visible(version, snapshot_at=0) is False
 
 
def test_snapshot_after_commit_can_see_it() -> None:
    commit_order.clear()
    commit(txn_id=9)
    version = RowVersion(value="v", xmin=9)
    assert is_visible(version, snapshot_at=1) is True
 
 
# => Run: pytest -- Output: 2 passed

2 passed

"""Example 38: An Update Creates a New Version."""
# MVCC never mutates a row version in place (co-21) -- every update appends a new one.
 
from dataclasses import dataclass  # => a plain, typed record for one MVCC row version
 
 
@dataclass  # => a plain, typed record -- no custom __init__ needed
class RowVersion:  # => one version of one row
    value: str  # => this version's actual data
    xmin: int  # => the transaction id that created this version
    xmax: int | None = None  # => the transaction id that deleted it, or None if still live
 
 
def update(versions: list[RowVersion], new_value: str, txn_id: int) -> RowVersion:  # => append, never mutate
    current = versions[-1]  # => the current, live version -- about to be superseded
    current.xmax = txn_id  # => tag the OLD version as superseded, but do NOT change its value field
    new_version = RowVersion(value=new_value, xmin=txn_id)  # => a genuinely NEW object
    versions.append(new_version)  # => appended, not swapped in place
    return new_version  # => hand back the freshly created version
 
 
versions: list[RowVersion] = [RowVersion(value="v1", xmin=1)]  # => the original version
original = versions[0]  # => keep a reference to compare against later
update(versions, new_value="v2", txn_id=2)  # => first update
update(versions, new_value="v3", txn_id=3)  # => second update
print(len(versions))  # => Output: 3
print([v.value for v in versions])  # => Output: ['v1', 'v2', 'v3']
 
assert len(versions) == 3  # => two updates produced TWO new versions, plus the original -- three total
assert original.value == "v1"  # => the ORIGINAL object's value was never mutated in place
assert versions[0] is original  # => it is literally the same object, still sitting in the chain
assert len({id(v) for v in versions}) == 3  # => three genuinely distinct objects, not aliases of each other
print("ex-38 OK")  # => Output: ex-38 OK

3
['v1', 'v2', 'v3']
ex-38 OK

"""Example 38: pytest verification for an Update Creating a New Version."""
 
from example import RowVersion, update
 
 
def test_update_appends_rather_than_overwrites() -> None:
    versions = [RowVersion(value="v1", xmin=1)]
    update(versions, new_value="v2", txn_id=2)
    assert len(versions) == 2
    assert versions[0].value == "v1"  # => the original object's value is untouched
 
 
def test_three_versions_are_three_distinct_objects() -> None:
    versions = [RowVersion(value="v1", xmin=1)]
    update(versions, new_value="v2", txn_id=2)
    update(versions, new_value="v3", txn_id=3)
    assert len({id(v) for v in versions}) == 3  # => three genuinely distinct objects, not aliases
 
 
# => Run: pytest -- Output: 2 passed

2 passed

"""Example 39: Readers Don't Block Writers."""
# Snapshot isolation (co-22) means readers need no lock at all.
 
from dataclasses import dataclass  # => a plain, typed record for one MVCC row version
 
 
@dataclass  # => a plain, typed record -- no custom __init__ needed
class RowVersion:  # => one version of one row
    value: str  # => this version's actual data
    xmin: int  # => transaction id that created this version
    xmax: int | None = None  # => transaction id that deleted it, or None if still live
 
 
commit_order: dict[int, int] = {1: 0}  # => txn 1 already committed the row's original version
versions: list[RowVersion] = [RowVersion(value="original", xmin=1)]  # => one live version so far
 
 
def snapshot_read(snapshot_at: int) -> str | None:  # => acquires NO lock -- pure read of existing data
    for version in reversed(versions):  # => newest-to-oldest, first visible match wins
        creator_pos = commit_order.get(version.xmin)  # => when (if ever) this version's creator committed
        if creator_pos is not None and creator_pos < snapshot_at:  # => created before this snapshot
            return version.value  # => the first (newest) visible version found
    return None  # => no version of this row was visible to the snapshot at all
 
 
def writer_update(new_value: str, txn_id: int) -> None:  # => a concurrent writer, appends a new version
    versions[-1].xmax = txn_id  # => tag the current version as superseded
    versions.append(RowVersion(value=new_value, xmin=txn_id))  # => append the new version -- no lock taken
    commit_order[txn_id] = len(commit_order)  # => commits immediately for this example's purposes
 
 
reader_snapshot_at = len(commit_order)  # => the reader's snapshot position, taken BEFORE the writer runs
reader_result = snapshot_read(reader_snapshot_at)  # => the reader's read -- acquired NO lock to do this
writer_update("written-concurrently", txn_id=2)  # => the writer proceeds WITHOUT waiting on the reader
print(reader_result)  # => Output: original
print(len(versions))  # => Output: 2
 
assert reader_result == "original"  # => the reader's result reflects its OWN snapshot, unaffected
assert len(versions) == 2  # => the writer's append succeeded -- it was never blocked by the reader
print("ex-39 OK")  # => Output: ex-39 OK

original
2
ex-39 OK

"""Example 39: pytest verification for Readers Not Blocking Writers."""
 
from example import RowVersion, commit_order, snapshot_read, versions, writer_update
 
 
def test_writer_can_proceed_after_a_snapshot_is_taken() -> None:
    commit_order.clear()
    commit_order[1] = 0
    versions.clear()
    versions.append(RowVersion(value="orig", xmin=1))
    writer_update("new", txn_id=2)  # => no exception, no wait -- the writer just proceeds
    assert len(versions) == 2
 
 
def test_reader_snapshot_is_unaffected_by_a_later_writer() -> None:
    commit_order.clear()
    commit_order[1] = 0
    versions.clear()
    versions.append(RowVersion(value="orig", xmin=1))
    snapshot_at = len(commit_order)
    result = snapshot_read(snapshot_at)
    writer_update("new", txn_id=2)
    assert result == "orig"  # => computed BEFORE the writer ran, and stays that way
 
 
# => Run: pytest -- Output: 2 passed

2 passed

"""Example 40: Vacuum Reclaims Dead Versions."""
 
from dataclasses import dataclass  # => a plain, typed record for one MVCC row version
 
 
@dataclass  # => a plain, typed record -- no custom __init__ needed
class RowVersion:  # => one version of one row
    value: str  # => this version's actual data
    xmin: int  # => the transaction id that created this version
    xmax: int | None = None  # => the transaction id that deleted it, or None if still live
 
 
commit_order: dict[int, int] = {1: 0, 2: 1, 3: 2}  # => three transactions, each already committed
versions: list[RowVersion] = [  # => three versions of the same logical row
    RowVersion(value="v1", xmin=1, xmax=2),  # => superseded by txn 2 -- may be dead, depending on readers
    RowVersion(value="v2", xmin=2, xmax=3),  # => superseded by txn 3 -- may be dead, depending on readers
    RowVersion(value="v3", xmin=3),  # => the current, live version -- xmax is None, never dead
]  # => end of the version-chain fixture
 
 
def vacuum(versions: list[RowVersion], oldest_active_snapshot: int) -> list[RowVersion]:  # => co-23: reclaim
    live: list[RowVersion] = []  # => survivors after vacuuming
    for version in versions:  # => decide each version's fate independently
        if version.xmax is None:  # => never superseded -- always kept
            live.append(version)  # => keep it -- still the current row
            continue  # => move on to the next version, nothing more to check
        deleter_pos = commit_order.get(version.xmax)  # => when (if ever) the deleter committed
        if deleter_pos is None or deleter_pos >= oldest_active_snapshot:  # => some active reader might need it
            live.append(version)  # => not yet safe to reclaim
        # => else: deleter_pos < oldest_active_snapshot -- NO active reader can possibly still see it
    return live  # => the pruned, still-valid version set
 
 
before_count = len(versions)  # => 3 versions before vacuuming
after = vacuum(versions, oldest_active_snapshot=2)  # => the oldest active snapshot is at position 2
print(len(after))  # => Output: 2
 
assert before_count == 3  # => confirms the starting point before reclaiming anything
assert len(after) == 2  # => v1 was reclaimed -- its delete (position 1) predates the oldest active snapshot
assert [v.value for v in after] == ["v2", "v3"]  # => the two still-relevant versions survive
print("ex-40 OK")  # => Output: ex-40 OK

2
ex-40 OK

"""Example 40: pytest verification for Vacuum Reclaiming Dead Versions."""
 
from example import RowVersion, commit_order, vacuum
 
 
def test_a_version_dead_before_every_active_snapshot_is_reclaimed() -> None:
    commit_order.clear()
    commit_order.update({1: 0, 2: 1})
    versions = [RowVersion(value="v1", xmin=1, xmax=2), RowVersion(value="v2", xmin=2)]
    survivors = vacuum(versions, oldest_active_snapshot=2)
    assert len(survivors) == 1
 
 
def test_a_version_needed_by_an_active_snapshot_is_kept() -> None:
    commit_order.clear()
    commit_order.update({1: 0, 2: 1})
    versions = [RowVersion(value="v1", xmin=1, xmax=2), RowVersion(value="v2", xmin=2)]
    survivors = vacuum(versions, oldest_active_snapshot=0)  # => a reader at position 0 still needs v1
    assert len(survivors) == 2
 
 
# => Run: pytest -- Output: 2 passed

2 passed

"""Example 41: LRU-K vs Plain LRU on a Scan-Then-Hot Pattern."""
# LRU-K (co-05) ranks a page by its K-th most recent access, not just its latest one.
 
from collections import OrderedDict  # => stdlib ordered-dict, gives free "most-recently-touched" order
 
CAPACITY = 3  # => the buffer pool holds at most 3 frames in this example
 
 
def simulate_lru(accesses: list[str], capacity: int) -> list[str]:  # => plain LRU eviction
    cache: OrderedDict[str, None] = OrderedDict()  # => key order IS recency order -- newest at the end
    for page in accesses:  # => walk the access trace in order
        if page in cache:  # => a hit -- refresh its recency
            cache.move_to_end(page)  # => moves it to the "most recently used" end
        else:  # => a miss -- must load the page
            if len(cache) >= capacity:  # => pool is full -- must evict first
                cache.popitem(last=False)  # => evicts the LEAST recently used page (the front)
            cache[page] = None  # => insert as most recently used
    return list(cache.keys())  # => final resident set, in recency order
 
 
def simulate_lru_k(accesses: list[str], capacity: int, k: int) -> list[str]:  # => LRU-K eviction
    cache: dict[str, None] = {}  # => the resident set -- insertion order tracked by dict semantics
    history: dict[str, list[int]] = {}  # => page -> its last k access timestamps, oldest first
    for t, page in enumerate(accesses, start=1):  # => t is a monotonic logical timestamp
        hist = history.setdefault(page, [])  # => this page's access-time history so far
        hist.append(t)  # => record this access
        if len(hist) > k:  # => only the last K references matter for the K-distance rule
            hist.pop(0)  # => drop the oldest, keep exactly K entries
        if page not in cache and len(cache) >= capacity:  # => a miss into a full pool -- must evict
            def backward_k_distance(candidate: str) -> float:  # => how "cold" a page looks to LRU-K
                h = history[candidate]  # => that candidate page's own access-time history
                if len(h) < k:  # => fewer than K references ever seen -- treat as infinitely cold
                    return float("inf")  # => a page seen once during a scan looks maximally evictable
                return t - h[0]  # => distance back to the K-th most recent reference
            victim = max(cache, key=lambda p: (backward_k_distance(p), -history[p][-1]))  # => coldest wins
            del cache[victim]  # => evict whichever page looks coldest under the K-distance rule
        cache[page] = None  # => insert (or refresh) this page as resident
    return list(cache.keys())  # => final resident set, insertion order (hot inserted once, never evicted)
 
 
accesses = ["hot", "hot", "p1", "p2", "p3", "p4", "p5"]  # => hot touched twice, then a long distinct scan
lru_result = simulate_lru(accesses, CAPACITY)  # => run the trace through plain LRU
lru_k_result = simulate_lru_k(accesses, CAPACITY, k=2)  # => and through LRU-K, for comparison
print(lru_result)  # => Output: ['p3', 'p4', 'p5']
print(lru_k_result)  # => Output: ['hot', 'p4', 'p5']
 
assert "hot" not in lru_result  # => plain LRU evicted the hot page during the one-off scan
assert "hot" in lru_k_result  # => LRU-K correctly kept the hot page resident
print("ex-41 OK")  # => Output: ex-41 OK

['p3', 'p4', 'p5']
['hot', 'p4', 'p5']
ex-41 OK

"""Example 41: pytest verification for LRU-K vs Plain LRU."""
 
from example import simulate_lru, simulate_lru_k
 
 
def test_plain_lru_evicts_the_hot_page_during_a_scan() -> None:
    accesses = ["hot", "hot", "a", "b", "c", "d"]
    result = simulate_lru(accesses, capacity=2)
    assert "hot" not in result
 
 
def test_lru_k_keeps_the_hot_page_during_the_same_scan() -> None:
    accesses = ["hot", "hot", "a", "b", "c", "d"]
    result = simulate_lru_k(accesses, capacity=2, k=2)
    assert "hot" in result
 
 
# => Run: pytest -- Output: 2 passed

2 passed

"""Example 42: Buffer Pool Hit Ratio."""
# Hit ratio (co-06) is the standard buffer-pool health metric.
 
from dataclasses import dataclass, field  # => typed pool + counters
 
 
@dataclass  # => a plain, typed record -- no custom __init__ needed
class BufferPool:  # => a minimal pool that just tracks resident pages and hit/miss counts
    frames: set[int] = field(default_factory=set[int])  # => currently resident page ids
    hits: int = 0  # => count of accesses that found the page already resident
    misses: int = 0  # => count of accesses that had to load the page
 
 
def access(pool: BufferPool, page_id: int) -> None:  # => one workload access against the pool
    if page_id in pool.frames:  # => already resident -- a hit
        pool.hits += 1  # => tally the hit -- no load needed
    else:  # => not resident -- a miss, must be loaded
        pool.misses += 1  # => tally the miss before loading
        pool.frames.add(page_id)  # => now resident for any future access
 
 
def hit_ratio(pool: BufferPool) -> float:  # => co-06: the standard buffer-pool efficiency metric
    total = pool.hits + pool.misses  # => total accesses attempted
    return pool.hits / total if total else 0.0  # => guard against dividing by zero on an empty workload
 
 
pool = BufferPool()  # => a fresh, empty pool
workload = [1, 2, 1, 3, 1, 2]  # => a mix of repeated (hits) and new (misses) page accesses
for page_id in workload:  # => replay the whole workload through the pool
    access(pool, page_id)  # => feed each workload access through the pool
print(pool.hits)  # => Output: 3
print(pool.misses)  # => Output: 3
ratio = hit_ratio(pool)  # => compute the final hit ratio for this workload
print(ratio)  # => Output: 0.5
 
assert pool.hits == 3  # => pages 1 (two repeat hits) and 2 (one repeat hit) -- three hits total
assert pool.misses == 3  # => pages 1, 2, 3 each loaded once -- three misses total
assert ratio == pool.hits / (pool.hits + pool.misses)  # => the exact formula the spec requires
print("ex-42 OK")  # => Output: ex-42 OK

3
3
0.5
ex-42 OK

"""Example 42: pytest verification for Buffer Pool Hit Ratio."""
 
from example import BufferPool, access, hit_ratio
 
 
def test_hit_ratio_equals_hits_over_total() -> None:
    pool = BufferPool()
    for page_id in [1, 1, 2]:
        access(pool, page_id)
    assert hit_ratio(pool) == pool.hits / (pool.hits + pool.misses)
 
 
def test_all_misses_gives_zero_hit_ratio() -> None:
    pool = BufferPool()
    for page_id in [1, 2, 3]:
        access(pool, page_id)
    assert hit_ratio(pool) == 0.0
 
 
# => Run: pytest -- Output: 2 passed

2 passed

"""Example 43: B-Tree Bulk Load vs Insert-One-by-One."""
# Bulk load and incremental insert (co-08) can differ in shape but must agree on every lookup.
 
import bisect  # => stdlib binary-search helper for sorted-list insertion
 
LEAF_CAPACITY = 4  # => each leaf page holds at most 4 keys before it must split
 
 
def bulk_load(sorted_keys: list[int], capacity: int) -> list[list[int]]:  # => bottom-up, one linear pass
    return [  # => a list comprehension -- one leaf per capacity-sized slice
        sorted_keys[i : i + capacity] for i in range(0, len(sorted_keys), capacity)  # => a fixed-size slice
    ]  # => end of the bulk-load comprehension  # => chunk ALREADY-SORTED input directly -- no comparisons, no splits, just slicing
 
 
def insert_one(leaves: list[list[int]], key: int, capacity: int) -> None:  # => classic incremental insert
    if not leaves:  # => the very first key ever inserted into an empty tree
        leaves.append([key])  # => the tree's first (and only) leaf so far
        return  # => nothing more to do for the very first key
    leaf_index = 0  # => find which leaf this key belongs in, by comparing to each leaf's first key
    for i, leaf in enumerate(leaves):  # => walk leaves left to right
        if key < leaf[0]:  # => key belongs BEFORE this leaf -- stop at the previous one
            break  # => leaf_index already points at the right leaf
        leaf_index = i  # => keep tracking the rightmost leaf whose start is still <= key
    leaf = leaves[leaf_index]  # => the leaf this key will be inserted into
    bisect.insort(leaf, key)  # => insert keeping the leaf internally sorted
    if len(leaf) > capacity:  # => the leaf overflowed -- split it in two, like a real B-tree
        mid = len(leaf) // 2  # => the split point -- roughly half the keys on each side
        leaves[leaf_index] = leaf[:mid]  # => left half stays at this position
        leaves.insert(leaf_index + 1, leaf[mid:])  # => right half becomes a brand-new leaf
 
 
def lookup(leaves: list[list[int]], key: int) -> bool:  # => point lookup across the leaf chain
    return any(key in leaf for leaf in leaves)  # => a real B-tree binary-searches per leaf; O(n) here for clarity
 
 
sorted_keys = list(range(0, 20, 2))  # => [0, 2, 4, ..., 18] -- ten pre-sorted keys
bulk_leaves = bulk_load(sorted_keys, LEAF_CAPACITY)  # => one linear pass, chunked geometry
incremental_leaves: list[list[int]] = []  # => built up key-by-key instead
for key in sorted_keys:  # => insert the SAME keys, one at a time
    insert_one(incremental_leaves, key, LEAF_CAPACITY)  # => grow the tree incrementally, key by key
 
print(bulk_leaves)  # => Output: [[0, 2, 4, 6], [8, 10, 12, 14], [16, 18]]
print(incremental_leaves)  # => Output: [[0, 2], [4, 6], [8, 10], [12, 14, 16, 18]]
 
test_keys = sorted_keys + [1, 3, 99]  # => every real key, plus some keys that should be absent
bulk_answers = [lookup(bulk_leaves, k) for k in test_keys]  # => query the bulk-loaded tree
incremental_answers = [lookup(incremental_leaves, k) for k in test_keys]  # => query the incremental tree
assert bulk_answers == incremental_answers  # => SAME lookup answers despite DIFFERENT leaf geometry
print("ex-43 OK")  # => Output: ex-43 OK

[[0, 2, 4, 6], [8, 10, 12, 14], [16, 18]]
[[0, 2], [4, 6], [8, 10], [12, 14, 16, 18]]
ex-43 OK

"""Example 43: pytest verification for B-Tree Bulk Load vs Insert-One-by-One."""
 
from example import bulk_load, insert_one, lookup
 
 
def test_bulk_and_incremental_agree_on_every_lookup() -> None:
    sorted_keys = list(range(0, 12, 2))
    bulk_leaves = bulk_load(sorted_keys, capacity=4)
    incremental_leaves: list[list[int]] = []
    for key in sorted_keys:
        insert_one(incremental_leaves, key, capacity=4)
    for probe in sorted_keys + [1, 99]:
        assert lookup(bulk_leaves, probe) == lookup(incremental_leaves, probe)
 
 
def test_absent_key_is_not_found_in_either_structure() -> None:
    bulk_leaves = bulk_load([0, 2, 4], capacity=4)
    assert lookup(bulk_leaves, 3) is False
 
 
# => Run: pytest -- Output: 2 passed

2 passed

"""Example 44: Force B-Tree Splits Up to the Root."""
# A B-tree only grows taller by splitting a FULL root (co-09) -- every other split stays local.
 
import bisect  # => stdlib binary-search helper for sorted-list insertion
 
 
class BTreeNode:  # => a single B-tree node -- either an internal node or a leaf
    def __init__(self, leaf: bool = True) -> None:  # => leaf=True until proven otherwise
        self.keys: list[int] = []  # => this node's own separator/data keys, always kept sorted
        self.children: list["BTreeNode"] = []  # => empty for leaves, len(keys)+1 for internal nodes
        self.leaf: bool = leaf  # => whether this node has no children at all
 
 
class BTree:  # => a minimal B-tree with proactive split-on-the-way-down insertion (co-09)
    def __init__(self, t: int) -> None:  # => t is the minimum degree -- max keys per node is 2t-1
        self.root: BTreeNode = BTreeNode(leaf=True)  # => starts as a single empty leaf
        self.t: int = t  # => stored for use by split/insert below
 
    def height(self) -> int:  # => counts edges from root down to a leaf (all leaves are equidistant)
        node, h = self.root, 0  # => start at the root with height 0
        while not node.leaf:  # => descend the leftmost path until a leaf is reached
            node = node.children[0]  # => step down one level, always via the first child
            h += 1  # => one more edge crossed
        return h  # => the tree's current height, in edges from root to any leaf
 
    def insert(self, key: int) -> None:  # => the public entry point -- handles a full root specially
        root = self.root  # => the current root, before any possible split
        if len(root.keys) == 2 * self.t - 1:  # => the root itself is full -- MUST split before inserting
            new_root = BTreeNode(leaf=False)  # => a brand-new root, one level taller than before
            new_root.children.append(root)  # => the old root becomes the new root's only child, for now
            self._split_child(new_root, 0)  # => splits it in two, promoting a median key up
            self.root = new_root  # => the tree's height has now increased by exactly one
        self._insert_nonfull(self.root, key)  # => descend and insert, splitting full children on the way
 
    def _split_child(self, parent: BTreeNode, i: int) -> None:  # => splits parent.children[i] in two
        t = self.t  # => the minimum degree, cached locally for readability below
        child = parent.children[i]  # => the full node being split
        new_node = BTreeNode(leaf=child.leaf)  # => the right half -- same leaf-ness as the original
        mid_key = child.keys[t - 1]  # => the median key -- it moves UP into the parent
        new_node.keys = child.keys[t:]  # => the right half of the keys goes to the new sibling
        child.keys = child.keys[: t - 1]  # => the original node keeps only its left half
        if not child.leaf:  # => internal nodes must also split their children pointers
            new_node.children = child.children[t:]  # => right-half children follow their keys
            child.children = child.children[:t]  # => left-half children stay behind
        parent.children.insert(i + 1, new_node)  # => the new sibling takes its place beside the original
        parent.keys.insert(i, mid_key)  # => the median key becomes a new separator in the parent
 
    def _insert_nonfull(self, node: BTreeNode, key: int) -> None:  # => assumes node is never full itself
        if node.leaf:  # => base case -- leaves just get the key inserted in sorted order
            bisect.insort(node.keys, key)  # => keeps the leaf's keys sorted after the insert
            return  # => nothing further to do once a leaf has absorbed the key
        i = len(node.keys) - 1  # => find which child subtree the key belongs under
        while i >= 0 and key < node.keys[i]:  # => walk right-to-left until the right slot is found
            i -= 1  # => keep moving left while the key is smaller than this separator
        i += 1  # => i now indexes the correct child to descend into
        if len(node.children[i].keys) == 2 * self.t - 1:  # => that child is full -- split it FIRST
            self._split_child(node, i)  # => proactively split before descending, never after
            if key > node.keys[i]:  # => the split may shift which child the key now belongs under
                i += 1  # => the key now belongs in the newly created right sibling instead
        self._insert_nonfull(node.children[i], key)  # => recurse into the now-guaranteed-non-full child
 
 
tree = BTree(t=2)  # => t=2 means max 3 keys per node, forcing splits quickly for this example
for key in range(1, 9):  # => insert keys 1 through 8 -- enough to fill the root once already
    tree.insert(key)  # => each insert may cascade zero or more splits below the root
height_before = tree.height()  # => height BEFORE the key that forces a root split
print(height_before)  # => Output: 1
 
tree.insert(9)  # => this ninth key overflows the (now full) root, forcing a root split
height_after = tree.height()  # => height AFTER the forced root split
print(height_after)  # => Output: 2
print(len(tree.root.children))  # => Output: 2
 
assert height_after == height_before + 1  # => the root split grew the tree by exactly one level
assert len(tree.root.children) == 2  # => a freshly split root always starts with exactly two children
print("ex-44 OK")  # => Output: ex-44 OK

1
2
2
ex-44 OK

"""Example 44: pytest verification for Forcing B-Tree Splits Up to the Root."""
 
from example import BTree
 
 
def test_root_split_increases_height_by_exactly_one() -> None:
    tree = BTree(t=2)
    for key in range(1, 9):
        tree.insert(key)
    before = tree.height()
    tree.insert(9)
    assert tree.height() == before + 1
 
 
def test_split_root_has_exactly_two_children() -> None:
    tree = BTree(t=2)
    for key in range(1, 10):
        tree.insert(key)
    assert len(tree.root.children) == 2
 
 
# => Run: pytest -- Output: 2 passed

2 passed

"""Example 45: B-Tree Leaf Underflow -- Merge or Borrow."""
# Borrowing is tried first (co-09) -- merging is only the fallback when no sibling can spare a key.
 
CAPACITY = 4  # => the fixed leaf capacity used throughout this course's B-tree examples
MIN_KEYS = CAPACITY // 2  # => a leaf below this many keys has UNDERFLOWED and must be repaired
 
 
def delete(leaves: list[list[int]], key: int) -> None:  # => removes key, then repairs any underflow
    leaf_index = next(i for i, leaf in enumerate(leaves) if key in leaf)  # => which leaf holds the key
    leaves[leaf_index].remove(key)  # => the actual removal
    leaf = leaves[leaf_index]  # => the (possibly now underflowed) leaf that lost a key
    if len(leaf) >= MIN_KEYS or len(leaves) == 1:  # => still valid, or it is the only leaf left -- done
        return  # => no repair needed -- occupancy is still within bounds
    if leaf_index + 1 < len(leaves) and len(leaves[leaf_index + 1]) > MIN_KEYS:  # => right sibling can spare one
        borrowed = leaves[leaf_index + 1].pop(0)  # => take its smallest key
        leaf.append(borrowed)  # => and give it to the underflowed leaf
        return  # => borrowing alone repaired the underflow -- no merge needed
    if leaf_index > 0 and len(leaves[leaf_index - 1]) > MIN_KEYS:  # => else try the left sibling instead
        borrowed = leaves[leaf_index - 1].pop()  # => take its largest key
        leaf.insert(0, borrowed)  # => and give it to the underflowed leaf, keeping sort order
        return  # => borrowing from the left sibling also repaired the underflow
    if leaf_index + 1 < len(leaves):  # => no sibling could spare a key -- MERGE with the right sibling
        leaf.extend(leaves.pop(leaf_index + 1))  # => absorb the whole right sibling's keys
    else:  # => no right sibling exists -- merge with the left one instead
        leaves[leaf_index - 1].extend(leaves.pop(leaf_index))  # => absorb THIS leaf into the left one
 
 
def is_valid(leaves: list[list[int]]) -> bool:  # => every leaf is sorted, and non-solo leaves meet MIN_KEYS
    for leaf in leaves:  # => check each leaf independently
        if leaf != sorted(leaf):  # => a leaf's keys must always stay in ascending order
            return False  # => a leaf out of sort order is never a valid B-tree state
        if len(leaf) < MIN_KEYS and len(leaves) > 1:  # => underflowed, UNLESS it is the sole remaining leaf
            return False  # => an underflowed leaf with siblings means the repair logic has a bug
    return True  # => every leaf passed both the sort-order and occupancy checks
 
 
leaves = [[1, 2, 3, 4], [5, 6, 7, 8]]  # => two full leaves, each right at capacity
assert is_valid(leaves)  # => sanity check before the deletions begin
 
delete(leaves, 1)  # => leaf 0 drops to [2,3,4] -- still >= MIN_KEYS(2), no repair needed
delete(leaves, 2)  # => leaf 0 drops to [3,4] -- exactly MIN_KEYS, still valid
delete(leaves, 3)  # => leaf 0 drops to [4] -- now UNDER MIN_KEYS(2): underflow triggers repair
print(leaves)  # => Output: [[4, 5], [6, 7, 8]]
 
assert is_valid(leaves)  # => the repair (a borrow, in this case) restored a valid tree
print("ex-45 OK")  # => Output: ex-45 OK

[[4, 5], [6, 7, 8]]
ex-45 OK

"""Example 45: pytest verification for B-Tree Leaf Underflow Repair."""
 
from example import delete, is_valid
 
 
def test_underflow_is_repaired_by_borrowing() -> None:
    leaves = [[1, 2, 3, 4], [5, 6, 7, 8]]
    delete(leaves, 1)
    delete(leaves, 2)
    delete(leaves, 3)
    assert is_valid(leaves)
 
 
def test_underflow_falls_back_to_merge_when_no_sibling_can_spare_a_key() -> None:
    leaves = [[1, 2], [3, 4]]
    delete(leaves, 1)  # => leaf 0 drops to [2] -- neither sibling has a spare key to lend
    assert is_valid(leaves)
    assert len(leaves) == 1  # => the underflowed leaf was merged away entirely
 
 
# => Run: pytest -- Output: 2 passed

2 passed

"""Example 46: Clustered Index -- Leaf Holds the Full Row."""
# A clustered index (co-27) stores the full row directly in the leaf.
 
from dataclasses import dataclass  # => a plain, typed row
 
 
@dataclass  # => a plain, typed record -- no custom __init__ needed
class Row:  # => a full table row, exactly as it would be stored
    id: int  # => the primary key -- also this row's clustering key
    name: str  # => an ordinary column, stored right alongside the PK
    email: str  # => another ordinary column, stored right alongside the PK
 
 
fetch_count: int = 0  # => counts SEPARATE storage fetches -- a clustered lookup should need only one
 
 
def clustered_lookup(index: dict[int, Row], pk: int) -> Row | None:  # => co-27: leaf itself holds the row
    global fetch_count  # => a module-level counter, mutated here to measure fetch cost
    fetch_count += 1  # => ONE fetch: the leaf lookup IS the row fetch, nothing further to do
    return index.get(pk)  # => the full row comes back directly from the index entry
 
 
index: dict[int, Row] = {  # => the clustered index -- keyed by PK, VALUED by the entire row
    1: Row(id=1, name="Alice", email="alice@example.com"),  # => the leaf entry for PK 1 IS the whole row
    2: Row(id=2, name="Bob", email="bob@example.com"),  # => same for PK 2 -- no separate heap exists
}  # => end of the clustered-index fixture
 
fetch_count = 0  # => reset before measuring this specific lookup
row = clustered_lookup(index, pk=1)  # => a single primary-key lookup
print(row)  # => Output: Row(id=1, name='Alice', email='alice@example.com')
print(fetch_count)  # => Output: 1
 
assert row is not None and row.name == "Alice"  # => the correct row came back
assert fetch_count == 1  # => exactly one fetch -- no second trip to a separate heap was ever needed
print("ex-46 OK")  # => Output: ex-46 OK

Row(id=1, name='Alice', email='alice@example.com')
1
ex-46 OK

"""Example 46: pytest verification for Clustered Index Leaves Holding the Full Row."""
 
from example import Row, clustered_lookup
 
 
def test_pk_lookup_returns_the_full_row() -> None:
    index = {5: Row(id=5, name="Carol", email="carol@example.com")}
    row = clustered_lookup(index, pk=5)
    assert row is not None
    assert row.email == "carol@example.com"
 
 
def test_pk_lookup_needs_exactly_one_fetch() -> None:
    import example
 
    example.fetch_count = 0
    index = {1: Row(id=1, name="A", email="a@example.com")}
    clustered_lookup(index, pk=1)
    assert example.fetch_count == 1
 
 
# => Run: pytest -- Output: 2 passed

2 passed

"""Example 47: Heap Table + Secondary Index via (page, slot) Pointer."""
# A heap table's secondary index (co-27) stores only a pointer, not the row.
 
from dataclasses import dataclass  # => a plain, typed pointer
 
 
@dataclass(frozen=True)  # => immutable -- a pointer value should never mutate after creation
class RowPointer:  # => a secondary index entry never stores the row itself -- only WHERE to find it
    page: int  # => which heap page the row physically lives on
    slot: int  # => which slot within that page's slot array (co-01/co-02 layout)
 
 
heap: dict[int, list[str]] = {  # => page_id -> list of row values, indexed by slot number
    0: ["alice@example.com", "bob@example.com"],  # => page 0 holds two rows
    1: ["carol@example.com"],  # => page 1 holds one row
}  # => end of the heap fixture
secondary_index: dict[str, RowPointer] = {  # => keyed by the INDEXED column (email), not the PK
    "alice@example.com": RowPointer(page=0, slot=0),  # => points AT the heap, doesn't duplicate the row
    "bob@example.com": RowPointer(page=0, slot=1),  # => same page, different slot
    "carol@example.com": RowPointer(page=1, slot=0),  # => a different page entirely
}  # => end of the secondary-index fixture
 
 
def secondary_lookup(email: str) -> str | None:  # => co-27: two steps -- index, THEN heap fetch
    pointer = secondary_index.get(email)  # => step 1: find WHERE the row lives
    if pointer is None:  # => not found in the index at all
        return None  # => the indexed value simply does not exist
    return heap[pointer.page][pointer.slot]  # => step 2: fetch the actual row from the heap page
 
 
result = secondary_lookup("bob@example.com")  # => resolve through both steps to the heap row
print(result)  # => Output: bob@example.com
 
assert result == "bob@example.com"  # => the secondary index correctly resolved to the heap row
assert secondary_index["bob@example.com"] == RowPointer(page=0, slot=1)  # => the pointer itself is exact
print("ex-47 OK")  # => Output: ex-47 OK

bob@example.com
ex-47 OK

"""Example 47: pytest verification for Heap Secondary Index Pointers."""
 
from example import secondary_lookup
 
 
def test_secondary_lookup_resolves_through_the_pointer_to_the_heap_row() -> None:
    result = secondary_lookup("carol@example.com")
    assert result == "carol@example.com"
 
 
def test_missing_key_returns_none_without_touching_the_heap() -> None:
    result = secondary_lookup("nobody@example.com")
    assert result is None
 
 
# => Run: pytest -- Output: 2 passed

2 passed

"""Example 48: LSM Size-Tiered Compaction."""
# Size-tiered compaction (co-13) merges same-size SSTables into one.
 
from dataclasses import dataclass, field  # => a typed SSTable representation
 
 
@dataclass  # => a plain, typed record -- no custom __init__ needed
class SSTable:  # => an immutable, sorted, on-disk segment (co-11/co-12 territory)
    data: dict[str, str] = field(default_factory=dict[str, str])  # => key -> value, sorted by key on disk
 
 
def compact(tables: list[SSTable]) -> SSTable:  # => size-tiered: merge SAME-SIZE tables into ONE bigger one
    merged: dict[str, str] = {}  # => the result of merging every input table
    for table in tables:  # => later tables (newer) override earlier ones on key conflicts
        merged.update(table.data)  # => co-12: newest-wins -- a later table's value replaces an earlier one
    return SSTable(data=merged)  # => one new, larger table replaces however many went in
 
 
tables = [  # => three small SSTables from the same size tier
    SSTable(data={"a": "v1", "b": "v1"}),  # => oldest SSTable
    SSTable(data={"c": "v1", "d": "v1"}),  # => same tier, disjoint keys
    SSTable(data={"a": "v2"}),  # => newest SSTable -- overwrites key "a"
]  # => end of the SSTable fixture
segment_count_before = len(tables)  # => three separate segments before compaction
merged = compact(tables)  # => run size-tiered compaction over all three tables
segment_count_after = 1  # => size-tiered compaction always produces exactly one output segment
print(segment_count_before)  # => Output: 3
print(segment_count_after)  # => Output: 1
print(sorted(merged.data.items()))  # => Output: [('a', 'v2'), ('b', 'v1'), ('c', 'v1'), ('d', 'v1')]
 
assert segment_count_after < segment_count_before  # => the merged segment count genuinely dropped
assert merged.data["a"] == "v2"  # => every key survived, and the NEWEST value for "a" won
assert set(merged.data.keys()) == {"a", "b", "c", "d"}  # => no key was lost during the merge
print("ex-48 OK")  # => Output: ex-48 OK

3
1
[('a', 'v2'), ('b', 'v1'), ('c', 'v1'), ('d', 'v1')]
ex-48 OK

"""Example 48: pytest verification for LSM Size-Tiered Compaction."""
 
from example import SSTable, compact
 
 
def test_compaction_reduces_segment_count_to_one() -> None:
    tables = [SSTable(data={"x": "1"}), SSTable(data={"y": "2"})]
    merged = compact(tables)
    assert isinstance(merged, SSTable)
 
 
def test_compaction_preserves_every_key_with_newest_value_winning() -> None:
    tables = [SSTable(data={"k": "old"}), SSTable(data={"k": "new"})]
    merged = compact(tables)
    assert merged.data["k"] == "new"
 
 
# => Run: pytest -- Output: 2 passed

2 passed

"""Example 49: LSM Leveled Compaction."""
# Leveled compaction (co-13) guarantees NO key-range overlap between tables within one level.
 
from dataclasses import dataclass  # => a typed SSTable with an explicit key range
 
 
@dataclass  # => a plain, typed record -- no custom __init__ needed
class SSTable:  # => a sorted, immutable segment, tagged with its own key range
    data: dict[str, str]  # => key -> value, sorted by key on disk
    min_key: str  # => the smallest key this table contains -- used to detect overlap
    max_key: str  # => the largest key this table contains -- used to detect overlap
 
 
def overlaps(a: SSTable, b: SSTable) -> bool:  # => do two tables' key ranges intersect at all?
    return a.min_key <= b.max_key and b.min_key <= a.max_key  # => standard interval-overlap test
 
 
def push_down(incoming: SSTable, level: list[SSTable]) -> list[SSTable]:  # => co-13: merge overlaps away
    overlapping = [t for t in level if overlaps(incoming, t)]  # => every existing table this one touches
    non_overlapping = [t for t in level if t not in overlapping]  # => tables entirely untouched by the push
    merged_data: dict[str, str] = {}  # => the combined contents of the incoming table and its overlaps
    for table in overlapping:  # => older tables in this level, oldest key precedence first
        merged_data.update(table.data)  # => start from what was already in the level
    merged_data.update(incoming.data)  # => the incoming (newer) table's keys win any conflicts
    keys = sorted(merged_data)  # => the merged key range, sorted, needed for min/max below
    merged = SSTable(data=merged_data, min_key=keys[0], max_key=keys[-1])  # => one new, overlap-free table
    return non_overlapping + [merged]  # => the level after the push: untouched tables plus the merged one
 
 
level = [  # => two disjoint existing tables in this level
    SSTable(data={"a": "v1", "b": "v1"}, min_key="a", max_key="b"),  # => an existing table in this level
    SSTable(data={"m": "v1", "n": "v1"}, min_key="m", max_key="n"),  # => a DISJOINT existing table
]  # => end of the level fixture
incoming = SSTable(data={"b": "v2", "c": "v1"}, min_key="b", max_key="c")  # => overlaps the first table
 
new_level = push_down(incoming, level)  # => run leveled compaction's push-down step
print(len(new_level))  # => Output: 2
 
for i, table_a in enumerate(new_level):  # => check EVERY pair of tables in the resulting level
    for table_b in new_level[i + 1 :]:  # => compare each table only against the ones after it
        assert not overlaps(table_a, table_b)  # => no two tables in the level may share any key range
print("ex-49 OK")  # => Output: ex-49 OK

2
ex-49 OK

"""Example 49: pytest verification for LSM Leveled Compaction."""
 
from example import SSTable, overlaps, push_down
 
 
def test_pushed_down_level_has_no_key_overlap() -> None:
    level = [SSTable(data={"a": "1"}, min_key="a", max_key="a")]
    incoming = SSTable(data={"a": "2", "b": "1"}, min_key="a", max_key="b")
    result = push_down(incoming, level)
    for i, t1 in enumerate(result):
        for t2 in result[i + 1 :]:
            assert not overlaps(t1, t2)
 
 
def test_disjoint_incoming_table_stays_separate() -> None:
    level = [SSTable(data={"a": "1"}, min_key="a", max_key="a")]
    incoming = SSTable(data={"z": "1"}, min_key="z", max_key="z")
    result = push_down(incoming, level)
    assert len(result) == 2
 
 
# => Run: pytest -- Output: 2 passed

2 passed

"""Example 50: Count Write Amplification."""
# Write amplification (co-14) is bytes ACTUALLY written to disk / bytes the application asked to write.
 
application_bytes = 0  # => the true, logical amount of data the application wrote
disk_bytes_written = 0  # => every byte ACTUALLY written to storage, across every compaction pass
 
 
def flush_memtable(data: dict[str, str]) -> None:  # => the memtable's first write to disk (a new SSTable)
    global application_bytes, disk_bytes_written  # => mutate both module-level counters
    size = sum(len(k) + len(v) for k, v in data.items())  # => rough byte size of this batch
    application_bytes += size  # => this many bytes came from the application itself
    disk_bytes_written += size  # => the FIRST write of these bytes to disk
 
 
def compact(tables: list[dict[str, str]]) -> dict[str, str]:  # => merges N tables, REWRITING every byte
    global disk_bytes_written  # => this counter grows on EVERY compaction, not just the first flush
    merged: dict[str, str] = {}  # => the compacted result -- newest table's values win
    for table in tables:  # => later tables' values win on key conflicts
        merged.update(table)  # => later tables in the list override earlier ones on conflict
    size = sum(len(k) + len(v) for k, v in merged.items())  # => the merged table's own byte size
    disk_bytes_written += size  # => compaction WRITES these bytes again -- this is the amplification
    return merged  # => hand back the merged table for the next compaction round, if any
 
 
flush_memtable({"a": "v1", "b": "v1"})  # => first flush -- application writes these bytes once
flush_memtable({"c": "v1"})  # => second flush -- more application bytes
table1 = {"a": "v1", "b": "v1"}  # => a plain dict standing in for a flushed SSTable
table2 = {"c": "v1"}  # => a second flushed SSTable, disjoint keys from table1
merged = compact([table1, table2])  # => compaction #1 rewrites everything flushed so far
merged = compact([merged, {"a": "v2"}])  # => compaction #2 rewrites it AGAIN, plus a new update
 
amplification = disk_bytes_written / application_bytes  # => the exact ratio the spec asks to verify
print(application_bytes)  # => Output: 9
print(disk_bytes_written)  # => Output: 27
print(round(amplification, 2))  # => Output: 3.0
 
assert amplification > 1  # => compaction rewrote bytes multiple times -- more disk I/O than logical writes
print("ex-50 OK")  # => Output: ex-50 OK

9
27
3.0
ex-50 OK

"""Example 50: pytest verification for Write Amplification Counting."""
 
import example
 
 
def test_write_amplification_exceeds_one() -> None:
    example.application_bytes = 0
    example.disk_bytes_written = 0
    example.flush_memtable({"x": "1"})
    example.compact([{"x": "1"}])
    assert example.disk_bytes_written / example.application_bytes > 1
 
 
def test_a_single_flush_with_no_compaction_has_amplification_one() -> None:
    example.application_bytes = 0
    example.disk_bytes_written = 0
    example.flush_memtable({"x": "1"})
    assert example.disk_bytes_written / example.application_bytes == 1
 
 
# => Run: pytest -- Output: 2 passed