Hi Hannu,
On Sun, Oct 25, 2026 at 12:48 PM Hannu Krosing <[email protected]>
wrote:
> Did you try to reach the limits where UNION DISTINCT ON fails to finish ?
Not to exhaustion, but the largest shapes I tried all finish on v3,
one run each with a 120 s cap:
grid 1000x1000 (1M nodes): 62.5 s, 1999 iterations
random digraph, 1M nodes, 4M edges: 18.5 s
adversarial DAG, 800 nodes: 11.0 s
The DAG is the slow one per node: edge i->j costs (j-i)^2, so the
cheapest path is the unit chain and every node improves once per
wave; time grows about n^3 (27 ms at 100 nodes, 1.4 s at 400).
The one shape that does not finish is a negative cycle: with no
fixpoint it keeps improving costs until statement_timeout, as
Bellman-Ford without cycle detection. If there are shapes you care
about, I can run them.
> Did you have you any insights about why the reversal of relative
> performance happens at 200k groups ?
>
> Is it about Hash vs Ordered grouping behaviour or something else?
Both: hash vs ordered, plus an estimate problem on the UNION side,
which the attached diff fixes. I re-measured on master 126976a8ff with
v3, on a ladder of 1k to 500k groups, in two independent passes with
all cells alternated. Percentages below are first pass / second pass;
the tables are in measurements.txt.
UNION DISTINCT ON plans its HashAggregate for as many groups as there
are input rows. build_setop_child_paths() estimates each branch over
its whole target list (prepunion.c:683 with v3 applied), and with a
unique id every branch looks fully distinct, so the estimate is
2,000,000 whatever the key and the hash table is sized for it: 40 MB
with 200k actual groups, against 16 MB with the diff, which takes the
estimate over the DISTINCT ON key only, summed over the branches as
for a plain UNION. Against the classic rewrite over UNION ALL, at
200k groups:
v3 as posted: +6.9% / -1.5% (the second within noise)
v3 with the diff: -14.5% / -21.2%
and with the diff -44% / -46% at 100k groups, -68% / -68% at 1k. The
other callers of build_setop_child_paths() pass NIL and behave as
before. make check and isolation pass on -O2 and on cassert+ASan, and
the oracle from my first mail gives 4,500 of 4,500 cases OK.
On the inline side the reversal from my October 17 mail does not
reproduce on 126976a8ff with this data: the HashAgg path is
-25.1% / -30.6% against classic DISTINCT ON at 200k groups, and
roughly even at 500k (+5.2% / +0.1%).
Where hash still loses, it is hash vs ordered:
UNION at 500k groups, with the diff: +26% / +22%
inline at 200k, work_mem = 4MB (the table spills): +14% / +10%
inline at 500k, work_mem = 4MB: +36% / +34%
Counted with perf stat, the hash path runs fewer instructions than the
sort but takes more cache misses once the table outgrows the L3 (16 MB
here; 500k groups is 36 MB even with the diff). The planner cannot see
that: in cost_agg() (costsize.c:2889-2900) the in-memory hashed cost
grows with the input rows only, so it picks HashAggregate at every
size on the ladder, at a fifth to a third of the Sort+Unique cost. A
plain GROUP BY on master is costed the same way, so I would treat the
remaining crossover as master's costing rather than something for v3.
> I wonder if there is a nice way to fix it ?
>
> What would be a good name for the unnamed first branch to use in error
> messages ?
It turns out not to come from v3. With ge(src, dst, cost), plain
master gives the same HINT for
SELECT 1 AS node, 0 AS c UNION SELECT dst, src + cost FROM ge
ORDER BY cost;
The name comes from 585e31fcb6d (September 2025), which replaced the
"*SELECT* n" aliases with generated names. A set operation's branches
can never be referenced from its ORDER BY, so I think the hint should
not offer them at all, and a DETAIL could name the branch by position
("the second UNION branch"). That is master code, a small thread of
its own; I can put a patch together if you think it is worth it.
> Thanks, I will very likely ask for them at some point.
It is attached as harness.txt: the differential oracle, the ladder
above, a small runner for alternated rounds, and the recursive cases.
The diff is yours to fold into your next version however you like.
Regards,
Manu
diff --git a/src/backend/optimizer/prep/prepunion.c
b/src/backend/optimizer/prep/prepunion.c
index aea0677d2b..33b62561ef 100644
--- a/src/backend/optimizer/prep/prepunion.c
+++ b/src/backend/optimizer/prep/prepunion.c
@@ -55,6 +55,7 @@ static RelOptInfo *generate_recursion_path(SetOperationStmt
*setOp,
static void build_setop_child_paths(PlannerInfo *root, RelOptInfo *rel,
bool
trivial_tlist, List *child_tlist,
List
*interesting_pathkeys,
+ List
*groupList,
double
*pNumGroups);
static RelOptInfo *generate_union_paths(SetOperationStmt *op, PlannerInfo
*root,
List *refnames_tlist,
@@ -399,7 +400,7 @@ generate_recursion_path(SetOperationStmt *setOp,
PlannerInfo *root,
&lpath_trivial_tlist);
if (lrel->rtekind == RTE_SUBQUERY)
build_setop_child_paths(root, lrel, lpath_trivial_tlist,
lpath_tlist,
- NIL, NULL);
+ NIL, NIL, NULL);
lpath = lrel->cheapest_total_path;
/* The right path will want to look at the left one ... */
root->non_recursive_path = lpath;
@@ -411,7 +412,7 @@ generate_recursion_path(SetOperationStmt *setOp,
PlannerInfo *root,
&rpath_trivial_tlist);
if (rrel->rtekind == RTE_SUBQUERY)
build_setop_child_paths(root, rrel, rpath_trivial_tlist,
rpath_tlist,
- NIL, NULL);
+ NIL, NIL, NULL);
rpath = rrel->cheapest_total_path;
root->non_recursive_path = NULL;
@@ -486,13 +487,18 @@ generate_recursion_path(SetOperationStmt *setOp,
PlannerInfo *root,
*
* interesting_pathkeys: if not NIL, also include paths that suit these
* pathkeys, sorting any unsorted paths as required.
+ * groupList: if not NIL, the SortGroupClauses (tleSortGroupRef = output
+ * column number) that define a group, e.g. the DISTINCT ON key of a UNION
+ * DISTINCT ON; the estimate below is then made over those columns only.
+ * NIL means the whole target list.
* *pNumGroups: if not NULL, we estimate the number of distinct groups
* in the result, and store it there.
*/
static void
build_setop_child_paths(PlannerInfo *root, RelOptInfo *rel,
bool trivial_tlist, List
*child_tlist,
- List *interesting_pathkeys,
double *pNumGroups)
+ List *interesting_pathkeys,
List *groupList,
+ double *pNumGroups)
{
RelOptInfo *final_rel;
List *setop_pathkeys = rel->subroot->setop_pathkeys;
@@ -680,11 +686,26 @@ build_setop_child_paths(PlannerInfo *root, RelOptInfo
*rel,
subquery->hasAggs)
*pNumGroups = rel->cheapest_total_path->rows;
else
+ {
+ List *groupExprs =
get_tlist_exprs(subroot->parse->targetList,
+
false);
+
+ if (groupList != NIL)
+ {
+ List *keyExprs = NIL;
+
+ foreach_node(SortGroupClause, sgc, groupList)
+ keyExprs = lappend(keyExprs,
+
list_nth(groupExprs,
+
sgc->tleSortGroupRef - 1));
+ groupExprs = keyExprs;
+ }
*pNumGroups = estimate_num_groups(subroot,
-
get_tlist_exprs(subroot->parse->targetList, false),
+
groupExprs,
rel->cheapest_total_path->rows,
NULL,
NULL);
+ }
}
}
@@ -786,6 +807,7 @@ generate_union_paths(SetOperationStmt *op, PlannerInfo
*root,
if (rel->rtekind == RTE_SUBQUERY)
build_setop_child_paths(root, rel, trivial_tlist,
child_tlist,
union_pathkeys,
+
op->sortClauses ? groupList : NIL,
op->all
? NULL : &childGroups);
else
childGroups = rel->rows;
@@ -1167,12 +1189,12 @@ generate_nonunion_paths(SetOperationStmt *op,
PlannerInfo *root,
*/
if (lrel->rtekind == RTE_SUBQUERY)
build_setop_child_paths(root, lrel, lpath_trivial_tlist,
lpath_tlist,
-
nonunion_pathkeys, &dLeftGroups);
+
nonunion_pathkeys, NIL, &dLeftGroups);
else
dLeftGroups = lrel->rows;
if (rrel->rtekind == RTE_SUBQUERY)
build_setop_child_paths(root, rrel, rpath_trivial_tlist,
rpath_tlist,
-
nonunion_pathkeys, &dRightGroups);
+
nonunion_pathkeys, NIL, &dRightGroups);
else
dRightGroups = rrel->rows;
Many-groups follow-up: where HashAgg DISTINCT ON wins and loses
Builds: master 126976a8ff + Hannu's v3 0001..0005 ("v3"), the same plus the
attached nocfbot-union-distinct-on-estimate-on-v3.diff.txt ("v3+fix"), and
master alone ("master", for the classic forms). -O2. All in one lane:
4 Zen 5 cores sharing a 16 MB L3; rounds whose calibration micro-benchmark
ran >8% slower than idle were discarded and retried.
Data: two tables of 1,000,000 rows; gK has K distinct values (uniform);
zK is skewed (K * random()^3). work_mem = 64MB unless noted, no parallel.
Inline: SELECT DISTINCT ON (gK ORDER BY s DESC, id) gK, s, id FROM big1
vs classic SELECT DISTINCT ON (gK) ... ORDER BY gK, s DESC, id (master).
UNION: big1 UNION DISTINCT ON (gK ORDER BY s DESC, id) big2
vs classic DISTINCT ON over big1 UNION ALL big2 (master).
Two independent passes, every cell alternated with all the others; each
number below is a median in ms, first pass then second (10 / 12 clean
rounds). The second pass ran ~4% faster across the board, so compare
within a pass; the deltas are median differences with a bootstrap 95% CI,
per pass.
Inline DISTINCT ON (1M rows), ms v3 hash v3 sort classic
g1000 91 90 310 306 310 306
g10000 102 100 280 273 282 276
g50000 115 112 267 257 265 256
g100000 142 127 264 258 265 257
g200000 193 175 258 254 258 252
g500000 267 246 252 247 254 246
z10000 102 99 300 289 293 287
z200000 158 144 267 262 269 261
work_mem 4MB (the hash table spills):
g200000 203 193 178 175
g500000 253 239 186 179
192-byte text in the output:
g1000 102 97 483 569
g200000 350 343 420 412
UNION DISTINCT ON (2 x 1M rows), ms v3 hash v3+fix hash classic
g1000 201 197 186 188 590 590
g10000 231 222 206 204 533 527
g50000 291 263 233 229 491 490
g100000 385 333 275 263 490 486
g200000 511 475 409 380 478 483
g500000 643 606 587 561 466 461
z10000 226 217 202 200 558 552
z200000 442 388 313 295 502 496
Deltas (new vs reference), first pass / second pass:
inline 1k, v3 hash vs classic:
-70.5% [-72.4..-70.1] / -70.5% [-70.9..-70.3]
inline 200k, v3 hash vs classic:
-25.1% [-32.1..-13.6] / -30.6% [-31.8..-29.6]
inline 500k, v3 hash vs classic:
+5.2% [+0.4..+10.4] / +0.1% [-0.9..+1.6]
inline 200k, 4MB, v3 hash vs classic:
+14.2% [+8.9..+22.5] / +10.4% [+8.1..+12.7]
inline 500k, 4MB, v3 hash vs classic:
+35.9% [+30.8..+39.5] / +33.6% [+31.6..+35.1]
UNION 1k, v3+fix vs classic:
-68.4% [-69.7..-62.8] / -68.1% [-68.3..-67.6]
UNION 100k, v3+fix vs classic:
-43.8% [-46.7..-41.4] / -45.9% [-47.3..-45.3]
UNION 200k, v3 vs classic:
+6.9% [+4.6..+16.8] / -1.5% [-4.4..+3.4]
UNION 200k, v3+fix vs classic:
-14.5% [-18.1..-3.6] / -21.2% [-23.3..-15.8]
UNION 200k, v3+fix vs v3:
-20.0% [-27.1..-9.8] / -20.0% [-24.1..-14.1]
UNION 500k, v3+fix vs classic:
+26.0% [+20.6..+35.5] / +21.8% [+20.5..+24.1]
UNION z200k, v3+fix vs classic:
-37.6% [-40.2..-26.8] / -40.5% [-41.4..-37.4]
Tried and dropped, each on top of v3+fix (same passes; the fix does not
touch the inline path, so v3 is the inline reference):
Bump instead of AllocSet for the tuples, inline 200k:
-5.2% [-19.4..+9.3] / -2.6% [-6.6..+0.7]
sort keys cached in the entry, inline 1k:
-6.9% [-7.9..-5.2] / -13.0% [-13.8..-11.4]
sort keys cached in the entry, inline 200k:
+7.5% [-7.1..+20.5] / +10.8% [+9.0..+13.4]
charging the tie-break comparisons in create_agg_path(): no plan changes
anywhere on the ladder (HashAggregate stays at ~1/4 of the Sort+Unique cost).
Planner choice: HashAggregate at every point of the ladder, for v3 and
v3+fix alike; its estimated cost is a fifth to a third of Sort+Unique.
UNION DISTINCT ON, estimated groups and hash table memory (EXPLAIN ANALYZE):
groups K est. v3 est. fix actual memory v3 memory fix
-------- ------- -------- -------- --------- ----------
1000 2000000 2000 1000 32857 kB 153 kB
10000 2000000 19931 10000 33305 kB 1049 kB
50000 2000000 96801 50000 34841 kB 4121 kB
100000 2000000 184533 100000 36889 kB 8217 kB
200000 2000000 338055 199991 40985 kB 16409 kB
500000 2000000 678786 490827 53273 kB 36889 kB
Recursive UNION DISTINCT ON (node ORDER BY cost), shortest path from node 1,
one run each on v3, other lane (8 Zen 5c cores, 8 MB L3), 120 s cap:
grid 200x200 (40k nodes): 441 ms, 399 iterations
grid 500x500 (250k nodes): 7.1 s, 999 iterations
grid 1000x1000 (1M nodes): 62.5 s, 1999 iterations
random, 100k nodes, 400k edges: 1.2 s, 28 iterations
random, 1M nodes, 4M edges: 18.5 s, 33 iterations
adversarial DAG, 100 nodes: 27 ms, 100 iterations
adversarial DAG, 200 nodes: 187 ms, 200 iterations
adversarial DAG, 400 nodes: 1.4 s, 400 iterations
adversarial DAG, 800 nodes: 11.0 s, 800 iterations
negative cycle (3 nodes): never finishes (statement_timeout 20 s)
(adversarial DAG: edge i->j costs (j-i)^2, so the cheapest path is the unit
chain and every node improves once per wave; time grows ~n^3)
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()
