#!/usr/bin/env bash
# Verifies every claim in optimize-tables-vacuum-analyze.html against a throwaway
# PostgreSQL cluster. Touches no existing database. Cluster is removed on exit.
#
#   bash verify-optimize-tables-vacuum-analyze.sh
#
# Runtime is roughly 3-5 minutes, most of it waiting for autovacuum's 60s naptime.

set -u

PATH="/opt/homebrew/opt/postgresql@17/bin:/usr/local/opt/postgresql@17/bin:$PATH"
export PATH

PORT=55511
BASE=$(mktemp -d "${TMPDIR:-/tmp}/hw_vac_verify.XXXXXX")
DATA="$BASE/data"
# The Unix socket path has a 103-byte limit, so keep the socket dir short.
SOCK=$(mktemp -d /tmp/hw_vac_sock.XXXXXX)

PASS=0; FAIL=0; SKIP=0

pass() { echo "PASS $*"; PASS=$((PASS+1)); }
fail() { echo "FAIL $*"; FAIL=$((FAIL+1)); }
skip() { echo "SKIP $*"; SKIP=$((SKIP+1)); }

cleanup() {
  if [ -d "$DATA" ]; then
    pg_ctl -D "$DATA" -m immediate stop >/dev/null 2>&1
  fi
  rm -rf "$BASE" "$SOCK"
}
trap cleanup EXIT

command -v initdb >/dev/null 2>&1 || { echo "SKIP (initdb not found; install postgresql@17)"; echo "PASS=0 FAIL=0 SKIP=1"; exit 0; }

if (echo > /dev/tcp/127.0.0.1/$PORT) >/dev/null 2>&1; then
  echo "SKIP (port $PORT already in use)"
  echo "PASS=0 FAIL=0 SKIP=1"
  exit 0
fi

initdb -D "$DATA" -U postgres --auth=trust >"$BASE/initdb.log" 2>&1 \
  || { echo "SKIP (initdb failed; see $BASE/initdb.log)"; echo "PASS=0 FAIL=0 SKIP=1"; exit 0; }
pg_ctl -D "$DATA" -o "-p $PORT -k $SOCK" -l "$BASE/pg.log" -w start >/dev/null 2>&1 \
  || { echo "SKIP (server would not start; see $BASE/pg.log)"; echo "PASS=0 FAIL=0 SKIP=1"; exit 0; }

PG="psql -h 127.0.0.1 -p $PORT -U postgres -X"
psql -h 127.0.0.1 -p $PORT -U postgres -X -q -c "CREATE DATABASE hw_vac_db;" >/dev/null
Q="psql -h 127.0.0.1 -p $PORT -U postgres -X -q -d hw_vac_db"
QA="psql -h 127.0.0.1 -p $PORT -U postgres -X -q -A -t -d hw_vac_db"

flush() { $QA -c "SELECT pg_stat_force_next_flush();" >/dev/null 2>&1; }

VER=$($QA -c "SHOW server_version;")
case "$VER" in
  17.*) pass "server is PostgreSQL $VER" ;;
  *)    skip "server is PostgreSQL $VER, article was measured on 17.10" ;;
esac

# ---------------------------------------------------------------- interleaved bloat
$Q >/dev/null <<'SQL'
CREATE TABLE hw_vac_gap (id int primary key, payload text);
INSERT INTO hw_vac_gap SELECT g, repeat('x',200) FROM generate_series(1,300000) g;
DELETE FROM hw_vac_gap WHERE id % 2 = 0;
SQL
flush
SZ0=$($QA -c "SELECT pg_relation_size('hw_vac_gap');")
FN0=$($QA -c "SELECT pg_relation_filenode('hw_vac_gap');")
DEAD0=$($QA -c "SELECT n_dead_tup FROM pg_stat_user_tables WHERE relname='hw_vac_gap';")

[ "$DEAD0" -eq 150000 ] \
  && pass "deleting every 2nd row of 300000 leaves n_dead_tup=150000" \
  || fail "expected n_dead_tup=150000, got $DEAD0"

$Q -c "VACUUM hw_vac_gap;" >/dev/null
flush
SZ1=$($QA -c "SELECT pg_relation_size('hw_vac_gap');")
FN1=$($QA -c "SELECT pg_relation_filenode('hw_vac_gap');")
DEAD1=$($QA -c "SELECT n_dead_tup FROM pg_stat_user_tables WHERE relname='hw_vac_gap';")

[ "$DEAD1" -eq 0 ] \
  && pass "plain VACUUM clears dead tuples (n_dead_tup 150000 -> 0)" \
  || fail "plain VACUUM left n_dead_tup=$DEAD1"

[ "$SZ1" -eq "$SZ0" ] \
  && pass "plain VACUUM does not shrink an interleaved-bloat table ($SZ0 bytes before and after)" \
  || fail "plain VACUUM changed size $SZ0 -> $SZ1"

[ "$FN1" = "$FN0" ] \
  && pass "plain VACUUM does not change relfilenode (still $FN0)" \
  || fail "plain VACUUM changed relfilenode $FN0 -> $FN1"

LV0=$($QA -c "SELECT last_vacuum FROM pg_stat_user_tables WHERE relname='hw_vac_gap';")
VC0=$($QA -c "SELECT vacuum_count FROM pg_stat_user_tables WHERE relname='hw_vac_gap';")

$Q -c "VACUUM FULL hw_vac_gap;" >/dev/null
flush
SZ2=$($QA -c "SELECT pg_relation_size('hw_vac_gap');")
FN2=$($QA -c "SELECT pg_relation_filenode('hw_vac_gap');")
LV1=$($QA -c "SELECT last_vacuum FROM pg_stat_user_tables WHERE relname='hw_vac_gap';")
VC1=$($QA -c "SELECT vacuum_count FROM pg_stat_user_tables WHERE relname='hw_vac_gap';")

[ "$SZ2" -lt "$SZ1" ] \
  && pass "VACUUM FULL shrinks the file ($SZ1 -> $SZ2 bytes)" \
  || fail "VACUUM FULL did not shrink the file ($SZ1 -> $SZ2)"

[ "$FN2" != "$FN1" ] \
  && pass "VACUUM FULL rewrites into a new relfilenode ($FN1 -> $FN2)" \
  || fail "VACUUM FULL left relfilenode at $FN1"

if [ "$LV1" = "$LV0" ] && [ "$VC1" = "$VC0" ]; then
  pass "VACUUM FULL does not update last_vacuum/vacuum_count (count stayed $VC0)"
else
  fail "VACUUM FULL changed last_vacuum '$LV0'->'$LV1' or vacuum_count $VC0->$VC1"
fi

# ---------------------------------------------------------------- space is reused
$Q >/dev/null <<'SQL'
CREATE TABLE hw_vac_reuse (id int primary key, payload text);
INSERT INTO hw_vac_reuse SELECT g, repeat('x',200) FROM generate_series(1,300000) g;
UPDATE hw_vac_reuse SET payload = repeat('y',200);
VACUUM hw_vac_reuse;
SQL
RSZ0=$($QA -c "SELECT pg_relation_size('hw_vac_reuse');")
$Q -c "INSERT INTO hw_vac_reuse SELECT g, repeat('z',200) FROM generate_series(300001,600000) g;" >/dev/null
RSZ1=$($QA -c "SELECT pg_relation_size('hw_vac_reuse');")
RCNT=$($QA -c "SELECT count(*) FROM hw_vac_reuse;")

if [ "$RSZ1" -eq "$RSZ0" ] && [ "$RCNT" -eq 600000 ]; then
  pass "vacuumed space is reused: 300000 rows added, file stayed at $RSZ0 bytes, now $RCNT rows"
else
  fail "file grew $RSZ0 -> $RSZ1 while inserting into reclaimed space (count=$RCNT)"
fi

# ---------------------------------------------------------------- tail truncation
$Q >/dev/null <<'SQL'
CREATE TABLE hw_vac_tail (id int primary key, payload text);
INSERT INTO hw_vac_tail SELECT g, repeat('x',200) FROM generate_series(1,300000) g;
SQL
TSZ0=$($QA -c "SELECT pg_relation_size('hw_vac_tail');")
$Q >/dev/null <<'SQL'
DELETE FROM hw_vac_tail WHERE id > 150000;
VACUUM hw_vac_tail;
SQL
TSZ1=$($QA -c "SELECT pg_relation_size('hw_vac_tail');")

[ "$TSZ1" -lt "$TSZ0" ] \
  && pass "plain VACUUM truncates trailing free space ($TSZ0 -> $TSZ1 bytes)" \
  || fail "plain VACUUM did not truncate the tail ($TSZ0 -> $TSZ1)"

# ---------------------------------------------------------------- VACUUM vs statistics
$Q >/dev/null <<'SQL'
CREATE TABLE hw_vac_stats (id int, grp text);
INSERT INTO hw_vac_stats SELECT g, 'g'||(g%7) FROM generate_series(1,100000) g;
SQL
RT0=$($QA -c "SELECT reltuples::int FROM pg_class WHERE relname='hw_vac_stats';")
$Q -c "VACUUM hw_vac_stats;" >/dev/null
flush
RT1=$($QA -c "SELECT reltuples::int FROM pg_class WHERE relname='hw_vac_stats';")
ST1=$($QA -c "SELECT count(*) FROM pg_stats WHERE tablename='hw_vac_stats';")
LA1=$($QA -c "SELECT coalesce(last_analyze::text,'NULL') FROM pg_stat_user_tables WHERE relname='hw_vac_stats';")

if [ "$ST1" -eq 0 ] && [ "$LA1" = "NULL" ]; then
  pass "bare VACUUM collects no planner column statistics (pg_stats rows=0, last_analyze NULL)"
else
  fail "bare VACUUM produced pg_stats rows=$ST1, last_analyze=$LA1"
fi

if [ "$RT0" -eq -1 ] && [ "$RT1" -eq 100000 ]; then
  pass "bare VACUUM does update pg_class.reltuples (-1 -> 100000)"
else
  fail "reltuples went $RT0 -> $RT1, expected -1 -> 100000"
fi

$Q -c "VACUUM ANALYZE hw_vac_stats;" >/dev/null
flush
ST2=$($QA -c "SELECT count(*) FROM pg_stats WHERE tablename='hw_vac_stats';")
ND2=$($QA -c "SELECT n_distinct::int FROM pg_stats WHERE tablename='hw_vac_stats' AND attname='grp';")

if [ "$ST2" -gt 0 ] && [ "$ND2" -eq 7 ]; then
  pass "VACUUM ANALYZE populates pg_stats (rows=$ST2, n_distinct for grp = $ND2)"
else
  fail "VACUUM ANALYZE gave pg_stats rows=$ST2, n_distinct=$ND2"
fi

# ---------------------------------------------------------------- locking
$Q -c "BEGIN; SELECT count(*) FROM hw_vac_gap; SELECT pg_sleep(20);" >/dev/null 2>&1 &
HOLDER=$!
sleep 3
HELD=$($QA -c "SELECT mode FROM pg_locks WHERE relation='hw_vac_gap'::regclass AND mode='AccessShareLock' LIMIT 1;")
[ "$HELD" = "AccessShareLock" ] \
  && pass "an open reading transaction holds AccessShareLock on the table" \
  || skip "could not observe the reader's AccessShareLock (got '$HELD')"

VOUT=$(printf "SET lock_timeout='3s';\nVACUUM hw_vac_gap;\n" | $Q 2>&1)
case "$VOUT" in
  *"lock timeout"*) fail "plain VACUUM was blocked by a concurrent reader: $VOUT" ;;
  *ERROR*)          fail "plain VACUUM errored against a concurrent reader: $VOUT" ;;
  *)                pass "plain VACUUM runs while another session holds ACCESS SHARE" ;;
esac

AOUT=$(printf "SET lock_timeout='3s';\nANALYZE hw_vac_gap;\n" | $Q 2>&1)
case "$AOUT" in
  *ERROR*) fail "ANALYZE was blocked by a concurrent reader: $AOUT" ;;
  *)       pass "ANALYZE runs while another session holds ACCESS SHARE" ;;
esac

FOUT=$(printf "SET lock_timeout='3s';\nVACUUM FULL hw_vac_gap;\n" | $Q 2>&1)
case "$FOUT" in
  *"canceling statement due to lock timeout"*)
    pass "VACUUM FULL is blocked by a concurrent reader (canceling statement due to lock timeout)" ;;
  *) fail "VACUUM FULL was not blocked by a concurrent reader: $FOUT" ;;
esac
wait $HOLDER 2>/dev/null

# observe the actual lock modes while each command runs
$Q >/dev/null <<'SQL'
CREATE TABLE hw_vac_big AS SELECT g AS id, repeat('x',300) AS payload FROM generate_series(1,800000) g;
SQL
poll_lock() {
  local n=0 m=""
  while [ $n -lt 600 ]; do
    m=$($QA -c "SELECT string_agg(DISTINCT mode,',') FROM pg_locks WHERE relation='hw_vac_big'::regclass AND pid <> pg_backend_pid();" 2>/dev/null)
    [ -n "$m" ] && { echo "$m"; return 0; }
    n=$((n+1))
  done
  echo ""
}

( $Q -c "VACUUM FULL hw_vac_big;" >/dev/null 2>&1 ) &
MODE_FULL=$(poll_lock)
wait
case "$MODE_FULL" in
  *AccessExclusiveLock*) pass "VACUUM FULL holds AccessExclusiveLock (observed in pg_locks)" ;;
  "")                    skip "VACUUM FULL finished before pg_locks could be sampled" ;;
  *)                     fail "VACUUM FULL held '$MODE_FULL', expected AccessExclusiveLock" ;;
esac

$Q -c "UPDATE hw_vac_big SET payload=repeat('y',300) WHERE id % 3 = 0;" >/dev/null
( printf "SET vacuum_cost_delay='20ms';\nSET vacuum_cost_limit=10;\nVACUUM hw_vac_big;\n" | $Q >/dev/null 2>&1 ) &
VPID=$!
MODE_VAC=$(poll_lock)
$QA -c "SELECT pg_terminate_backend(pid) FROM pg_stat_activity WHERE backend_type='client backend' AND pid<>pg_backend_pid() AND query LIKE '%hw_vac_big%';" >/dev/null 2>&1
wait $VPID 2>/dev/null
case "$MODE_VAC" in
  *ShareUpdateExclusiveLock*) pass "plain VACUUM holds ShareUpdateExclusiveLock (observed in pg_locks)" ;;
  "")                         skip "plain VACUUM finished before pg_locks could be sampled" ;;
  *)                          fail "plain VACUUM held '$MODE_VAC', expected ShareUpdateExclusiveLock" ;;
esac

# ---------------------------------------------------------------- privileges
$Q >/dev/null <<'SQL'
CREATE ROLE hw_vac_user LOGIN;
GRANT CONNECT ON DATABASE hw_vac_db TO hw_vac_user;
GRANT USAGE ON SCHEMA public TO hw_vac_user;
GRANT SELECT ON hw_vac_stats TO hw_vac_user;
SQL
ISSUPER=$($QA -c "SELECT rolsuper FROM pg_roles WHERE rolname='hw_vac_user';")
DBOWNER=$($QA -c "SELECT pg_get_userbyid(datdba) FROM pg_database WHERE datname='hw_vac_db';")
UOUT=$(psql -h 127.0.0.1 -p $PORT -U hw_vac_user -X -q -d hw_vac_db -c "VACUUM hw_vac_stats;" 2>&1)

if [ "$ISSUPER" = "f" ] && [ "$DBOWNER" != "hw_vac_user" ] && \
   printf '%s' "$UOUT" | grep -q 'permission denied to vacuum'; then
  pass "non-superuser, non-db-owner without MAINTAIN is refused (warning, not error)"
else
  fail "expected a 'permission denied to vacuum' warning, got: $UOUT (super=$ISSUPER dbowner=$DBOWNER)"
fi

flush
MB0=$($QA -c "SELECT vacuum_count FROM pg_stat_user_tables WHERE relname='hw_vac_stats';")
$Q -c "GRANT MAINTAIN ON TABLE hw_vac_stats TO hw_vac_user;" >/dev/null
MOUT=$(psql -h 127.0.0.1 -p $PORT -U hw_vac_user -X -q -d hw_vac_db -c "VACUUM hw_vac_stats;" 2>&1)
flush
MB1=$($QA -c "SELECT vacuum_count FROM pg_stat_user_tables WHERE relname='hw_vac_stats';")

if [ "$MB1" -gt "$MB0" ] && ! printf '%s' "$MOUT" | grep -q 'permission denied'; then
  pass "MAINTAIN alone lets a non-owner vacuum (vacuum_count $MB0 -> $MB1)"
else
  fail "after GRANT MAINTAIN, vacuum_count $MB0 -> $MB1, output: $MOUT"
fi

# ---------------------------------------------------------------- parallel vacuum
$Q >/dev/null <<'SQL'
CREATE TABLE hw_vac_par (id int primary key, a int, b int, payload text);
INSERT INTO hw_vac_par SELECT g, g, g, repeat('p',100) FROM generate_series(1,400000) g;
CREATE INDEX hw_vac_par_a ON hw_vac_par(a);
CREATE INDEX hw_vac_par_b ON hw_vac_par(b);
SQL

PFULL=$(printf "VACUUM (FULL, PARALLEL 2) hw_vac_par;\n" | $Q 2>&1)
case "$PFULL" in
  *"VACUUM FULL cannot be performed in parallel"*)
    pass "VACUUM FULL rejects PARALLEL (VACUUM FULL cannot be performed in parallel)" ;;
  *) fail "expected a parallel/FULL error, got: $PFULL" ;;
esac

workers_for() {   # $1 = reloptions SQL to apply first
  $Q -c "$1" >/dev/null 2>&1
  $Q -c "UPDATE hw_vac_par SET a = a + 1;" >/dev/null
  printf "VACUUM (VERBOSE) hw_vac_par;\n" | $Q 2>&1 \
    | sed -n 's/.*launched \([0-9]*\) parallel vacuum workers.*/\1/p' | head -1
}
W_UNSET=$(workers_for "ALTER TABLE hw_vac_par RESET (parallel_workers);")
W_SET=$(workers_for "ALTER TABLE hw_vac_par SET (parallel_workers = 4);")

if [ -n "$W_UNSET" ] && [ "$W_SET" = "$W_UNSET" ]; then
  pass "parallel_workers storage parameter does not change parallel vacuum workers ($W_UNSET both ways)"
elif [ -z "$W_UNSET" ]; then
  skip "no parallel vacuum workers were launched on this machine, cannot compare"
else
  fail "worker count differed: unset=$W_UNSET, parallel_workers=4 -> $W_SET"
fi

# ---------------------------------------------------------------- autovacuum
AV=$($QA -c "SELECT setting FROM pg_settings WHERE name='autovacuum';")
[ "$AV" = "on" ] \
  && pass "autovacuum is on by default" \
  || fail "autovacuum is '$AV'"

DEFAULTS=$($QA -c "SELECT string_agg(name||'='||setting, ' ' ORDER BY name) FROM pg_settings WHERE name IN ('autovacuum_analyze_scale_factor','autovacuum_analyze_threshold','autovacuum_max_workers','autovacuum_naptime','autovacuum_vacuum_scale_factor','autovacuum_vacuum_threshold');")
EXPECTED="autovacuum_analyze_scale_factor=0.1 autovacuum_analyze_threshold=50 autovacuum_max_workers=3 autovacuum_naptime=60 autovacuum_vacuum_scale_factor=0.2 autovacuum_vacuum_threshold=50"
[ "$DEFAULTS" = "$EXPECTED" ] \
  && pass "shipped autovacuum defaults match the article ($DEFAULTS)" \
  || fail "defaults differ: $DEFAULTS"

$Q >/dev/null <<'SQL'
CREATE TABLE hw_vac_auto (id int primary key, payload text)
  WITH (autovacuum_vacuum_scale_factor = 0, autovacuum_vacuum_threshold = 50);
INSERT INTO hw_vac_auto SELECT g, repeat('a',100) FROM generate_series(1,5000) g;
SQL
$Q -c "UPDATE hw_vac_auto SET payload = repeat('b',100);" >/dev/null
START=$(date +%s)
FIRED=""
for _ in $(seq 1 90); do
  FIRED=$($QA -c "SELECT coalesce(last_autovacuum::text,'') FROM pg_stat_user_tables WHERE relname='hw_vac_auto';")
  [ -n "$FIRED" ] && break
  sleep 2
done
if [ -n "$FIRED" ]; then
  pass "autovacuum ran unprompted after $(( $(date +%s) - START ))s ($FIRED)"
else
  skip "autovacuum did not fire within 180s on this machine"
fi

echo "PASS=$PASS FAIL=$FAIL SKIP=$SKIP"
[ "$FAIL" -eq 0 ]