Harness for the DISTINCT ON / UNION DISTINCT ON measurements (plain text bundle).
Split at the '===== file:' lines.  Python 3, psql and the builds' bin directories:
  PGBIN_<name>=/path/to/bin for every build you want to compare.

diffgen.py  differential oracle: inline / UNION / recursive DISTINCT ON against
            row_number(), classic DISTINCT ON over UNION ALL, and Bellman-Ford
perf24.py   the many-groups ladder (data, queries, one timed run per call)
ladder.sh   alternated rounds over a list of build:query cells, median + CI95
rec24.py    recursive limits (grids, random graphs, adversarial DAG, negative
            cycle) and the error-HINT cases

===== file: diffgen.py =====
#!/usr/bin/env python3
"""Differential / metamorphic harness for the v3 DISTINCT ON series (Hannu Krosing).

Every case builds random tables (NULLs, varlenas around the 1-byte header
boundary, incompressible TOASTed text, numeric with varying scale), runs the
new operator, and checks it against a formulation PostgreSQL already computes:

  inline   SELECT DISTINCT ON (k ORDER BY s)   vs  row_number() OVER (PARTITION BY k ORDER BY s)
  union    A UNION DISTINCT ON (k ORDER BY s) B vs  classic DISTINCT ON over A UNION ALL B
  recursive UNION DISTINCT ON (node ORDER BY cost) vs Dijkstra / Bellman-Ford in Python

Ties are legal: the tolerant check only demands one row per key, that the row
exists in the input, and that its sort tuple equals the per-key optimum.  When
the spec is tie-free (id appended) the result is also compared row for row.

usage: diffgen.py --bindir B --variant V --out DIR --seed0 S --cases N
                  [--workers W] [--budget SEC] [--port0 P] [--families a,b]
       diffgen.py --bindir B --replay SEED          (print one case's SQL)
"""
import argparse
import json
import os
import random
import re
import shutil
import subprocess
import sys
import time
from decimal import Decimal
from multiprocessing import Process

FAMILIES = [('inline', 22), ('classic', 5), ('inline_where', 8), ('union', 18),
            ('union3', 10), ('union_mixed', 10), ('union_where', 7),
            ('recursive', 15), ('recursive_neg', 5),
            ('inline_spill', 0), ('union_spill', 0),      # targeted: only via --families
            ('union_merge', 0)]

GUCS = [
    {},
    {'enable_hashagg': 'off'},
    {'enable_sort': 'off'},
    {'enable_sort': 'off', 'work_mem': "'64kB'"},
    {'enable_sort': 'off', 'work_mem': "'64kB'", 'hash_mem_multiplier': '1'},
    {'enable_hashagg': 'off', 'work_mem': "'64kB'"},
    {'enable_seqscan': 'off'},
    {'max_parallel_workers_per_gather': '4', 'parallel_setup_cost': '0',
     'parallel_tuple_cost': '0', 'min_parallel_table_scan_size': '0'},
]

K2VALS = ["''", "'a'", "'A'", "'b'", "'ab'", "'ä'", "repeat('k', 125)",
          "repeat('k', 126)", "repeat('k', 127)", "repeat('k', 3000)"]

SETUP_SQL = r"""
CREATE OR REPLACE FUNCTION public.rlen(m int) RETURNS int VOLATILE LANGUAGE plpgsql AS $f$
DECLARE mm int := m;
BEGIN
  IF mm < 0 THEN mm := floor(random() * 4)::int; END IF;
  RETURN CASE mm WHEN 0 THEN floor(random() * 11)::int
                 WHEN 1 THEN 100 + floor(random() * 41)::int
                 WHEN 2 THEN 1500 + floor(random() * 1001)::int
                 ELSE 2500 + floor(random() * 6501)::int END;
END $f$;
CREATE OR REPLACE FUNCTION public.rtxt(n int) RETURNS text VOLATILE LANGUAGE sql AS $f$
  SELECT coalesce(left(string_agg(md5(random()::text), ''), n), '')
  FROM generate_series(0, n / 32) $f$;
"""

CRASH_MARKS = ('server closed the connection unexpectedly', 'connection to server was lost',
               'terminating connection because of crash', 'no connection to the server')
LAST_SPEC = None
LOG_MARKS = ('AddressSanitizer', 'runtime error:', 'TRAP:', 'terminated by signal',
             'LeakSanitizer', 'UndefinedBehaviorSanitizer')


# ---------------------------------------------------------------- generation

def gen_table(rng, name, off, spill=False, small=False):
    n = rng.choice([0, 1, 2, 5, 20, 100, 300, 1000, 2000])
    if spill:
        n = rng.choice([5000, 20000])
    pn = rng.choice([0, 0.05, 0.2, 0.5])
    dk1 = rng.choice([1, 2, 5, 20, 1000])
    if spill:
        dk1 = rng.choice([500, 5000, 50000])
    ds1 = rng.choice([1, 3, 10, 1000000])
    ds3 = rng.choice([1, 10, 1000])
    k2 = rng.sample(K2VALS, rng.randint(1, len(K2VALS)))
    s2mode = rng.choice([-1, 0, 1, 2, 3])
    plmode = rng.choice([-1, 0, 0, 1, 2, 3])
    storage = rng.choice(['', '', 'EXTERNAL', 'MAIN', 'PLAIN'])
    fseed = rng.random() * 2 - 1
    sql = [f"CREATE TABLE {name} (id int, k1 int, k2 text, s1 int, s2 text, s3 numeric, "
           f"pl text, b bool);"]
    if small:                                # indexed sort columns: btree rows must stay < 8 kB
        s2mode = 0 if s2mode in (-1, 2, 3) else s2mode
        plmode = 0 if plmode in (-1, 2, 3) else plmode
    if storage == 'PLAIN':
        plmode = 0 if plmode != 1 else 1      # PLAIN cannot hold > 8kB rows
        s2mode = 0 if s2mode not in (0, 1) else s2mode
    if storage:
        sql.append(f"ALTER TABLE {name} ALTER COLUMN pl SET STORAGE {storage}, "
                   f"ALTER COLUMN s2 SET STORAGE {storage};")
    sql.append(f"SELECT setseed({fseed:.6f}) \\g /dev/null")
    sql.append(
        f"INSERT INTO {name} SELECT g + {off}, "
        f"CASE WHEN random() < {pn} THEN NULL ELSE floor(random() * {dk1})::int END, "
        f"CASE WHEN random() < {pn} THEN NULL ELSE (ARRAY[{', '.join(k2)}])"
        f"[1 + floor(random() * {len(k2)})::int] END, "
        f"CASE WHEN random() < {pn} THEN NULL ELSE floor(random() * {ds1})::int END, "
        f"CASE WHEN random() < {pn} THEN NULL ELSE public.rtxt(public.rlen({s2mode})) END, "
        f"CASE WHEN random() < {pn} THEN NULL ELSE "
        f"round((random() * {ds3})::numeric, floor(random() * 4)::int) END, "
        f"public.rtxt(public.rlen({plmode})), random() < 0.5 "
        f"FROM generate_series(1, {n}) g;")
    if spill:
        # payload length is a function of the tie-breaker: successive "better" rows for a
        # key are sometimes shorter (in-place overwrite) and sometimes longer (realloc)
        sql.append(f"UPDATE {name} SET pl = public.rtxt(((coalesce(s1, 0)::bigint * 7919) % 2500)::int), "
                   f"s2 = CASE WHEN s2 IS NULL THEN NULL ELSE public.rtxt((coalesce(s1, 0) * 31) % 300) END;")
    if rng.random() < 0.3:
        cols = rng.sample(['k1', 'k2', 's1', 's3', 'b'], rng.randint(1, 2))
        sql.append(f"CREATE INDEX ON {name} ({', '.join(cols)});")
    return '\n'.join(sql)


def sort_item(rng, e):
    return e + rng.choice(['', ' ASC', ' DESC']) + rng.choice(['', '', ' NULLS FIRST', ' NULLS LAST'])


def gen_spec(rng, key_pool, sort_pool, tie_free):
    keys = rng.sample(key_pool, rng.choice([1, 1, 1, 2]))
    if rng.random() < 0.05:
        keys.append(keys[0])
    sorts = [(e, sort_item(rng, e)) for e in rng.sample(sort_pool, rng.choice([1, 1, 2, 3]))]
    if tie_free and not any(e == 'id' for e, _ in sorts):
        sorts.append(('id', rng.choice(['id', 'id DESC'])))
    return keys, sorts


def spec_sql(keys, sorts, order=True):
    if order and sorts:
        return f"DISTINCT ON ({', '.join(keys)} ORDER BY {', '.join(s for _, s in sorts)})"
    return f"DISTINCT ON ({', '.join(keys)})"


def tolerant_check(T, K, sorts, sel, with_best=True):
    """sel: select list producing r's columns from T rows."""
    Kl = ', '.join(K)
    Sx = ', '.join(e for e, _ in sorts)
    So = ', '.join(s for _, s in sorts)
    best = 'true'
    if with_best and sorts:
        mine = f"SELECT {Kl}, {Sx} FROM ({T}) t WHERE t.id IN (SELECT r.id FROM r)"
        orc = (f"SELECT {Kl}, {Sx} FROM (SELECT t.*, row_number() OVER (PARTITION BY {Kl} "
               f"ORDER BY {So}) AS rn__ FROM ({T}) t) t WHERE rn__ = 1")
        best = (f"(NOT EXISTS (({mine}) EXCEPT ALL ({orc})) AND "
                f"NOT EXISTS (({orc}) EXCEPT ALL ({mine})))")
    return (f"SELECT 'CHK|' || n_r || '|' || n_k || '|' || ids_ok || '|' || rows_ok || '|' || best_ok "
            f"FROM (SELECT (SELECT count(*) FROM r) AS n_r, "
            f"(SELECT count(*) FROM (SELECT DISTINCT {Kl} FROM ({T}) t) x) AS n_k, "
            f"(SELECT count(DISTINCT id) = count(*) FROM r) AS ids_ok, "
            f"NOT EXISTS (SELECT * FROM r EXCEPT ALL SELECT {sel} FROM ({T}) t "
            f"WHERE t.id IN (SELECT r.id FROM r)) AS rows_ok, "
            f"{best} AS best_ok) z;")


def exact_check(O):
    return (f"SELECT 'EXA|' || (SELECT count(*) FROM r) || '|' || (SELECT count(*) FROM ({O}) o) "
            f"|| '|' || (NOT EXISTS (SELECT * FROM r EXCEPT ALL SELECT * FROM ({O}) o) AND "
            f"NOT EXISTS (SELECT * FROM ({O}) o EXCEPT ALL SELECT * FROM r));")


def rand_pred(rng, cols):
    c = rng.choice([x for x in cols if x != 'id'] or ['id'])
    return {
        'k1': f"k1 {rng.choice(['=', '<', '>', '<>'])} {rng.randint(0, 5)}",
        'k2': rng.choice(["k2 IS NULL", "k2 > 'a'", "length(k2) > 100", "k2 = ''"]),
        's1': rng.choice(["s1 IS NULL", f"s1 < {rng.randint(0, 10)}", "s1 % 2 = 0"]),
        's2': rng.choice(["s2 IS NULL", "length(s2) > 120", "s2 < '8'"]),
        's3': rng.choice(["s3 IS NOT NULL", f"s3 > {rng.randint(0, 10)}"]),
        'pl': rng.choice(["length(pl) > 2000", "pl < '5'"]),
        'b': rng.choice(["b", "NOT b", "b IS NULL"]),
        'lpl': "lpl > 100", 's2x': "s2x IS NULL", 's3d': "s3d > 3",
    }.get(c, 'true')


def build_inline(rng, family):
    tie_free = family == 'inline_where' or rng.random() < 0.3
    keys, sorts = gen_spec(rng, ['k1', 'k2', 'b', '(k1 % 3)', 'lower(k2)'],
                           ['s1', 's2', 's3', 'lower(s2)', 'length(pl)', 'pl', 'k1', 'k2', 'id', 'b'],
                           tie_free)
    extra = [('k1', 'k1'), ('k2', 'k2'), ('s1', 's1'), ('s2', 's2'), ('s3', 's3'), ('pl', 'pl'),
             ('b', 'b'), ('length(pl)', 'lpl'), ("s2 || '#'", 's2x'), ('s3 * 2', 's3d')]
    proj = rng.sample(extra, rng.randint(0, 5)) + [('id', 'id')]
    rng.shuffle(proj)
    sel = ', '.join(f"{e} AS {a}" for e, a in proj)
    where = rng.choice(['', '', f" WHERE {rand_pred(rng, ['k1', 's1', 's2', 'b'])}"])
    T = f"SELECT * FROM a{where}"
    keys_sql = list(keys)
    for i, k in enumerate(keys_sql):          # occasional positional reference
        pos = [j for j, (e, _) in enumerate(proj) if e == k]
        if pos and rng.random() < 0.15:
            keys_sql[i] = str(pos[0] + 1)
    if family == 'classic':
        ks = list(dict.fromkeys(keys))
        q = (f"SELECT DISTINCT ON ({', '.join(ks)}) {sel} FROM a{where} ORDER BY "
             f"{', '.join(ks + [s for _, s in sorts])}")
        keys = ks
    else:
        outer = rng.choice(['', '', f" ORDER BY {rng.choice(proj)[1]}"])
        q = f"SELECT {spec_sql(keys_sql, sorts)} {sel} FROM a{where}{outer}"
    checks = [tolerant_check(T, keys, sorts, sel)]
    orc = (f"SELECT {sel} FROM (SELECT t.*, row_number() OVER (PARTITION BY {', '.join(keys)} "
           f"ORDER BY {', '.join(s for _, s in sorts)}) AS rn__ FROM ({T}) t) t WHERE rn__ = 1")
    if family == 'inline_where':
        pred = rand_pred(rng, [a for _, a in proj])
        q = f"SELECT * FROM ({q}) s WHERE {pred}"
        checks = [exact_check(f"SELECT * FROM ({orc}) s WHERE {pred}")]
    elif tie_free:
        checks.append(exact_check(orc))
    return ['a'], q, checks


def build_union(rng, family):
    nt = 3 if family in ('union3', 'union_mixed') else 2
    tie_free = family in ('union_mixed', 'union_where') or rng.random() < 0.3
    keys, sorts = gen_spec(rng, ['k1', 'k2', 'b'], ['s1', 's2', 's3', 'pl', 'k1', 'k2', 'id', 'b'],
                           tie_free)
    need = set(keys) | {e for e, _ in sorts} | {'id'}
    pool = [c for c in ['k1', 'k2', 's1', 's2', 's3', 'pl', 'b'] if c not in need]
    cols = sorted(need) + rng.sample(pool, min(rng.randint(0, 3), len(pool)))   # sorted: no hash order
    rng.shuffle(cols)
    cl = ', '.join(cols)
    tabs = ['a', 'b', 'c'][:nt]
    br = {t: f"SELECT {cl} FROM {t}" + rng.choice(['', '', f" WHERE {rand_pred(rng, cols)}"])
          for t in tabs}
    keys_sql = [str(cols.index(k) + 1) if rng.random() < 0.15 else k for k in keys]
    with_order = rng.random() < 0.9
    global LAST_SPEC
    LAST_SPEC = (keys, sorts)
    sp = spec_sql(keys_sql, sorts, with_order)
    T2 = f"{br['a']} UNION ALL {br['b']}"
    O1 = (f"SELECT DISTINCT ON ({', '.join(keys)}) * FROM ({T2}) o ORDER BY "
          f"{', '.join(keys + [s for _, s in sorts])}")
    if family == 'union':
        outer = rng.choice(['', '', f" ORDER BY {rng.choice(cols)}"])
        q = f"{br['a']} UNION {sp} {br['b']}{outer}"
        checks = [tolerant_check(T2, keys, sorts, cl, with_order)]
        if tie_free and with_order:
            checks.append(exact_check(O1))
        return tabs, q, checks
    if family == 'union_where':
        pred = rand_pred(rng, cols)
        q = f"SELECT * FROM ({br['a']} UNION {sp} {br['b']}) s WHERE {pred}"
        if not with_order:
            return tabs, f"{br['a']} UNION {sp} {br['b']}", [tolerant_check(T2, keys, sorts, cl, False)]
        return tabs, q, [exact_check(f"SELECT * FROM ({O1}) s WHERE {pred}")]
    if family == 'union3':
        if rng.random() < 0.5:                   # same spec, left- or right-nested
            T3 = f"{T2} UNION ALL {br['c']}"
            if rng.random() < 0.5:
                q = f"{br['a']} UNION {sp} {br['b']} UNION {sp} {br['c']}"
            else:
                q = f"{br['a']} UNION {sp} ({br['b']} UNION {sp} {br['c']})"
            return tabs, q, [tolerant_check(T3, keys, sorts, cl, with_order)]
        # inner spec tie-free, outer spec different (needs exact inner oracle)
        if not with_order:
            sp = spec_sql(keys_sql, sorts, True)
        if not any(e == 'id' for e, _ in sorts):
            sorts = sorts + [('id', 'id')]
            sp = spec_sql(keys_sql, sorts, True)
            O1 = (f"SELECT DISTINCT ON ({', '.join(keys)}) * FROM ({T2}) o ORDER BY "
                  f"{', '.join(keys + [s for _, s in sorts])}")
        keys2 = [k for k in rng.sample(['k1', 'k2', 'b'], 1) if k in cols] or [keys[0]]
        sorts2 = [(e, sort_item(rng, e)) for e in rng.sample([c for c in cols], 1)]
        q = f"({br['a']} UNION {sp} {br['b']}) UNION {spec_sql(keys2, sorts2)} {br['c']}"
        T3 = f"SELECT * FROM ({O1}) x1 UNION ALL {br['c']}"
        return tabs, q, [tolerant_check(T3, keys2, sorts2, cl, True)]
    # union_mixed: tie-free inner, composed with an ordinary set operation
    if not with_order:
        sp = spec_sql(keys_sql, sorts, True)
    op = rng.choice(['UNION ALL', 'UNION', 'INTERSECT', 'INTERSECT ALL', 'EXCEPT', 'EXCEPT ALL'])
    other = rng.choice([br['c'], br['a'], f"SELECT {cl} FROM a"])
    if rng.random() < 0.5:
        q = f"({br['a']} UNION {sp} {br['b']}) {op} {other}"
        O = f"SELECT * FROM ({O1}) x1 {op} {other}"
    else:
        q = f"{other} {op} ({br['a']} UNION {sp} {br['b']})"
        O = f"{other} {op} SELECT * FROM ({O1}) x1"
    if rng.random() < 0.5:                   # outer qual over the nested set operation
        pred = rand_pred(rng, cols)
        q = f"SELECT * FROM ({q}) s WHERE {pred}"
        O = f"SELECT * FROM ({O}) s WHERE {pred}"
    return tabs, q, [exact_check(O)]


def gen_graph(rng, neg):
    n = rng.choice([2, 3, 5, 10, 20, 40, 80])
    m = rng.randint(0, n * rng.choice([1, 2, 4]))
    textkey = rng.random() < 0.3
    numeric = rng.random() < 0.3
    names = ([f"n{i}{'x' * rng.choice([0, 0, 5, 130])}" for i in range(n)] if textkey
             else list(range(1, n + 1)))
    h = [rng.randint(-20, 20) for _ in range(n)]
    edges = []
    for _ in range(m):
        u, v = rng.randrange(n), rng.randrange(n)
        w = Decimal(rng.randint(0, 30)) if not numeric else \
            Decimal(rng.randint(0, 3000)) / Decimal(rng.choice([1, 10, 100]))
        if neg:
            w = w + h[u] - h[v]           # potentials: negative edges, no negative cycle
        edges.append((names[u], names[v], w))
    seeds = []
    for _ in range(rng.choice([1, 1, 1, 2, 3])):
        seeds.append((names[rng.randrange(n)], Decimal(rng.choice([0, 0, 5, 100]))))
    return names, edges, seeds, textkey, numeric


def shortest(names, edges, seeds):
    """Bellman-Ford (handles the potential-shifted negative edges)."""
    dist = {}
    for s, c in seeds:
        if s not in dist or c < dist[s]:
            dist[s] = c
    for _ in range(len(names) + 1):
        changed = False
        for u, v, w in edges:
            if u in dist and (v not in dist or dist[u] + w < dist[v]):
                dist[v] = dist[u] + w
                changed = True
        if not changed:
            return dist
    raise RuntimeError('negative cycle in generator')


def build_recursive(rng, family):
    names, edges, seeds, textkey, numeric = gen_graph(rng, family == 'recursive_neg')
    kt = 'text' if textkey else 'int'
    ct = 'numeric' if numeric else 'int'
    if not numeric:
        edges = [(u, v, int(w)) for u, v, w in edges]
        seeds = [(s, int(c)) for s, c in seeds]
    lit = (lambda x: f"'{x}'") if textkey else str
    sql = [f"CREATE TABLE e (src {kt}, dst {kt}, w {ct});",
           f"CREATE TABLE seeds (node {kt}, cost {ct});"]
    if edges:
        sql.append("INSERT INTO e VALUES " + ', '.join(f"({lit(u)}, {lit(v)}, {w})"
                                                       for u, v, w in edges) + ';')
    sql.append("INSERT INTO seeds VALUES " + ', '.join(f"({lit(s)}, {c})" for s, c in seeds) + ';')
    if rng.random() < 0.3:
        sql.append("CREATE INDEX ON e (src);")
    path = rng.random() < 0.6
    bound = None
    if family == 'recursive' and rng.random() < 0.2:
        bound = rng.randint(5, 200)
    pcol = ", path" if path else ""
    pseed = f", ARRAY[node]" if path else ""
    pstep = ", sp.path || e.dst" if path else ""
    join = rng.choice(["FROM sp JOIN e ON sp.node = e.src",
                       "FROM e JOIN sp ON e.src = sp.node",
                       "FROM sp, e WHERE sp.node = e.src"])
    bw = ""
    if bound is not None:
        bw = (" AND " if "WHERE" in join else " WHERE ") + f"sp.cost + e.w < {bound}"
    order = rng.choice(["cost", "cost ASC", "cost ASC NULLS LAST"])
    q = (f"WITH RECURSIVE sp(node, cost{pcol}) AS (SELECT node, cost{pseed} FROM seeds "
         f"UNION DISTINCT ON (node ORDER BY {order}) SELECT e.dst, sp.cost + e.w{pstep} "
         f"{join}{bw})")
    if rng.random() < 0.25:                  # SEARCH / CYCLE are accepted by v3: same distances
        q += rng.choice([" SEARCH DEPTH FIRST BY node SET seq__",
                         " SEARCH BREADTH FIRST BY node SET seq__",
                         " CYCLE node SET is_cycle__ USING cpath__"])
    q += f" SELECT node, cost{', path' if path else ''} FROM sp"
    dist = shortest(names, edges, seeds)
    if bound is not None:
        seedset = {s for s, _ in seeds}
        dist = {k: v for k, v in dist.items() if v < bound or k in seedset}
        # seeds keep their best seed cost unless a cheaper in-bound path exists
        sc = {}
        for s, c in seeds:
            sc[s] = min(sc.get(s, c), c)
        d2 = shortest(names, [(u, v, w) for u, v, w in edges], seeds)
        dist = {}
        # bounded search: Bellman-Ford restricted to relaxations under the bound
        dist = dict(sc)
        for _ in range(len(names) + 1):
            ch = False
            for u, v, w in edges:
                if u in dist and dist[u] + w < bound and (v not in dist or dist[u] + w < dist[v]):
                    dist[v] = dist[u] + w
                    ch = True
            if not ch:
                break
        del d2
    expect = {'dist': {str(k): str(v) for k, v in dist.items()}, 'path': path,
              'edges': [(str(u), str(v), str(w)) for u, v, w in edges],
              'seeds': [(str(s), str(c)) for s, c in seeds]}
    return '\n'.join(sql), q, expect


def build_case(seed, families=None):
    rng = random.Random(seed)
    fams = [f for f in FAMILIES if not families or f[0] in families]
    # zero-weight (targeted) families only count when asked for explicitly; with no filter the
    # cumulative weights equal the original list, so a seed maps to the same case everywhere
    family = rng.choices([f for f, _ in fams], [(w or 1) if families else w for _, w in fams])[0]
    guc = dict(rng.choice(GUCS))
    merge = family == 'union_merge'
    if merge:                                # pre-sorted branches: MergeAppend -> Unique
        family = 'union'
        guc = {'enable_hashagg': 'off', 'enable_seqscan': 'off', 'enable_bitmapscan': 'off'}
    spill = family.endswith('_spill')
    if spill:
        family = family[:-len('_spill')]
        guc = {'enable_sort': 'off', 'work_mem': "'64kB'", 'hash_mem_multiplier': '1'}
    if family.startswith('recursive'):
        guc = dict(rng.choice([{}, {'enable_hashagg': 'off'}, {'work_mem': "'64kB'"}]))
        setup, q, expect = build_recursive(rng, family)
        return family, guc, setup, q, None, expect
    if family in ('inline', 'classic', 'inline_where'):
        tabs, q, checks = build_inline(rng, family)
    else:
        tabs, q, checks = build_union(rng, family)
    setup = '\n'.join(gen_table(rng, t, i * 1000000, spill, merge) for i, t in enumerate(tabs))
    if merge:
        ks, ss = LAST_SPEC
        icols = list(dict.fromkeys(ks)) + [x for _, x in ss]
        setup += ''.join(f"\nCREATE INDEX ON {t} ({', '.join(icols)});" for t in tabs)
    view = rng.random() < 0.15
    return family, guc, setup, q, checks, {'view': view}


def case_sql(seed, families=None):
    family, guc, setup, q, checks, expect = build_case(seed, families)
    if os.environ.get('DIFFGEN_MUTATE'):     # harness self-test: break q, keep the oracle
        if family.startswith('recursive'):
            q = q.replace('sp.cost + e.w', 'sp.cost + e.w + 1', 1)
        elif family != 'classic':
            q = q.replace(' ORDER BY ', ' ORDER BY id DESC, ', 1)
    gs = '\n'.join(f"SET {k} = {v};" for k, v in guc.items())
    out = ["\\set VERBOSITY verbose", "\\echo @@PHASE setup",
           "SET client_min_messages = warning;", "DROP SCHEMA IF EXISTS c CASCADE;",
           "CREATE SCHEMA c;", "SET search_path = c, public;", setup, "ANALYZE;",
           "SET statement_timeout = '20s';", gs]
    view = checks is not None and expect.get('view')
    if view:
        out += ["\\echo @@PHASE view", f"CREATE VIEW v AS {q};",
                "DO $d$ BEGIN EXECUTE 'CREATE VIEW v2 AS ' || pg_get_viewdef('v'::regclass); END $d$;",
                "SELECT 'RT|' || (pg_get_viewdef('v'::regclass) = pg_get_viewdef('v2'::regclass));"]
        qrun = "SELECT * FROM v"
    else:
        qrun = q
    out += ["\\echo @@PHASE explain",
            f"EXPLAIN (ANALYZE, COSTS OFF, TIMING OFF, SUMMARY OFF) {qrun};"]
    if checks is None:
        out += ["\\echo @@PHASE query", f"{qrun};"]
    else:
        out += ["\\echo @@PHASE query", f"CREATE TEMP TABLE r AS {qrun};",
                "\\echo @@PHASE check", "RESET ALL;", "SET search_path = c, public;",
                "SET statement_timeout = '60s';"] + checks
        if view:
            out += ["DROP TABLE r;", "CREATE TEMP TABLE r AS SELECT * FROM v2;"] + checks
    out.append("\\echo @@PHASE end")
    return family, guc, q, '\n'.join(out) + '\n', expect


# ------------------------------------------------------------------- checking

def parse_phases(text):
    ph, cur = {}, 'pre'
    for line in text.splitlines():
        if line.startswith('@@PHASE '):
            cur = line.split()[1]
            ph.setdefault(cur, [])
            continue
        ph.setdefault(cur, []).append(line)
    return ph


ERR_RE = re.compile(r'ERROR:\s+([0-9A-Z]{5}):\s*(.*)')


def check_recursive(lines, expect):
    rows = [l for l in lines if '|' in l and not l.startswith('psql:')]
    got, problems = {}, []
    for l in rows:
        parts = l.split('|')
        node, cost = parts[0], parts[1]
        if node in got:
            problems.append(f'duplicate key {node}')
        got[node] = (Decimal(cost), parts[2] if len(parts) > 2 else None)
    want = {k: Decimal(v) for k, v in expect['dist'].items()}
    if set(got) != set(want):
        problems.append(f'key set differs: missing={sorted(set(want) - set(got))[:5]} '
                        f'extra={sorted(set(got) - set(want))[:5]}')
    for k in set(got) & set(want):
        if got[k][0] != want[k]:
            problems.append(f'cost {k}: got {got[k][0]} want {want[k]}')
    if expect['path']:
        wmin = {}
        for u, v, w in expect['edges']:
            wmin[(u, v)] = min(wmin.get((u, v), Decimal(w)), Decimal(w))
        seedc = {}
        for s, c in expect['seeds']:
            seedc[s] = min(seedc.get(s, Decimal(c)), Decimal(c))
        for k, (c, p) in got.items():
            nodes = p.strip('{}').split(',')
            if nodes[-1] != k or nodes[0] not in seedc:
                problems.append(f'path {k}: bad endpoints {p[:80]}')
                continue
            tot = seedc[nodes[0]]
            ok = True
            for x, y in zip(nodes, nodes[1:]):
                if (x, y) not in wmin:
                    ok = False
                    break
                tot += wmin[(x, y)]
            if not ok or tot > c:
                problems.append(f'path {k}: invalid or costlier than reported ({p[:80]})')
    return problems


def judge(family, text, expect, logtail, elapsed):
    ph = parse_phases(text)
    res = {'plan': [], 'errors': []}
    plan = '\n'.join(ph.get('explain', []))
    for tag in ('HashAggregate', 'Unique', 'Sort', 'Incremental Sort', 'Merge Append', 'Append',
                'Gather', 'Recursive Union', 'SetOp', 'Disk Usage', 'Batches', 'WindowAgg',
                'Limit', 'Index'):
        if tag in plan:
            res['plan'].append(tag)
    if any(m in text or m in logtail for m in ('Disk quota exceeded', 'No space left on device')):
        res['status'] = 'ENV_ERROR'          # tmpfs full: an environment fault, rerun the seed
        return res
    if any(m in logtail for m in LOG_MARKS):
        res['status'] = 'SANITIZER_OR_TRAP'
        return res
    if any(m in text for m in CRASH_MARKS):
        res['status'] = 'CRASH'
        return res
    for name in ('setup', 'view', 'explain', 'query', 'check'):
        for l in ph.get(name, []):
            m = ERR_RE.search(l)
            if m:
                res['errors'].append((name, m.group(1), m.group(2)[:200]))
    errs = res['errors']
    if any(e[1].startswith('XX') for e in errs):
        res['status'] = 'INTERNAL_ERROR'
        return res
    if any(e[0] == 'setup' for e in errs):
        res['status'] = 'HARNESS_ERROR'
        return res
    if any(e[1] == '57014' for e in errs):
        res['status'] = 'TIMEOUT'
        return res
    if any(e[0] in ('view', 'explain', 'query') for e in errs):
        res['status'] = 'REJECTED'
        return res
    if any(e[0] == 'check' for e in errs):
        res['status'] = 'HARNESS_ERROR'
        return res
    if family.startswith('recursive'):
        probs = check_recursive(ph.get('query', []), expect)
        res['status'] = 'DIVERGE' if probs else 'OK'
        if probs:
            res['detail'] = probs[:10]
        return res
    bad = []
    nchk = 0
    for l in ph.get('check', []) + ph.get('view', []):
        if l.startswith('CHK|'):
            nchk += 1
            _, n_r, n_k, ids_ok, rows_ok, best_ok = l.split('|')
            if not (n_r == n_k and ids_ok == 'true' and rows_ok == 'true' and best_ok == 'true'):
                bad.append(l)
        elif l.startswith('EXA|'):
            nchk += 1
            if not l.endswith('|true'):
                bad.append(l)
        elif l.startswith('RT|'):
            if l != 'RT|true':
                bad.append('viewdef round trip differs')
    if nchk == 0:
        res['status'] = 'HARNESS_ERROR'
        res['detail'] = ['no check output']
        return res
    res['status'] = 'DIVERGE' if bad else 'OK'
    if bad:
        res['detail'] = bad
    return res


# -------------------------------------------------------------------- cluster

class Cluster:
    def __init__(self, bindir, data, port):
        self.bindir, self.data, self.port = bindir, data, port
        self.sock = data + '-sock'
        self.log = data + '.log'
        self.env = dict(os.environ,
                        ASAN_OPTIONS='detect_leaks=0:abort_on_error=1:print_stacktrace=1'
                                     ':disable_coredump=1' + os.environ.get('DIFFGEN_ASAN_EXTRA', ''),
                        UBSAN_OPTIONS='print_stacktrace=1:halt_on_error=1',
                        PGHOST=self.sock, PGPORT=str(port), PGUSER='postgres',
                        PGDATABASE='postgres', LC_ALL='C')

    def b(self, x):
        return os.path.join(self.bindir, x)

    def init(self):
        shutil.rmtree(self.data, ignore_errors=True)
        shutil.rmtree(self.sock, ignore_errors=True)
        os.makedirs(self.sock)
        subprocess.run([self.b('initdb'), '-D', self.data, '-U', 'postgres', '--no-sync',
                        '-E', 'UTF8', '--locale=C', '--no-instructions'],
                       env=self.env, check=True, stdout=subprocess.DEVNULL)
        with open(os.path.join(self.data, 'postgresql.conf'), 'a') as f:
            f.write(f"port = {self.port}\nunix_socket_directories = '{self.sock}'\n"
                    "listen_addresses = ''\nfsync = off\nsynchronous_commit = off\n"
                    "full_page_writes = off\nshared_buffers = 128MB\nmax_connections = 10\n"
                    "log_min_messages = warning\nlog_line_prefix = '%m [%p] '\n"
                    "max_parallel_workers = 4\nmax_worker_processes = 8\n")

    def start(self):
        subprocess.run([self.b('pg_ctl'), '-D', self.data, '-l', self.log, '-w', '-t', '120',
                        'start'], env=self.env, check=True, stdout=subprocess.DEVNULL)

    def stop(self):
        subprocess.run([self.b('pg_ctl'), '-D', self.data, '-m', 'immediate', '-w', 'stop'],
                       env=self.env, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)

    def wait_ready(self):
        for _ in range(240):
            r = subprocess.run([self.b('pg_isready'), '-q'], env=self.env)
            if r.returncode == 0:
                return True
            time.sleep(0.5)
        return False

    def psql(self, sql, timeout=300):
        try:
            p = subprocess.run([self.b('psql'), '-X', '-q', '-A', '-t', '-F', '|',
                                '-v', 'ON_ERROR_STOP=0', '-f', '-'],
                               input=sql, env=self.env, stdout=subprocess.PIPE,
                               stderr=subprocess.STDOUT, text=True,   # one pipe: errors stay
                               timeout=timeout, errors='replace')     # inside their phase
            return p.stdout
        except subprocess.TimeoutExpired:
            return '@@PHASE end\nERROR:  57014: harness wall-clock timeout'


def worker(args, w):
    # unique per port: a second run of the same variant must never touch a live cluster
    # (measured 2026-10-17: a one-seed test with the same path wiped NO5 worker 0 mid-run)
    data = f"/dev/shm/hannu-{args.variant}-p{args.port0 + w}"
    cl = Cluster(args.bindir, data, args.port0 + w)
    cl.init()
    cl.start()
    cl.psql(SETUP_SQL)
    fails = os.path.join(args.out, 'fail')
    os.makedirs(fails, exist_ok=True)
    resf = open(os.path.join(args.out, f'results-w{w}.jsonl'), 'a')
    t0 = time.time()
    fams = args.families.split(',') if args.families else None
    for seed in range(args.seed0 + w, args.seed0 + args.cases, args.workers):
        if args.budget and time.time() - t0 > args.budget:
            break
        try:
            family, guc, q, sql, expect = case_sql(seed, fams)
        except Exception as ex:                  # generator bug: record it, keep the worker alive
            resf.write(json.dumps({'seed': seed, 'variant': args.variant, 'family': '?',
                                   'status': 'HARNESS_ERROR', 'errors': [['gen', '', repr(ex)]],
                                   'ms': 0, 'plan': [], 'guc': {}}) + '\n')
            resf.flush()
            continue
        logpos = os.path.getsize(cl.log) if os.path.exists(cl.log) else 0
        ts = time.time()
        out = cl.psql(sql)
        el = time.time() - ts
        with open(cl.log, errors='replace') as f:
            f.seek(logpos)
            tail = f.read()
        res = judge(family, out, expect, tail, el)
        rec = {'seed': seed, 'variant': args.variant, 'family': family, 'guc': guc,
               'status': res['status'], 'ms': int(el * 1000), 'plan': res['plan'],
               'errors': res['errors'][:3]}
        if 'detail' in res:
            rec['detail'] = res['detail']
        if res['status'] not in ('OK', 'REJECTED'):
            base = os.path.join(fails, f"{args.variant}-{family}-{seed}")
            with open(base + '.sql', 'w') as f:
                f.write(f"-- seed {seed} family {family} variant {args.variant}\n"
                        f"-- query: {q}\n" + sql)
            with open(base + '.out', 'w') as f:
                f.write(out)
            if tail.strip():
                with open(base + '.log', 'w') as f:
                    f.write(tail)
        resf.write(json.dumps(rec) + '\n')
        resf.flush()
        if res['status'] in ('CRASH', 'SANITIZER_OR_TRAP', 'HARNESS_ERROR'):
            # fresh cluster after any crash: fsync=off + full_page_writes=off make crash
            # recovery unreliable, and one damaged catalog would poison every later seed
            cl.stop()
            cl.init()
            cl.start()
            cl.psql(SETUP_SQL)
    cl.stop()
    shutil.rmtree(data, ignore_errors=True)
    shutil.rmtree(cl.sock, ignore_errors=True)


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument('--bindir', required=True)
    ap.add_argument('--variant', default='x')
    ap.add_argument('--out')
    ap.add_argument('--seed0', type=int, default=0)
    ap.add_argument('--cases', type=int, default=100)
    ap.add_argument('--workers', type=int, default=4)
    ap.add_argument('--budget', type=float, default=0)
    ap.add_argument('--port0', type=int, default=61500)
    ap.add_argument('--families', default='')
    ap.add_argument('--replay', type=int)
    args = ap.parse_args()
    if args.replay is not None:
        fams = args.families.split(',') if args.families else None
        print(case_sql(args.replay, fams)[3])
        return
    os.makedirs(args.out, exist_ok=True)
    ps = [Process(target=worker, args=(args, w)) for w in range(args.workers)]
    for p in ps:
        p.start()
    for p in ps:
        p.join()


if __name__ == '__main__':
    main()

===== file: perf24.py =====
#!/usr/bin/env python3
"""Cardinality ladder for the many-groups follow-up (2026-10-24).

psql's \\timing of the measured statement, a fresh backend per run.  Each build's bin
directory comes from the environment: PGBIN_base, PGBIN_v3, PGBIN_d1, ... (any names).
Data directories go under $PGDATA_ROOT (default /tmp).

  perf24.py start <build>          initdb + start + load
  perf24.py one   <build> <qid>    one timed run; prints ms as last field
  perf24.py plan  <build> <qid>    EXPLAIN (costs) with the query's GUCs
  perf24.py explain <build> <qid>  EXPLAIN ANALYZE (TIMING OFF)
  perf24.py stop  <build>
  perf24.py list [base]            query ids (base: only those valid on master)

Data: big1, big2: 1M rows each.  gK: K distinct values, uniform.  zK: K values, skewed
(K * random()^3: a few huge groups and a long tail).  bigw: 1M rows with a 192-byte text
column w, for the wide-tuple variant.  Queries return count(*) and sum(s).
"""
import os
import re
import shutil
import subprocess
import sys

BUILDS = sorted(k[len('PGBIN_'):] for k in os.environ if k.startswith('PGBIN_'))
PORT = {b: int(os.environ.get('PGPORT0', '62010')) + i for i, b in enumerate(BUILDS)}
UNIFORM = [1000, 10000, 50000, 100000, 200000, 500000]
SKEWED = [10000, 200000]

LOAD = """
SET client_min_messages = warning;
SELECT setseed(0.17);
CREATE TABLE big1 AS SELECT g AS id, %(cols)s, (random() * 1000000)::int AS s, md5(g::text) AS pl
  FROM generate_series(1, 1000000) g;
CREATE TABLE big2 AS SELECT g + 1000000 AS id, %(cols)s, (random() * 1000000)::int AS s,
  md5(g::text) AS pl FROM generate_series(1, 1000000) g;
SELECT setseed(0.29);
CREATE TABLE bigw AS SELECT g AS id, (random() * 999)::int AS g1000,
  (random() * 199999)::int AS g200000, (random() * 1000000)::int AS s,
  repeat(md5(g::text), 6) AS w FROM generate_series(1, 1000000) g;
VACUUM ANALYZE;
""" % {'cols': ', '.join([f"(random() * {k - 1})::int AS g{k}" for k in UNIFORM] +
                         [f"(({k} - 1) * power(random(), 3))::int AS z{k}" for k in SKEWED])}

W = {'work_mem': "'64MB'", 'max_parallel_workers_per_gather': '0'}


def queries():
    q = {}
    cols = [f'g{k}' for k in UNIFORM] + [f'z{k}' for k in SKEWED]
    for c in cols:
        for h in ('on', 'off'):
            gu = dict(W, enable_hashagg=h)
            q[f'inline_{c}_hash{h}'] = (gu, f"SELECT count(*), sum(s) FROM (SELECT DISTINCT ON "
                                           f"({c} ORDER BY s DESC, id) {c}, s, id FROM big1) x")
            q[f'union_new_{c}_hash{h}'] = (gu, f"SELECT count(*), sum(s) FROM (SELECT {c}, s, id "
                                              f"FROM big1 UNION DISTINCT ON ({c} ORDER BY s DESC, id) "
                                              f"SELECT {c}, s, id FROM big2) x")
        q[f'classic_{c}'] = (W, f"SELECT count(*), sum(s) FROM (SELECT DISTINCT ON ({c}) {c}, s, id "
                                f"FROM big1 ORDER BY {c}, s DESC, id) x")
        q[f'union_classic_{c}'] = (W, f"SELECT count(*), sum(s) FROM (SELECT DISTINCT ON ({c}) {c}, s, "
                                      f"id FROM (SELECT {c}, s, id FROM big1 UNION ALL SELECT {c}, s, id "
                                      f"FROM big2) u ORDER BY {c}, s DESC, id) x")
    # wide tuple: the representative carries a 192-byte text
    for k in (1000, 200000):
        c = f'g{k}'
        for h in ('on', 'off'):
            q[f'wide_{c}_hash{h}'] = (dict(W, enable_hashagg=h),
                                      f"SELECT count(*), sum(s), sum(length(w)) FROM (SELECT DISTINCT ON "
                                      f"({c} ORDER BY s DESC, id) {c}, s, id, w FROM bigw) x")
        q[f'wide_classic_{c}'] = (W, f"SELECT count(*), sum(s), sum(length(w)) FROM (SELECT DISTINCT ON "
                                     f"({c}) {c}, s, id, w FROM bigw ORDER BY {c}, s DESC, id) x")
    # small work_mem: the hash table spills (0005 changed that path)
    S = {'work_mem': "'4MB'", 'max_parallel_workers_per_gather': '0'}
    for k in (200000, 500000):
        c = f'g{k}'
        for h in ('on', 'off'):
            q[f'inline_{c}_wm4_hash{h}'] = (dict(S, enable_hashagg=h),
                                            f"SELECT count(*), sum(s) FROM (SELECT DISTINCT ON "
                                            f"({c} ORDER BY s DESC, id) {c}, s, id FROM big1) x")
            q[f'union_new_{c}_wm4_hash{h}'] = (dict(S, enable_hashagg=h),
                                               f"SELECT count(*), sum(s) FROM (SELECT {c}, s, id FROM big1 "
                                               f"UNION DISTINCT ON ({c} ORDER BY s DESC, id) "
                                               f"SELECT {c}, s, id FROM big2) x")
        q[f'classic_{c}_wm4'] = (S, f"SELECT count(*), sum(s) FROM (SELECT DISTINCT ON ({c}) {c}, s, id "
                                    f"FROM big1 ORDER BY {c}, s DESC, id) x")
        q[f'union_classic_{c}_wm4'] = (S, f"SELECT count(*), sum(s) FROM (SELECT DISTINCT ON ({c}) {c}, "
                                          f"s, id FROM (SELECT {c}, s, id FROM big1 UNION ALL SELECT {c}, "
                                          f"s, id FROM big2) u ORDER BY {c}, s DESC, id) x")
    return q


BASE_OK = ('classic_', 'union_classic_', 'wide_classic_')


def env(b):
    d = os.path.join(os.environ.get('PGDATA_ROOT', '/tmp'), f"ladder-{b}")
    return d, dict(os.environ, PGHOST=d + '-sock', PGPORT=str(PORT[b]), PGUSER='postgres',
                   PGDATABASE='postgres', LC_ALL='C')


def binp(b, x):
    return os.path.join(os.environ[f'PGBIN_{b}'], x)


def psql(b, script, timeout=None):
    d, e = env(b)
    p = subprocess.run([binp(b, 'psql'), '-X', '-q', '-A', '-t', '-v', 'ON_ERROR_STOP=0', '-f', '-'],
                       input=script, env=e, stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
                       text=True, timeout=timeout)
    return p.stdout


def start(b):
    d, e = env(b)
    shutil.rmtree(d, ignore_errors=True)
    shutil.rmtree(d + '-sock', ignore_errors=True)
    os.makedirs(d + '-sock')
    subprocess.run([binp(b, 'initdb'), '-D', d, '-U', 'postgres', '--no-sync', '-E', 'UTF8',
                    '--locale=C', '--no-instructions'], env=e, check=True,
                   stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
    with open(d + '/postgresql.conf', 'a') as f:
        f.write(f"port = {PORT[b]}\nunix_socket_directories = '{d}-sock'\nlisten_addresses = ''\n"
                "shared_buffers = 1GB\nfsync = off\nsynchronous_commit = off\nmax_wal_size = 8GB\n"
                "jit = off\nautovacuum = off\nmax_parallel_workers_per_gather = 0\n"
                "temp_file_limit = '4GB'\n")
    subprocess.run([binp(b, 'pg_ctl'), '-D', d, '-l', d + '.log', '-w', 'start'], env=e, check=True,
                   stdout=subprocess.DEVNULL)
    out = psql(b, LOAD)
    if 'ERROR' in out:
        print(out)
        sys.exit(1)
    psql(b, "SELECT count(*) FROM big1; SELECT count(*) FROM big2; SELECT count(*) FROM bigw;")
    print('started', b)


def stop(b):
    d, e = env(b)
    subprocess.run([binp(b, 'pg_ctl'), '-D', d, '-m', 'fast', '-w', 'stop'], env=e,
                   stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
    shutil.rmtree(d, ignore_errors=True)
    shutil.rmtree(d + '-sock', ignore_errors=True)
    if os.path.exists(d + '.log'):
        os.remove(d + '.log')


def script(gucs, sql, cap):
    s = ''.join(f"SET {k} = {val};\n" for k, val in gucs.items())
    if cap:
        s += f"SET statement_timeout = '{cap}s';\n"
    return s + "\\timing on\n" + sql + ";\n"


TIME_RE = re.compile(r'^Time: ([0-9.]+) ms', re.M)


def one(b, qid):
    gucs, sql = queries()[qid]
    out = psql(b, script(gucs, sql, 600))
    t = TIME_RE.findall(out)
    if not t or 'ERROR' in out:
        print('FAILED', out.strip().replace('\n', ' ')[:300])
        sys.exit(1)
    res = [l for l in out.splitlines() if '|' in l]
    print(qid, res[-1] if res else '', t[-1])


def plan(b, qid, analyze=False):
    gucs, sql = queries()[qid]
    opt = "ANALYZE, BUFFERS OFF, TIMING OFF" if analyze else "COSTS ON"
    out = psql(b, script(gucs, f"EXPLAIN ({opt}) " + sql, 300))
    print(f"== {b} {qid}")
    print('\n'.join(l for l in out.splitlines() if not l.startswith('Time:')))


if __name__ == '__main__':
    cmd = sys.argv[1]
    if cmd == 'list':
        for k in queries():
            if len(sys.argv) < 3 or k.startswith(BASE_OK):
                print(k)
    elif cmd == 'start':
        start(sys.argv[2])
    elif cmd == 'stop':
        stop(sys.argv[2])
    elif cmd == 'one':
        one(sys.argv[2], sys.argv[3])
    elif cmd == 'plan':
        plan(sys.argv[2], sys.argv[3])
    elif cmd == 'explain':
        plan(sys.argv[2], sys.argv[3], analyze=True)

===== file: ladder.sh =====
#!/bin/bash
# Run every cell of a ladder N times, alternated (one sample per cell per round, the order
# rotated by one each round, so drift hits all cells alike), then print median and a
# bootstrap 95% CI per cell.  Each cell is "build:query_id" as listed by perf24.py.
#   PGBIN_base=... PGBIN_v3=... ./ladder.sh OUTDIR N base:classic_g200000 v3:inline_g200000_hashon ...
# Our runs additionally pinned everything to one CCX and discarded rounds whose
# calibration micro-benchmark ran >8% slower than idle; that sentinel is not included.
set -u
H=$(cd "$(dirname "$0")" && pwd); OUT=$1; N=$2; shift 2; CELLS=("$@"); mkdir -p "$OUT"
BS=$(printf '%s\n' "${CELLS[@]}" | cut -d: -f1 | sort -u)
trap 'for b in $BS; do python3 "$H/perf24.py" stop $b; done' EXIT
for b in $BS; do python3 "$H/perf24.py" start $b || exit 1; done
n=${#CELLS[@]}
for r in $(seq 0 $((N - 1))); do
  for i in $(seq 0 $((n - 1))); do
    c=${CELLS[$(( (i + r) % n ))]}
    python3 "$H/perf24.py" one "${c%%:*}" "${c#*:}" | awk '{print $NF}' >> "$OUT/${c/:/-}.txt"
  done
done
python3 - "$OUT" "${CELLS[@]}" <<'PY'
import random, statistics, sys
out, cells = sys.argv[1], sys.argv[2:]
rng = random.Random(0)
for c in cells:
    xs = [float(l) for l in open(f"{out}/{c.replace(':', '-', 1)}.txt") if l.strip()]
    meds = sorted(statistics.median(rng.choices(xs, k=len(xs))) for _ in range(10000))
    print(f"{c:40s} median {statistics.median(xs):9.2f} ms  CI95 {meds[250]:.2f}..{meds[9750]:.2f}  n={len(xs)}")
PY

===== file: rec24.py =====
#!/usr/bin/env python3
"""Where does recursive UNION DISTINCT ON stop finishing?  And the unnamed_subquery HINT.
Order of magnitude only (one run each, 120 s cap, temp_file_limit 4GB); not timing claims.
  rec24.py <build> <outfile>        build as in perf24.py (PGBIN_<build>)
"""
import os
import re
import sys

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import perf24  # noqa: E402

# own data directories and ports, so it can run next to ladder.sh
_env = perf24.env


def _rec_env(b):
    d, e = _env(b)
    d = d.replace('ladder-', 'rec-')
    e = dict(e, PGHOST=d + '-sock', PGPORT=str(62150 + list(perf24.BUILDS).index(b)))
    return d, e


perf24.env = _rec_env

LOAD = r"""
SET client_min_messages = warning;
CREATE TABLE grid_e (n int, src int, dst int, w int);
DO $$ DECLARE n int; BEGIN
  FOREACH n IN ARRAY ARRAY[200, 500, 1000] LOOP
    INSERT INTO grid_e SELECT n, r * n + c + 1, (r + m.dr) * n + (c + m.dc) + 1,
           ((r * 7 + c * 13 + m.d * 5) % 9) + 1
    FROM generate_series(0, n - 1) r, generate_series(0, n - 1) c,
         (VALUES (1, 0, 1), (0, 1, 2)) AS m(dc, dr, d)
    WHERE r + m.dr <= n - 1 AND c + m.dc <= n - 1;
  END LOOP; END $$;
INSERT INTO grid_e SELECT n, dst, src, w FROM grid_e;
CREATE INDEX ON grid_e (n, src);
SELECT setseed(0.42);
CREATE TABLE rand_e (v int, src int, dst int, w int);
DO $$ DECLARE v int; BEGIN
  FOREACH v IN ARRAY ARRAY[100000, 300000, 1000000] LOOP
    INSERT INTO rand_e SELECT v, 1 + floor(random() * v)::int, 1 + floor(random() * v)::int,
                              1 + floor(random() * 100)::int FROM generate_series(1, v * 4);
  END LOOP; END $$;
CREATE INDEX ON rand_e (v, src);
-- adversarial: complete DAG, edge i->j costs (j-i)^2, so the cheapest path to j is the
-- unit chain and every earlier wave delivers a worse cost: node j improves ~j times
CREATE TABLE dag_e (n int, src int, dst int, w int);
INSERT INTO dag_e SELECT n, i, j, (j - i) * (j - i)
  FROM unnest(ARRAY[100, 200, 400, 800]) n, generate_series(1, n) i, generate_series(1, n) j
  WHERE i < j;
CREATE INDEX ON dag_e (n, src);
-- negative cycle 1 -> 2 -> 3 -> 1, total -1
CREATE TABLE neg_e (n int, src int, dst int, w int);
INSERT INTO neg_e VALUES (1, 1, 2, 1), (1, 2, 3, 1), (1, 3, 1, -3);
CREATE TABLE ge (src int, dst int, cost int);
INSERT INTO ge VALUES (1, 2, 5);
VACUUM ANALYZE;
"""


def sp(tab, col, val):
    return (f"WITH RECURSIVE sp(node, cost) AS (SELECT 1 AS node, 0 AS cost "
            f"UNION DISTINCT ON (node ORDER BY cost) SELECT e.dst, sp.cost + e.w "
            f"FROM sp JOIN {tab} e ON e.{col} = {val} AND e.src = sp.node) "
            f"SELECT count(*), sum(cost) FROM sp")


CASES = ([(f'grid{n}', 'grid_e', 'n', n, 120) for n in (200, 500, 1000)] +
         [(f'rand{v}', 'rand_e', 'v', v, 120) for v in (100000, 300000, 1000000)] +
         [(f'dag{n}', 'dag_e', 'n', n, 120) for n in (100, 200, 400, 800)] +
         [('negcycle', 'neg_e', 'n', 1, 20)])

HINTS = [
    ('H1 recursive, seed without names',
     "WITH RECURSIVE s(node, cost) AS (SELECT 1, 0 UNION DISTINCT ON (node ORDER BY cost) "
     "SELECT e.dst, s.cost + e.cost FROM s JOIN ge e ON s.node = e.src) SELECT * FROM s"),
    ('H2 non-recursive, branch 2 has a column "cost"',
     "SELECT 1 AS node, 0 AS c UNION DISTINCT ON (node ORDER BY cost) SELECT dst, cost FROM ge"),
    ('H3 plain UNION ... ORDER BY, same branches',
     "SELECT 1 AS node, 0 AS c UNION SELECT dst, cost FROM ge ORDER BY cost"),
    ('H4 non-recursive, branch 2 has no "cost"',
     "SELECT 1 AS node, 0 AS c UNION DISTINCT ON (node ORDER BY cost) SELECT dst, src + cost FROM ge"),
    ('H5 plain UNION ... ORDER BY, branch 2 has no "cost"',
     "SELECT 1 AS node, 0 AS c UNION SELECT dst, src + cost FROM ge ORDER BY cost"),
]


def main():
    b, outf = sys.argv[1], sys.argv[2]
    perf24.LOAD = LOAD          # start() loads perf24.LOAD
    perf24.start(b)
    out = []
    try:
        for name, sql in HINTS:
            r = perf24.psql(b, "\\set VERBOSITY default\n" + sql + ";\n", 60)
            out.append(f"== {b} {name}\n{sql}\n{r.strip()}\n")
        if b != 'base':
            for name, tab, col, val, cap in CASES:
                s = f"SET temp_file_limit = '4GB';\nSET statement_timeout = '{cap}s';\n\\timing on\n"
                r = perf24.psql(b, s + sp(tab, col, val) + ";\n", cap + 30)
                t = re.findall(r'^Time: ([0-9.]+) ms', r, re.M)
                res = [l for l in r.splitlines() if '|' in l]
                if 'statement timeout' in r:
                    line = f"TIMEOUT >{cap}s"
                elif 'temp_file_limit' in r:
                    line = "TEMP >4GB"
                elif 'ERROR' in r:
                    line = "ERROR " + r.strip().replace('\n', ' ')[:200]
                else:
                    line = f"{t[-1] if t else '?'} ms result={res[-1] if res else ''}"
                    ex = perf24.psql(b, s + "EXPLAIN (ANALYZE, TIMING OFF, BUFFERS OFF) " +
                                     sp(tab, col, val) + ";\n", cap + 30)
                    wt = re.search(r'WorkTable Scan on sp .*?\(actual rows=([0-9.]+) loops=(\d+)\)', ex)
                    ru = re.search(r'Recursive Union \(actual rows=([0-9.]+)', ex)
                    if wt:
                        line += (f" iterations={wt.group(2)} worktable_rows="
                                 f"{int(float(wt.group(1)) * int(wt.group(2)))}"
                                 f" union_rows={ru.group(1) if ru else '?'}")
                out.append(f"{name:10s} {line}")
                print(out[-1], flush=True)
    finally:
        perf24.stop(b)
    open(outf, 'w').write('\n'.join(out) + '\n')


if __name__ == '__main__':
    main()