#!/bin/bash
# toast-oracle.sh BINDIR_PARENT OUTFILE [fill|read|both] [DATADIR]
#   BINDIR_PARENT is an install prefix (the directory holding bin/).
# Differential oracle for "0001 changes no behavior": builds a corpus of
# TOAST values and reads every one of them through every reader we know of,
# writing one line per (value, reader) to OUTFILE.  Two builds are compared
# with cmp on their OUTFILEs, e.g.
#   bash toast-oracle.sh.txt /path/to/master out-master.txt
#   bash toast-oracle.sh.txt /path/to/master+0001 out-0001.txt
#   cmp out-master.txt out-0001.txt
#   both : initdb, fill, read, stop (default)
#   fill : initdb + fill only, leave the cluster stopped in DATADIR
#   read : start an existing DATADIR (e.g. after pg_upgrade), read, stop
# Corpus: pglz and lz4, external compressed, external uncompressed
# (STORAGE EXTERNAL), inline compressed (STORAGE MAIN), text and bytea,
# sizes around the TOAST chunk size (1996 and 1992 bytes, +-2, times 1..3).
# NOSLOT=1 skips the logical slot (pg_upgrade refuses a slot with pending WAL).
set -eu
I=$1; OUT=$2; MODE=${3:-both}
D=${4:-$(mktemp -d ${TMPDIR:-/tmp}/oracle.XXXXXX)}
P=${P:-$((61000 + RANDOM % 900))}
export ASAN_OPTIONS=detect_leaks=0:abort_on_error=1:log_path=$D/asan
Q() { "$I/bin/psql" -X -qAt -h $D -p $P -U postgres -d postgres -v ON_ERROR_STOP=1 "$@"; }
start() { "$I/bin/pg_ctl" -D $D/data -l $D/log -w -t 120 -o "-p $P -k $D -c listen_addresses=''" start >/dev/null; }
stop() { "$I/bin/pg_ctl" -D $D/data -m fast -w stop >/dev/null; }
fill() {
  "$I/bin/initdb" -D $D/data -A trust --no-sync --no-data-checksums -U postgres -E UTF8 --locale=C >/dev/null
  cat >> $D/data/postgresql.conf <<CONF
shared_buffers = 64MB
max_connections = 20
autovacuum = off
jit = off
wal_level = logical
max_replication_slots = 4
fsync = off
CONF
  start
  Q ${NOSLOT:+-v noslot=1} <<'SQL'
CREATE EXTENSION amcheck;
-- deterministic, poorly compressible filler of exactly n bytes
CREATE FUNCTION noise(seed int, n int) RETURNS text LANGUAGE sql IMMUTABLE AS $$
  SELECT substr(string_agg(md5((seed::bigint * 100003 + i)::text), ''), 1, n)
  FROM generate_series(0, n / 32 + 1) i $$;
-- compressible: a short alphabet, repeated with drift
CREATE FUNCTION soft(seed int, n int) RETURNS text LANGUAGE sql IMMUTABLE AS $$
  SELECT substr(string_agg(repeat(chr(97 + (seed + i) % 7), 1 + i % 5) || (i % 97)::text, ''), 1, n)
  FROM generate_series(0, n) i $$;
CREATE TABLE sizes(n int);
INSERT INTO sizes SELECT k * c + d FROM generate_series(1, 3) k,
  (VALUES (1996), (1992)) cs(c), generate_series(-2, 2) d;
INSERT INTO sizes VALUES (1), (100), (2000), (8000), (8192), (65536), (200000), (1048576);
CREATE TABLE c (id serial, kind text, n int,
  tp text COMPRESSION pglz, tl text COMPRESSION lz4,
  bp bytea COMPRESSION pglz, bl bytea COMPRESSION lz4);
ALTER TABLE c ALTER tp SET STORAGE EXTENDED, ALTER tl SET STORAGE EXTENDED;
CREATE TABLE u (id serial, kind text, n int, t text, b bytea);
ALTER TABLE u ALTER t SET STORAGE EXTERNAL, ALTER b SET STORAGE EXTERNAL;
CREATE TABLE m (id serial, kind text, n int, tp text COMPRESSION pglz, tl text COMPRESSION lz4);
ALTER TABLE m ALTER tp SET STORAGE MAIN, ALTER tl SET STORAGE MAIN;
\if :{?noslot}
\else
SELECT 'slot' FROM pg_create_logical_replication_slot('o', 'test_decoding');
\endif
-- external compressed (large soft values), and poorly compressible ones
INSERT INTO c (kind, n, tp, tl, bp, bl)
  SELECT 'soft', n, soft(n, n), soft(n + 1, n), convert_to(soft(n + 2, n), 'UTF8'),
         convert_to(soft(n + 3, n), 'UTF8') FROM sizes;
INSERT INTO c (kind, n, tp, tl, bp, bl)
  SELECT 'noise', n, noise(n, n), noise(n + 1, n), convert_to(noise(n + 2, n), 'UTF8'),
         convert_to(noise(n + 3, n), 'UTF8') FROM sizes;
-- external uncompressed
INSERT INTO u (kind, n, t, b)
  SELECT 'noise', n, noise(n + 4, n), convert_to(noise(n + 5, n), 'UTF8') FROM sizes;
INSERT INTO u (kind, n, t, b)
  SELECT 'soft', n, soft(n + 4, n), convert_to(soft(n + 5, n), 'UTF8') FROM sizes;
-- inline compressed (MAIN keeps them in line while they fit)
INSERT INTO m (kind, n, tp, tl)
  SELECT 'soft', n, soft(n + 6, n), soft(n + 7, n) FROM sizes WHERE n <= 8192;
ALTER TABLE c REPLICA IDENTITY FULL;
UPDATE c SET n = n WHERE id % 3 = 0;
DELETE FROM c WHERE id % 7 = 0;
CHECKPOINT;
SQL
}
read_all() {
  local offs="0,1,1990,1991,1992,1995,1996,1997,3983,3984,3991,3992,3993,5987,5988,5989,100000"
  local lens="1,2,4,100,1996,1997,4000"
  Q <<SQL > $OUT
\\pset tuples_only on
-- every column of every table: text-level readers
WITH v AS (
  SELECT 'c' t, id, kind, n, 'tp' col, tp::bytea x, tp::text s FROM c UNION ALL
  SELECT 'c', id, kind, n, 'tl', tl::bytea, tl FROM c UNION ALL
  SELECT 'u', id, kind, n, 't', t::bytea, t FROM u UNION ALL
  SELECT 'm', id, kind, n, 'tp', tp::bytea, tp FROM m UNION ALL
  SELECT 'm', id, kind, n, 'tl', tl::bytea, tl FROM m)
SELECT format('%s.%s id=%s %s n=%s md5=%s len=%s oct=%s', t, col, id, kind, n, md5(s),
              length(s), octet_length(s)) FROM v ORDER BY t, col, id;
SELECT format('size c id=%s tp=%s/%s/%s tl=%s/%s/%s bp=%s/%s/%s bl=%s/%s/%s', id,
  pg_column_size(tp), pg_column_compression(tp), pg_column_toast_chunk_id(tp) IS NOT NULL,
  pg_column_size(tl), pg_column_compression(tl), pg_column_toast_chunk_id(tl) IS NOT NULL,
  pg_column_size(bp), pg_column_compression(bp), pg_column_toast_chunk_id(bp) IS NOT NULL,
  pg_column_size(bl), pg_column_compression(bl), pg_column_toast_chunk_id(bl) IS NOT NULL)
  FROM c ORDER BY id;
SELECT format('size u id=%s t=%s/%s/%s b=%s/%s/%s', id,
  pg_column_size(t), pg_column_compression(t), pg_column_toast_chunk_id(t) IS NOT NULL,
  pg_column_size(b), pg_column_compression(b), pg_column_toast_chunk_id(b) IS NOT NULL)
  FROM u ORDER BY id;
SELECT format('size m id=%s tp=%s/%s/%s tl=%s/%s/%s', id,
  pg_column_size(tp), pg_column_compression(tp), pg_column_toast_chunk_id(tp) IS NOT NULL,
  pg_column_size(tl), pg_column_compression(tl), pg_column_toast_chunk_id(tl) IS NOT NULL)
  FROM m ORDER BY id;
-- chunk ids: equal across builds when the corpus is filled the same way
SELECT format('chunkid c id=%s %s %s %s %s', id, pg_column_toast_chunk_id(tp),
  pg_column_toast_chunk_id(tl), pg_column_toast_chunk_id(bp), pg_column_toast_chunk_id(bl))
  FROM c ORDER BY id;
-- substr / bytea substring at chunk-boundary offsets and lengths
SELECT format('substr c.%s id=%s o=%s l=%s %s', col, id, o, l, md5(x))
FROM (SELECT id, 'tp' col, o, l, substr(tp, o, l)::bytea x FROM c,
        unnest('{$offs}'::int[]) o, unnest('{$lens}'::int[]) l UNION ALL
      SELECT id, 'tl', o, l, substr(tl, o, l)::bytea FROM c,
        unnest('{$offs}'::int[]) o, unnest('{$lens}'::int[]) l UNION ALL
      SELECT id, 'bp', o, l, substring(bp FROM o FOR l) FROM c,
        unnest('{$offs}'::int[]) o, unnest('{$lens}'::int[]) l UNION ALL
      SELECT id, 'bl', o, l, substring(bl FROM o FOR l) FROM c,
        unnest('{$offs}'::int[]) o, unnest('{$lens}'::int[]) l) s
ORDER BY col, id, o, l;
SELECT format('substr u.%s id=%s o=%s l=%s %s', col, id, o, l, md5(x))
FROM (SELECT id, 't' col, o, l, substr(t, o, l)::bytea x FROM u,
        unnest('{$offs}'::int[]) o, unnest('{$lens}'::int[]) l UNION ALL
      SELECT id, 'b', o, l, substring(b FROM o FOR l) FROM u,
        unnest('{$offs}'::int[]) o, unnest('{$lens}'::int[]) l) s
ORDER BY col, id, o, l;
SELECT format('substr m.%s id=%s o=%s l=%s %s', col, id, o, l, md5(x))
FROM (SELECT id, 'tp' col, o, l, substr(tp, o, l)::bytea x FROM m,
        unnest('{$offs}'::int[]) o, unnest('{$lens}'::int[]) l UNION ALL
      SELECT id, 'tl', o, l, substr(tl, o, l)::bytea FROM m,
        unnest('{$offs}'::int[]) o, unnest('{$lens}'::int[]) l) s
ORDER BY col, id, o, l;
-- prefix slices (pglz's partial-fetch path): left()
SELECT format('left c id=%s l=%s %s %s', id, l, md5(left(tp, l)), md5(left(tl, l)))
  FROM c, unnest('{1,10,1000,1996,2000,5000}'::int[]) l ORDER BY id, l;
SELECT format('amcheck %s %s %s', r, coalesce(blkno::text, '-'), coalesce(msg, 'none'))
FROM (SELECT 'c' r, * FROM verify_heapam('c', check_toast => true) UNION ALL
      SELECT 'u', * FROM verify_heapam('u', check_toast => true) UNION ALL
      SELECT 'm', * FROM verify_heapam('m', check_toast => true)) s;
SELECT format('amcheck-count %s', (SELECT count(*) FROM verify_heapam('c', check_toast => true))
  + (SELECT count(*) FROM verify_heapam('u', check_toast => true))
  + (SELECT count(*) FROM verify_heapam('m', check_toast => true)));
SQL
  # logical decoding of the fill (INSERT, UPDATE, DELETE with REPLICA IDENTITY FULL):
  # reassembles TOAST chunks in the reorder buffer.  Values are hashed per change.
  if Q -c "SELECT 1 FROM pg_replication_slots WHERE slot_name = 'o'" | grep -q 1; then
    Q -c "SELECT format('decode %s %s %s', row_number() OVER (), length(data), md5(data))
          FROM pg_logical_slot_peek_changes('o', NULL, NULL, 'include-xids', '0')" >> $OUT
  fi
}
case $MODE in
  both) fill; read_all; stop ;;
  fill) fill; stop ;;
  read) start; read_all; stop ;;
esac
echo "lines=$([ -f $OUT ] && wc -l < $OUT || echo -) asan=$(cat $D/asan.* 2>/dev/null | grep -c AddressSanitizer || true) traps=$(grep -c -E 'TRAP|PANIC' $D/log || true) datadir=$D"
[ $MODE = both ] && rm -rf $D || true