"""Weekly small measurements, 2026-11-08 (first run of the weekly refill). PostgreSQL 18.6 lab, port 5418."""
import os, sys, subprocess, tempfile
sys.path.insert(0, os.path.expanduser("~/products/c13/minis"))
from lab import *

HERE = os.path.dirname(os.path.abspath(__file__))
setup(os.path.join(HERE, "results.json"))
start()
save_env()

# base table: 2M orders
psql("DROP TABLE IF EXISTS w_orders, w_parent, w_child, w_logged, w_unlogged, w_ins, w_ev, w_upd, w_dd CASCADE;"
     "CREATE TABLE w_orders (id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY, customer_id int NOT NULL, status text NOT NULL, amount int NOT NULL, created_at timestamptz NOT NULL, note text);"
     "INSERT INTO w_orders (customer_id,status,amount,created_at,note) SELECT 1+g%50000, CASE WHEN g%20=0 THEN 'open' ELSE 'done' END, g%1000, timestamptz '2026-01-01' + g*interval '10 seconds', md5(g::text) FROM generate_series(1,2000000) g;"
     "VACUUM ANALYZE w_orders")

# 1. partial index: 5% open rows
psql("CREATE INDEX w_full ON w_orders (customer_id) ; CREATE INDEX w_part ON w_orders (customer_id) WHERE status = 'open'; ANALYZE w_orders")
r = {"full_index_mb": mb("w_full"), "partial_index_mb": mb("w_part"),
     "open_rows": int(q1("select count(*) from w_orders where status='open'"))}
r["with_partial"] = runs("SELECT count(*) FROM w_orders WHERE customer_id = 777 AND status = 'open'")
psql("DROP INDEX w_part")
r["with_full_only"] = runs("SELECT count(*) FROM w_orders WHERE customer_id = 777 AND status = 'open'")
psql("DROP INDEX w_full")
save("partial_index", r)

# 2. BRIN vs B-tree on append-only timestamp, one day range
day = "created_at >= '2026-03-01' AND created_at < '2026-03-02'"
psql("CREATE INDEX w_brin ON w_orders USING brin (created_at)")
b = {"brin_mb": mb("w_brin"), "brin": runs(f"SELECT count(*), sum(amount) FROM w_orders WHERE {day}")}
psql("DROP INDEX w_brin; CREATE INDEX w_bt ON w_orders (created_at)")
b["btree_mb"] = mb("w_bt"); b["btree"] = runs(f"SELECT count(*), sum(amount) FROM w_orders WHERE {day}")
b["rows_in_day"] = int(q1(f"select count(*) from w_orders where {day}"))
psql("DROP INDEX w_bt")
save("brin_vs_btree", b)

# 3. INCLUDE: sum(amount) for one customer
psql("CREATE INDEX w_c ON w_orders (customer_id); VACUUM ANALYZE w_orders")
c = {"plain": runs("SELECT sum(amount) FROM w_orders WHERE customer_id = 4242")}
psql("DROP INDEX w_c; CREATE INDEX w_ci ON w_orders (customer_id) INCLUDE (amount); VACUUM ANALYZE w_orders")
c["include"] = runs("SELECT sum(amount) FROM w_orders WHERE customer_id = 4242")
c["plain_mb_note"] = "see include_mb"; c["include_mb"] = mb("w_ci")
save("include_covering", c)

# 4. visibility map: index-only scan heap fetches before/after VACUUM on freshly loaded table
psql("DROP TABLE IF EXISTS w_vm; CREATE TABLE w_vm (id int, v int); ALTER TABLE w_vm SET (autovacuum_enabled = off);"
     "INSERT INTO w_vm SELECT g, g%100 FROM generate_series(1,2000000) g; CREATE INDEX w_vm_i ON w_vm (id) INCLUDE (v); ANALYZE w_vm")
def ios(sql):
    j = json.loads(psql("SET enable_bitmapscan=off; SET enable_seqscan=off; EXPLAIN (ANALYZE, BUFFERS, TIMING OFF, FORMAT JSON) " + sql))[0]
    p = j["Plan"]
    while p.get("Node Type") not in ("Index Only Scan", "Index Scan") and p.get("Plans"):
        p = p["Plans"][0]
    return {"node": p["Node Type"], "heap_fetches": p.get("Heap Fetches"), "ms": j["Execution Time"],
            "buffers": j["Plan"].get("Shared Hit Blocks", 0) + j["Plan"].get("Shared Read Blocks", 0)}
q = "SELECT sum(v) FROM w_vm WHERE id BETWEEN 100000 AND 300000"
vm = {"before_vacuum": [ios(q) for _ in range(4)][1:]}
psql("VACUUM w_vm")
vm["after_vacuum"] = [ios(q) for _ in range(4)][1:]
psql("DROP TABLE w_vm")
save("visibility_map", vm)

# 5. FK without index on child: delete 200 parents (their children deleted first? no: ON DELETE CASCADE child rows)
def fk_case(with_index):
    psql("DROP TABLE IF EXISTS w_child, w_parent; CREATE TABLE w_parent (id int PRIMARY KEY);"
         "CREATE TABLE w_child (id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY, parent_id int REFERENCES w_parent ON DELETE CASCADE, x int);"
         "INSERT INTO w_parent SELECT generate_series(1,10000); INSERT INTO w_child (parent_id,x) SELECT 1+g%10000, g FROM generate_series(1,1000000) g;"
         + ("CREATE INDEX ON w_child (parent_id);" if with_index else "") + "VACUUM ANALYZE w_parent; VACUUM ANALYZE w_child")
    return timed("DELETE FROM w_parent WHERE id <= 200")
save("fk_no_index_delete", {"no_index_s": [fk_case(False) for _ in range(2)], "with_index_s": [fk_case(True) for _ in range(2)],
                            "child_rows_deleted": 20000, "parents_deleted": 200, "child_rows": 1000000})
psql("DROP TABLE IF EXISTS w_child, w_parent")

# 6. logged vs unlogged insert: time and WAL
def ins(kind):
    psql(f"DROP TABLE IF EXISTS w_l; CREATE {kind} TABLE w_l (id bigint, t text)")
    l = wal_lsn()
    s = timed("INSERT INTO w_l SELECT g, md5(g::text) FROM generate_series(1,2000000) g")
    return {"s": s, "wal_mb": wal_mb_since(l)}
save("unlogged_insert", {"logged": [ins("") for _ in range(3)], "unlogged": [ins("UNLOGGED") for _ in range(3)]})
psql("DROP TABLE IF EXISTS w_l")

# 7. COPY vs multi-row INSERT vs single-row INSERTs (200k rows, one transaction each)
N = 200000
tmp = tempfile.mkdtemp()
csvp = os.path.join(tmp, "rows.csv")
with open(csvp, "w") as f:
    for i in range(1, N + 1):
        f.write(f"{i},row{i}\n")
single = os.path.join(tmp, "single.sql"); multi = os.path.join(tmp, "multi.sql")
with open(single, "w") as f:
    f.write("BEGIN;\n"); [f.write(f"INSERT INTO w_ins VALUES ({i},'row{i}');\n") for i in range(1, N + 1)]; f.write("COMMIT;\n")
with open(multi, "w") as f:
    f.write("BEGIN;\n")
    for s0 in range(1, N + 1, 1000):
        f.write("INSERT INTO w_ins VALUES " + ",".join(f"({i},'row{i}')" for i in range(s0, s0 + 1000)) + ";\n")
    f.write("COMMIT;\n")
def load(kind):
    psql("DROP TABLE IF EXISTS w_ins; CREATE TABLE w_ins (id int, t text)")
    t = time.perf_counter()
    if kind == "copy":
        subprocess.run(BASE + ["-c", f"\\copy w_ins FROM '{csvp}' WITH (FORMAT csv)"], check=True, capture_output=True)
    else:
        subprocess.run(BASE + ["-f", single if kind == "single" else multi], check=True, capture_output=True)
    return round(time.perf_counter() - t, 3)
save("copy_vs_insert", {k: [load(k) for _ in range(3)] for k in ("copy", "multi", "single")} | {"rows": N, "note": "client psql on same machine, unix socket, one transaction each"})
psql("DROP TABLE IF EXISTS w_ins")

# 8. synchronous_commit: 10k single-row commits with pgbench, 1 client
psql("DROP TABLE IF EXISTS w_sc; CREATE TABLE w_sc (id bigint GENERATED ALWAYS AS IDENTITY, t timestamptz)")
scr = os.path.join(tmp, "sc.sql"); open(scr, "w").write("INSERT INTO w_sc (t) VALUES (now());\n")
def pgb(sc):
    env = dict(os.environ, PGOPTIONS=f"-c synchronous_commit={sc}")
    out = subprocess.run([f"{PG}/pgbench", "-n", "-c", "1", "-t", "10000", "-f", scr, "-h", HOST, "-p", "5418", "-U", "postgres", "lab"],
                         capture_output=True, text=True, env=env, check=True).stdout
    tps = [l for l in out.splitlines() if l.startswith("tps")][0]
    lat = [l for l in out.splitlines() if l.startswith("latency average")][0]
    return {"tps_line": tps, "latency_line": lat}
save("synchronous_commit", {"on": [pgb("on") for _ in range(3)], "off": [pgb("off") for _ in range(3)], "fsync": q1("show fsync"), "wal_sync_method": q1("show wal_sync_method")})
psql("DROP TABLE IF EXISTS w_sc")

# 9. top-N heapsort: ORDER BY amount, created_at LIMIT 10 vs no limit (no index)
save("topn_heapsort", {
    "limit10": runs("SELECT id FROM w_orders ORDER BY note LIMIT 10", 3, "SET max_parallel_workers_per_gather=0;"),
    "full_sort": runs("SELECT count(*) FROM (SELECT id FROM w_orders ORDER BY note OFFSET 0) s", 3, "SET max_parallel_workers_per_gather=0;"),
    "limit10_method": psql("SET max_parallel_workers_per_gather=0; EXPLAIN (ANALYZE, TIMING OFF, COSTS OFF) SELECT id FROM w_orders ORDER BY note LIMIT 10")})

# 10. work_mem and sort method
def sortm(wm):
    plan = psql(f"SET max_parallel_workers_per_gather=0; SET work_mem='{wm}'; EXPLAIN (ANALYZE, TIMING OFF, COSTS OFF) SELECT count(*) FROM (SELECT note FROM w_orders ORDER BY note OFFSET 0) s")
    return {"run": runs("SELECT count(*) FROM (SELECT note FROM w_orders ORDER BY note OFFSET 0) s", 3, f"SET max_parallel_workers_per_gather=0; SET work_mem='{wm}';"),
            "sort_method": [l.strip() for l in plan.splitlines() if "Sort Method" in l]}
save("work_mem_sort", {"4MB": sortm("4MB"), "64MB": sortm("64MB"), "256MB": sortm("256MB")})

# 11. parallel workers
save("parallel_workers", {str(n): runs("SELECT status, sum(amount) FROM w_orders WHERE note LIKE '%ab%' GROUP BY status", 3, f"SET max_parallel_workers_per_gather={n};") for n in (0, 2, 4)}
     | {"cpus": subprocess.run(["sysctl", "-n", "hw.ncpu"], capture_output=True, text=True).stdout.strip()})

# 12. UPDATE every row: table size
psql("DROP TABLE IF EXISTS w_upd; CREATE TABLE w_upd AS SELECT * FROM w_orders; ALTER TABLE w_upd SET (autovacuum_enabled = off); VACUUM ANALYZE w_upd")
u = {"before_mb": mb("w_upd")}
u["update_s"] = timed("UPDATE w_upd SET amount = amount + 1"); u["after_update_mb"] = mb("w_upd")
psql("VACUUM w_upd"); u["after_vacuum_mb"] = mb("w_upd")
u["second_update_s"] = timed("UPDATE w_upd SET amount = amount + 1"); u["after_second_update_mb"] = mb("w_upd")
psql("DROP TABLE w_upd")
save("update_all_rows", u)

# 13. B-tree deduplication on a low-cardinality column
psql("CREATE INDEX w_dd_on ON w_orders (status); CREATE INDEX w_dd_off ON w_orders (status) WITH (deduplicate_items = off);"
     "CREATE INDEX w_cust_on ON w_orders (customer_id); CREATE INDEX w_cust_off ON w_orders (customer_id) WITH (deduplicate_items = off)")
save("btree_dedup", {"status_on_mb": mb("w_dd_on"), "status_off_mb": mb("w_dd_off"), "customer_on_mb": mb("w_cust_on"), "customer_off_mb": mb("w_cust_off"),
                     "pk_mb": mb("w_orders_pkey"), "distinct_status": 2, "distinct_customer": 50000, "rows": 2000000})
psql("DROP INDEX w_dd_on, w_dd_off, w_cust_on, w_cust_off")

psql("DROP TABLE IF EXISTS w_orders CASCADE")
stop()
print("done")