SQL はどうやってディスク上のバイト列に届くのか — クエリーエンジンをスクラッチで実装する

from dataclasses import dataclass
 
KEYWORDS = {
    "SELECT", "FROM", "WHERE", "AND", "OR", "NOT",
    "JOIN", "ON", "AS", "ORDER", "BY", "ASC", "DESC", "LIMIT",
    "GROUP", "HAVING",
}
TWO_CHAR_OPS = (">=", "<=", "<>", "!=")
 
 
@dataclass(frozen=True)
class Token:
    kind: str   # KEYWORD / IDENT / NUMBER / STRING / OP / PUNCT / EOF
    text: str
    pos: int
 
 
def tokenize(sql: str) -> list[Token]:
    out: list[Token] = []
    i, n = 0, len(sql)
    while i < n:
        if sql[i].isspace():
            i += 1
            continue
 
        start, c = i, sql[i]
 
        if c.isalpha() or c == "_":
            while i < n and (sql[i].isalnum() or sql[i] == "_"):
                i += 1
            word = sql[start:i]
            kind = "KEYWORD" if word.upper() in KEYWORDS else "IDENT"
            out.append(Token(kind, word, start))
 
        elif c.isdigit():
            while i < n and sql[i].isdigit():
                i += 1
            # 小数点は「後ろに数字が続くとき」だけ取り込む。
            # そうしないと "t.1" のようなドット参照を壊す。
            if i + 1 < n and sql[i] == "." and sql[i + 1].isdigit():
                i += 1
                while i < n and sql[i].isdigit():
                    i += 1
            out.append(Token("NUMBER", sql[start:i], start))
 
        elif c == "'":
            i += 1
            while i < n and sql[i] != "'":
                i += 1
            if i >= n:
                raise SyntaxError("unterminated string literal")
            i += 1                       # 閉じ引用符
            out.append(Token("STRING", sql[start:i], start))
 
        elif sql[i:i + 2] in TWO_CHAR_OPS:   # ← 1文字演算子より先に試すのが肝
            out.append(Token("OP", sql[i:i + 2], start))
            i += 2
 
        elif c in "=<>+-*/%":
            out.append(Token("OP", c, start))
            i += 1
 
        elif c in ",().;":
            out.append(Token("PUNCT", c, start))
            i += 1
 
        else:
            raise SyntaxError(f"unexpected character {c!r} at position {i}")
 
    out.append(Token("EOF", "", n))
    return out

@dataclass
class ColRef:                       # 未解決の列参照
    table: str | None
    column: str
 
@dataclass
class Const:
    value: object
 
@dataclass
class BinOp:
    op: str
    left: object
    right: object
 
@dataclass
class FuncCall:
    name: str
    args: list
    star: bool = False       # COUNT(*) の * を表す
 
@dataclass
class TableRef:
    name: str
    alias: str | None
 
@dataclass
class JoinExpr:
    left: object
    right: object
    on: object
 
@dataclass
class SortKey:
    expr: object
    desc: bool
 
@dataclass
class Select:
    targets: list
    frm: object
    where: object | None
    group_by: list
    having: object | None
    order_by: list
    limit: int | None

PRECEDENCE = {
    "OR": 1,
    "AND": 2,
    "=": 3, "<>": 3, "!=": 3, "<": 3, ">": 3, "<=": 3, ">=": 3,
    "+": 4, "-": 4,
    "*": 5, "/": 5,
}
 
 
class Parser:
    def __init__(self, tokens: list[Token]):
        self.toks, self.i = tokens, 0
 
    def peek(self) -> Token:
        return self.toks[self.i]
 
    def advance(self) -> Token:
        tok = self.toks[self.i]
        self.i += 1
        return tok
 
    def accept(self, text: str) -> Token | None:
        if self.peek().text.upper() == text.upper():
            return self.advance()
        return None
 
    def expect(self, text: str) -> Token:
        tok = self.accept(text)
        if tok is None:
            raise SyntaxError(f"expected {text!r}, found {self.peek().text!r}")
        return tok
 
    # ── 文の骨格：再帰下降 ────────────────────────────
    def parse_select(self) -> Select:
        self.expect("SELECT")
        targets = [self.parse_expr()]
        while self.accept(","):
            targets.append(self.parse_expr())
 
        self.expect("FROM")
        frm = self.parse_from()
 
        where = self.parse_expr() if self.accept("WHERE") else None
 
        group_by = []
        if self.accept("GROUP"):
            self.expect("BY")
            group_by.append(self.parse_expr())
            while self.accept(","):
                group_by.append(self.parse_expr())
 
        having = self.parse_expr() if self.accept("HAVING") else None
 
        order_by = []
        if self.accept("ORDER"):
            self.expect("BY")
            while True:
                expr = self.parse_expr()
                desc = self.accept("DESC") is not None
                if not desc:
                    self.accept("ASC")
                order_by.append(SortKey(expr, desc))
                if not self.accept(","):
                    break
 
        limit = int(self.advance().text) if self.accept("LIMIT") else None
        return Select(targets, frm, where, group_by, having, order_by, limit)
 
    def parse_from(self):
        node = self.parse_table_ref()
        while self.accept("JOIN"):
            right = self.parse_table_ref()
            self.expect("ON")
            node = JoinExpr(node, right, self.parse_expr())
        return node
 
    def parse_table_ref(self) -> TableRef:
        name = self.advance().text
        if self.accept("AS"):
            return TableRef(name, self.advance().text)
        if self.peek().kind == "IDENT":          # AS 省略形
            return TableRef(name, self.advance().text)
        return TableRef(name, None)
 
    # ── 式：優先順位登り ─────────────────────────────
    def parse_expr(self, min_prec: int = 0):
        left = self.parse_primary()
        while True:
            op = self.peek().text.upper()
            prec = PRECEDENCE.get(op, -1)
            if prec < 0 or prec < min_prec:
                return left
            self.advance()
            # 左結合なので、右辺には「自分より強い演算子」しか許さない
            right = self.parse_expr(prec + 1)
            left = BinOp(op, left, right)
 
    def parse_primary(self):
        tok = self.advance()
        if tok.kind == "NUMBER":
            return Const(float(tok.text) if "." in tok.text else int(tok.text))
        if tok.kind == "STRING":
            return Const(tok.text[1:-1])
        if tok.kind == "IDENT":
            if self.peek().text == "(":          # 集約関数の呼び出し
                self.advance()
                if self.accept("*"):
                    self.expect(")")
                    return FuncCall(tok.text.upper(), [], star=True)
                args = [self.parse_expr()]
                while self.accept(","):
                    args.append(self.parse_expr())
                self.expect(")")
                return FuncCall(tok.text.upper(), args)
            if self.peek().text == ".":
                self.advance()
                return ColRef(tok.text, self.advance().text)
            return ColRef(None, tok.text)
        if tok.text == "(":
            inner = self.parse_expr()
            self.expect(")")
            return inner
        raise SyntaxError(f"unexpected token {tok.text!r}")

@dataclass
class Column:
    name: str
    type: str            # "int" / "text" / "numeric"
 
@dataclass
class Relation:
    name: str
    columns: list[Column]
    n_rows: int          # pg_class.reltuples 相当
    n_pages: int         # pg_class.relpages 相当
    oid: int = 0         # pg_class.oid
    relfilenode: int = 0 # 実ファイル名。VACUUM FULL 等で変わりうる
 
CATALOG = {
    "users": Relation(
        "users",
        [Column("id", "int"), Column("name", "text"), Column("age", "int")],
        n_rows=100_000, n_pages=1_000, oid=16384, relfilenode=16384,
    ),
    "orders": Relation(
        "orders",
        [Column("id", "int"), Column("user_id", "int"), Column("total", "numeric")],
        n_rows=1_000_000, n_pages=10_000, oid=16390, relfilenode=16390,
    ),
}

@dataclass(frozen=True)
class Var:
    rel: int        # レンジテーブル内の添字
    attno: int      # 列番号（0起点。PostgreSQL 本体は1起点）
    type: str
    label: str      # 表示用。実行には使わない
 
 
COMPARISONS = {"=", "<>", "!=", "<", ">", "<=", ">="}
NUMERIC_TYPES = {"int", "numeric"}
 
 
class Binder:
    def __init__(self, catalog):
        self.catalog = catalog
        self.range_table: list[tuple[str, Relation]] = []
 
    def collect_from(self, node) -> None:
        if isinstance(node, TableRef):
            if node.name not in self.catalog:
                raise NameError(f'relation "{node.name}" does not exist')
            rel = self.catalog[node.name]
            self.range_table.append((node.alias or node.name, rel))
        elif isinstance(node, JoinExpr):
            self.collect_from(node.left)
            self.collect_from(node.right)
        else:
            raise TypeError(node)
 
    def bind_expr(self, expr) -> tuple[object, str]:
        if isinstance(expr, Const):
            if isinstance(expr.value, bool):
                return expr, "bool"
            if isinstance(expr.value, int):
                return expr, "int"
            if isinstance(expr.value, float):
                return expr, "numeric"
            return expr, "text"
 
        if isinstance(expr, ColRef):
            hits = []
            for rel_idx, (alias, rel) in enumerate(self.range_table):
                if expr.table is not None and expr.table != alias:
                    continue
                for attno, col in enumerate(rel.columns):
                    if col.name == expr.column:
                        hits.append(Var(rel_idx, attno, col.type,
                                        f"{alias}.{col.name}"))
            if not hits:
                raise NameError(f'column "{expr.column}" does not exist')
            if len(hits) > 1:
                raise NameError(f'column reference "{expr.column}" is ambiguous')
            return hits[0], hits[0].type
 
        if isinstance(expr, BinOp):
            left, lt = self.bind_expr(expr.left)
            right, rt = self.bind_expr(expr.right)
 
            if expr.op in ("AND", "OR"):
                if (lt, rt) != ("bool", "bool"):
                    raise TypeError(f"argument of {expr.op} must be boolean")
                return BinOp(expr.op, left, right), "bool"
 
            if lt != rt and not {lt, rt} <= NUMERIC_TYPES:
                raise TypeError(
                    f"operator does not exist: {lt} {expr.op} {rt}")
 
            result = "bool" if expr.op in COMPARISONS else lt
            return BinOp(expr.op, left, right), result
 
        if isinstance(expr, FuncCall):
            args = [self.bind_expr(a) for a in expr.args]
            bound = FuncCall(expr.name, [a for a, _ in args], expr.star)
            if expr.name == "COUNT":
                return bound, "int"                # COUNT は常に整数
            if not args:
                raise TypeError(f"{expr.name} requires an argument")
            arg_type = args[0][1]
            if expr.name == "AVG":
                return bound, "numeric"            # 整数の平均でも小数になる
            if expr.name in ("SUM", "MIN", "MAX"):
                # MIN/MAX は実際にこの型。SUM だけ簡略化してある——PostgreSQL は
                # 桁溢れを避けるため sum(int4) を int8、sum(int8) を numeric に広げる
                return bound, arg_type
            raise NameError(f'function "{expr.name}" does not exist')
 
        raise TypeError(f"cannot bind {expr!r}")

@dataclass
class Scan:
    rel: int
    relation: Relation
    alias: str
    predicate: object | None = None    # 走査しながら評価する条件
 
@dataclass
class Filter:
    child: object
    predicate: object
 
@dataclass
class Join:
    left: object
    right: object
    predicate: object
 
@dataclass
class Project:
    child: object
    exprs: list
 
@dataclass
class Sort:
    child: object
    keys: list
 
@dataclass
class Aggregate:
    child: object
    group_keys: list
    aggs: list           # [FuncCall("COUNT", [], star=True), …]
 
@dataclass
class Limit:
    child: object
    n: int

Limit(10)
└─ Sort(o.total DESC)
   └─ Project(u.name, o.total)
      └─ Filter(u.age > 30 AND o.total > 100)
         └─ Join(u.id = o.user_id)
            ├─ Scan(users AS u)
            └─ Scan(orders AS o)

def conjuncts(expr) -> list:
    """AND の連鎖を平坦なリストに分解する。"""
    if isinstance(expr, BinOp) and expr.op == "AND":
        return conjuncts(expr.left) + conjuncts(expr.right)
    return [expr]
 
 
def and_all(preds: list):
    if not preds:
        return None
    node = preds[0]
    for p in preds[1:]:
        node = BinOp("AND", node, p)
    return node
 
 
def used_rels(expr) -> set[int]:
    """式が参照しているレンジテーブル添字の集合。"""
    if isinstance(expr, Var):
        return {expr.rel}
    if isinstance(expr, BinOp):
        return used_rels(expr.left) | used_rels(expr.right)
    return set()
 
 
def rels_of(node) -> set[int]:
    if isinstance(node, Scan):
        return {node.rel}
    if isinstance(node, Join):
        return rels_of(node.left) | rels_of(node.right)
    return rels_of(node.child)
 
 
def push_down(node, preds: list):
    """preds をできるだけ深い位置へ降ろす。
       戻り値は (書き換えた部分木, まだ降ろせなかった述語)。"""
    if isinstance(node, Scan):
        mine, stays = [], []
        for p in preds:
            (mine if used_rels(p) <= {node.rel} else stays).append(p)
        node.predicate = and_all(mine)
        return node, stays
 
    if isinstance(node, Join):
        node.left, preds = push_down(node.left, preds)
        node.right, preds = push_down(node.right, preds)
        here, stays = [], []
        scope = rels_of(node)
        for p in preds:
            (here if used_rels(p) <= scope else stays).append(p)
        return (Filter(node, and_all(here)) if here else node), stays
 
    raise TypeError(node)

Limit(10)
└─ Sort(o.total DESC)
   └─ Project(u.name, o.total)
      └─ Join(u.id = o.user_id)
         ├─ Scan(users)   filter: age > 30
         └─ Scan(orders)  filter: total > 100

def transitive_closure(preds: list) -> list:
    """x = y の同値クラスを作り、定数条件をクラス全体へ複写する。"""
    classes: list[set] = []                     # 同値と判明した Var の集合
    for p in preds:
        if (isinstance(p, BinOp) and p.op == "="
                and isinstance(p.left, Var) and isinstance(p.right, Var)):
            merged = {p.left, p.right}
            rest = []
            for c in classes:                   # 重なるクラスを統合する
                (merged.update(c) if merged & c else rest.append(c))
            classes = rest + [merged]
 
    derived = []
    for p in preds:
        if isinstance(p, BinOp) and isinstance(p.left, Var) and isinstance(p.right, Const):
            for c in classes:
                if p.left in c:
                    derived += [BinOp(p.op, v, p.right) for v in c if v != p.left]
    return preds + derived

-- 相関サブクエリ：外側の1行ごとに内側を評価する（ナイーブには N 回）
SELECT u.name FROM users u
WHERE EXISTS (SELECT 1 FROM orders o WHERE o.user_id = u.id AND o.total > 100);
 
-- 半結合（semi join）に書き換え：ハッシュ表を1回作れば済む
SELECT u.name FROM users u
SEMI JOIN orders o ON o.user_id = u.id AND o.total > 100;

AND    | T | F | U          OR     | T | F | U
-------+---+---+---        -------+---+---+---
   T   | T | F | U             T   | T | T | T
   F   | F | F | F             F   | T | F | U
   U   | U | F | U             U   | T | U | U

-- 元のクエリ：注文の無いユーザも残る
SELECT u.name, o.total
FROM users u LEFT JOIN orders o ON u.id = o.user_id
WHERE o.total > 100;          -- ← NULL 側を落とすので内部結合と同義になる
 
-- こちらは条件を WHERE（結合の外）へ移せない
SELECT u.name, o.total
FROM users u LEFT JOIN orders o ON u.id = o.user_id AND o.total > 100;

@dataclass
class ColumnStats:
    n_distinct: int
    bounds: list         # 等深度ヒストグラムの境界値。len(bounds) - 1 個のバケット
 
 
def selectivity_gt(stats: ColumnStats, value) -> float:
    """col > value の選択率。バケット内は一様分布を仮定して線形補間する。"""
    bounds = stats.bounds
    nb = len(bounds) - 1                     # バケット数
    if value < bounds[0]:
        return 1.0
    if value >= bounds[-1]:
        return 0.0
    for i in range(nb):
        lo, hi = bounds[i], bounds[i + 1]
        if lo <= value < hi:
            above_in_bucket = (hi - value) / (hi - lo)
            return (nb - i - 1) / nb + above_in_bucket / nb
    return 0.0

age_stats = ColumnStats(
    n_distinct=63,
    bounds=[18, 22, 25, 27, 29, 31, 34, 38, 44, 53, 80],
)
selectivity_gt(age_stats, 30)     # → 0.55

def join_cardinality(left_rows, right_rows, ndv_left, ndv_right) -> float:
    """含有仮定（containment assumption）に基づく等結合の見積り。
       値の少ない側のすべての値が、多い側にも現れると仮定する。"""
    return left_rows * right_rows / max(ndv_left, ndv_right)

import math
 
SEQ_PAGE_COST        = 1.0      # 順次読みでページを1枚読む
RANDOM_PAGE_COST     = 4.0      # ランダム読みでページを1枚読む
CPU_TUPLE_COST       = 0.01     # タプルを1件処理する
CPU_INDEX_TUPLE_COST = 0.005    # 索引エントリを1件処理する
CPU_OPERATOR_COST    = 0.0025   # 演算子や関数を1回評価する

def seq_scan_cost(rel: Relation, n_quals: int) -> float:
    """全ブロックを順に読み、全タプルに対して n_quals 個の条件を評価する。"""
    return (rel.n_pages * SEQ_PAGE_COST
            + rel.n_rows * CPU_TUPLE_COST
            + rel.n_rows * n_quals * CPU_OPERATOR_COST)

seq_scan_cost(CATALOG["users"],  1)   # 1,000 + 1,000 +   250 =  2,250
seq_scan_cost(CATALOG["orders"], 1)   # 10,000 + 10,000 + 2,500 = 22,500

def index_pages_fetched(tuples_fetched, pages, index_pages,
                        total_table_pages, effective_cache_size=524_288):
    """Mackert-Lohman。有限バッファのもとで実際に読むページ数。
       effective_cache_size の既定は 4GB = 524,288 ページ。"""
    T = float(pages) if pages > 1 else 1.0
    # クエリ内の全テーブルとこの索引でキャッシュを奪い合うと仮定し、T の割合で按分する
    total_pages = max(total_table_pages + index_pages, 1.0)
    b = max(1.0, math.ceil(effective_cache_size * T / total_pages))
 
    if T <= b:                                  # 表全体がキャッシュに収まる
        pf = (2.0 * T * tuples_fetched) / (2.0 * T + tuples_fetched)
        return T if pf >= T else math.ceil(pf)
 
    lim = (2.0 * T * b) / (2.0 * T - b)         # 再読みが発生し始める点
    if tuples_fetched <= lim:
        return math.ceil((2.0 * T * tuples_fetched) / (2.0 * T + tuples_fetched))
    return math.ceil(b + (tuples_fetched - lim) * (T - b) / T)

def index_scan_cost(rel, index_pages, selectivity, correlation=0.0,
                    random_page_cost=RANDOM_PAGE_COST, total_table_pages=11_000):
    tuples = rel.n_rows * selectivity
 
    # 索引側：該当するエントリを読む
    idx_pages = math.ceil(selectivity * index_pages)
    idx_cost = (idx_pages * random_page_cost
                + tuples * (CPU_INDEX_TUPLE_COST + CPU_OPERATOR_COST))
 
    # ヒープ側：無相関と完全相関の2つの端点を相関の二乗で補間する
    pf = index_pages_fetched(tuples, rel.n_pages, index_pages, total_table_pages)
    max_io = pf * random_page_cost                       # correlation = 0
    seq_pages = math.ceil(selectivity * rel.n_pages)
    min_io = random_page_cost + max(seq_pages - 1, 0) * SEQ_PAGE_COST  # = 1
    heap_cost = max_io + correlation ** 2 * (min_io - max_io)
 
    return idx_cost + heap_cost + tuples * CPU_TUPLE_COST

@dataclass
class Plan:
    rows: float
    cost: float
    desc: str
 
 
def hash_join_cost(build: Plan, probe: Plan, out_rows: float) -> float:
    return (build.cost + probe.cost
            + build.rows * CPU_OPERATOR_COST     # ハッシュ表の構築
            + probe.rows * CPU_OPERATOR_COST     # プローブ
            + out_rows   * CPU_TUPLE_COST)       # 出力タプルの生成
 
 
def sort_cost(input_cost: float, rows: float, limit: int | None = None) -> float:
    """PostgreSQL に倣い、比較1回のコストを 2 * cpu_operator_cost とする。
       LIMIT が付くと上位 n 件だけを保つヒープになり、log の底が小さくなる。"""
    comparison_cost = 2.0 * CPU_OPERATOR_COST
    keep = rows if limit is None else min(rows, limit)
    return input_cost + comparison_cost * rows * math.log2(max(keep, 2.0))
 
 
def merge_join_cost(left: Plan, right: Plan, out_rows: float) -> float:
    ls = sort_cost(left.cost, left.rows)
    rs = sort_cost(right.cost, right.rows)
    return (ls + rs
            + (left.rows + right.rows) * CPU_OPERATOR_COST
            + out_rows * CPU_TUPLE_COST)

Hash Join  (cost=2387.50..25687.50 rows=55000 width=36)

@dataclass
class CostedPlan:
    rows: float
    startup_cost: float
    total_cost: float
 
 
def cost_for(plan: CostedPlan, tuples_to_fetch: float) -> float:
    """部分取得のコスト。起動コストはどうせ払うので按分しない。"""
    frac = min(tuples_to_fetch, plan.rows) / plan.rows
    return plan.startup_cost + (plan.total_cost - plan.startup_cost) * frac

def hash_join_startup(build: Plan) -> float:
    """build を読み切り、ハッシュ表を埋め切るまでが startup。"""
    return build.cost + build.rows * CPU_OPERATOR_COST

def dp_join_order(rels, cardinality, connected):
    """部分集合の小さい順に最適プランを確定していく（DPsize）。"""
    n = len(rels)
    best = {}
    for i in range(n):
        best[1 << i] = Plan(rows=rels[i].n_rows, cost=0.0, desc=rels[i].name)
 
    for size in range(2, n + 1):
        for mask in range(1, 1 << n):
            if bin(mask).count("1") != size:
                continue
            if not connected(mask):          # 直積になる部分集合は刈る
                continue
            rows = cardinality(mask)
            for right in submasks(mask):     # mask の真部分集合を列挙
                left = mask ^ right
                if left == 0 or right == 0:
                    continue
                if left not in best or right not in best:
                    continue
                if not has_join_edge(left, right):
                    continue
                cost = rows + best[left].cost + best[right].cost   # C_out
                if mask not in best or cost < best[mask].cost:
                    best[mask] = Plan(rows, cost,
                                      f"({best[left].desc} ⋈ {best[right].desc})")
    return best[(1 << n) - 1]
 
 
def submasks(mask: int):
    """mask の空でない真部分集合をすべて列挙する定番のイディオム。"""
    sub = (mask - 1) & mask
    while sub > 0:
        yield sub
        sub = (sub - 1) & mask

Limit(10)                              rows=10       cost≈26,601
└─ Sort(o.total DESC) top-N heapsort   rows=10       cost≈26,601
   └─ HashJoin(u.id = o.user_id)       rows=55,000   cost≈25,688
      ├─ SeqScan(orders)               rows=100,000  cost≈22,500   [probe]
      │    Filter: total > 100
      └─ Hash                          rows=55,000
         └─ SeqScan(users)             rows=55,000   cost≈2,250    [build]
              Filter: age > 30

DEFAULT_INEQ_SEL = 0.3333333333333333    # selfuncs.h
DEFAULT_EQ_SEL   = 0.005
UNKNOWN = object()                       # 値未定のバインドパラメータ
 
 
def selectivity_gt(stats: ColumnStats, value) -> float:
    if value is UNKNOWN:
        return DEFAULT_INEQ_SEL          # ヒストグラムを引けないので定数を返す
    ...                                  # 以降は第6章と同じ

class Operator:
    def open(self) -> None: ...          # 状態を初期化する
    def next(self) -> dict | None: ...   # 1タプル返す。もう無ければ None
    def close(self) -> None: ...         # 資源を解放する

class SeqScanExec:
    def __init__(self, rows, predicate=None):
        self.rows, self.predicate = rows, predicate
        self.cursor = 0
 
    def open(self):
        self.cursor = 0
 
    def next(self):
        while self.cursor < len(self.rows):
            row = self.rows[self.cursor]
            self.cursor += 1
            if self.predicate is None or self.predicate(row):
                return row
        return None
 
    def close(self):
        pass
 
 
class NestedLoopJoinExec:
    def __init__(self, outer, inner, predicate):
        self.outer, self.inner, self.predicate = outer, inner, predicate
        self.outer_tuple = None
 
    def open(self):
        self.outer.open()
        self.inner.open()
        self.outer_tuple = None
 
    def next(self):
        while True:
            if self.outer_tuple is None:
                self.outer_tuple = self.outer.next()
                if self.outer_tuple is None:
                    return None
                self.inner.open()          # 内側を先頭に巻き戻す（rescan）
            inner_tuple = self.inner.next()
            if inner_tuple is None:
                self.outer_tuple = None    # 内側を1周した。次の外側へ
                continue
            if self.predicate(self.outer_tuple, inner_tuple):
                return {**self.outer_tuple, **inner_tuple}
 
    def close(self):
        self.outer.close()
        self.inner.close()

from collections import defaultdict
 
class HashJoinExec:
    def __init__(self, build, probe, build_key, probe_key):
        self.build, self.probe = build, probe
        self.build_key, self.probe_key = build_key, probe_key
 
    def open(self):
        # ── build フェーズ：ここで build 側を最後まで読み切る（ブロッキング）
        self.build.open()
        self.table = defaultdict(list)
        while (row := self.build.next()) is not None:
            self.table[self.build_key(row)].append(row)
        self.build.close()
 
        # ── probe フェーズの準備
        self.probe.open()
        self.matches = iter(())
        self.current = None
 
    def next(self):
        while True:
            m = next(self.matches, None)
            if m is not None:
                return {**m, **self.current}
            self.current = self.probe.next()
            if self.current is None:
                return None
            self.matches = iter(self.table.get(self.probe_key(self.current), ()))
 
    def close(self):
        self.probe.close()
        self.table = None

import heapq
import itertools
 
class TopNExec:
    def __init__(self, child, key, n, desc=True):
        self.child, self.key, self.n, self.desc = child, key, n, desc
 
    def open(self):
        self.child.open()
        heap = []                       # (整列キー, 連番, 行) の最小ヒープ
        seq = itertools.count()
        while (row := self.child.next()) is not None:
            k = self.key(row)
            if not self.desc:
                k = -k                  # ASC のときは符号を反転して向きを揃える
            if len(heap) < self.n:
                heapq.heappush(heap, (k, next(seq), row))
            elif k > heap[0][0]:        # 現在の最下位より良いものだけ入れ替える
                heapq.heapreplace(heap, (k, next(seq), row))
        heap.sort(key=lambda x: x[0], reverse=True)
        self.out = iter([row for _, _, row in heap])
 
    def next(self):
        return next(self.out, None)
 
    def close(self):
        self.out = None

 0  EEOP_SCAN_FETCHSOME       last_var=3     # 参照する最大 attnum までしか deform しない
 1  EEOP_SCAN_VAR             attnum=3       # total
 2  EEOP_CONST                value=100
 3  EEOP_FUNCEXPR_STRICT      numeric_gt     # 引数のどちらかが NULL なら関数を呼ばない
 4  EEOP_QUAL                 jumpdone=9     # FALSE も NULL も即座に不合格
 5  EEOP_SCAN_VAR             attnum=2       # user_id
 6  EEOP_CONST                value=42
 7  EEOP_FUNCEXPR_STRICT      int4eq
 8  EEOP_QUAL                 jumpdone=9
 9  EEOP_DONE

def sort_external(rows, key, work_mem_rows, tmp):
    """run 生成 → マージ。tuplesort.c の骨格。"""
    runs, buf = [], []
    for row in rows:
        buf.append(row)
        if len(buf) == work_mem_rows:      # work_mem を使い切った
            buf.sort(key=key)
            runs.append(tmp.spill(buf))    # ソート済みの塊を一時ファイルへ
            buf = []
    if buf:
        buf.sort(key=key)
        runs.append(tmp.spill(buf))
 
    # 各 run から1行ずつヒープに載せ、最小値を取り出しては補充する
    return heapq.merge(*(tmp.read(r) for r in runs), key=key)

Sort Method: top-N heapsort  Memory: 25kB
Sort Method: quicksort       Memory: 3400kB
Sort Method: external merge  Disk: 341000kB

class HashAggExec:
    """ブロッキング。グループごとの状態をハッシュ表に持つ。"""
 
    def __init__(self, child, group_keys, init, accum, final):
        self.child, self.group_keys = child, group_keys
        self.init, self.accum, self.final = init, accum, final
 
    def open(self):
        self.child.open()
        table = {}
        while (row := self.child.next()) is not None:      # ← 全部読む
            k = tuple(row[c] for c in self.group_keys)
            # `table.get(k) or self.init()` と書いてはいけない。累算器の状態が
            # 偽値（0 や False）になった瞬間に初期値へ巻き戻ってしまう。
            state = table[k] if k in table else self.init()
            table[k] = self.accum(state, row)
        self.out = iter([(k, self.final(s)) for k, s in table.items()])
 
    def next(self):
        return next(self.out, None)
 
    def close(self):
        self.child.close()
 
 
class SortedAggExec:
    """非ブロッキング。キーが変わった瞬間に1行出せる。入力はソート済みが前提。"""
 
    def __init__(self, child, group_keys, init, accum, final):
        self.child, self.group_keys = child, group_keys
        self.init, self.accum, self.final = init, accum, final
 
    def open(self):
        self.child.open()
        self.cur_key, self.state = None, None
 
    def next(self):
        while True:
            row = self.child.next()
            if row is None:
                if self.cur_key is None:
                    return None
                out = (self.cur_key, self.final(self.state))    # 最後のグループ
                self.cur_key = None
                return out
 
            k = tuple(row[c] for c in self.group_keys)
            if self.cur_key is None:
                self.cur_key = k
                self.state = self.accum(self.init(), row)
            elif k == self.cur_key:
                self.state = self.accum(self.state, row)
            else:
                out = (self.cur_key, self.final(self.state))    # キーが変わった
                self.cur_key = k
                self.state = self.accum(self.init(), row)
                return out
 
    def close(self):
        self.child.close()

HashAggregate  (actual rows=100000 loops=1)
  Batches: 5  Memory Usage: 6273kB  Disk Usage: 21456kB

Gather  (cost=3387.50..24429.00 rows=55000 width=36)
  Workers Planned: 2
  ->  Hash Join  (cost=2387.50..17929.00 rows=22917 width=36)
        Hash Cond: (o.user_id = u.id)
        ->  Parallel Seq Scan on orders o  (cost=0.00..15208.33 rows=41667 width=20)
              Filter: (total > '100'::numeric)
        ->  Hash  (cost=2250.00..2250.00 rows=55000 width=24)
              ->  Seq Scan on users u  (cost=0.00..2250.00 rows=55000 width=24)
                    Filter: (age > 30)

SELECT ctid, id, name FROM users LIMIT 3;
--  ctid  | id  | name
-- -------+-----+------
--  (0,1) |   1 | Ann
--  (0,2) |   2 | Bob
--  (0,3) |   3 | Cid

PAGE_SIZE = 8192
 
class HeapScan:
    """ブロック単位でヒープファイルを走査する。"""
 
    def __init__(self, rel, buffer_pool, snapshot):
        self.rel, self.pool, self.snapshot = rel, buffer_pool, snapshot
 
    def open(self):
        self.blkno = 0
        self.slot = 0
        self.page = None
 
    def next(self):
        while self.blkno < self.rel.n_pages:
            if self.page is None:
                self.page = self.pool.read_buffer(self.rel, self.blkno)  # pin される
                self.slot = 0
 
            while self.slot < self.page.n_slots:
                slot, raw = self.slot, self.page.item(self.slot)
                self.slot += 1
                if raw is None:            # LP_UNUSED / LP_DEAD は飛ばす
                    continue
                tup = parse_tuple(raw)
                if not visible(tup, self.snapshot):    # MVCC の可視性判定
                    continue
                return tup, (self.blkno, slot + 1)
 
            self.pool.release_buffer(self.rel, self.blkno)   # unpin
            self.page = None
            self.blkno += 1
        return None

def visible(tup, snapshot) -> bool:
    """PostgreSQL の HeapTupleSatisfiesMVCC を大幅に簡略化したもの。"""
    if not committed(tup.xmin, snapshot):
        return False                       # まだコミットされていない挿入
    if tup.xmin >= snapshot.xmax or tup.xmin in snapshot.in_progress:
        return False                       # スナップショット取得より後の挿入
    if tup.xmax == 0:
        return True                        # 削除されていない
    if not committed(tup.xmax, snapshot):
        return True                        # 削除がまだコミットされていない
    if tup.xmax >= snapshot.xmax or tup.xmax in snapshot.in_progress:
        return True                        # スナップショット取得より後の削除
    return False

MAX_USAGE = 5                # PostgreSQL の BM_MAX_USAGE_COUNT
 
 
@dataclass
class Frame:
    page: object | None = None    # Page オブジェクト（第15章で定義）
    tag: tuple | None = None
    usage: int = 0
    pin: int = 0
    dirty: bool = False
 
 
class BufferPool:
    def __init__(self, n_frames: int, storage):
        self.frames = [Frame() for _ in range(n_frames)]
        self.table = {}          # buffer tag -> フレーム番号
        self.hand = 0            # clock-sweep の針
        self.storage = storage
 
    def read_buffer(self, rel, blkno) -> "Page":
        tag = (rel.relfilenode, "main", blkno)
        idx = self.table.get(tag)
        if idx is not None:                       # ── ヒット
            f = self.frames[idx]
            f.usage = min(f.usage + 1, MAX_USAGE)
            f.pin += 1
            return f.page
 
        idx = self.evict_one()                    # ── ミス
        victim = self.frames[idx]
        if victim.tag is not None:
            if victim.dirty:
                old_node, _, old_blkno = victim.tag
                self.storage.flush_wal_up_to(victim.page.lsn)   # WAL 先行書き込み
                self.storage.write_page(old_node, old_blkno, victim.page)
            del self.table[victim.tag]
 
        victim.page = Page(self.storage.read_page(rel.relfilenode, blkno))
        victim.tag, victim.usage, victim.pin, victim.dirty = tag, 1, 1, False
        self.table[tag] = idx
        return victim.page
 
    def release_buffer(self, rel, blkno) -> None:
        """pin を一つ返す。ゼロになったフレームは再び置換の候補に戻る。"""
        idx = self.table[(rel.relfilenode, "main", blkno)]
        self.frames[idx].pin -= 1
 
    def evict_one(self) -> int:
        """clock-sweep（second chance）で犠牲フレームを選ぶ。"""
        for _ in range(len(self.frames) * (MAX_USAGE + 2)):
            f = self.frames[self.hand]
            here = self.hand
            self.hand = (self.hand + 1) % len(self.frames)
            if f.pin > 0:
                continue                    # 使用中のページは追い出せない
            if f.usage > 0:
                f.usage -= 1                # 猶予を与えて次へ
                continue
            return here
        raise RuntimeError("no unpinned buffers available")

+--------------------------------------------------+
| PageHeaderData (24B)                             |
+--------------------------------------------------+
| ItemIdData[0] | ItemIdData[1] | ItemIdData[2] |… |  ← 前から伸びる
+--------------------------------------------------+
|                                                  |
|                   free space                     |
|                                                  |
+--------------------------------------------------+
| … | tuple 2 | tuple 1 | tuple 0                  |  ← 後ろから伸びる
+--------------------------------------------------+

/* src/include/storage/itemid.h */
typedef struct ItemIdData
{
    unsigned    lp_off:15,      /* ページ先頭からのオフセット */
                lp_flags:2,     /* 状態 */
                lp_len:15;      /* タプルのバイト長 */
} ItemIdData;

import struct
 
PAGE_SIZE = 8192
SIZEOF_PAGE_HEADER = 24
SIZEOF_ITEM_ID = 4
 
LP_UNUSED, LP_NORMAL, LP_REDIRECT, LP_DEAD = 0, 1, 2, 3
 
 
def parse_page(buf: bytes):
    """8192 バイトのページから有効なタプルのバイト列を取り出す。"""
    # PageHeaderData: pd_lsn(8) pd_checksum(2) pd_flags(2)
    #                 pd_lower(2) pd_upper(2) pd_special(2)
    #                 pd_pagesize_version(2) pd_prune_xid(4)
    pd_lower, pd_upper = struct.unpack_from("<HH", buf, 12)
    n_slots = (pd_lower - SIZEOF_PAGE_HEADER) // SIZEOF_ITEM_ID
 
    for slot in range(n_slots):
        (item_id,) = struct.unpack_from(
            "<I", buf, SIZEOF_PAGE_HEADER + slot * SIZEOF_ITEM_ID)
        # リトルエンディアン環境の一般的なビットフィールド配置
        lp_off = item_id & 0x7FFF
        lp_flags = (item_id >> 15) & 0x0003
        lp_len = (item_id >> 17) & 0x7FFF
        if lp_flags != LP_NORMAL:
            continue
        yield slot + 1, buf[lp_off:lp_off + lp_len]   # TID の offset は1起点

class Page:
    def __init__(self, buf: bytes):
        self.buf = buf
        (pd_lower,) = struct.unpack_from("<H", buf, 12)
        self.n_slots = (pd_lower - SIZEOF_PAGE_HEADER) // SIZEOF_ITEM_ID
        live = dict(parse_page(buf))                     # {ラインポインタ番号: raw}
        # 添字 i をラインポインタ番号 i+1 に揃える。詰めてしまうと TID がずれる
        self.slots = [live.get(i + 1) for i in range(self.n_slots)]
        (self.lsn,) = struct.unpack_from("<Q", buf, 0)   # pd_lsn
 
    def item(self, i: int) -> bytes | None:
        """0起点のスロット添字。LP_UNUSED / LP_DEAD なら None。"""
        return self.slots[i]