-- Semana 9: tamaños y buffer pool con chasqui_big (2 millones de envíos).
-- No cambia datos: solo instala pg_buffercache si falta y saca páginas del
-- buffer pool.
--   psql -d chasqui_big -f big.sql

CREATE EXTENSION IF NOT EXISTS pg_buffercache;

-- 1. Tamaño de shipment: bytes, páginas, relpages y filas por página.
-- name: big_size
SELECT pg_relation_size('shipment') AS bytes,
       pg_size_pretty(pg_relation_size('shipment')) AS pretty,
       pg_relation_size('shipment') / 8192 AS pages,
       relpages,
       reltuples::bigint AS reltuples,
       round(reltuples / relpages) AS rows_per_page
FROM pg_class
WHERE relname = 'shipment';

SET max_parallel_workers_per_gather = 0;

-- 2. Una tabla chica (customer: 2040 páginas) entra en el buffer pool:
--    el segundo scan ya no lee nada.
-- name: big_evict_customer
SELECT * FROM pg_buffercache_evict_relation('customer');

-- name: big_customer_1
EXPLAIN (ANALYZE, BUFFERS, COSTS OFF)
SELECT count(*) FROM customer;

-- name: big_customer_2
EXPLAIN (ANALYZE, BUFFERS, COSTS OFF)
SELECT count(*) FROM customer;

-- 3. Una tabla grande (shipment: más de la cuarta parte de shared_buffers)
--    se recorre con un buffer ring: el segundo scan vuelve a leer casi todo.
-- name: big_evict_shipment
SELECT * FROM pg_buffercache_evict_relation('shipment');

-- name: big_shipment_1
EXPLAIN (ANALYZE, BUFFERS, COSTS OFF)
SELECT count(*) FROM shipment;

-- name: big_shipment_2
EXPLAIN (ANALYZE, BUFFERS, COSTS OFF)
SELECT count(*) FROM shipment;

-- 4. Cuántas páginas de shipment quedaron en el buffer pool.
-- name: big_shipment_buffers
SELECT count(*) AS buffers
FROM pg_buffercache b
  JOIN pg_class c ON b.relfilenode = pg_relation_filenode(c.oid)
WHERE b.reldatabase = (SELECT oid FROM pg_database WHERE datname = current_database())
  AND c.relname = 'shipment';

-- Semana 9: el buffer pool de PostgreSQL (shared_buffers).
-- Se corre sobre chasqui_w09 después de make:
--   psql -d chasqui_w09 -f buffers.sql
-- pg_buffercache ve el buffer pool de todo el servidor: los números cambian
-- según lo que hayan hecho las otras bases de datos.

-- 1. Tamaño del buffer pool, en marcos de 8 KiB y en total.
-- name: shared_buffers
SELECT setting AS buffers, unit, pg_size_pretty(setting::bigint * 8192) AS total
FROM pg_settings
WHERE name = 'shared_buffers';

-- 2. Las relaciones de esta base de datos con más páginas en el buffer pool,
--    y cuántas de esas páginas están sucias.
-- name: top_relations
SELECT c.relname,
       count(*) AS buffers,
       count(*) FILTER (WHERE b.isdirty) AS dirty
FROM pg_buffercache b
  JOIN pg_class c ON b.relfilenode = pg_relation_filenode(c.oid)
WHERE b.reldatabase = (SELECT oid FROM pg_database WHERE datname = current_database())
GROUP BY c.relname
ORDER BY buffers DESC, c.relname
LIMIT 10;

-- 3. Caché fría y caliente. Primero se sacan del buffer pool las páginas de
--    shipment (las sucias se escriben antes). Sin workers en paralelo, para
--    que el plan sea más corto.
SET max_parallel_workers_per_gather = 0;

-- name: evict
SELECT * FROM pg_buffercache_evict_relation('shipment');

-- name: cold
EXPLAIN (ANALYZE, BUFFERS, COSTS OFF, TIMING OFF)
SELECT count(*) FROM shipment;

-- name: warm
EXPLAIN (ANALYZE, BUFFERS, COSTS OFF, TIMING OFF)
SELECT count(*) FROM shipment;

-- 4. Usage count: el contador que usa el reemplazo clock de PostgreSQL.
-- name: usage_counts
SELECT usagecount, count(*) AS buffers
FROM pg_buffercache b
  JOIN pg_class c ON b.relfilenode = pg_relation_filenode(c.oid)
WHERE b.reldatabase = (SELECT oid FROM pg_database WHERE datname = current_database())
  AND c.relname = 'shipment'
GROUP BY usagecount
ORDER BY usagecount;

// Muestra una página de heap leída e interpretada con pgpage.mjs.
// Uso (dentro del contenedor): PGDATABASE=chasqui_w09 node dump.mjs agency 0
import pg from 'pg';
import { fetchRawPage, parseHeader, items, freeSpace, slotCount } from './pgpage.mjs';

const [relation = 'agency', block = '0'] = process.argv.slice(2);
const client = new pg.Client();
await client.connect();
try {
  const page = await fetchRawPage(client, relation, Number(block));
  const h = parseHeader(page);
  console.log(`${relation}, página ${block}: lower=${h.lower} upper=${h.upper} ` +
    `special=${h.special} slots=${slotCount(page)} libre=${freeSpace(page)} bytes`);
  for (const it of items(page)) {
    const t = it.tuple;
    const extra = t ? ` xmin=${t.t_xmin} xmax=${t.t_xmax} ctid=(${t.t_ctid.block},${t.t_ctid.offset})` : '';
    console.log(`  lp=${it.lp} off=${it.lp_off} flags=${it.lp_flags} len=${it.lp_len}${extra}`);
  }
} finally {
  await client.end();
}

-- Semana 9: las páginas de una tabla de PostgreSQL por dentro.
-- Se corre sobre chasqui_w09 después de make:
--   psql -d chasqui_w09 -f pages.sql
-- Las marcas "-- name:" las usan las pruebas para correr el mismo texto.

-- 1. Dónde está el archivo de la tabla, relativo al directorio de datos.
-- name: filepath
SELECT pg_relation_filepath('agency') AS path;

-- 2. Tamaño de cada tabla en bytes y en páginas de 8 KiB, y cuántas filas
--    caben en promedio en una página. relpages y reltuples los actualizan
--    VACUUM y ANALYZE; pg_relation_size mide el archivo ahora.
-- name: sizes
SELECT relname,
       pg_relation_size(oid) AS bytes,
       pg_relation_size(oid) / current_setting('block_size')::int AS pages,
       relpages,
       reltuples::bigint AS reltuples,
       round(reltuples / relpages) AS rows_per_page
FROM pg_class
WHERE relnamespace = 'public'::regnamespace AND relkind = 'r'
ORDER BY pages DESC;

-- 3. Además del archivo principal (main), una tabla tiene un free space map
--    (fsm) y un visibility map (vm), cada uno en su propio archivo.
-- name: forks
SELECT pg_relation_size('shipment', 'main') AS main,
       pg_relation_size('shipment', 'fsm') AS fsm,
       pg_relation_size('shipment', 'vm') AS vm;

-- 4. El encabezado de la página 0 de agency. El espacio libre es upper - lower.
-- name: header
SELECT lower, upper, special, pagesize, upper - lower AS free
FROM page_header(get_raw_page('agency', 0));

-- 5. Los slots (line pointers) y el encabezado de cada tupla.
-- name: slots
SELECT lp, lp_off, lp_flags, lp_len, t_xmin, t_xmax, t_ctid
FROM heap_page_items(get_raw_page('agency', 0))
ORDER BY lp
LIMIT 5;

-- 6. El último slot apunta a la tupla que está más cerca del inicio: los
--    slots crecen hacia adelante y las tuplas hacia atrás.
-- name: last_slots
SELECT lp, lp_off, lp_len
FROM heap_page_items(get_raw_page('agency', 0))
ORDER BY lp DESC
LIMIT 3;

-- 7. ctid es el record id de PostgreSQL: (página, slot).
-- name: ctid
SELECT ctid, agency_id, name
FROM agency
ORDER BY agency_id
LIMIT 5;

-- 8. Los bytes de tres filas de customer. La tercera tiene phone NULL: tiene
--    bitmap de nulos (t_bits) y sus datos no guardan nada para phone.
-- name: customer_rows
SELECT ctid, customer_id, dni, first_name, last_name, phone, city
FROM customer
ORDER BY customer_id
LIMIT 3;

-- name: customer_bytes
SELECT lp, lp_len, t_hoff, t_bits, t_data
FROM heap_page_items(get_raw_page('customer', 0))
WHERE lp <= 3
ORDER BY lp;

-- 9. El diccionario de datos: los atributos de customer en pg_attribute.
--    attlen = -1 significa longitud variable (varlena).
-- name: attributes
SELECT attnum, attname, atttypid::regtype AS type, attlen, attalign, attnotnull
FROM pg_attribute
WHERE attrelid = 'customer'::regclass AND attnum > 0
ORDER BY attnum;

-- 10. El free space map: espacio libre aproximado de cada página, en
--     múltiplos de 32 bytes (8192 / 256).
-- name: free_space
SELECT count(*) AS pages, min(avail), max(avail), round(avg(avail)) AS avg
FROM pg_freespace('shipment');

-- name: free_space_agency
SELECT blkno, avail FROM pg_freespace('agency');

-- 11. Tamaño de las tuplas de shipment en su primera página.
-- name: shipment_tuples
SELECT lp_len, count(*) AS tuples
FROM heap_page_items(get_raw_page('shipment', 0))
GROUP BY lp_len
ORDER BY lp_len;

-- name: shipment_header
SELECT lower, upper, upper - lower AS free
FROM page_header(get_raw_page('shipment', 0));

// Lector de páginas de heap de PostgreSQL, solo lectura.
//
// Recibe los 8192 bytes de una página tal como los devuelve
// get_raw_page('tabla', n) (extensión pageinspect) y los interpreta a mano,
// byte por byte, siguiendo la documentación "Database Page Layout":
//
//   0       24                lower          upper                 special  8192
//   | header | line pointers -> |  espacio libre  | <- tuplas          | (vacío)
//
// No escribe nada: sirve para ver que una slotted page de verdad es un
// arreglo de bytes con un formato fijo. Los enteros están en little-endian
// (el contenedor corre en x86-64 o ARM64, ambos little-endian).

export const PAGE_SIZE = 8192;
export const PAGE_HEADER_SIZE = 24;
export const LINE_POINTER_SIZE = 4;

// Valores de lp_flags (src/include/storage/itemid.h).
export const LP_UNUSED = 0; // slot libre
export const LP_NORMAL = 1; // apunta a una tupla
export const LP_REDIRECT = 2; // apunta a otro slot (HOT)
export const LP_DEAD = 3; // tupla muerta, falta que VACUUM libere el slot

// Bit de t_infomask que indica que la tupla tiene bitmap de nulos.
const HEAP_HASNULL = 0x0001;
const HEAP_NATTS_MASK = 0x07ff;

/** Lee el encabezado de 24 bytes. Los nombres son los de page_header(). */
export function parseHeader(page) {
  checkPage(page);
  const xlogid = page.readUInt32LE(0);
  const xrecoff = page.readUInt32LE(4);
  const sizeVersion = page.readUInt16LE(18);
  return {
    lsn: `${hex(xlogid)}/${hex(xrecoff)}`,
    checksum: page.readInt16LE(8),
    flags: page.readUInt16LE(10),
    lower: page.readUInt16LE(12),
    upper: page.readUInt16LE(14),
    special: page.readUInt16LE(16),
    pagesize: sizeVersion & 0xff00,
    version: sizeVersion & 0x00ff,
    prune_xid: page.readUInt32LE(20),
  };
}

/** Número de slots: los line pointers ocupan de 24 hasta lower. */
export function slotCount(page) {
  return (parseHeader(page).lower - PAGE_HEADER_SIZE) / LINE_POINTER_SIZE;
}

/** Bytes libres entre el último line pointer y la primera tupla. */
export function freeSpace(page) {
  const { lower, upper } = parseHeader(page);
  return upper - lower;
}

/**
 * Lee el line pointer del slot lp (desde 1, como ctid). Son 32 bits:
 * lp_off (15 bits), lp_flags (2 bits) y lp_len (15 bits).
 */
export function linePointer(page, lp) {
  const n = slotCount(page);
  if (!Number.isInteger(lp) || lp < 1 || lp > n) {
    throw new RangeError(`slot ${lp} fuera de rango: la página tiene ${n} slots`);
  }
  const v = page.readUInt32LE(PAGE_HEADER_SIZE + (lp - 1) * LINE_POINTER_SIZE);
  return { lp, lp_off: v & 0x7fff, lp_flags: (v >>> 15) & 0x3, lp_len: v >>> 17 };
}

/**
 * Lee el encabezado de una tupla (HeapTupleHeaderData, 23 bytes más el
 * bitmap de nulos) que empieza en el byte off de la página.
 */
export function parseTupleHeader(page, off) {
  const blockHi = page.readUInt16LE(off + 12);
  const blockLo = page.readUInt16LE(off + 14);
  const infomask2 = page.readUInt16LE(off + 18);
  const infomask = page.readUInt16LE(off + 20);
  const natts = infomask2 & HEAP_NATTS_MASK;
  let bits = null;
  if (infomask & HEAP_HASNULL) {
    // Un bit por atributo, el menos significativo primero; 1 = no es NULL.
    bits = '';
    for (let i = 0; i < Math.ceil(natts / 8) * 8; i++) {
      bits += (page[off + 23 + (i >> 3)] >> (i & 7)) & 1;
    }
  }
  return {
    t_xmin: page.readUInt32LE(off),
    t_xmax: page.readUInt32LE(off + 4),
    t_ctid: { block: blockHi * 65536 + blockLo, offset: page.readUInt16LE(off + 16) },
    t_infomask2: infomask2,
    t_infomask: infomask,
    t_hoff: page[off + 22],
    natts,
    t_bits: bits,
  };
}

/**
 * Recorre todos los slots. Para los LP_NORMAL agrega el encabezado de la
 * tupla y sus datos (los bytes después de t_hoff).
 */
export function items(page) {
  const out = [];
  for (let lp = 1; lp <= slotCount(page); lp++) {
    const item = linePointer(page, lp);
    if (item.lp_flags === LP_NORMAL) {
      const header = parseTupleHeader(page, item.lp_off);
      item.tuple = header;
      item.data = page.subarray(item.lp_off + header.t_hoff, item.lp_off + item.lp_len);
    }
    out.push(item);
  }
  return out;
}

/**
 * Decodifica los datos de una tupla cuyas columnas son 'int4' o 'text' (o
 * char/varchar, que se guardan igual). Solo entiende textos cortos, con
 * encabezado de 1 byte (hasta 126 bytes de contenido): es lo que tienen las
 * tablas pequeñas de Chasqui. columns es una lista de tipos, en el orden de
 * la tabla; tuple es el resultado de parseTupleHeader.
 */
export function decodeTuple(data, tuple, columns) {
  const values = [];
  let pos = 0;
  for (let i = 0; i < columns.length; i++) {
    if (i >= tuple.natts || (tuple.t_bits && tuple.t_bits[i] === '0')) {
      values.push(null); // los NULL no ocupan bytes en los datos
      continue;
    }
    if (columns[i] === 'int4') {
      pos = align(pos, 4);
      values.push(data.readInt32LE(pos));
      pos += 4;
    } else if (columns[i] === 'text') {
      // Un byte de relleno vale 0; un encabezado corto es impar (bit bajo en 1)
      // y no se alinea. Un encabezado de 4 bytes se alinea a 4.
      if (data[pos] === 0) pos = align(pos, 4);
      const head = data[pos];
      if ((head & 0x01) === 0) {
        throw new Error(`atributo ${i + 1}: solo se leen textos con encabezado de 1 byte`);
      }
      const total = head >>> 1; // longitud total, con el byte de encabezado
      values.push(data.toString('utf8', pos + 1, pos + total));
      pos += total;
    } else {
      throw new Error(`tipo no soportado: ${columns[i]}`);
    }
  }
  return values;
}

function align(n, a) {
  return Math.ceil(n / a) * a;
}

function hex(n) {
  return n.toString(16).toUpperCase();
}

function checkPage(page) {
  if (!Buffer.isBuffer(page) || page.length !== PAGE_SIZE) {
    throw new TypeError(`se esperaba un Buffer de ${PAGE_SIZE} bytes`);
  }
}

/** Trae una página con get_raw_page; pg devuelve el bytea como Buffer. */
export async function fetchRawPage(db, relation, block) {
  const { rows } = await db.query('SELECT get_raw_page($1, $2) AS page', [relation, block]);
  return rows[0].page;
}

// Simulador de políticas de reemplazo de un buffer pool.
//
// Solo modela la decisión de qué marco (frame) desalojar: recibe una cadena
// de referencias (los números de página que pide el motor, en orden) y la
// cantidad de marcos, y cuenta aciertos (hits) y fallos (misses). No hay pin
// counts, ni páginas sucias, ni disco: todas las páginas se pueden desalojar
// en cualquier momento. Es el modelo con que se comparan políticas en
// Silberschatz, sec. 13.5.2, y en la nota 5 de CS186.
//
// Políticas:
//   'lru'    desaloja la página usada hace más tiempo
//   'mru'    desaloja la página usada más recientemente
//   'clock'  aproximación de LRU: un bit de referencia por marco y una
//            manecilla que da vueltas (la variante de la nota 5 de CS186)

export const POLICIES = ['lru', 'mru', 'clock'];

/**
 * Simula la cadena refs con `frames` marcos. Devuelve
 * { hits, misses, trace }, donde trace tiene un paso por referencia:
 * { page, hit, evicted, frames } (evicted es null si no hubo desalojo y
 * frames es el contenido de los marcos después del paso; null = vacío).
 * Con { trace: false } no se guarda el detalle (para cadenas largas).
 */
export function simulate(policy, frames, refs, { trace: keepTrace = true } = {}) {
  if (!POLICIES.includes(policy)) {
    throw new Error(`política desconocida: ${policy} (use ${POLICIES.join(', ')})`);
  }
  if (!Number.isInteger(frames) || frames < 1) {
    throw new RangeError('se necesita al menos un marco');
  }
  const state = {
    pages: new Array(frames).fill(null), // qué página hay en cada marco
    where: new Map(), // página -> marco (la "tabla de páginas" del buffer pool)
    recency: new Map(), // para LRU: páginas de la menos a la más reciente
    mostRecent: null, // para MRU: la última página pedida
    ref: new Array(frames).fill(0), // para clock: bit de referencia por marco
    hand: 0, // para clock: la manecilla
    used: 0, // marcos ocupados; se llenan en orden 0, 1, 2, ...
  };
  const trace = [];
  let hits = 0;
  for (const page of refs) {
    let frame = state.where.get(page);
    const hit = frame !== undefined;
    let evicted = null;
    if (hit) {
      hits++;
    } else {
      if (state.used < frames) {
        frame = state.used++; // primero se usan los marcos vacíos
      } else {
        frame = chooseVictim(policy, state);
        evicted = state.pages[frame];
        state.where.delete(evicted);
        state.recency.delete(evicted);
      }
      state.pages[frame] = page;
      state.where.set(page, frame);
    }
    state.recency.delete(page); // vuelve a entrar al final: la más reciente
    state.recency.set(page, true);
    state.mostRecent = page;
    state.ref[frame] = 1;
    if (keepTrace) trace.push({ page, hit, evicted, frames: [...state.pages] });
  }
  return { hits, misses: refs.length - hits, trace };
}

function chooseVictim(policy, state) {
  if (policy === 'lru') return state.where.get(first(state.recency));
  if (policy === 'mru') return state.where.get(state.mostRecent);
  // clock: la manecilla avanza; si el bit está en 1 lo baja a 0 (segunda
  // oportunidad), si está en 0 ese marco es la víctima. Termina a lo más en
  // dos vueltas.
  const n = state.pages.length;
  for (;;) {
    const f = state.hand;
    state.hand = (state.hand + 1) % n;
    if (state.ref[f] === 1) state.ref[f] = 0;
    else return f;
  }
}

function first(map) {
  return map.keys().next().value;
}

/** Una cadena que recorre las páginas 1..n, `times` veces (un scan repetido). */
export function repeatedScan(n, times) {
  const one = Array.from({ length: n }, (_, i) => i + 1);
  return Array.from({ length: times }, () => one).flat();
}

// Muestra paso a paso una cadena de referencias con una política.
// Uso: node replay.mjs <lru|mru|clock> <marcos> <página> <página> ...
//   node replay.mjs clock 3 1 2 3 4 1 2 5 1 2 3 4 5
import { simulate } from './replacement.mjs';

const [policy = 'clock', frames = '3', ...pages] = process.argv.slice(2);
const refs = pages.length > 0 ? pages.map(Number) : [1, 2, 3, 4, 1, 2, 5, 1, 2, 3, 4, 5];
const { hits, misses, trace } = simulate(policy, Number(frames), refs);

console.log(`${policy}, ${frames} marcos`);
for (const { page, hit, evicted, frames: f } of trace) {
  const shown = f.map((p) => (p === null ? '-' : String(p)).padStart(2)).join(' ');
  const what = hit ? 'hit ' : 'miss';
  const out = evicted === null ? '' : `  sale ${evicted}`;
  console.log(`  pide ${String(page).padStart(2)}  ${what}  [${shown} ]${out}`);
}
console.log(`hits=${hits} misses=${misses}`);

// Lee un archivo .sql con consultas marcadas "-- name: nombre" y devuelve un
// objeto { nombre: texto }. Así las pruebas corren el mismo texto que el
// archivo que se muestra en clase. Cada consulta va desde la marca hasta la
// primera línea que termina en punto y coma; lo que sigue (comentarios,
// BEGIN, SET, VACUUM) no es parte de ella.
import { readFileSync } from 'node:fs';

export function namedQueries(file) {
  const text = readFileSync(new URL(file, import.meta.url), 'utf8');
  const parts = text.split(/^-- name: (\w+)\n/m);
  const queries = {};
  for (let i = 1; i < parts.length; i += 2) {
    const end = parts[i + 1].search(/;\s*$/m);
    queries[parts[i]] = parts[i + 1].slice(0, end + 1).trim();
  }
  return queries;
}

-- Semana 9: un UPDATE no sobrescribe la fila en su lugar.
-- Se corre una sola vez sobre chasqui_w09 recién creada con make (cambia el
-- nombre de la agencia 1; para repetir el experimento, vuelvan a correr make):
--   psql -d chasqui_w09 -f update.sql
-- Los números de transacción (t_xmin, t_xmax) serán otros en cada corrida.

-- name: before
SELECT ctid, agency_id, name FROM agency WHERE agency_id = 1;

BEGIN;

-- name: update
UPDATE agency SET name = name || ' (centro)' WHERE agency_id = 1;

-- La fila nueva quedó en otro slot.
-- name: after
SELECT ctid, agency_id, name FROM agency WHERE agency_id = 1;

-- En la página están las dos versiones. La vieja tiene t_xmax (la transacción
-- que la reemplazó) y su t_ctid apunta a la nueva.
-- name: versions
SELECT lp, lp_off, lp_flags, lp_len, t_xmin, t_xmax, t_ctid
FROM heap_page_items(get_raw_page('agency', 0))
WHERE lp IN (1, 2, 29, 30)
ORDER BY lp;

-- name: header_after_update
SELECT lower, upper, upper - lower AS free
FROM page_header(get_raw_page('agency', 0));

COMMIT;

-- VACUUM no puede correr dentro de una transacción.
VACUUM agency;

-- El slot 1 quedó libre (lp_flags = 0) y la página se compactó: el slot 2
-- cambió de lp_off pero su ctid sigue siendo (0,2).
-- name: after_vacuum
SELECT lp, lp_off, lp_flags, lp_len, t_xmin, t_xmax, t_ctid
FROM heap_page_items(get_raw_page('agency', 0))
WHERE lp IN (1, 2, 29, 30)
ORDER BY lp;

-- name: header_after_vacuum
SELECT lower, upper, upper - lower AS free
FROM page_header(get_raw_page('agency', 0));

-- name: ctid_2
SELECT ctid, agency_id, name FROM agency WHERE agency_id IN (1, 2) ORDER BY agency_id;

// Pruebas de buffers.sql: el buffer pool de PostgreSQL visto con
// pg_buffercache y EXPLAIN (BUFFERS). Solo saca del buffer pool páginas de
// chasqui_w09; no mira cifras de todo el servidor, que dependen de las demás
// bases de datos.
import { test, before, after } from 'node:test';
import assert from 'node:assert/strict';
import pg from 'pg';
import { namedQueries } from './sqlfile.mjs';

const q = namedQueries('./buffers.sql');
let client;
before(async () => {
  client = new pg.Client();
  await client.connect();
  await client.query('SET max_parallel_workers_per_gather = 0');
});
after(() => client.end());

// Corre EXPLAIN (ANALYZE, BUFFERS) en JSON y devuelve los buffers del scan.
async function scanBuffers(table) {
  const { rows } = await client.query(
    `EXPLAIN (ANALYZE, BUFFERS, FORMAT JSON) SELECT count(*) FROM ${table}`,
  );
  const scan = rows[0]['QUERY PLAN'][0].Plan.Plans[0];
  assert.equal(scan['Node Type'], 'Seq Scan');
  return { hit: scan['Shared Hit Blocks'], read: scan['Shared Read Blocks'] };
}

test('shared_buffers son 16384 marcos de 8 KiB (128 MB)', async () => {
  const { rows } = await client.query(q.shared_buffers);
  assert.deepEqual(rows[0], { buffers: '16384', unit: '8kB', total: '128 MB' });
});

test('top_relations: lista relaciones de esta base de datos', async () => {
  await scanBuffers('shipment');
  const { rows } = await client.query(q.top_relations);
  assert.ok(rows.length > 0 && rows.length <= 10);
  assert.ok(rows.some((r) => r.relname === 'shipment'));
  for (const r of rows) assert.ok(Number(r.dirty) <= Number(r.buffers));
});

for (const table of ['shipment', 'customer', 'trip']) {
  test(`${table}: el primer scan lee del disco, el segundo encuentra todo en memoria`, async () => {
    const { rows } = await client.query(
      'SELECT relpages FROM pg_class WHERE relname = $1', [table],
    );
    const pages = rows[0].relpages;
    await client.query('SELECT * FROM pg_buffercache_evict_relation($1)', [table]);
    assert.deepEqual(await scanBuffers(table), { hit: 0, read: pages });
    assert.deepEqual(await scanBuffers(table), { hit: pages, read: 0 });
  });
}

test('el texto de buffers.sql: cold con read=148 y warm con hit=148', async () => {
  await client.query(q.evict);
  const cold = (await client.query(q.cold)).rows.map((r) => r['QUERY PLAN']).join('\n');
  const warm = (await client.query(q.warm)).rows.map((r) => r['QUERY PLAN']).join('\n');
  assert.match(cold, /Seq Scan on shipment[\s\S]*Buffers: shared read=148\n/);
  assert.match(warm, /Seq Scan on shipment[\s\S]*Buffers: shared hit=148\n/);
});

test('cada scan sube el usage count de las páginas', async () => {
  await client.query(q.evict);
  await scanBuffers('shipment');
  const once = (await client.query(q.usage_counts)).rows;
  assert.deepEqual(once, [{ usagecount: 1, buffers: '148' }]);
  await scanBuffers('shipment');
  const twice = (await client.query(q.usage_counts)).rows;
  assert.deepEqual(twice, [{ usagecount: 2, buffers: '148' }]);
});

// Pruebas de pages.sql y update.sql: lo que el apunte dice de las páginas de
// chasqui_w09. Necesitan la base recién creada (make test la crea): update.sql
// cambia la agencia 1, y las pruebas de este archivo corren en orden.
import { test, before, after } from 'node:test';
import assert from 'node:assert/strict';
import pg from 'pg';
import { namedQueries } from './sqlfile.mjs';

const q = { ...namedQueries('./pages.sql'), ...namedQueries('./update.sql') };
let client;
before(async () => {
  client = new pg.Client();
  await client.connect();
});
after(() => client.end());

const all = async (name) => (await client.query(q[name])).rows;
const one = async (name) => (await all(name))[0];

test('el archivo de agency está en base/<oid de la base>/<relfilenode>', async () => {
  const { path } = await one('filepath');
  const [ids] = (await client.query(
    `SELECT (SELECT oid FROM pg_database WHERE datname = current_database()) AS db,
            pg_relation_filenode('agency') AS node`,
  )).rows;
  assert.equal(path, `base/${ids.db}/${ids.node}`);
});

test('tamaños: páginas = bytes / 8192 = relpages, y filas por página', async () => {
  const rows = await all('sizes');
  const by = Object.fromEntries(rows.map((r) => [r.relname, r]));
  for (const r of rows) {
    assert.equal(Number(r.bytes) % 8192, 0);
    assert.equal(Number(r.pages), r.relpages, r.relname);
  }
  assert.deepEqual(
    [by.shipment.pages, by.shipment.reltuples, by.shipment.rows_per_page],
    ['148', '12000', 81],
  );
  assert.deepEqual([by.agency.pages, by.agency.reltuples], ['1', '29']);
  assert.equal(by.customer.pages, '19');
  assert.equal(by.shipment_event.pages, '342');
});

test('shipment tiene tres archivos: main, fsm y vm', async () => {
  assert.deepEqual(await one('forks'), { main: '1212416', fsm: '24576', vm: '8192' });
});

test('encabezado de agency: 29 slots, lower = 24 + 4 · 29, libre = upper - lower', async () => {
  assert.deepEqual(await one('header'), { lower: 140, upper: 6280, special: 8192, pagesize: 8192, free: 6140 });
  const { count } = (await client.query(
    "SELECT count(*)::int FROM heap_page_items(get_raw_page('agency', 0))",
  )).rows[0];
  assert.equal(count, 29);
  assert.equal(24 + 4 * count, 140);
});

test('slots: el primero apunta al final de la página, el último a upper', async () => {
  const first = await all('slots');
  assert.deepEqual(first.map((r) => [r.lp, r.lp_off, r.lp_len]), [
    [1, 8128, 57], [2, 8072, 56], [3, 8016, 49], [4, 7952, 64], [5, 7880, 72],
  ]);
  // 8192 - 57 = 8135, redondeado hacia abajo a múltiplo de 8 = 8128.
  assert.equal(Math.floor((8192 - 57) / 8) * 8, 8128);
  // Después de VACUUM (FREEZE) ninguna tupla tiene t_xmax.
  assert.ok(first.every((r) => r.t_xmax === '0'));
  const last = await all('last_slots');
  assert.deepEqual(last[0], { lp: 29, lp_off: 6280, lp_len: 60 });
});

test('ctid = (página, slot)', async () => {
  const rows = await all('ctid');
  assert.deepEqual(rows.map((r) => r.ctid), ['(0,1)', '(0,2)', '(0,3)', '(0,4)', '(0,5)']);
});

test('los bytes de customer: entero de 4 bytes, textos con encabezado de 1 byte, bitmap de nulos', async () => {
  const people = await all('customer_rows');
  assert.equal(people[2].phone, null);
  const rows = await all('customer_bytes');
  assert.deepEqual(rows.map((r) => [r.lp_len, r.t_hoff, r.t_bits]), [
    [77, 24, null], [75, 24, null], [65, 24, '11110100'],
  ]);
  const data = rows[0].t_data;
  assert.equal(data.readInt32LE(0), 1); // customer_id = 1, little-endian
  assert.equal(data[4], 0x13); // encabezado corto: (0x13 >> 1) = 9 bytes con él
  assert.equal(data.toString('utf8', 5, 13), '61830975');
  assert.equal(data.length, 77 - 24);
});

test('pg_attribute: solo customer_id tiene longitud fija; char(8) es varlena', async () => {
  const rows = await all('attributes');
  assert.deepEqual(rows.map((r) => [r.attname, r.attlen]), [
    ['customer_id', 4], ['dni', -1], ['first_name', -1], ['last_name', -1], ['phone', -1], ['city', -1],
  ]);
});

test('free space map: en múltiplos de 32 bytes, por debajo del espacio libre real', async () => {
  const fsm = await one('free_space');
  assert.deepEqual(fsm, { pages: '148', min: 0, max: 1856, avg: '63' });
  const { avail } = await one('free_space_agency');
  assert.equal(avail, 6112);
  assert.equal(avail % 32, 0);
  assert.ok(avail <= 6140 && 6140 - avail < 32);
});

test('shipment: 81 tuplas de 88 o 96 bytes en la página 0', async () => {
  const sizes = await all('shipment_tuples');
  assert.deepEqual(sizes.map((r) => [r.lp_len, r.tuples]), [[88, '2'], [96, '79']]);
  assert.deepEqual(await one('shipment_header'), { lower: 348, upper: 432, free: 84 });
  assert.equal(24 + 4 * 81, 348);
  assert.equal(8192 - (2 * 88 + 79 * 96), 432);
});

// ------------------------------------------------------------- update.sql

test('UPDATE deja dos versiones; VACUUM libera el slot viejo y compacta', async () => {
  assert.equal((await one('before')).ctid, '(0,1)');
  await client.query('BEGIN');
  try {
    await client.query(q.update);
    const after = await one('after');
    assert.equal(after.ctid, '(0,30)');
    assert.equal(after.name, 'Lima - La Victoria (centro)');
    const v = Object.fromEntries((await all('versions')).map((r) => [r.lp, r]));
    const { xid } = (await client.query('SELECT txid_current()::text AS xid')).rows[0];
    assert.equal(v[1].t_xmax, xid); // la vieja: la reemplazó esta transacción
    assert.equal(v[1].t_ctid, '(0,30)'); // y apunta a la nueva
    assert.equal(v[30].t_xmin, xid);
    assert.equal(v[30].t_xmax, '0');
    assert.deepEqual([v[1].lp_off, v[2].lp_off, v[30].lp_off], [8128, 8072, 6208]);
    assert.deepEqual(await one('header_after_update'), { lower: 144, upper: 6208, free: 6064 });
    await client.query('COMMIT');
  } catch (err) {
    await client.query('ROLLBACK');
    throw err;
  }
  await client.query('VACUUM agency');
  const v = Object.fromEntries((await all('after_vacuum')).map((r) => [r.lp, r]));
  assert.deepEqual([v[1].lp_flags, v[1].lp_off, v[1].lp_len], [0, 0, 0]);
  assert.deepEqual([v[2].lp_off, v[29].lp_off, v[30].lp_off], [8136, 6344, 6272]);
  assert.deepEqual(await one('header_after_vacuum'), { lower: 144, upper: 6272, free: 6128 });
  const ctids = await all('ctid_2');
  assert.deepEqual(ctids.map((r) => r.ctid), ['(0,30)', '(0,2)']);
});

// Pruebas de pgpage.mjs: lo que el lector saca de los bytes tiene que
// coincidir con lo que dicen page_header() y heap_page_items() de pageinspect.
import { test, before, after } from 'node:test';
import assert from 'node:assert/strict';
import pg from 'pg';
import {
  PAGE_SIZE, fetchRawPage, parseHeader, slotCount, freeSpace, linePointer, items, decodeTuple,
} from './pgpage.mjs';

let pool;
before(() => {
  pool = new pg.Pool();
});
after(() => pool.end());

const all = async (text, values) => (await pool.query(text, values)).rows;

// Algunas páginas de tablas distintas: la única de agency, la primera y la
// última de shipment, una del medio de shipment_event.
const PAGES = [['agency', 0], ['customer', 0], ['shipment', 0], ['shipment', 147], ['shipment_event', 200]];

test('get_raw_page devuelve un Buffer de 8192 bytes', async () => {
  const page = await fetchRawPage(pool, 'agency', 0);
  assert.ok(Buffer.isBuffer(page));
  assert.equal(page.length, PAGE_SIZE);
});

for (const [rel, block] of PAGES) {
  test(`el encabezado de ${rel}, página ${block}, coincide con page_header()`, async () => {
    const page = await fetchRawPage(pool, rel, block);
    const [expected] = await all(
      `SELECT lsn::text AS lsn, checksum, flags, lower, upper, special, pagesize, version,
              prune_xid::text::bigint AS prune_xid
       FROM page_header(get_raw_page($1, $2))`,
      [rel, block],
    );
    assert.deepEqual(parseHeader(page), { ...expected, prune_xid: Number(expected.prune_xid) });
    assert.equal(freeSpace(page), expected.upper - expected.lower);
  });

  test(`los slots de ${rel}, página ${block}, coinciden con heap_page_items()`, async () => {
    const page = await fetchRawPage(pool, rel, block);
    const rows = await all(
      `SELECT lp, lp_off, lp_flags, lp_len, t_xmin::text::bigint AS t_xmin,
              t_xmax::text::bigint AS t_xmax, t_ctid::text AS t_ctid,
              t_infomask2, t_infomask, t_hoff, t_bits
       FROM heap_page_items(get_raw_page($1, $2))
       ORDER BY lp`,
      [rel, block],
    );
    const mine = items(page);
    assert.equal(slotCount(page), rows.length);
    assert.equal(mine.length, rows.length);
    for (let i = 0; i < rows.length; i++) {
      const r = rows[i];
      const m = mine[i];
      assert.deepEqual(
        { lp: m.lp, lp_off: m.lp_off, lp_flags: m.lp_flags, lp_len: m.lp_len },
        { lp: r.lp, lp_off: r.lp_off, lp_flags: r.lp_flags, lp_len: r.lp_len },
      );
      if (r.lp_flags !== 1) continue;
      const t = m.tuple;
      assert.equal(t.t_xmin, Number(r.t_xmin));
      assert.equal(t.t_xmax, Number(r.t_xmax));
      assert.equal(`(${t.t_ctid.block},${t.t_ctid.offset})`, r.t_ctid);
      assert.equal(t.t_infomask2, r.t_infomask2);
      assert.equal(t.t_infomask, r.t_infomask);
      assert.equal(t.t_hoff, r.t_hoff);
      assert.equal(t.t_bits, r.t_bits);
    }
  });
}

test('los slots crecen desde el byte 24 y las tuplas desde el final', async () => {
  const page = await fetchRawPage(pool, 'shipment', 0);
  const h = parseHeader(page);
  assert.equal(h.lower, 24 + 4 * slotCount(page));
  const offsets = items(page).map((it) => it.lp_off);
  // Cada tupla nueva quedó antes que la anterior (la página se llenó con COPY).
  for (let i = 1; i < offsets.length; i++) assert.ok(offsets[i] < offsets[i - 1]);
  assert.equal(Math.min(...offsets), h.upper);
});

test('un slot fuera de rango es un error', async () => {
  const page = await fetchRawPage(pool, 'agency', 0);
  assert.throws(() => linePointer(page, 0), RangeError);
  assert.throws(() => linePointer(page, slotCount(page) + 1), /fuera de rango/);
});

test('decodeTuple lee las filas de customer igual que SELECT, con y sin NULL', async () => {
  const page = await fetchRawPage(pool, 'customer', 0);
  const types = ['int4', 'text', 'text', 'text', 'text', 'text'];
  const fromPage = items(page)
    .filter((it) => it.tuple)
    .map((it) => decodeTuple(it.data, it.tuple, types));
  const rows = await all(
    `SELECT customer_id, dni, first_name, last_name, phone, city
     FROM customer WHERE ctid >= '(0,0)' AND ctid < '(1,0)' ORDER BY ctid`,
  );
  assert.deepEqual(fromPage, rows.map((r) => Object.values(r)));
  assert.ok(fromPage.some((v) => v[4] === null), 'en la página 0 hay algún teléfono NULL');
});

test('decodeTuple lee agency entera', async () => {
  const page = await fetchRawPage(pool, 'agency', 0);
  const fromPage = items(page)
    .filter((it) => it.tuple)
    .map((it) => decodeTuple(it.data, it.tuple, ['int4', 'text', 'text', 'text']));
  const rows = await all('SELECT agency_id, name, city, region FROM agency ORDER BY ctid');
  assert.deepEqual(fromPage, rows.map((r) => Object.values(r)));
});

// Pruebas del simulador de reemplazo (replacement.mjs). No usan la base.
import { test } from 'node:test';
import assert from 'node:assert/strict';
import { simulate, repeatedScan } from './replacement.mjs';

const REFS = [1, 2, 3, 4, 1, 2, 5, 1, 2, 3, 4, 5];
const hitsOf = (r) => r.trace.map((s) => (s.hit ? 'h' : 'm')).join('');

test('LRU con 3 marcos sobre la cadena del apunte', () => {
  const r = simulate('lru', 3, REFS);
  assert.deepEqual([r.hits, r.misses], [2, 10]);
  assert.equal(hitsOf(r), 'mmmmmmmhhmmm');
  assert.deepEqual(r.trace[3], { page: 4, hit: false, evicted: 1, frames: [4, 2, 3] });
});

test('clock con 3 marcos: se parece a LRU pero no es igual', () => {
  const r = simulate('clock', 3, REFS);
  assert.deepEqual([r.hits, r.misses], [3, 9]);
  assert.equal(hitsOf(r), 'mmmmmmmhhmmh');
  // Al pedir 3, la manecilla baja los bits de 1, 2 y 5 (en ese orden) y desaloja 1.
  assert.equal(r.trace[9].evicted, 1);
});

test('MRU con 3 marcos', () => {
  const r = simulate('mru', 3, REFS);
  assert.deepEqual([r.hits, r.misses], [5, 7]);
  assert.equal(r.trace[3].evicted, 3);
});

test('si todo entra, solo fallan las primeras referencias', () => {
  for (const policy of ['lru', 'mru', 'clock']) {
    const r = simulate(policy, 5, REFS);
    assert.deepEqual([r.hits, r.misses], [7, 5], policy);
  }
});

test('sequential flooding: un scan repetido de 4 páginas con 3 marcos', () => {
  const refs = repeatedScan(4, 3);
  assert.deepEqual(refs, [1, 2, 3, 4, 1, 2, 3, 4, 1, 2, 3, 4]);
  assert.equal(simulate('lru', 3, refs).hits, 0);
  assert.equal(simulate('clock', 3, refs).hits, 0);
  assert.equal(simulate('mru', 3, refs).hits, 6);
});

test('pregunta de la guía: dos scans de una tabla más grande que el buffer pool', () => {
  // shipment de chasqui_big: 24 667 páginas; un buffer pool de 10 000 marcos.
  const refs = repeatedScan(24667, 2);
  const opts = { trace: false };
  assert.equal(simulate('lru', 10000, refs, opts).hits, 0);
  assert.equal(simulate('clock', 10000, refs, opts).hits, 0);
  assert.equal(simulate('mru', 10000, refs, opts).hits, 10000);
});

test('errores claros', () => {
  assert.throws(() => simulate('fifo', 3, REFS), /política desconocida/);
  assert.throws(() => simulate('lru', 0, REFS), RangeError);
});

# Semana 9: almacenamiento físico, páginas con slots y buffer pool.
# Se usa dentro del contenedor del curso.
#
#   make          crea la base de datos chasqui_w09 con los datos de ejemplo,
#                 instala pageinspect, pg_buffercache y pg_freespacemap, y hace
#                 VACUUM (FREEZE, ANALYZE) para que las páginas queden en un
#                 estado conocido (el apunte de la semana explica por qué)
#   make test     la vuelve a crear y corre las pruebas (*.test.mjs)
#   make clean    borra chasqui_w09
#
# pages.sql, update.sql y buffers.sql se corren sobre chasqui_w09 con
#   psql -d chasqui_w09 -f pages.sql
# big.sql se corre aparte sobre chasqui_big (no cambia sus datos):
#   psql -d chasqui_big -f big.sql

DB = chasqui_w09
PSQL = psql -X -q -v ON_ERROR_STOP=1 -d $(DB)

.PHONY: all test clean

all:
	chasqui-load $(DB)
	$(PSQL) -c "CREATE EXTENSION IF NOT EXISTS pageinspect" \
	        -c "CREATE EXTENSION IF NOT EXISTS pg_buffercache" \
	        -c "CREATE EXTENSION IF NOT EXISTS pg_freespacemap" \
	        -c "VACUUM (FREEZE, ANALYZE)"

# Las pruebas cambian datos y vacían buffers: se corren una después de otra.
test: all
	PGDATABASE=$(DB) node --test --test-concurrency=1 --test-reporter=spec

clean:
	dropdb --if-exists $(DB)