Beginner Examples

"""Example 1: Allocate a Fixed-Size Page -- the unit of I/O every storage engine reads and writes in."""
 
# A database never reads or writes a single row directly -- it reads and
# writes whole PAGES (co-01). Postgres/SQL Server default to 8 KB pages;
# InnoDB defaults to 16 KB -- the exact size is engine-specific, so this
# constant is named, not a hardcoded literal sprinkled through the file.
PAGE_SIZE: int = 4096  # => 4 KB, a convenient small-scale page size for this course
 
 
def new_page() -> bytearray:  # => allocates one zero-filled page buffer
    return bytearray(PAGE_SIZE)  # => bytearray(n) is mutable AND zero-initialized
 
 
page: bytearray = new_page()  # => a single fresh page
print(len(page))  # => Output: 4096
print(page[:8])  # => Output: bytearray(b'\x00\x00\x00\x00\x00\x00\x00\x00')
 
assert len(page) == PAGE_SIZE  # => the page's byte length always equals the page-size constant
assert all(byte == 0 for byte in page)  # => a brand-new page is entirely zero-filled before any write
print("ex-01 OK")  # => Output: ex-01 OK

4096
bytearray(b'\x00\x00\x00\x00\x00\x00\x00\x00')
ex-01 OK

"""Example 1: pytest verification for Fixed-Size Page Allocation."""
 
from example import PAGE_SIZE, new_page
 
 
def test_page_length_equals_page_size() -> None:
    page = new_page()  # => allocate via the SAME constructor example.py uses
    assert len(page) == PAGE_SIZE  # => every allocated page is exactly PAGE_SIZE bytes
 
 
def test_fresh_page_is_zero_filled() -> None:
    page = new_page()  # => a second, independent allocation
    assert page == bytearray(PAGE_SIZE)  # => no stale bytes leak in from a prior allocation
 
 
# => Run: pytest -- Output: 2 passed

2 passed

"""Example 2: Page Header Pack and Unpack -- pd_lower/pd_upper via struct."""
 
import struct
 
# A page's header stores two offsets: pd_lower (end of the slot array, grows
# UP from the front) and pd_upper (start of tuple data, grows DOWN from the
# back) (co-02). "<HH" packs both as little-endian unsigned 16-bit integers --
# 4 bytes total, enough to address a page far larger than this course's 4 KB.
HEADER_FORMAT: str = "<HH"  # => two uint16 fields: pd_lower, pd_upper
HEADER_SIZE: int = struct.calcsize(HEADER_FORMAT)  # => 4 bytes
 
 
def pack_header(pd_lower: int, pd_upper: int) -> bytes:  # => serializes the header
    return struct.pack(HEADER_FORMAT, pd_lower, pd_upper)  # => exactly HEADER_SIZE bytes
 
 
def unpack_header(raw: bytes) -> tuple[int, int]:  # => deserializes the header
    lo, hi = struct.unpack(HEADER_FORMAT, raw[:HEADER_SIZE])
    return int(lo), int(hi)  # => (pd_lower, pd_upper), as plain Python ints
 
 
packed: bytes = pack_header(HEADER_SIZE, 4096)  # => a fresh page: slots start right after the header
print(packed)  # => Output: b'\x04\x00\x00\x10'
pd_lower, pd_upper = unpack_header(packed)  # => round-trip back to Python ints
print((pd_lower, pd_upper))  # => Output: (4, 4096)
 
assert (pd_lower, pd_upper) == (HEADER_SIZE, 4096)  # => unpack(pack(x)) == x -- the round-trip holds
print("ex-02 OK")  # => Output: ex-02 OK

b'\x04\x00\x00\x10'
(4, 4096)
ex-02 OK

"""Example 2: pytest verification for Page Header Pack and Unpack."""
 
from example import HEADER_SIZE, pack_header, unpack_header
 
 
def test_header_size_is_four_bytes() -> None:
    assert HEADER_SIZE == 4  # => two uint16 fields pack into exactly 4 bytes
 
 
def test_pack_unpack_round_trips() -> None:
    packed = pack_header(100, 200)
    assert unpack_header(packed) == (100, 200)  # => unpack(pack(x)) always reproduces x exactly
 
 
def test_different_values_round_trip_independently() -> None:
    packed = pack_header(4, 4096)
    assert unpack_header(packed) == (4, 4096)  # => a second, independent pair round-trips too
 
 
# => Run: pytest -- Output: 3 passed

3 passed

"""Example 3: Append a Slot to the Slot Array -- grows from the front of the page."""
 
import struct
 
PAGE_SIZE: int = 4096
HEADER_SIZE: int = 4  # => pd_lower(uint16) + pd_upper(uint16)
SLOT_SIZE: int = 4  # => offset(uint16) + length(uint16) per slot
 
# The slot array grows FORWARD from right after the header; each slot is a
# fixed 4-byte (offset, length) pair pointing at one record's bytes elsewhere
# on the page (co-02). Appending a slot means: write the pair at pd_lower,
# then advance pd_lower past it.
 
 
def new_page() -> bytearray:  # => allocates one empty page with a fresh header already packed in
    page = bytearray(PAGE_SIZE)  # => zero-filled buffer
    struct.pack_into("<HH", page, 0, HEADER_SIZE, PAGE_SIZE)  # => empty page: no slots, no tuples yet
    return page  # => ready for inserts
 
 
def header(page: bytearray) -> tuple[int, int]:  # => reads the two header offsets back out
    lo, hi = struct.unpack_from("<HH", page, 0)  # => reads pd_lower, pd_upper in place
    return int(lo), int(hi)  # => plain Python ints, not struct's internal tuple type
 
 
def append_slot(page: bytearray, offset: int, length: int) -> int:  # => returns the new pd_lower
    pd_lower, pd_upper = header(page)  # => unpack the header before writing anything
    struct.pack_into("<HH", page, pd_lower, offset, length)  # => write the new slot at the OLD pd_lower
    pd_lower += SLOT_SIZE  # => the slot array just grew by one 4-byte entry
    struct.pack_into("<HH", page, 0, pd_lower, pd_upper)  # => persist the advanced pd_lower
    return pd_lower  # => callers use this to know how many slots now exist
 
 
page: bytearray = new_page()  # => one fresh page to append slots onto
before_lower, _ = header(page)  # => starts at HEADER_SIZE -- zero slots so far
after_first = append_slot(page, offset=4090, length=6)  # => slot 0 -> a 6-byte record near the back
after_second = append_slot(page, offset=4080, length=10)  # => slot 1 -> a 10-byte record
print((before_lower, after_first, after_second))  # => Output: (4, 8, 12)
 
assert after_first == before_lower + SLOT_SIZE  # => one append advances pd_lower by exactly SLOT_SIZE
assert after_second == before_lower + 2 * SLOT_SIZE  # => two appends advance it by 2 * SLOT_SIZE
slot_count = (after_second - HEADER_SIZE) // SLOT_SIZE  # => slot count derived from pd_lower's growth
assert slot_count == 2  # => exactly two slots now live in the slot array
print("ex-03 OK")  # => Output: ex-03 OK

(4, 8, 12)
ex-03 OK

"""Example 3: pytest verification for Slot Array Append."""
 
from example import HEADER_SIZE, SLOT_SIZE, append_slot, header, new_page
 
 
def test_first_append_advances_pd_lower_by_slot_size() -> None:
    page = new_page()
    before, _ = header(page)
    after = append_slot(page, offset=100, length=10)
    assert after == before + SLOT_SIZE  # => exactly one slot's worth of growth
 
 
def test_three_appends_grow_pd_lower_by_three_slots() -> None:
    page = new_page()
    before, _ = header(page)
    after = before
    for _ in range(3):
        after = append_slot(page, offset=200, length=5)
    assert after == before + 3 * SLOT_SIZE  # => three appends -> three slots' worth of growth
    assert (after - HEADER_SIZE) // SLOT_SIZE == 3  # => slot count derived from pd_lower matches
 
 
# => Run: pytest -- Output: 2 passed

2 passed

"""Example 4: Insert a Fixed-Size Record from the Back of the Page."""
 
import struct  # => stdlib module for packing/unpacking fixed-layout binary fields
 
PAGE_SIZE: int = 4096  # => same 4 KB page size used throughout this course
HEADER_SIZE: int = 4  # => pd_lower(uint16) + pd_upper(uint16), same layout as Example 2
 
 
def new_page() -> bytearray:  # => allocates one empty page with a fresh header already packed in
    page = bytearray(PAGE_SIZE)  # => zero-filled buffer
    struct.pack_into("<HH", page, 0, HEADER_SIZE, PAGE_SIZE)  # => empty page: no slots, no tuples yet
    return page  # => ready for inserts
 
 
def header(page: bytearray) -> tuple[int, int]:  # => reads the two header offsets back out
    lo, hi = struct.unpack_from("<HH", page, 0)  # => reads pd_lower, pd_upper in place
    return int(lo), int(hi)  # => plain Python ints, not struct's internal tuple type
 
 
def insert_record_from_back(page: bytearray, record: bytes) -> int:  # => returns the record's new offset
    pd_lower, pd_upper = header(page)  # => unpack the header before writing anything
    new_upper = pd_upper - len(record)  # => tuple data grows DOWNWARD, so subtract to make room
    page[new_upper:pd_upper] = record  # => the record's bytes land just before the OLD pd_upper
    struct.pack_into("<HH", page, 0, pd_lower, new_upper)  # => persist the lowered pd_upper
    return new_upper  # => also the record's own offset -- pd_upper only ever moves down
 
 
page: bytearray = new_page()  # => one fresh page
_, upper_before = header(page)  # => 4096: nothing stored yet
offset = insert_record_from_back(page, b"row-001")  # => a 7-byte record
_, upper_after = header(page)  # => re-read the header after the insert
print((upper_before, offset, upper_after))  # => Output: (4096, 4089, 4089)
# => notice offset == upper_after: the record IS the newly-reserved tail of the page
 
assert upper_after == upper_before - len(b"row-001")  # => pd_upper drops by exactly the record's size
assert page[offset : offset + 7] == b"row-001"  # => the record's bytes are readable back at its offset
print("ex-04 OK")  # => Output: ex-04 OK

(4096, 4089, 4089)
ex-04 OK

"""Example 4: pytest verification for Insert a Fixed-Size Record from the Back."""
 
from example import header, insert_record_from_back, new_page
 
 
def test_pd_upper_shrinks_by_record_length() -> None:
    page = new_page()
    _, before = header(page)
    insert_record_from_back(page, b"12345")
    _, after = header(page)
    assert after == before - 5  # => a 5-byte record shrinks pd_upper by exactly 5
 
 
def test_record_reads_back_unchanged() -> None:
    page = new_page()
    offset = insert_record_from_back(page, b"payload!")
    assert page[offset : offset + len(b"payload!")] == b"payload!"  # => bytes round-trip exactly
 
 
# => Run: pytest -- Output: 2 passed

2 passed

"""Example 5: Read a Record by Slot Index -- the indirection that makes a slot a stable handle."""
 
import struct  # => stdlib module for packing/unpacking fixed-layout binary fields
 
PAGE_SIZE: int = 4096  # => same 4 KB page size used throughout this course
HEADER_SIZE: int = 4  # => pd_lower(uint16) + pd_upper(uint16)
SLOT_SIZE: int = 4  # => offset(uint16) + length(uint16) per slot
 
 
def new_page() -> bytearray:  # => allocates one empty page with a fresh header already packed in
    page = bytearray(PAGE_SIZE)  # => zero-filled buffer
    struct.pack_into("<HH", page, 0, HEADER_SIZE, PAGE_SIZE)  # => empty page: no slots, no tuples yet
    return page  # => ready for inserts
 
 
def header(page: bytearray) -> tuple[int, int]:  # => reads the two header offsets back out
    lo, hi = struct.unpack_from("<HH", page, 0)  # => reads pd_lower, pd_upper in place
    return int(lo), int(hi)  # => plain Python ints, not struct's internal tuple type
 
 
def insert(page: bytearray, record: bytes) -> int:  # => inserts a record, returns its NEW SLOT INDEX
    pd_lower, pd_upper = header(page)  # => unpack the header before writing anything
    new_upper = pd_upper - len(record)  # => reserve space from the back
    page[new_upper:pd_upper] = record  # => write the record bytes
    struct.pack_into("<HH", page, pd_lower, new_upper, len(record))  # => the slot points AT the record
    slot_index = (pd_lower - HEADER_SIZE) // SLOT_SIZE  # => zero-based index of the slot just written
    struct.pack_into("<HH", page, 0, pd_lower + SLOT_SIZE, new_upper)  # => advance pd_lower, pd_upper
    return slot_index  # => the handle callers use to read this record back later
 
 
def read_by_slot(page: bytearray, slot_index: int) -> bytes:  # => resolves a slot index to record bytes
    slot_pos = HEADER_SIZE + slot_index * SLOT_SIZE  # => byte offset of THIS slot in the array
    off, length = struct.unpack_from("<HH", page, slot_pos)  # => the slot's own (offset, length) pair
    return bytes(page[int(off) : int(off) + int(length)])  # => slice out exactly `length` bytes
 
 
page: bytearray = new_page()  # => one fresh page to insert two records into
idx_a = insert(page, b"alice")  # => slot 0
idx_b = insert(page, b"bob-two")  # => slot 1
print((idx_a, idx_b))  # => Output: (0, 1)
# => slot indexes, NOT byte offsets -- the indirection ex-08 later depends on
print(read_by_slot(page, idx_a))  # => Output: b'alice'
print(read_by_slot(page, idx_b))  # => Output: b'bob-two'
 
assert read_by_slot(page, idx_a) == b"alice"  # => slot 0 resolves to exactly the bytes inserted there
assert read_by_slot(page, idx_b) == b"bob-two"  # => slot 1 resolves independently of slot 0
print("ex-05 OK")  # => Output: ex-05 OK

(0, 1)
b'alice'
b'bob-two'
ex-05 OK

"""Example 5: pytest verification for Read a Record by Slot Index."""
 
from example import insert, new_page, read_by_slot
 
 
def test_each_slot_resolves_to_its_own_record() -> None:
    page = new_page()
    idx1 = insert(page, b"one")
    idx2 = insert(page, b"two-two")
    assert read_by_slot(page, idx1) == b"one"
    assert read_by_slot(page, idx2) == b"two-two"
 
 
def test_slot_indexes_are_sequential() -> None:
    page = new_page()
    indexes = [insert(page, str(i).encode()) for i in range(4)]
    assert indexes == [0, 1, 2, 3]  # => slot indexes assign in insertion order
 
 
# => Run: pytest -- Output: 2 passed

2 passed

"""Example 6: Guard Against Insufficient Free Space -- insert raises before it corrupts the page."""
 
import struct  # => stdlib module for packing/unpacking fixed-layout binary fields
 
PAGE_SIZE: int = 4096  # => same 4 KB page size used throughout this course
HEADER_SIZE: int = 4  # => pd_lower(uint16) + pd_upper(uint16)
SLOT_SIZE: int = 4  # => offset(uint16) + length(uint16) per slot
 
 
class PageFullError(Exception):  # => raised instead of silently overwriting live data
    """Raised when a page has too little free space for a new record + slot."""  # => documents the contract
 
 
def new_page() -> bytearray:  # => allocates one empty page with a fresh header already packed in
    page = bytearray(PAGE_SIZE)  # => zero-filled buffer
    struct.pack_into("<HH", page, 0, HEADER_SIZE, PAGE_SIZE)  # => empty page: no slots, no tuples yet
    return page  # => ready for inserts
 
 
def header(page: bytearray) -> tuple[int, int]:  # => reads the two header offsets back out
    lo, hi = struct.unpack_from("<HH", page, 0)  # => reads pd_lower, pd_upper in place
    return int(lo), int(hi)  # => plain Python ints, not struct's internal tuple type
 
 
def free_space(page: bytearray) -> int:  # => the gap between the slot array and the tuple data
    pd_lower, pd_upper = header(page)  # => unpack the header before writing anything
    return pd_upper - pd_lower  # => shrinks from BOTH ends as slots and records are added
 
 
def insert(page: bytearray, record: bytes) -> int:  # => guarded insert: raises instead of corrupting
    pd_lower, pd_upper = header(page)  # => unpack the header before writing anything
    needed = len(record) + SLOT_SIZE  # => the record AND its slot both need room
    if free_space(page) < needed:  # => the guard: check BEFORE writing anything
        raise PageFullError(f"need {needed} bytes, have {free_space(page)}")  # => no write happened
    new_upper = pd_upper - len(record)  # => safe to reserve room now -- the guard already passed
    page[new_upper:pd_upper] = record  # => write the record's bytes into the reserved gap
    struct.pack_into("<HH", page, pd_lower, new_upper, len(record))  # => point a new slot at it
    struct.pack_into("<HH", page, 0, pd_lower + SLOT_SIZE, new_upper)  # => persist the advanced header
    return new_upper  # => the record's offset, same convention as ex-04
 
 
page: bytearray = new_page()  # => one fresh, empty page
print(free_space(page))  # => Output: 4092
insert(page, b"x" * 4000)  # => a near-page-sized record: leaves little free space
print(free_space(page))  # => Output: 88
# => 88 bytes left is not nearly enough for a 200-byte record plus its 4-byte slot
 
raised = False  # => flips to True only if the guard actually fires below
try:
    insert(page, b"y" * 200)  # => 200 + SLOT_SIZE (204) exceeds the remaining 88 bytes
except PageFullError:  # => the exact exception free_space's guard raises
    raised = True  # => confirms the guard fired instead of writing past the page
assert raised  # => the guard fired instead of silently corrupting adjacent bytes
# => nothing on the page changed as a result of the rejected insert attempt
print("ex-06 OK")  # => Output: ex-06 OK

4092
88
ex-06 OK

"""Example 6: pytest verification for the Free-Space Guard."""
 
import pytest
 
from example import PageFullError, free_space, insert, new_page
 
 
def test_free_space_shrinks_after_insert() -> None:
    page = new_page()
    before = free_space(page)
    insert(page, b"x" * 100)
    assert free_space(page) < before  # => both the slot AND the record consumed free space
 
 
def test_insert_raises_when_free_space_is_insufficient() -> None:
    page = new_page()
    insert(page, b"x" * 4000)
    with pytest.raises(PageFullError):
        insert(page, b"y" * 200)  # => the remaining free space can't fit this record + its slot
 
 
# => Run: pytest -- Output: 2 passed

2 passed

"""Example 7: Variable-Length Records Addressed by Slot."""
 
import struct  # => stdlib module for packing/unpacking fixed-layout binary fields
 
PAGE_SIZE: int = 4096  # => same 4 KB page size used throughout this course
HEADER_SIZE: int = 4  # => pd_lower(uint16) + pd_upper(uint16)
SLOT_SIZE: int = 4  # => offset(uint16) + length(uint16) per slot
 
 
def new_page() -> bytearray:  # => allocates one empty page with a fresh header already packed in
    page = bytearray(PAGE_SIZE)  # => zero-filled buffer
    struct.pack_into("<HH", page, 0, HEADER_SIZE, PAGE_SIZE)  # => empty page: no slots, no tuples yet
    return page  # => ready for inserts
 
 
def header(page: bytearray) -> tuple[int, int]:  # => reads the two header offsets back out
    lo, hi = struct.unpack_from("<HH", page, 0)  # => reads pd_lower, pd_upper in place
    return int(lo), int(hi)  # => plain Python ints, not struct's internal tuple type
 
 
def insert(page: bytearray, record: bytes) -> int:  # => each call may store a DIFFERENT-length record
    pd_lower, pd_upper = header(page)  # => unpack the header before writing anything
    new_upper = pd_upper - len(record)  # => length varies per call -- no fixed record stride
    page[new_upper:pd_upper] = record  # => write exactly len(record) bytes, whatever that length is
    struct.pack_into("<HH", page, pd_lower, new_upper, len(record))  # => length travels WITH the slot
    struct.pack_into("<HH", page, 0, pd_lower + SLOT_SIZE, new_upper)  # => persist the advanced header
    return (pd_lower - HEADER_SIZE) // SLOT_SIZE  # => this record's new slot index
 
 
def read(page: bytearray, slot_index: int) -> bytes:  # => reads a record at whatever length ITS slot says
    slot_pos = HEADER_SIZE + slot_index * SLOT_SIZE  # => byte offset of this slot in the array
    off, length = struct.unpack_from("<HH", page, slot_pos)  # => each slot's OWN length, not a global one
    return bytes(page[int(off) : int(off) + int(length)])  # => slice out exactly `length` bytes
 
 
page: bytearray = new_page()  # => one fresh page to hold both records
short_idx = insert(page, b"hi")  # => a 2-byte record
long_idx = insert(page, b"a much longer variable-length row")  # => a 33-byte record
print(read(page, short_idx))  # => Output: b'hi'
print(read(page, long_idx))  # => Output: b'a much longer variable-length row'
# => two wildly different lengths, same page, same slot-based read function
 
assert len(read(page, short_idx)) == 2  # => the short record reads back at its OWN length
assert len(read(page, long_idx)) == 33  # => the long record reads back at ITS OWN length, unaffected
print("ex-07 OK")  # => Output: ex-07 OK

b'hi'
b'a much longer variable-length row'
ex-07 OK

"""Example 7: pytest verification for Variable-Length Records."""
 
from example import insert, new_page, read
 
 
def test_short_and_long_records_coexist() -> None:
    page = new_page()
    short_idx = insert(page, b"hi")
    long_idx = insert(page, b"a genuinely much longer row of bytes")
    assert read(page, short_idx) == b"hi"
    assert read(page, long_idx) == b"a genuinely much longer row of bytes"
 
 
def test_each_record_reads_at_its_own_length() -> None:
    page = new_page()
    idx = insert(page, b"12345")
    assert len(read(page, idx)) == 5  # => the slot's OWN length field governs, not a fixed stride
 
 
# => Run: pytest -- Output: 2 passed

2 passed

"""Example 8: In-Page Compaction After a Delete -- surviving slots still resolve correctly."""
 
import struct  # => stdlib module for packing/unpacking fixed-layout binary fields
 
PAGE_SIZE: int = 4096  # => same 4 KB page size used throughout this course
HEADER_SIZE: int = 4  # => pd_lower(uint16) + pd_upper(uint16)
SLOT_SIZE: int = 4  # => offset(uint16) + length(uint16) per slot
DEAD: int = 0xFFFF  # => sentinel offset marking a slot as deleted (tombstoned, not removed)
 
 
def new_page() -> bytearray:  # => allocates one empty page with a fresh header already packed in
    page = bytearray(PAGE_SIZE)  # => zero-filled buffer
    struct.pack_into("<HH", page, 0, HEADER_SIZE, PAGE_SIZE)  # => empty page: no slots, no tuples yet
    return page  # => ready for inserts
 
 
def header(page: bytearray) -> tuple[int, int]:  # => reads the two header offsets back out
    lo, hi = struct.unpack_from("<HH", page, 0)  # => reads pd_lower, pd_upper in place
    return int(lo), int(hi)  # => plain Python ints, not struct's internal tuple type
 
 
def slot_count(page: bytearray) -> int:  # => how many slots exist -- live AND deleted
    pd_lower, _ = header(page)  # => only pd_lower is needed for a slot count
    return (pd_lower - HEADER_SIZE) // SLOT_SIZE  # => each slot occupies exactly SLOT_SIZE bytes
 
 
def insert(page: bytearray, record: bytes) -> int:  # => appends a record, returns its new slot index
    pd_lower, pd_upper = header(page)  # => unpack the header before writing anything
    new_upper = pd_upper - len(record)  # => reserve room from the back
    page[new_upper:pd_upper] = record  # => write the record's bytes
    struct.pack_into("<HH", page, pd_lower, new_upper, len(record))  # => point a new slot at it
    struct.pack_into("<HH", page, 0, pd_lower + SLOT_SIZE, new_upper)  # => persist the advanced header
    return (pd_lower - HEADER_SIZE) // SLOT_SIZE  # => this record's new slot index
 
 
def delete(page: bytearray, slot_index: int) -> None:  # => tombstones the slot; bytes stay until compaction
    slot_pos = HEADER_SIZE + slot_index * SLOT_SIZE  # => byte offset of this slot in the array
    _, length = struct.unpack_from("<HH", page, slot_pos)  # => keep the length -- only the offset dies
    struct.pack_into("<HH", page, slot_pos, DEAD, length)  # => offset -> DEAD marks it unreadable
 
 
def read(page: bytearray, slot_index: int) -> bytes:  # => resolves a live slot to its record bytes
    slot_pos = HEADER_SIZE + slot_index * SLOT_SIZE  # => byte offset of this slot in the array
    off, length = struct.unpack_from("<HH", page, slot_pos)  # => this slot's (offset, length) pair
    if off == DEAD:  # => a tombstoned slot has no live bytes to read
        raise KeyError(f"slot {slot_index} is deleted")  # => fail loudly, never return stale bytes
    return bytes(page[int(off) : int(off) + int(length)])  # => slice out exactly `length` bytes
 
 
def compact(page: bytearray) -> None:  # => squeezes out the gap a deleted record left behind
    pd_lower, _ = header(page)  # => the slot array itself does NOT shrink during compaction
    live: list[tuple[int, bytes]] = []  # => (slot_index, bytes) for every non-deleted slot
    for i in range(slot_count(page)):  # => walk every slot, live or dead
        slot_pos = HEADER_SIZE + i * SLOT_SIZE  # => byte offset of slot i
        off, length = struct.unpack_from("<HH", page, slot_pos)  # => this slot's current (offset, length)
        if off != DEAD:  # => only live slots get carried forward into the compacted layout
            live.append((i, bytes(page[int(off) : int(off) + int(length)])))  # => snapshot its bytes NOW
    write_at = PAGE_SIZE  # => rewrite tuple data from the very back, tightly packed
    for i, data in live:  # => re-lay live records back-to-back, closing the gap the delete left
        write_at -= len(data)  # => reserve this record's own length before writing it
        page[write_at : write_at + len(data)] = data  # => write the record at its NEW, compacted offset
        slot_pos = HEADER_SIZE + i * SLOT_SIZE  # => the SAME slot position as before -- index unchanged
        struct.pack_into("<HH", page, slot_pos, write_at, len(data))  # => slot now points at the NEW offset
    struct.pack_into("<HH", page, 0, pd_lower, write_at)  # => pd_upper reclaims the freed gap
 
 
page: bytearray = new_page()  # => one fresh page for a, b, c
a = insert(page, b"row-a")  # => slot 0
b = insert(page, b"row-b-middle")  # => slot 1 -- this one gets deleted below
c = insert(page, b"row-c")  # => slot 2
delete(page, b)  # => tombstones slot 1; its bytes still physically sit on the page for now
compact(page)  # => squeezes slot 1's gap out and rewrites slots 0 and 2 at new offsets
print(read(page, a))  # => Output: b'row-a'
print(read(page, c))  # => Output: b'row-c'
 
assert read(page, a) == b"row-a"  # => slot a still resolves correctly after compaction moved its bytes
assert read(page, c) == b"row-c"  # => slot c still resolves correctly too
_, pd_upper_after = header(page)  # => re-read the header after compaction
assert pd_upper_after == PAGE_SIZE - len(b"row-a") - len(b"row-c")  # => the deleted record's gap is gone
print("ex-08 OK")  # => Output: ex-08 OK
# => the slot INDEX never changed for either survivor -- only its internal offset did

b'row-a'
b'row-c'
ex-08 OK

"""Example 8: pytest verification for In-Page Compaction."""
 
from example import compact, delete, header, insert, new_page, read
 
 
def test_surviving_slot_resolves_after_compaction() -> None:
    page = new_page()
    a = insert(page, b"keep-a")
    b = insert(page, b"delete-me")
    delete(page, b)
    compact(page)
    assert read(page, a) == b"keep-a"  # => slot a's offset was rewritten but its bytes stayed correct
 
 
def test_compaction_reclaims_the_deleted_gap() -> None:
    page = new_page()
    insert(page, b"keep-a")
    b = insert(page, b"delete-me")
    delete(page, b)
    compact(page)
    _, pd_upper = header(page)
    assert pd_upper == 4096 - len(b"keep-a")  # => only the surviving record's bytes remain reserved
 
 
# => Run: pytest -- Output: 2 passed

2 passed

"""Example 9: Detect Page Corruption with a Checksum."""
 
import zlib  # => stdlib module -- CRC32 is fast and good enough for a corruption DETECTOR
 
PAGE_SIZE: int = 4096  # => same 4 KB page size used throughout this course
 
# A checksum over the whole page is a cheap corruption DETECTOR, not a fix
# (co-26) -- CRC32 is not cryptographically strong, but flipping even one bit
# anywhere it covers changes the result, which is all detection needs.
 
 
def checksum(page: bytes | bytearray) -> int:  # => a single deterministic integer summarizing every byte
    return zlib.crc32(bytes(page))  # => same input bytes always produce the same checksum
 
 
page = bytearray(PAGE_SIZE)  # => start from a fresh, zero-filled page
page[100:110] = b"hello-row!"  # => simulate SOME real content living on the page
original = checksum(page)  # => the checksum of the page as it exists right now
print(hex(original))  # => Output: 0x9fc341db (a genuine, captured hex CRC for this exact byte layout)
 
corrupted = bytearray(page)  # => an independent COPY -- `page` itself stays untouched
corrupted[0] ^= 0x01  # => flip a SINGLE bit in a byte the CRC actually covers
corrupted_sum = checksum(corrupted)  # => recompute over the corrupted copy
 
assert corrupted_sum != original  # => one flipped bit changes the checksum -- corruption is detectable
assert checksum(bytes(page)) == original  # => an UNCHANGED page always reproduces the same checksum
print("ex-09 OK")  # => Output: ex-09 OK

0x930aab07
ex-09 OK

"""Example 9: pytest verification for Page Checksum Detection."""
 
from example import checksum
 
 
def test_identical_pages_checksum_identically() -> None:
    page = bytearray(4096)
    page[50:60] = b"same-bytes"
    assert checksum(bytes(page)) == checksum(bytes(page))  # => deterministic: same bytes, same checksum
 
 
def test_single_bit_flip_changes_the_checksum() -> None:
    page = bytearray(4096)
    page[50:60] = b"same-bytes"
    flipped = bytearray(page)
    flipped[0] ^= 0x01
    assert checksum(bytes(page)) != checksum(bytes(flipped))  # => one bit is enough to detect corruption
 
 
# => Run: pytest -- Output: 2 passed

2 passed

"""Example 10: Buffer Pool Page-Table Lookup -- page-id maps to a resident frame."""
 
from dataclasses import dataclass  # => a plain, typed record for one buffer-pool frame
 
 
@dataclass
class Frame:  # => one buffer-pool slot holding a page's in-memory copy
    page_id: int  # => which on-disk page this frame currently holds
    data: bytes  # => that page's bytes, as loaded into memory
 
 
# The page table maps page_id -> Frame (co-04) -- it is the buffer pool's
# core index, and a lookup miss (returning None) is exactly the signal that
# tells a fetch it must go to disk.
page_table: dict[int, Frame] = {}  # => empty pool: nothing resident yet
page_table[7] = Frame(page_id=7, data=b"resident-page-7")  # => page 7 is loaded into the pool
page_table[9] = Frame(page_id=9, data=b"resident-page-9")  # => page 9 is loaded too
 
found = page_table.get(7)  # => a lookup for a RESIDENT page id
print(found)  # => Output: Frame(page_id=7, data=b'resident-page-7')
missing = page_table.get(3)  # => a lookup for a page id NOT currently resident
print(missing)  # => Output: None
 
assert found is not None and found.page_id == 7  # => a resident lookup returns its own frame
assert missing is None  # => a non-resident lookup returns None, never a stale/wrong frame
print("ex-10 OK")  # => Output: ex-10 OK

Frame(page_id=7, data=b'resident-page-7')
None
ex-10 OK

"""Example 10: pytest verification for Buffer Pool Page-Table Lookup."""
 
from example import Frame
 
 
def test_resident_lookup_returns_its_frame() -> None:
    page_table: dict[int, Frame] = {1: Frame(page_id=1, data=b"one")}
    assert page_table.get(1) is not None
    assert page_table[1].data == b"one"
 
 
def test_non_resident_lookup_returns_none() -> None:
    page_table: dict[int, Frame] = {1: Frame(page_id=1, data=b"one")}
    assert page_table.get(99) is None  # => no frame exists for page 99
 
 
# => Run: pytest -- Output: 2 passed

2 passed

"""Example 11: Buffer Fetch on a Miss -- loads from the fake disk and counts the miss."""
 
from dataclasses import dataclass, field  # => a typed, mutable buffer-pool model
 
 
@dataclass  # => auto-generates __init__ from the fields below
class BufferPool:  # => models the whole read path: disk, resident frames, hit/miss counters
    disk: dict[int, bytes]  # => stand-in for a real file on disk, keyed by page id
    frames: dict[int, bytes] = field(default_factory=dict[int, bytes])  # => resident pages
    hits: int = 0  # => counts fetches that found the page already resident
    misses: int = 0  # => counts fetches that had to load from disk
 
    def fetch(self, page_id: int) -> bytes:  # => the ONLY read path -- callers never touch disk directly
        if page_id in self.frames:  # => already resident: no disk access needed
            self.hits += 1  # => counted as a hit -- disk untouched
            return self.frames[page_id]  # => serve straight from memory
        self.misses += 1  # => not resident: this IS a miss
        data = self.disk[page_id]  # => load from the fake disk
        self.frames[page_id] = data  # => cache it so the NEXT fetch of this page is a hit
        return data  # => the newly-loaded bytes
 
 
pool = BufferPool(disk={1: b"page-one", 2: b"page-two"})  # => a fake disk with two pages, empty pool
print(pool.fetch(1))  # => Output: b'page-one'
print((pool.hits, pool.misses))  # => Output: (0, 1)
# => one miss, zero hits -- this was the very first fetch of page 1
 
assert pool.misses == 1  # => the first fetch of an unloaded page is always a miss
assert 1 in pool.frames  # => after the miss, the page IS now resident for future fetches
print("ex-11 OK")  # => Output: ex-11 OK

b'page-one'
(0, 1)
ex-11 OK

"""Example 11: pytest verification for Buffer Fetch on a Miss."""
 
from example import BufferPool
 
 
def test_first_fetch_of_a_page_is_a_miss() -> None:
    pool = BufferPool(disk={10: b"ten"})
    pool.fetch(10)
    assert pool.misses == 1 and pool.hits == 0
 
 
def test_miss_loads_the_page_into_frames() -> None:
    pool = BufferPool(disk={10: b"ten"})
    pool.fetch(10)
    assert 10 in pool.frames  # => now resident for future fetches
 
 
# => Run: pytest -- Output: 2 passed

2 passed

"""Example 12: Buffer Fetch on a Hit -- no disk read when the page is already resident."""
 
from dataclasses import dataclass, field  # => a typed, mutable buffer-pool model
 
 
@dataclass  # => auto-generates __init__ from the fields below
class BufferPool:  # => same shape as ex-11, plus a disk_reads audit trail
    disk: dict[int, bytes]  # => stand-in for a real file on disk, keyed by page id
    frames: dict[int, bytes] = field(default_factory=dict[int, bytes])  # => resident pages
    hits: int = 0  # => counts fetches that found the page already resident
    misses: int = 0  # => counts fetches that had to load from disk
    disk_reads: list[int] = field(default_factory=list[int])  # => records EVERY page_id read from disk
 
    def fetch(self, page_id: int) -> bytes:  # => same read path as ex-11, now tracking disk_reads too
        if page_id in self.frames:  # => already resident: no disk access needed
            self.hits += 1  # => resident: counted as a hit, disk untouched
            return self.frames[page_id]  # => serve straight from memory
        self.misses += 1  # => not resident: this IS a miss
        self.disk_reads.append(page_id)  # => only a miss ever appends here
        data = self.disk[page_id]  # => the ONLY line in this file that reads the fake disk
        self.frames[page_id] = data  # => now resident -- the NEXT fetch of this page is a hit
        return data  # => the newly-loaded bytes
 
 
pool = BufferPool(disk={5: b"page-five"})  # => a fake disk with one page, empty pool
pool.fetch(5)  # => first fetch: a miss, loads page 5 into frames
second = pool.fetch(5)  # => second fetch: page 5 is now resident
print(second)  # => Output: b'page-five'
print((pool.hits, pool.misses, pool.disk_reads))  # => Output: (1, 1, [5])
# => two fetches total, but disk_reads has only ONE entry -- the second fetch never touched it
 
assert pool.hits == 1  # => exactly one hit occurred (the second fetch)
assert pool.disk_reads == [5]  # => disk was read exactly once, never again for the same page
print("ex-12 OK")  # => Output: ex-12 OK

b'page-five'
(1, 1, [5])
ex-12 OK

"""Example 12: pytest verification for Buffer Fetch on a Hit."""
 
from example import BufferPool
 
 
def test_second_fetch_of_same_page_is_a_hit() -> None:
    pool = BufferPool(disk={5: b"five"})
    pool.fetch(5)
    pool.fetch(5)
    assert pool.hits == 1 and pool.misses == 1
 
 
def test_disk_is_read_at_most_once_per_page() -> None:
    pool = BufferPool(disk={5: b"five"})
    for _ in range(5):
        pool.fetch(5)
    assert pool.disk_reads == [5]  # => 5 fetches, but disk touched only on the very first one
 
 
# => Run: pytest -- Output: 2 passed

2 passed

"""Example 13: Pin Count Guards Against Eviction -- a pinned frame is never a victim."""
 
from dataclasses import dataclass  # => a plain, typed record for one buffer-pool frame
 
 
@dataclass  # => auto-generates __init__ from the fields below
class Frame:  # => a buffer-pool frame that tracks how many callers currently need it
    page_id: int  # => which on-disk page this frame currently holds
    pin_count: int = 0  # => how many active readers/writers currently need this page
 
 
class PinnedFrameError(Exception):  # => raised instead of silently evicting in-use data
    """Raised when eviction is attempted on a frame with pin_count > 0."""  # => documents the contract
 
 
def pin(frame: Frame) -> None:  # => a caller starting work on the page increments the pin count
    frame.pin_count += 1  # => one more caller now depends on this frame staying resident
 
 
def unpin(frame: Frame) -> None:  # => a caller finishing work decrements it back down
    frame.pin_count -= 1  # => one fewer caller depends on it -- may reach zero and become evictable
 
 
def evict(frames: list[Frame], victim_id: int) -> None:  # => removes a frame, but only if it is safe to
    victim = next(f for f in frames if f.page_id == victim_id)  # => locate the requested frame
    if victim.pin_count > 0:  # => the guard: refuse to evict anything still in use
        raise PinnedFrameError(f"page {victim_id} is pinned ({victim.pin_count} pins)")  # => refuse
    frames.remove(victim)  # => only reached once pin_count is exactly zero
 
 
frames = [Frame(page_id=1), Frame(page_id=2)]  # => two candidate frames, neither pinned yet
pin(frames[0])  # => page 1 is now in active use
raised = False  # => flips to True only if the guard actually fires below
try:  # => the guard should raise here, not silently succeed
    evict(frames, victim_id=1)  # => attempting to evict a PINNED page
except PinnedFrameError:  # => the exact exception the guard raises
    raised = True  # => confirms the guard fired instead of evicting in-use data
assert raised  # => eviction was refused while the pin was held
evict(frames, victim_id=2)  # => page 2 has zero pins -- this succeeds
assert [f.page_id for f in frames] == [1]  # => only the unpinned page was actually removed
# => the pin count, not any eviction-policy ranking, had the final say over what got removed
print("ex-13 OK")  # => Output: ex-13 OK

ex-13 OK

"""Example 13: pytest verification for the Pin Count Guard."""
 
import pytest
 
from example import Frame, PinnedFrameError, evict, pin, unpin
 
 
def test_pinned_frame_cannot_be_evicted() -> None:
    frames = [Frame(page_id=1)]
    pin(frames[0])
    with pytest.raises(PinnedFrameError):
        evict(frames, victim_id=1)
 
 
def test_unpinned_frame_can_be_evicted() -> None:
    frames = [Frame(page_id=1)]
    pin(frames[0])
    unpin(frames[0])  # => pin_count is back to zero
    evict(frames, victim_id=1)
    assert frames == []
 
 
# => Run: pytest -- Output: 2 passed

2 passed

"""Example 14: Flush a Dirty Page Before Eviction."""
 
from dataclasses import dataclass  # => a plain, typed record for one buffer-pool frame
 
 
@dataclass  # => auto-generates __init__ from the fields below
class Frame:  # => a buffer-pool frame that tracks whether it needs flushing
    page_id: int  # => which on-disk page this frame currently holds
    data: bytes  # => this frame's current in-memory bytes
    dirty: bool = False  # => True once a WRITE has touched this frame since it was loaded
 
 
def write(frame: Frame, new_data: bytes) -> None:  # => the only way a frame's bytes change in memory
    frame.data = new_data  # => the in-memory copy now diverges from whatever disk still holds
    frame.dirty = True  # => marks it as needing a flush before it can safely leave memory
 
 
def evict(frame: Frame, disk: dict[int, bytes]) -> None:  # => removes a frame from memory, safely
    if frame.dirty:  # => a dirty frame's changes only exist in memory so far
        disk[frame.page_id] = frame.data  # => flush BEFORE the frame is discarded
    # => a non-dirty (clean) frame already matches disk -- nothing to flush
 
 
disk: dict[int, bytes] = {7: b"original-on-disk"}  # => a fake disk with page 7's original bytes
frame = Frame(page_id=7, data=b"original-on-disk")  # => the same bytes, now also resident in memory
write(frame, b"modified-in-memory")  # => the frame diverges from disk and becomes dirty
assert disk[7] == b"original-on-disk"  # => disk has NOT changed yet -- only the in-memory frame has
evict(frame, disk)  # => eviction flushes the dirty frame first
print(disk[7])  # => Output: b'modified-in-memory'
 
assert disk[7] == b"modified-in-memory"  # => the flush happened BEFORE the frame left memory
print("ex-14 OK")  # => Output: ex-14 OK

b'modified-in-memory'
ex-14 OK

"""Example 14: pytest verification for Flushing a Dirty Page Before Eviction."""
 
from example import Frame, evict, write
 
 
def test_clean_frame_needs_no_flush() -> None:
    disk: dict[int, bytes] = {1: b"same"}
    frame = Frame(page_id=1, data=b"same")  # => never written -- dirty stays False
    evict(frame, disk)
    assert disk[1] == b"same"  # => unchanged, because there was nothing to flush
 
 
def test_dirty_frame_is_flushed_before_eviction() -> None:
    disk: dict[int, bytes] = {1: b"old"}
    frame = Frame(page_id=1, data=b"old")
    write(frame, b"new")
    evict(frame, disk)
    assert disk[1] == b"new"  # => the write reached disk exactly at eviction time
 
 
# => Run: pytest -- Output: 2 passed

2 passed

"""Example 15: LRU Eviction -- the least-recently-used frame is always the victim."""
 
from collections import OrderedDict  # => a dict that also remembers insertion/move-to-end order
 
 
class LRUCache:  # => insertion/access order IS the recency order, via OrderedDict
    def __init__(self, capacity: int) -> None:  # => capacity is the max number of resident pages
        self.capacity = capacity  # => stored so touch() knows when it has overflowed
        self.pool: OrderedDict[int, bytes] = OrderedDict()  # => page_id -> bytes, oldest-first order
        self.evicted: list[int] = []  # => records every page_id ever evicted, in eviction order
 
    def touch(self, page_id: int, data: bytes) -> None:  # => access (or load) a page: marks it MOST recent
        if page_id in self.pool:  # => already resident: this is a re-access, not a fresh load
            self.pool.move_to_end(page_id)  # => re-access moves it to the "most recently used" end
        self.pool[page_id] = data  # => insert (or refresh) this page at the most-recent end
        if len(self.pool) > self.capacity:  # => over capacity: evict the LEAST recently used
            victim_id, _ = self.pool.popitem(last=False)  # => removes the OLDEST (front) entry
            self.evicted.append(victim_id)  # => record it for this example's own verification below
 
 
cache = LRUCache(capacity=3)  # => room for 3 resident pages
cache.touch(1, b"a")  # => page 1 loaded
cache.touch(2, b"b")  # => page 2 loaded
cache.touch(3, b"c")  # => pool is now full: order is [1, 2, 3], oldest-first
cache.touch(1, b"a")  # => RE-touching page 1 moves it to most-recent -- no eviction yet (still 3 entries)
cache.touch(4, b"d")  # => now over capacity: evicts the current LRU entry
print(cache.evicted)  # => Output: [2]
print(list(cache.pool.keys()))  # => Output: [3, 1, 4]
 
assert cache.evicted == [2]  # => page 2 was evicted -- it was LRU once page 1 got re-touched before it
assert 1 in cache.pool and 2 not in cache.pool  # => the re-touched page survived; the untouched one did not
# => recency, not insertion order, is exactly what OrderedDict.move_to_end tracks here
print("ex-15 OK")  # => Output: ex-15 OK

[2]
[3, 1, 4]
ex-15 OK

"""Example 15: pytest verification for LRU Eviction."""
 
from example import LRUCache
 
 
def test_re_touched_page_survives_eviction() -> None:
    cache = LRUCache(capacity=2)
    cache.touch(1, b"a")
    cache.touch(2, b"b")
    cache.touch(1, b"a")  # => re-touch page 1 -- it is now the most recent
    cache.touch(3, b"c")  # => forces an eviction
    assert cache.evicted == [2]  # => page 2, not page 1, was the true LRU victim
 
 
def test_capacity_is_never_exceeded() -> None:
    cache = LRUCache(capacity=2)
    for page_id in range(5):
        cache.touch(page_id, b"x")
    assert len(cache.pool) == 2  # => the pool never grows past its declared capacity
 
 
# => Run: pytest -- Output: 2 passed

2 passed

"""Example 16: CLOCK (Second-Chance) Eviction -- a reference bit approximates LRU cheaply."""
 
from dataclasses import dataclass  # => a plain, typed record for one buffer-pool frame
 
 
@dataclass  # => auto-generates __init__ from the fields below
class Frame:  # => a buffer-pool frame carrying one cheap reference bit
    page_id: int  # => which on-disk page this frame currently holds
    referenced: bool = False  # => set True on every access; CLOCK's cheap stand-in for "recently used"
 
 
def clock_evict(frames: list[Frame], hand: int) -> tuple[int, int]:  # => returns (victim_id, new_hand)
    while True:  # => the "clock hand" sweeps frames in a circle until it finds a victim
        frame = frames[hand]  # => the frame currently under the sweeping hand
        if frame.referenced:  # => this frame gets a "second chance" -- clear the bit and move on
            frame.referenced = False  # => second chance spent -- next sweep WOULD evict it
            hand = (hand + 1) % len(frames)  # => advance the hand, wrapping around circularly
        else:  # => this frame's bit was already clear -- it becomes the victim
            victim_id = frame.page_id  # => remember which page is being evicted
            del frames[hand]  # => remove it from the pool entirely
            return victim_id, hand % max(len(frames), 1)  # => report the victim and where the hand stops next
 
 
frames = [  # => three candidate frames for the clock hand to sweep across
    Frame(page_id=1, referenced=True),  # => recently accessed
    Frame(page_id=2),  # => NOT recently accessed -- referenced defaults to False
    Frame(page_id=3, referenced=True),  # => recently accessed
]
victim, hand = clock_evict(frames, hand=0)  # => the hand starts sweeping from index 0
print(victim)  # => Output: 2
# => the hand never even had to inspect page 1 or 3's bit a second time to find this victim
print([(f.page_id, f.referenced) for f in frames])  # => Output: [(1, False), (3, True)]
 
assert victim == 2  # => page 2's bit was clear -- it becomes the victim once the hand reaches it
assert frames[0].referenced is False  # => page 1 got a SECOND CHANCE: its bit was cleared, not evicted
# => a cleared bit today means page 1 WOULD be the next victim if the hand sweeps again
print("ex-16 OK")  # => Output: ex-16 OK

2
[(1, False), (3, True)]
ex-16 OK

"""Example 16: pytest verification for CLOCK (Second-Chance) Eviction."""
 
from example import Frame, clock_evict
 
 
def test_unreferenced_frame_is_evicted_first() -> None:
    frames = [Frame(page_id=1, referenced=True), Frame(page_id=2), Frame(page_id=3, referenced=True)]
    victim, _ = clock_evict(frames, hand=0)
    assert victim == 2
 
 
def test_referenced_frame_gets_its_bit_cleared_not_evicted() -> None:
    frames = [Frame(page_id=1, referenced=True), Frame(page_id=2), Frame(page_id=3, referenced=True)]
    clock_evict(frames, hand=0)
    survivor = next(f for f in frames if f.page_id == 1)
    assert survivor.referenced is False  # => it survived, but its second chance is now spent
 
 
# => Run: pytest -- Output: 2 passed

2 passed

"""Example 17: B-Tree Leaf -- Sorted Insert."""
 
import bisect
 
# A B-tree (and B+-tree) leaf holds its keys in SORTED order at all times
# (co-07) -- bisect.insort finds the correct insertion point and shifts the
# rest of the list over, so the invariant holds after every single insert,
# not just once a whole batch finishes.
 
 
def leaf_insert(leaf: list[int], key: int) -> None:  # => keeps `leaf` sorted after this call returns
    bisect.insort(leaf, key)  # => O(n) insert-and-shift; O(log n) just to FIND the position
 
 
leaf: list[int] = []  # => a single leaf node modeled as a plain sorted list of keys
for key in [40, 10, 30, 20]:  # => insert in a deliberately UNsorted order
    leaf_insert(leaf, key)
    print(leaf)  # => Output: one growing, always-sorted list per insert
 
assert leaf == [10, 20, 30, 40]  # => the leaf stays sorted after every single insertion, not just at the end
assert leaf == sorted(leaf)  # => equivalent way to say the same thing: leaf never needs a final re-sort
print("ex-17 OK")  # => Output: ex-17 OK

[40]
[10, 40]
[10, 30, 40]
[10, 20, 30, 40]
ex-17 OK

"""Example 17: pytest verification for B-Tree Leaf Sorted Insert."""
 
from example import leaf_insert
 
 
def test_insert_keeps_a_list_sorted() -> None:
    leaf: list[int] = []
    for key in [5, 1, 4, 2, 3]:
        leaf_insert(leaf, key)
    assert leaf == [1, 2, 3, 4, 5]
 
 
def test_leaf_is_sorted_after_every_single_insert() -> None:
    leaf: list[int] = []
    for key in [9, 3, 7]:
        leaf_insert(leaf, key)
        assert leaf == sorted(leaf)  # => the invariant holds at EVERY step, not just at the end
 
 
# => Run: pytest -- Output: 2 passed

2 passed

"""Example 18: B-Tree Point Lookup."""
 
import bisect
 
# Searching a single sorted leaf for one key is the base case every deeper
# B-tree lookup eventually bottoms out in (co-08) -- bisect_left finds the
# key's position in O(log n) comparisons, and an exact-match check tells
# present from absent.
leaf: list[int] = [10, 20, 30, 40, 50]  # => a pre-sorted leaf node
 
 
def lookup(leaf: list[int], key: int) -> int | None:  # => returns the key if present, else None
    i = bisect.bisect_left(leaf, key)  # => O(log n): the position key WOULD occupy if present
    if i < len(leaf) and leaf[i] == key:  # => confirm the position actually holds this exact key
        return leaf[i]
    return None  # => key would fall here, but it does not actually exist in the leaf
 
 
print(lookup(leaf, 30))  # => Output: 30
print(lookup(leaf, 25))  # => Output: None
 
assert lookup(leaf, 30) == 30  # => a present key is found and returned
assert lookup(leaf, 25) is None  # => an absent key returns None, not an incorrect neighbor
print("ex-18 OK")  # => Output: ex-18 OK

30
None
ex-18 OK

"""Example 18: pytest verification for B-Tree Point Lookup."""
 
from example import lookup
 
 
def test_present_key_is_found() -> None:
    assert lookup([10, 20, 30, 40, 50], 40) == 40
 
 
def test_absent_key_returns_none() -> None:
    assert lookup([10, 20, 30, 40, 50], 15) is None
 
 
# => Run: pytest -- Output: 2 passed

2 passed

"""Example 19: B-Tree Height from Fanout."""
 
import math  # => stdlib module for the log/ceil formula this example checks itself against
 
# Tree height grows LOGARITHMICALLY with the key count, in the base of the
# fanout (co-08) -- a high fanout keeps the tree shallow, so a lookup only
# ever touches a handful of pages even over millions of keys. This example
# builds actual bottom-up levels and checks the resulting level count against
# the ceil(log_fanout(N)) formula (illustrative numbers, not vendor constants).
 
FANOUT: int = 10  # => each internal node groups up to 10 children below it
 
 
def build_levels(keys: list[int], fanout: int) -> list[list[int]]:  # => bottom-up: leaf level first
    level: list[int] = keys  # => start at the leaf level -- the raw key list itself
    levels: list[list[int]] = [level]  # => levels[0] is always the leaf level
    while len(level) > 1:  # => keep grouping representatives until one root-level group remains
        level = [level[i] for i in range(0, len(level), fanout)]  # => one representative key per group
        levels.append(level)  # => this new, smaller group becomes the next level up
    return levels  # => bottom-to-top: levels[0] is leaves, levels[-1] is the root
 
 
keys: list[int] = list(range(100))  # => N = 100 keys
levels = build_levels(keys, FANOUT)  # => actually build the levels, not just compute a formula
level_count = len(levels)  # => leaf level + every internal level up to (and including) the root
internal_levels = math.ceil(math.log(len(keys), FANOUT))  # => ceil(log_fanout(N)) -- levels ABOVE the leaf
print(level_count)  # => Output: 3
print(internal_levels)  # => Output: 2
 
assert level_count == internal_levels + 1  # => +1 accounts for the leaf level the pure formula excludes
print("ex-19 OK")  # => Output: ex-19 OK

3
2
ex-19 OK

"""Example 19: pytest verification for B-Tree Height from Fanout."""
 
import math
 
from example import FANOUT, build_levels
 
 
def test_level_count_matches_the_fanout_formula() -> None:
    keys = list(range(1000))
    levels = build_levels(keys, FANOUT)
    expected = math.ceil(math.log(len(keys), FANOUT))
    assert len(levels) == expected + 1  # => +1 for the leaf level the formula excludes
 
 
def test_higher_fanout_yields_a_shallower_tree() -> None:
    keys = list(range(1000))
    shallow = build_levels(keys, fanout=100)
    deep = build_levels(keys, fanout=2)
    assert len(shallow) < len(deep)  # => a higher fanout keeps the tree shallower for the SAME key count
 
 
# => Run: pytest -- Output: 2 passed

2 passed

"""Example 20: B-Tree Leaf Split and Separator Promotion."""
 
import bisect  # => stdlib module for the sorted insert this example performs before splitting
 
# When a leaf's key count exceeds its capacity, it SPLITS into two leaves and
# promotes the median key upward as a separator (co-09) -- the tree stays
# balanced because every leaf keeps roughly half its former keys, never a
# lopsided remainder.
MAX_KEYS: int = 4  # => a small capacity so this example can trigger a split with few inserts
 
 
def insert_and_maybe_split(
    leaf: list[int], key: int
) -> tuple[list[int], list[int] | None, int | None]:  # => (left, right-or-None, separator-or-None)
    bisect.insort(leaf, key)  # => insert first, keeping the leaf sorted
    if len(leaf) <= MAX_KEYS:  # => still under capacity: no split needed
        return leaf, None, None  # => the caller's own leaf, unchanged in identity, no split occurred
    mid = len(leaf) // 2  # => the median position -- splits as evenly as possible
    left, separator, right = leaf[:mid], leaf[mid], leaf[mid + 1 :]  # => slice into three pieces
    return left, right, separator  # => two leaves plus the promoted key between them
 
 
leaf: list[int] = [10, 20, 30, 40]  # => already at MAX_KEYS capacity
left, right, separator = insert_and_maybe_split(leaf, 25)  # => this insert forces a split
print((left, separator, right))  # => Output: ([10, 20], 25, [30, 40])
 
assert left == [10, 20]  # => the left leaf keeps the smaller half
assert right == [30, 40]  # => the right leaf keeps the larger half
assert separator == 25  # => the median key is promoted, not simply dropped or duplicated
print("ex-20 OK")  # => Output: ex-20 OK

([10, 20], 25, [30, 40])
ex-20 OK

"""Example 20: pytest verification for B-Tree Leaf Split."""
 
from example import insert_and_maybe_split
 
 
def test_split_produces_two_leaves_and_a_separator() -> None:
    leaf = [10, 20, 30, 40]
    left, right, separator = insert_and_maybe_split(leaf, 25)
    assert left is not None and right is not None and separator is not None
 
 
def test_no_split_when_under_capacity() -> None:
    leaf = [10, 20]
    left, right, separator = insert_and_maybe_split(leaf, 15)
    assert right is None and separator is None  # => still under MAX_KEYS -- nothing was split
    assert left == [10, 15, 20]
 
 
# => Run: pytest -- Output: 2 passed

2 passed

"""Example 21: B-Tree Range Scan via Sibling Links."""
 
from dataclasses import dataclass  # => a plain, typed record for one leaf node
 
 
@dataclass
class Leaf:  # => a leaf node with a NEXT pointer chaining it to its right sibling
    keys: list[int]  # => this leaf's own sorted keys
    next: "Leaf | None" = None  # => the sibling immediately to the right, or None if this is the last leaf
 
 
# Sibling-linked leaves let a range scan walk forward leaf-to-leaf without
# ever re-descending from the root (co-10) -- only the STARTING leaf needs a
# root-to-leaf search; every following leaf is reached via `next` alone.
 
 
def range_scan(start: Leaf, lo: int, hi: int) -> list[int]:  # => inclusive [lo, hi] range scan
    result: list[int] = []  # => accumulates every matching key, in sorted order
    node: Leaf | None = start  # => begin at the leaf the caller already descended to
    while node is not None:  # => walk sideways via `next`, never back up to the root
        for key in node.keys:  # => scan this leaf's own keys
            if lo <= key <= hi:  # => inclusive bounds check
                result.append(key)
        if node.keys and node.keys[-1] > hi:  # => this leaf's max key already exceeds hi -- stop early
            break
        node = node.next  # => hop sideways to the next leaf -- no root descent needed
    return result  # => a contiguous, sorted run possibly spanning several leaves
 
 
leaf_a = Leaf(keys=[10, 20, 30])  # => the first leaf in the chain
leaf_b = Leaf(keys=[40, 50, 60])  # => the second leaf in the chain
leaf_c = Leaf(keys=[70, 80, 90])  # => the third leaf in the chain
leaf_a.next, leaf_b.next = leaf_b, leaf_c  # => chain: a -> b -> c
 
result = range_scan(leaf_a, 25, 75)  # => starts inside leaf_a, ends inside leaf_c
print(result)  # => Output: [30, 40, 50, 60, 70]
 
assert result == [30, 40, 50, 60, 70]  # => a contiguous sorted run spanning THREE separate leaf nodes
# => the scan never had to climb back up to leaf_a's parent to reach leaf_b or leaf_c
print("ex-21 OK")  # => Output: ex-21 OK

[30, 40, 50, 60, 70]
ex-21 OK

"""Example 21: pytest verification for B-Tree Range Scan via Sibling Links."""
 
from example import Leaf, range_scan
 
 
def test_range_scan_spans_multiple_leaves() -> None:
    leaf_a = Leaf(keys=[1, 2, 3])
    leaf_b = Leaf(keys=[4, 5, 6])
    leaf_a.next = leaf_b
    assert range_scan(leaf_a, 2, 5) == [2, 3, 4, 5]
 
 
def test_range_scan_stops_once_past_hi() -> None:
    leaf_a = Leaf(keys=[1, 2, 3])
    leaf_b = Leaf(keys=[100, 200])
    leaf_a.next = leaf_b
    assert range_scan(leaf_a, 1, 3) == [1, 2, 3]  # => never even reads leaf_b's keys
 
 
# => Run: pytest -- Output: 2 passed

2 passed

"""Example 22: B+-Tree -- Values Live Only in Leaves."""
 
from dataclasses import dataclass  # => plain, typed records for both node kinds
 
 
@dataclass
class LeafNode:  # => the ONLY place a key-value pair actually lives
    entries: dict[int, str]  # => key -> value, the real data this tree stores
 
 
@dataclass
class InternalNode:  # => carries ROUTING keys only -- no values, ever
    keys: list[int]  # => separator keys used to pick which child to descend into
    children: list["InternalNode | LeafNode"]  # => one more child than there are keys
 
 
# A B+-tree keeps values EXCLUSIVELY in leaves; internal nodes hold only
# routing keys used to steer a search downward (co-07) -- unlike a classical
# B-tree, which may store a value at any node, including internal ones.
leaf1 = LeafNode(entries={10: "row-10", 20: "row-20"})  # => the LEFT leaf's real data
leaf2 = LeafNode(entries={30: "row-30", 40: "row-40"})  # => the RIGHT leaf's real data
root = InternalNode(keys=[30], children=[leaf1, leaf2])  # => keys route; children hold the real data
 
 
def has_any_value(node: "InternalNode | LeafNode") -> bool:  # => walks the whole tree checking the rule
    if isinstance(node, LeafNode):  # => base case: a leaf either has entries or it doesn't
        return len(node.entries) > 0
    return any(has_any_value(child) for child in node.children)  # => recurse: true if ANY child has values
 
 
assert not hasattr(root, "entries")  # => the internal node has NO entries field -- values physically can't live there
assert has_any_value(root)  # => but the tree AS A WHOLE does hold values -- all of them inside its leaves
print(root.keys)  # => Output: [30]
print(leaf1.entries)  # => Output: {10: 'row-10', 20: 'row-20'}
# => root.keys routes a search; leaf1.entries and leaf2.entries hold the actual rows
print("ex-22 OK")  # => Output: ex-22 OK

[30]
{10: 'row-10', 20: 'row-20'}
ex-22 OK

"""Example 22: pytest verification for B+-Tree Values-in-Leaves."""
 
from example import InternalNode, LeafNode, has_any_value
 
 
def test_internal_node_has_no_entries_attribute() -> None:
    node = InternalNode(keys=[5], children=[])
    assert not hasattr(node, "entries")
 
 
def test_values_are_reachable_through_leaves() -> None:
    leaf = LeafNode(entries={1: "one"})
    root = InternalNode(keys=[1], children=[leaf])
    assert has_any_value(root) is True
 
 
# => Run: pytest -- Output: 2 passed

2 passed

"""Example 23: Memtable -- Sorted Insert."""
 
import bisect  # => stdlib module used to find each new key's correct sorted position
 
# An LSM engine buffers every write in an in-memory SORTED structure -- the
# memtable -- before anything touches disk (co-11). RocksDB's default
# memtable implementation is a skiplist; this example models the same
# "always sorted" contract with a plain list kept sorted via bisect.
memtable: list[tuple[str, str]] = []  # => (key, value) pairs, kept sorted BY KEY
 
 
def memtable_put(key: str, value: str) -> None:  # => insert, or overwrite an existing key in place
    keys = [k for k, _ in memtable]  # => extract just the keys to find the insertion point
    i = bisect.bisect_left(keys, key)  # => O(log n) to FIND the position; the list insert itself is O(n)
    if i < len(memtable) and memtable[i][0] == key:  # => key already present: overwrite in place
        memtable[i] = (key, value)
    else:
        memtable.insert(i, (key, value))  # => new key: insert at the sorted position
 
 
for k, v in [("charlie", "3"), ("alice", "1"), ("bob", "2")]:  # => inserted out of order
    memtable_put(k, v)
print(memtable)  # => Output: [('alice', '1'), ('bob', '2'), ('charlie', '3')]
 
assert [k for k, _ in memtable] == ["alice", "bob", "charlie"]  # => iteration is ALWAYS in sorted-key order
print("ex-23 OK")  # => Output: ex-23 OK

[('alice', '1'), ('bob', '2'), ('charlie', '3')]
ex-23 OK

"""Example 23: pytest verification for Memtable Sorted Insert."""
 
from example import memtable, memtable_put
 
 
def test_memtable_iterates_in_sorted_key_order() -> None:
    memtable.clear()
    for k, v in [("z", "26"), ("a", "1"), ("m", "13")]:
        memtable_put(k, v)
    assert [k for k, _ in memtable] == ["a", "m", "z"]
 
 
def test_put_on_existing_key_overwrites_in_place() -> None:
    memtable.clear()
    memtable_put("a", "1")
    memtable_put("a", "one")
    assert memtable == [("a", "one")]  # => no duplicate entry -- the value was updated, not appended
 
 
# => Run: pytest -- Output: 2 passed

2 passed

"""Example 24: Flush a Memtable to an SSTable."""
 
# Once a memtable grows past a size threshold, it is flushed to disk as an
# IMMUTABLE, sorted SSTable, and a fresh empty memtable takes over writes
# (co-11) -- "immutable" is the key word: an SSTable is never edited again,
# only replaced by a later compaction.
 
 
def flush(memtable: list[tuple[str, str]]) -> list[tuple[str, str]]:  # => returns a NEW sorted SSTable
    sstable = sorted(memtable, key=lambda pair: pair[0])  # => sort once, at flush time, not on every write
    memtable.clear()  # => the memtable is now empty and ready for new writes
    return sstable
 
 
memtable: list[tuple[str, str]] = [("bob", "2"), ("alice", "1"), ("charlie", "3")]
sstable = flush(memtable)
print(sstable)  # => Output: [('alice', '1'), ('bob', '2'), ('charlie', '3')]
print(memtable)  # => Output: []
 
assert sstable == sorted(sstable)  # => the flushed segment is sorted by key
assert memtable == []  # => the memtable was cleared -- future writes go into a fresh one
print("ex-24 OK")  # => Output: ex-24 OK

[('alice', '1'), ('bob', '2'), ('charlie', '3')]
[]
ex-24 OK

"""Example 24: pytest verification for Flushing a Memtable to an SSTable."""
 
from example import flush
 
 
def test_flush_returns_a_sorted_sstable() -> None:
    mt = [("c", "3"), ("a", "1"), ("b", "2")]
    sstable = flush(mt)
    assert sstable == [("a", "1"), ("b", "2"), ("c", "3")]
 
 
def test_flush_clears_the_source_memtable() -> None:
    mt = [("a", "1")]
    flush(mt)
    assert mt == []
 
 
# => Run: pytest -- Output: 2 passed