-- Semana 11: los mismos joins sobre la tabla grande (2 millones de envíos).
-- Se corre sobre la copia de chasqui_big que crea make big:
--   psql -d chasqui_w11_big -f big.sql
-- Los tiempos y el reparto entre hit y read cambian en cada corrida.
SET max_parallel_workers_per_gather = 0;

-- 1. Los números reales de b y n (compárenlos con los supuestos del apunte).
-- name: sizes
SELECT relname, relpages AS b, reltuples::bigint AS n
FROM pg_class
WHERE relname IN ('customer', 'shipment', 'shipment_sender_idx')
ORDER BY relname;

-- name: height
SELECT level, level + 1 AS h FROM bt_metap('shipment_sender_idx');

-- 2. El join completo con work_mem = 4MB: la hash table de customer no entra
--    en memoria y el hash join se parte en 2 batches.
-- name: hash
EXPLAIN (ANALYZE, BUFFERS, COSTS OFF, TIMING OFF, SUMMARY ON)
SELECT c.city, sum(s.price)
FROM customer c JOIN shipment s ON s.sender_id = c.customer_id
GROUP BY c.city;

-- 3. Con work_mem = 256MB entra entera: un solo batch, sin archivos
--    temporales.
SET work_mem = '256MB';

-- name: hash_big_mem
EXPLAIN (ANALYZE, BUFFERS, COSTS OFF, TIMING OFF, SUMMARY ON)
SELECT c.city, sum(s.price)
FROM customer c JOIN shipment s ON s.sender_id = c.customer_id
GROUP BY c.city;

RESET work_mem;

-- 4. Sin hash join: merge join. El orden de shipment por sender_id sale del
--    índice, pero cada fila es una visita a una página del heap.
SET enable_hashjoin = off;

-- name: merge
EXPLAIN (ANALYZE, BUFFERS, COSTS OFF, TIMING OFF, SUMMARY ON)
SELECT c.city, sum(s.price)
FROM customer c JOIN shipment s ON s.sender_id = c.customer_id
GROUP BY c.city;

-- 5. Merge join sin index scans: hay que ordenar shipment (external merge).
SET enable_indexscan = off;

-- name: merge_sort
EXPLAIN (ANALYZE, BUFFERS, COSTS OFF, TIMING OFF, SUMMARY ON)
SELECT c.city, sum(s.price)
FROM customer c JOIN shipment s ON s.sender_id = c.customer_id
GROUP BY c.city;

RESET enable_hashjoin;
RESET enable_indexscan;

-- 6. Un join selectivo: Iquitos tiene 38 clientes entre 201 800. Index
--    nested loop con el índice en shipment(sender_id).
-- name: cities
SELECT city, count(*) FROM customer GROUP BY city ORDER BY 2 LIMIT 3;

-- name: selective
EXPLAIN (ANALYZE, BUFFERS, COSTS OFF, TIMING OFF, SUMMARY ON)
SELECT c.customer_id, c.last_name, s.shipment_id, s.price
FROM customer c JOIN shipment s ON s.sender_id = c.customer_id
WHERE c.city = 'Iquitos';

-- 7. Una ciudad con 7083 clientes: el hash join vuelve a ganar.
-- name: less_selective
EXPLAIN (ANALYZE, BUFFERS, COSTS OFF, TIMING OFF, SUMMARY ON)
SELECT c.customer_id, c.last_name, s.shipment_id, s.price
FROM customer c JOIN shipment s ON s.sender_id = c.customer_id
WHERE c.city = 'Huaraz';

// Fórmulas de costo de Silberschatz, cap. 15, para revisar las cuentas del
// apunte. Cada función devuelve transferencias de bloques y seeks; time()
// los convierte a milisegundos con t_T y t_S. Una relación se describe con
// { b, n }: páginas y tuplas. M son las páginas de memoria disponibles.

export const HDD = { tT: 0.1, tS: 4 }; // ms, los valores del libro
export const SSD = { tT: 0.01, tS: 0.09 };

export function time({ transfers, seeks }, { tT, tS } = HDD) {
  return transfers * tT + seeks * tS;
}

// ⌈log_k(x)⌉ con enteros, para no depender del redondeo de Math.log.
export function ceilLog(k, x) {
  let passes = 0;
  for (let reach = 1; reach < x; reach *= k) passes++;
  return passes;
}

// --- Selección (sec. 15.3) ---

// A1: leer todo el archivo, un seek al comienzo.
export const linearScan = (r) => ({ transfers: r.b, seeks: 1 });

// A2: clustered index, igualdad sobre una clave: bajar el árbol (h páginas)
// y leer una página de datos, cada una con su seek.
export const clusteredKeyEq = (h) => ({ transfers: h + 1, seeks: h + 1 });

// A4: índice secundario, igualdad sobre una no clave: una página por tupla
// encontrada, en el peor caso cada una en un lugar distinto.
export const secondaryEq = (h, n) => ({ transfers: h + n, seeks: h + n });

// --- Ordenamiento externo (sec. 15.4) ---

// bb: páginas de buffer por run en la fusión. Sin contar la escritura final.
export function externalSort(b, M, bb = 1) {
  const runs = Math.ceil(b / M);
  if (runs <= 1) return { runs: 1, ways: 0, passes: 0, transfers: b, seeks: 1 };
  const ways = Math.floor(M / bb) - 1;
  const passes = ceilLog(ways, runs);
  return {
    runs,
    ways,
    passes,
    transfers: b * (2 * passes + 1),
    seeks: 2 * runs + Math.ceil(b / bb) * (2 * passes - 1),
  };
}

// --- Joins (sec. 15.5); r es la relación de afuera (outer) ---

export const nestedLoop = (r, s) => ({
  transfers: r.n * s.b + r.b,
  seeks: r.n + r.b,
});

// M − 2 páginas para el bloque de r, una para s y una para la salida.
export function blockNestedLoop(r, s, M) {
  if (M >= r.b + 2) return { blocks: 1, transfers: r.b + s.b, seeks: 2 };
  const blocks = Math.ceil(r.b / (M - 2));
  return { blocks, transfers: blocks * s.b + r.b, seeks: 2 * blocks };
}

// Se lee r entera y se hacen `lookups` búsquedas en el índice de s, cada una
// de `pagesPerLookup` páginas (altura del índice + páginas de datos).
export function indexNestedLoop(r, lookups, pagesPerLookup) {
  const transfers = r.b + lookups * pagesPerLookup;
  return { transfers, seeks: transfers };
}

// Ordena cada entrada que no venga ordenada y escribe el resultado, luego
// lee las dos una vez para fusionarlas.
export function mergeJoin(r, s, M, { rSorted = false, sSorted = false } = {}) {
  const sortCost = (x) => {
    const sort = externalSort(x.b, M);
    return { ...sort, transfers: sort.transfers + x.b };
  };
  const sortR = rSorted ? null : sortCost(r);
  const sortS = sSorted ? null : sortCost(s);
  const merge = r.b + s.b;
  return {
    sortR,
    sortS,
    merge,
    transfers: (sortR?.transfers ?? 0) + (sortS?.transfers ?? 0) + merge,
  };
}

// build es la entrada con la que se arma la hash table. nh es el número de
// particiones; por omisión, ⌈b_build / M⌉ con un 20 % de holgura (el fudge
// factor del libro). bb: páginas de buffer por partición, para los seeks.
export function hashJoin(build, probe, M, nh, bb = 1) {
  if (M >= build.b + 2) return { nh: 0, recursive: false, transfers: build.b + probe.b };
  const parts = nh ?? Math.ceil(1.2 * Math.ceil(build.b / M));
  return {
    nh: parts,
    partitionPages: Math.ceil(build.b / parts),
    recursive: parts > M - 1 || Math.ceil(build.b / parts) > M - 2,
    transfers: 3 * (build.b + probe.b) + 4 * parts,
    seeks: 2 * (Math.ceil(build.b / bb) + Math.ceil(probe.b / bb)) + 2 * parts,
  };
}

// Operadores de consulta con la interfaz de iteradores (Silberschatz,
// sec. 15.7.2; nota 9 de CS186). Cada operador tiene:
//
//   await open()   prepara el operador (y abre a sus hijos)
//   next()         devuelve la siguiente fila (un objeto) o null si no hay más
//   close()        libera lo que haya usado (y cierra a sus hijos)
//
// Solo SeqScan habla con PostgreSQL, por eso open() es async: trae todas las
// filas de la tabla al abrir. El resto de operadores trabaja fila por fila
// sobre lo que les entregan sus hijos, sin saber de dónde viene.

export class SeqScan {
  // pool: un pg.Pool; table y columns se validan contra una lista fija,
  // porque los nombres de tabla no pueden ir como parámetros ($1) en SQL.
  constructor(pool, table, columns) {
    const allowed = { customer: true, shipment: true, agency: true, route: true, trip: true };
    if (!allowed[table]) throw new Error(`tabla no permitida: ${table}`);
    if (!columns.every((c) => /^[a-z_]+$/.test(c))) throw new Error('nombre de columna inválido');
    this.pool = pool;
    this.sql = `SELECT ${columns.join(', ')} FROM ${table}`;
  }

  async open() {
    const { rows } = await this.pool.query(this.sql);
    this.rows = rows;
    this.pos = 0;
  }

  next() {
    return this.pos < this.rows.length ? this.rows[this.pos++] : null;
  }

  close() {
    this.rows = null;
  }
}

// Devuelve las filas de child para las que predicate(row) es verdadero.
export class Filter {
  constructor(child, predicate) {
    this.child = child;
    this.predicate = predicate;
  }

  async open() {
    await this.child.open();
  }

  next() {
    // TODO: pedir filas a child hasta encontrar una que cumpla el predicado.
    throw new Error('Filter.next: por implementar');
  }

  close() {
    this.child.close();
  }
}

// Devuelve, de cada fila, solo las columnas pedidas (en ese orden).
export class Project {
  constructor(child, columns) {
    this.child = child;
    this.columns = columns;
  }

  async open() {
    await this.child.open();
  }

  next() {
    // TODO
    throw new Error('Project.next: por implementar');
  }

  close() {
    this.child.close();
  }
}

// Nested loop join: por cada fila de outer, recorre inner completo y devuelve
// la combinación ({ ...o, ...i }) de las parejas que cumplen predicate(o, i).
// Para "volver a empezar" el inner, ciérrenlo y ábranlo de nuevo, o guarden
// sus filas en open(): decidan y expliquen en un comentario qué costo tiene
// cada opción.
export class NestedLoopJoin {
  constructor(outer, inner, predicate) {
    this.outer = outer;
    this.inner = inner;
    this.predicate = predicate;
  }

  async open() {
    // TODO
    throw new Error('NestedLoopJoin.open: por implementar');
  }

  next() {
    // TODO
    throw new Error('NestedLoopJoin.next: por implementar');
  }

  close() {
    this.outer.close();
    this.inner.close();
  }
}

// Hash join en memoria: en open() lee todo build y arma un Map de
// buildKey(row) -> filas; en next() recorre probe y devuelve las
// combinaciones con la misma clave.
export class HashJoin {
  constructor(build, probe, buildKey, probeKey) {
    this.build = build;
    this.probe = probe;
    this.buildKey = buildKey;
    this.probeKey = probeKey;
  }

  async open() {
    // TODO (último paso del laboratorio)
    throw new Error('HashJoin.open: por implementar');
  }

  next() {
    throw new Error('HashJoin.next: por implementar');
  }

  close() {
    this.build.close();
    this.probe.close();
  }
}

// Ejecuta un plan completo y devuelve todas sus filas.
export async function run(plan) {
  await plan.open();
  const out = [];
  for (let row = plan.next(); row !== null; row = plan.next()) out.push(row);
  plan.close();
  return out;
}

-- Semana 11: algoritmos de join (sec. 15.5) en PostgreSQL.
-- Se corre sobre chasqui_w11 después de make:
--   psql -d chasqui_w11 -f joins.sql
-- La consulta de la sesión: cuánto facturan los remitentes de cada ciudad.
-- Se pide s.price para que el plan tenga que leer shipment (con count(*) le
-- bastaría el índice shipment_sender_idx).
SET max_parallel_workers_per_gather = 0;

-- 1. El plan que elige PostgreSQL: hash join con customer como build.
-- name: hash
EXPLAIN (ANALYZE, BUFFERS, COSTS OFF, TIMING OFF, SUMMARY OFF)
SELECT c.city, sum(s.price)
FROM customer c JOIN shipment s ON s.sender_id = c.customer_id
GROUP BY c.city;

-- 2. Sin hash join: merge join, con las dos entradas ordenadas por la clave.
SET enable_hashjoin = off;

-- name: merge
EXPLAIN (ANALYZE, BUFFERS, COSTS OFF, TIMING OFF, SUMMARY OFF)
SELECT c.city, sum(s.price)
FROM customer c JOIN shipment s ON s.sender_id = c.customer_id
GROUP BY c.city;

-- 3. Sin hash join ni merge join: nested loop.
SET enable_mergejoin = off;

-- name: nestloop
EXPLAIN (ANALYZE, BUFFERS, COSTS OFF, TIMING OFF, SUMMARY OFF)
SELECT c.city, sum(s.price)
FROM customer c JOIN shipment s ON s.sender_id = c.customer_id
GROUP BY c.city;

-- 4. Sin ninguno de los tres: el planificador igual tiene que hacer el join,
--    así que usa uno "desalentado" y lo marca con Disabled: true.
SET enable_nestloop = off;

-- name: disabled
EXPLAIN (COSTS OFF)
SELECT c.city, sum(s.price)
FROM customer c JOIN shipment s ON s.sender_id = c.customer_id
GROUP BY c.city;

RESET ALL;
SET max_parallel_workers_per_gather = 0;

-- 5. Hash join con poca memoria: work_mem = 64kB no alcanza para la hash
--    table de shipment, y el hash join se parte en batches (particiones) que
--    se escriben en archivos temporales.
SET work_mem = '64kB';

-- name: batches
EXPLAIN (ANALYZE, BUFFERS, COSTS OFF, TIMING OFF, SUMMARY OFF)
SELECT s.shipment_id, e.seq, e.kind
FROM shipment s JOIN shipment_event e ON e.shipment_id = s.shipment_id;

RESET work_mem;

-- 6. Un join selectivo: los envíos de los 25 clientes de Huancavelica. En
--    chasqui_w11 shipment tiene solo 148 páginas y leerla entera es barato:
--    gana el hash join (comparen con big.sql).
-- name: selective
EXPLAIN (ANALYZE, BUFFERS, COSTS OFF, TIMING OFF, SUMMARY OFF)
SELECT c.customer_id, c.last_name, s.shipment_id, s.price
FROM customer c JOIN shipment s ON s.sender_id = c.customer_id
WHERE c.city = 'Huancavelica';

-- 7. Un solo cliente afuera: index nested loop.
-- name: one_customer
EXPLAIN (ANALYZE, BUFFERS, COSTS OFF, TIMING OFF, SUMMARY OFF)
SELECT c.last_name, s.shipment_id, s.price
FROM customer c JOIN shipment s ON s.sender_id = c.customer_id
WHERE c.customer_id = 53;

-- 8. Equi-join con una condición extra: el hash usa la igualdad y la
--    desigualdad queda como Join Filter.
-- name: equi_filter
EXPLAIN (ANALYZE, BUFFERS OFF, COSTS OFF, TIMING OFF, SUMMARY OFF)
SELECT a.name, b.name
FROM agency a JOIN agency b ON a.region = b.region AND a.agency_id < b.agency_id;

-- 9. Un theta join sin igualdad: ni hash join ni merge join sirven; solo
--    queda el nested loop (con el inner materializado en memoria).
-- name: theta
EXPLAIN (ANALYZE, BUFFERS OFF, COSTS OFF, TIMING OFF, SUMMARY OFF)
SELECT a.name, r.route_id
FROM agency a JOIN route r ON r.distance_km > a.agency_id * 40;

// Tres operadores con la misma interfaz que iterators.mjs (open, next,
// close), que no son parte del laboratorio. Sirven para ver el pipelining:
// la raíz pide filas y los hijos trabajan solo lo necesario.

// Las filas de un arreglo en memoria, como un VALUES de SQL.
export class Values {
  constructor(rows) {
    this.rows = rows;
  }

  async open() {
    this.pos = 0;
  }

  next() {
    return this.pos < this.rows.length ? this.rows[this.pos++] : null;
  }

  close() {}
}

// Las primeras k filas de child. Cuando ya tiene k, no le pide más.
export class Limit {
  constructor(child, k) {
    this.child = child;
    this.k = k;
  }

  async open() {
    await this.child.open();
    this.count = 0;
  }

  next() {
    if (this.count >= this.k) return null;
    const row = this.child.next();
    if (row !== null) this.count++;
    return row;
  }

  close() {
    this.child.close();
  }
}

// Deja pasar las filas de child sin cambiarlas y cuenta las llamadas.
export class Counter {
  constructor(child) {
    this.child = child;
    this.stats = { opens: 0, nexts: 0, rows: 0, closes: 0 };
  }

  async open() {
    this.stats.opens++;
    await this.child.open();
  }

  next() {
    this.stats.nexts++;
    const row = this.child.next();
    if (row !== null) this.stats.rows++;
    return row;
  }

  close() {
    this.stats.closes++;
    this.child.close();
  }
}

// La misma idea con generadores de JavaScript: cada next() del generador
// corre el cuerpo hasta el siguiente yield. Si el consumidor deja de pedir
// (un break en for...of), el bloque finally hace las veces de close().
export function* values(rows, log = []) {
  try {
    for (const row of rows) {
      log.push(`values: entrega ${JSON.stringify(row)}`);
      yield row;
    }
  } finally {
    log.push('values: cerrado');
  }
}

export function* limit(input, k) {
  if (k <= 0) return;
  let count = 0;
  for (const row of input) {
    yield row;
    if (++count === k) return;
  }
}

-- Semana 11: otras operaciones (sec. 15.6) y evaluación de expresiones
-- (sec. 15.7) en PostgreSQL.
-- Se corre sobre chasqui_w11 después de make:
--   psql -d chasqui_w11 -f other.sql
SET max_parallel_workers_per_gather = 0;

-- 1. Eliminación de duplicados con hash: HashAggregate.
-- name: distinct_hash
EXPLAIN (ANALYZE, BUFFERS OFF, COSTS OFF, TIMING OFF, SUMMARY OFF)
SELECT DISTINCT city FROM customer;

-- 2. La misma consulta ordenando: Sort y después Unique, que descarta cada
--    tupla igual a la anterior.
SET enable_hashagg = off;

-- name: distinct_sort
EXPLAIN (ANALYZE, BUFFERS OFF, COSTS OFF, TIMING OFF, SUMMARY OFF)
SELECT DISTINCT city FROM customer;

-- 3. Agregación ordenando: GroupAggregate sobre la entrada ordenada.
-- name: group_sort
EXPLAIN (ANALYZE, BUFFERS OFF, COSTS OFF, TIMING OFF, SUMMARY OFF)
SELECT status, count(*), avg(price)
FROM shipment
GROUP BY status;

RESET enable_hashagg;

-- 4. Agregación con hash: una entrada por grupo en la hash table.
-- name: group_hash
EXPLAIN (ANALYZE, BUFFERS OFF, COSTS OFF, TIMING OFF, SUMMARY OFF)
SELECT status, count(*), avg(price)
FROM shipment
GROUP BY status;

-- name: group_result
SELECT status, count(*), round(avg(price), 2) AS avg_price
FROM shipment
GROUP BY status
ORDER BY status;

-- 5. Operaciones de conjuntos: INTERSECT con hash (HashSetOp) y UNION como
--    Append seguido de eliminación de duplicados.
-- name: intersect
EXPLAIN (ANALYZE, BUFFERS OFF, COSTS OFF, TIMING OFF, SUMMARY OFF)
SELECT city FROM customer
INTERSECT
SELECT city FROM agency;

-- name: union
EXPLAIN (ANALYZE, BUFFERS OFF, COSTS OFF, TIMING OFF, SUMMARY OFF)
SELECT city FROM customer
UNION
SELECT city FROM agency;

-- 6. Outer join: cuántos envíos mandó cada cliente, también los que no
--    mandaron ninguno (82 filas de customer sin pareja).
-- name: outer_join
EXPLAIN (ANALYZE, BUFFERS OFF, COSTS OFF, TIMING OFF, SUMMARY OFF)
SELECT c.customer_id, count(s.shipment_id)
FROM customer c LEFT JOIN shipment s ON s.sender_id = c.customer_id
GROUP BY c.customer_id;

-- 7. Pipelining: con LIMIT, el Seq Scan se detiene apenas el Limit tiene sus
--    5 filas. Lean las páginas (Buffers) y las filas de cada nodo.
-- name: limit_pipeline
EXPLAIN (ANALYZE, BUFFERS, COSTS OFF, TIMING OFF, SUMMARY OFF)
SELECT * FROM shipment WHERE status = 'returned' LIMIT 5;

-- 8. Un operador bloqueante: el Sort tiene que leer las 12 000 filas antes de
--    entregar la primera, aunque solo se pidan 5.
-- name: limit_blocking
EXPLAIN (ANALYZE, BUFFERS, COSTS OFF, TIMING OFF, SUMMARY OFF)
SELECT * FROM shipment WHERE status = 'returned' ORDER BY price LIMIT 5;

-- 9. Un CTE con MATERIALIZED se guarda en un tuplestore (CTE Scan), pero
--    PostgreSQL lo llena a pedido: con LIMIT tampoco lee toda la tabla.
-- name: cte_materialized
EXPLAIN (ANALYZE, BUFFERS, COSTS OFF, TIMING OFF, SUMMARY OFF)
WITH r AS MATERIALIZED (
  SELECT * FROM shipment WHERE status = 'returned'
)
SELECT * FROM r LIMIT 5;

// Pipelining por demanda: un Limit encima de un SeqScan pide solo las filas
// que necesita. Se corre dentro del contenedor, en code/week-11:
//   PGDATABASE=chasqui_w11 node pipeline.mjs
import pg from 'pg';
import { SeqScan, run } from './iterators.mjs';
import { Counter, Limit, values, limit } from './limit.mjs';

const pool = new pg.Pool();

// SeqScan anota cuántas filas trajo de PostgreSQL al abrir.
class MeasuredScan extends SeqScan {
  async open() {
    await super.open();
    this.fetched = this.rows.length;
  }
}

// 1. Con las clases: Limit(5) sobre un SeqScan de shipment.
const seq = new MeasuredScan(pool, 'shipment', ['shipment_id', 'price']);
const scan = new Counter(seq);
const rows = await run(new Limit(scan, 5));
console.log('filas del plan:', rows.map((r) => r.shipment_id).join(', '));
console.log('llamadas al SeqScan:', scan.stats);
console.log('filas que SeqScan.open() trajo de PostgreSQL:', seq.fetched);

// 2. Con generadores: el log muestra en qué orden trabaja cada operador.
const log = [];
for (const row of limit(values([{ id: 1 }, { id: 2 }, { id: 3 }, { id: 4 }], log), 2)) {
  log.push(`raíz: recibe ${JSON.stringify(row)}`);
}
console.log(log.join('\n'));

await pool.end();

-- Semana 11: medidas de costo y algoritmos de selección (secs. 15.2 y 15.3).
-- Se corre sobre chasqui_w11 después de make:
--   psql -d chasqui_w11 -f selection.sql
-- Los planes usan COSTS OFF y TIMING OFF para que la salida sea la misma en
-- cada corrida; lo que cambia entre corridas es el reparto entre hit y read.
SET max_parallel_workers_per_gather = 0;

-- 1. Los números del modelo de costo: páginas (b) y tuplas (n) de cada tabla,
--    según la última vez que se hizo ANALYZE.
-- name: sizes
SELECT relname, relpages AS b, reltuples::bigint AS n
FROM pg_class
WHERE relname IN ('customer', 'shipment')
ORDER BY relname;

-- 2. M: cuántas páginas de 8 KiB caben en work_mem.
-- name: memory
SELECT current_setting('work_mem') AS work_mem,
       current_setting('block_size') AS block_size,
       pg_size_bytes(current_setting('work_mem'))
         / current_setting('block_size')::int AS m;

-- 3. Altura de los B+ trees: level es el nivel de la raíz, contando las hojas
--    como nivel 0. La altura h_i del libro es level + 1.
-- name: heights
SELECT 'shipment_pkey' AS index, level, level + 1 AS h
FROM bt_metap('shipment_pkey')
UNION ALL
SELECT 'shipment_sender_idx', level, level + 1
FROM bt_metap('shipment_sender_idx');

-- 4. A1, búsqueda lineal: status no tiene índice.
-- name: a1
EXPLAIN (ANALYZE, BUFFERS, COSTS OFF, TIMING OFF, SUMMARY OFF)
SELECT * FROM shipment WHERE status = 'returned';

-- 5. Igualdad sobre la clave primaria.
-- name: pk_eq
EXPLAIN (ANALYZE, BUFFERS, COSTS OFF, TIMING OFF, SUMMARY OFF)
SELECT * FROM shipment WHERE shipment_id = 4242;

-- 6. A4, índice secundario con igualdad sobre una no clave: el remitente 53
--    tiene 10 envíos.
-- name: a4
EXPLAIN (ANALYZE, BUFFERS, COSTS OFF, TIMING OFF, SUMMARY OFF)
SELECT * FROM shipment WHERE sender_id = 53;

-- 7. Comparación sobre la clave primaria: la tabla se cargó en orden de
--    shipment_id, así que las 199 tuplas están en pocas páginas seguidas.
-- name: pk_range
EXPLAIN (ANALYZE, BUFFERS, COSTS OFF, TIMING OFF, SUMMARY OFF)
SELECT * FROM shipment WHERE shipment_id < 200;

-- 8. Qué tan ordenada está la tabla por cada columna (1 = en el mismo orden
--    que las páginas; 0 = sin relación).
-- name: correlation
SELECT attname, round(correlation::numeric, 3) AS correlation
FROM pg_stats
WHERE tablename = 'shipment'
  AND attname IN ('shipment_id', 'sender_id', 'created_at')
ORDER BY attname;

-- 9. Comparación sobre created_at, con un índice secundario: pocas tuplas,
--    pero repartidas en casi todas las páginas.
-- name: range_small
EXPLAIN (ANALYZE, BUFFERS, COSTS OFF, TIMING OFF, SUMMARY OFF)
SELECT * FROM shipment WHERE created_at >= '2026-06-20';

-- 10. La misma comparación, pero con el 69 % de las tuplas.
-- name: range_large
EXPLAIN (ANALYZE, BUFFERS, COSTS OFF, TIMING OFF, SUMMARY OFF)
SELECT * FROM shipment WHERE created_at >= '2026-03-01';

-- 11. Conjunción con dos índices: intersección de bitmaps (BitmapAnd).
-- name: conj
EXPLAIN (ANALYZE, BUFFERS, COSTS OFF, TIMING OFF, SUMMARY OFF)
SELECT * FROM shipment
WHERE sender_id BETWEEN 100 AND 140
  AND created_at >= '2026-06-01';

-- 12. Conjunción en la que basta un índice: el resto se filtra en el heap.
-- name: conj_one
EXPLAIN (ANALYZE, BUFFERS, COSTS OFF, TIMING OFF, SUMMARY OFF)
SELECT * FROM shipment
WHERE sender_id = 53
  AND status = 'delivered';

-- 13. Disyunción: unión de bitmaps (BitmapOr).
-- name: disj
EXPLAIN (ANALYZE, BUFFERS, COSTS OFF, TIMING OFF, SUMMARY OFF)
SELECT * FROM shipment
WHERE sender_id = 53
   OR created_at >= '2026-06-25';

-- 14. Disyunción con una condición sin índice: hay que leer todo.
-- name: disj_seq
EXPLAIN (ANALYZE, BUFFERS, COSTS OFF, TIMING OFF, SUMMARY OFF)
SELECT * FROM shipment
WHERE sender_id = 53
   OR status = 'returned';

-- 15. Negación: casi todas las tuplas cumplen, el índice no sirve.
-- name: neg
EXPLAIN (ANALYZE, BUFFERS, COSTS OFF, TIMING OFF, SUMMARY OFF)
SELECT * FROM shipment WHERE sender_id <> 53;

-- Semana 11: ordenamiento externo (sec. 15.4) en PostgreSQL.
-- Se corre sobre chasqui_w11 después de make:
--   psql -d chasqui_w11 -f sort.sql
-- price no tiene índice: para ORDER BY price hay que ordenar de verdad.
SET max_parallel_workers_per_gather = 0;

-- 1. Con work_mem = 4MB (M = 512 páginas), las 148 páginas de shipment
--    entran en memoria: un quicksort y listo.
-- name: in_memory
EXPLAIN (ANALYZE, BUFFERS, COSTS OFF, TIMING OFF, SUMMARY OFF)
SELECT * FROM shipment ORDER BY price;

-- 2. Con work_mem = 64kB (M = 8 páginas) no entran: external merge sort,
--    con los runs en archivos temporales (temp read / written).
SET work_mem = '64kB';

-- name: external
EXPLAIN (ANALYZE, BUFFERS, COSTS OFF, TIMING OFF, SUMMARY OFF)
SELECT * FROM shipment ORDER BY price;

-- 3. Con LIMIT no hace falta ordenar todo: basta un heap con las 10 menores
--    (top-N heapsort), que entra en memoria aunque work_mem sea chico.
-- name: top_n
EXPLAIN (ANALYZE, BUFFERS, COSTS OFF, TIMING OFF, SUMMARY OFF)
SELECT * FROM shipment ORDER BY price LIMIT 10;

-- 4. trace_sort escribe en el log (y, con client_min_messages = log, también
--    en psql) cada run que se genera y cada pasada de fusión.
SET trace_sort = on;
SET client_min_messages = log;

-- name: traced
EXPLAIN (ANALYZE, BUFFERS OFF, COSTS OFF, TIMING OFF, SUMMARY OFF)
SELECT * FROM shipment ORDER BY price;

RESET ALL;

// 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;
}

// Pruebas de big.sql sobre chasqui_w11_big. Si esa base de datos no existe
// (hay que crearla con make big), las pruebas se saltan.
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('./big.sql');
const DB = 'chasqui_w11_big';

let client = null;
before(async () => {
  const admin = new pg.Client({ database: 'postgres' });
  await admin.connect();
  const { rowCount } = await admin.query('SELECT 1 FROM pg_database WHERE datname = $1', [DB]);
  await admin.end();
  if (rowCount === 1) {
    client = new pg.Client({ database: DB });
    await client.connect();
  }
});
after(() => client?.end());

async function plan(sql, settings = {}) {
  await client.query('RESET ALL');
  await client.query('SET max_parallel_workers_per_gather = 0');
  for (const [k, v] of Object.entries(settings)) await client.query(`SET ${k} = '${v}'`);
  const { rows } = await client.query(sql);
  await client.query('RESET ALL');
  return rows.map((r) => r['QUERY PLAN']).join('\n');
}

test('big: tamaños, altura del índice y planes de join', async (t) => {
  if (!client) return t.skip(`no existe ${DB} (make big)`);
  const sizes = (await client.query(q.sizes)).rows;
  assert.deepEqual(sizes.map((r) => [r.relname, r.n]),
    [['customer', '201800'], ['shipment', '2012000'], ['shipment_sender_idx', '2012000']]);
  assert.equal((await client.query(q.height)).rows[0].h, '3');

  const hash = await plan(q.hash);
  assert.match(hash, /Hash Join \(actual rows=2012000\.00/);
  assert.match(hash, /Batches: 2 /);
  assert.match(hash, /temp read=\d+ written=\d+/);
  const bigMem = await plan(q.hash_big_mem, { work_mem: '256MB' });
  assert.match(bigMem, /Buckets: \d+ {2}Batches: 1 /);
  assert.doesNotMatch(bigMem, /temp/);

  const merge = await plan(q.merge, { enable_hashjoin: 'off' });
  assert.match(merge, /Merge Join[\s\S]*Index Scan using shipment_sender_idx/);
  const mergeSort = await plan(q.merge_sort, { enable_hashjoin: 'off', enable_indexscan: 'off' });
  assert.match(mergeSort, /Merge Join[\s\S]*Sort Method: external merge/);

  const cities = (await client.query(q.cities)).rows;
  assert.deepEqual(cities[0], { city: 'Iquitos', count: '38' });
  const sel = await plan(q.selective);
  assert.match(sel, /^Nested Loop \(actual rows=370\.00/);
  assert.match(sel, /Index Cond: \(sender_id = c\.customer_id\)/);
  assert.match(await plan(q.less_selective), /^Hash Join \(actual rows=71114\.00/);
});

// Las cuentas del apunte, hechas con las fórmulas de costs.mjs. Datos de
// referencia (supuestos): shipment con 33 000 páginas y 2·10⁶ tuplas,
// customer con 2000 páginas y 200 000 tuplas.
import { test } from 'node:test';
import assert from 'node:assert/strict';
import * as cost from './costs.mjs';

const shipment = { b: 33000, n: 2_000_000 };
const customer = { b: 2000, n: 200_000 };

test('ceilLog sin errores de redondeo', () => {
  assert.equal(cost.ceilLog(19, 361), 2);
  assert.equal(cost.ceilLog(19, 362), 3);
  assert.equal(cost.ceilLog(19, 1650), 3);
  assert.equal(cost.ceilLog(511, 65), 1);
  assert.equal(cost.ceilLog(7, 1), 0);
});

test('selección: A1, A2 y A4 en milisegundos', () => {
  assert.equal(cost.time(cost.linearScan(shipment)), 3304);
  assert.equal(cost.time(cost.clusteredKeyEq(3)), 16.4);
  assert.ok(Math.abs(cost.time(cost.secondaryEq(3, 10)) - 53.3) < 1e-9);
});

test('A4 deja de convenir frente a A1 a partir de unas 800 tuplas', () => {
  const a1 = cost.time(cost.linearScan(shipment));
  assert.ok(cost.time(cost.secondaryEq(3, 802)) < a1);
  assert.ok(cost.time(cost.secondaryEq(3, 803)) > a1);
});

test('external sort de shipment con M = 512, 102 y 20', () => {
  assert.deepEqual(cost.externalSort(33000, 512),
    { runs: 65, ways: 511, passes: 1, transfers: 99000, seeks: 33130 });
  const m20 = cost.externalSort(33000, 20);
  assert.deepEqual([m20.runs, m20.passes, m20.transfers], [1650, 3, 231000]);
  const m102 = cost.externalSort(33000, 102);
  assert.deepEqual([m102.runs, m102.passes, m102.transfers], [324, 2, 165000]);
});

test('external sort: si cabe en memoria, una sola lectura', () => {
  assert.deepEqual(cost.externalSort(148, 512), { runs: 1, ways: 0, passes: 0, transfers: 148, seeks: 1 });
});

test('external sort con b_b = 4: menos seeks, más pasadas', () => {
  const bb1 = cost.externalSort(33000, 20, 1);
  const bb4 = cost.externalSort(33000, 20, 4);
  assert.equal(bb4.ways, 4);
  assert.equal(bb4.passes, 6);
  assert.ok(bb4.seeks < bb1.seeks);
  assert.ok(bb4.transfers > bb1.transfers);
});

test('nested loop join: unos 6,6·10⁹ bloques, más de 7 días', () => {
  const c = cost.nestedLoop(customer, shipment);
  assert.equal(c.transfers, 6_600_002_000);
  assert.ok(cost.time(c) / 1000 / 86400 > 7);
});

test('block nested loop join: customer afuera, shipment afuera, todo en memoria', () => {
  assert.deepEqual(cost.blockNestedLoop(customer, shipment, 102), { blocks: 20, transfers: 662000, seeks: 40 });
  assert.equal(cost.blockNestedLoop(shipment, customer, 102).transfers, 693000);
  assert.equal(cost.blockNestedLoop(customer, shipment, 2002).transfers, 35000);
  assert.equal(Math.round(cost.time(cost.blockNestedLoop(customer, shipment, 102)) / 1000), 66);
});

test('index nested loop join: todo customer, 200 clientes y 10 000 clientes', () => {
  assert.equal(cost.indexNestedLoop(customer, 200000, 13).transfers, 2602000);
  assert.equal(cost.indexNestedLoop(customer, 200, 13).transfers, 4600);
  assert.equal(cost.indexNestedLoop(customer, 10000, 13).transfers, 132000);
});

test('merge join ordenando las dos entradas: unos 240 000 bloques', () => {
  const c = cost.mergeJoin(customer, shipment, 102);
  assert.deepEqual([c.sortR.runs, c.sortR.passes, c.sortR.transfers], [20, 1, 8000]);
  assert.deepEqual([c.sortS.runs, c.sortS.passes, c.sortS.transfers], [324, 2, 198000]);
  assert.equal(c.transfers, 241000);
  assert.equal(cost.mergeJoin(customer, shipment, 102, { rSorted: true, sSorted: true }).transfers, 35000);
});

test('hash join: unos 105 000 bloques con M = 102, 35 000 si el build entra', () => {
  const c = cost.hashJoin(customer, shipment, 102, 25);
  assert.equal(c.transfers, 105100);
  assert.equal(c.partitionPages, 80);
  assert.equal(c.recursive, false);
  assert.equal(cost.hashJoin(customer, shipment, 102).nh, 24);
  assert.equal(cost.hashJoin(customer, shipment, 2002).transfers, 35000);
});

test('hash join: con M = 40 hace falta particionar de forma recursiva', () => {
  assert.equal(cost.hashJoin(customer, shipment, 40).recursive, true);
});

test('las cuentas con los tamaños medidos en chasqui_big y M = 512', () => {
  const c = { b: 2040, n: 201800 };
  const s = { b: 24667, n: 2012000 };
  assert.equal(cost.blockNestedLoop(c, s, 512).transfers, 100708);
  assert.equal(cost.hashJoin(c, s, 512).transfers, 80141);
  assert.equal(cost.mergeJoin(c, s, 512).transfers, 133535);
});

// Pruebas de los iteradores: cada plan se compara con la misma consulta en
// PostgreSQL. Usan la base de datos chasqui_w11 (make test la crea).
import { test, after } from 'node:test';
import assert from 'node:assert/strict';
import pg from 'pg';
import { SeqScan, Filter, Project, NestedLoopJoin, HashJoin, run } from './iterators.mjs';

const pool = new pg.Pool();
after(() => pool.end());

// Orden estable para comparar conjuntos de filas.
const sorted = (rows) => rows.map((r) => JSON.stringify(r)).sort();

const CITY = 'Tumbes';
const expectedJoin = async () =>
  (await pool.query(
    `SELECT c.customer_id, c.last_name, s.shipment_id, s.price
     FROM customer c JOIN shipment s ON s.sender_id = c.customer_id
     WHERE c.city = $1`,
    [CITY],
  )).rows;

test('SeqScan devuelve todas las agencias', async () => {
  const rows = await run(new SeqScan(pool, 'agency', ['agency_id', 'name']));
  assert.equal(rows.length, 29);
});

test('Filter y Project', async () => {
  const plan = new Project(
    new Filter(new SeqScan(pool, 'agency', ['agency_id', 'name', 'region']), (r) => r.region === 'Puno'),
    ['name'],
  );
  assert.deepEqual(sorted(await run(plan)), sorted([{ name: 'Juliaca - Centro' }, { name: 'Puno - Centro' }]));
});

const joinPlan = (Join) => {
  const customers = new Filter(new SeqScan(pool, 'customer', ['customer_id', 'last_name', 'city']), (c) => c.city === CITY);
  const shipments = new SeqScan(pool, 'shipment', ['shipment_id', 'sender_id', 'price']);
  const join = Join === NestedLoopJoin
    ? new NestedLoopJoin(customers, shipments, (c, s) => s.sender_id === c.customer_id)
    : new HashJoin(customers, shipments, (c) => c.customer_id, (s) => s.sender_id);
  return new Project(join, ['customer_id', 'last_name', 'shipment_id', 'price']);
};

test('NestedLoopJoin: los envíos de los clientes de una ciudad', async () => {
  assert.deepEqual(sorted(await run(joinPlan(NestedLoopJoin))), sorted(await expectedJoin()));
});

test('HashJoin da el mismo resultado que NestedLoopJoin', async () => {
  assert.deepEqual(sorted(await run(joinPlan(HashJoin))), sorted(await expectedJoin()));
});

// Pruebas de limit.mjs: Limit no le pide al hijo más filas de las que
// necesita, y la versión con generadores cierra la fuente al terminar.
import { test, after } from 'node:test';
import assert from 'node:assert/strict';
import pg from 'pg';
import { SeqScan, run } from './iterators.mjs';
import { Values, Limit, Counter, values, limit } from './limit.mjs';

const pool = new pg.Pool();
after(() => pool.end());

const rows = [{ id: 1 }, { id: 2 }, { id: 3 }, { id: 4 }];

test('Limit(2) sobre Values: dos filas y dos llamadas a next()', async () => {
  const source = new Counter(new Values(rows));
  assert.deepEqual(await run(new Limit(source, 2)), [{ id: 1 }, { id: 2 }]);
  assert.deepEqual(source.stats, { opens: 1, nexts: 2, rows: 2, closes: 1 });
});

test('Limit más grande que la entrada: todas las filas y una llamada que da null', async () => {
  const source = new Counter(new Values(rows));
  assert.equal((await run(new Limit(source, 10))).length, 4);
  assert.equal(source.stats.nexts, 5);
});

test('Limit(0) no le pide nada al hijo', async () => {
  const source = new Counter(new Values(rows));
  assert.deepEqual(await run(new Limit(source, 0)), []);
  assert.equal(source.stats.nexts, 0);
});

test('Limit(5) sobre SeqScan de shipment: 5 llamadas a next()', async () => {
  const scan = new Counter(new SeqScan(pool, 'shipment', ['shipment_id']));
  assert.equal((await run(new Limit(scan, 5))).length, 5);
  assert.equal(scan.stats.nexts, 5);
});

test('generadores: limit pide solo lo necesario y cierra values', () => {
  const log = [];
  const out = [];
  for (const row of limit(values(rows, log), 2)) {
    log.push(`raíz: recibe ${JSON.stringify(row)}`);
    out.push(row);
  }
  assert.deepEqual(out, [{ id: 1 }, { id: 2 }]);
  assert.deepEqual(log, [
    'values: entrega {"id":1}',
    'raíz: recibe {"id":1}',
    'values: entrega {"id":2}',
    'raíz: recibe {"id":2}',
    'values: cerrado',
  ]);
});

// Pruebas de selection.sql, sort.sql, joins.sql y other.sql: corren el mismo
// texto que los archivos y revisan lo que dice el apunte de cada plan. No
// miran el reparto entre hit y read, que depende de la caché, sino el total.
import { test, before, after } from 'node:test';
import assert from 'node:assert/strict';
import pg from 'pg';
import { namedQueries } from './sqlfile.mjs';

const sel = namedQueries('./selection.sql');
const srt = namedQueries('./sort.sql');
const joi = namedQueries('./joins.sql');
const oth = namedQueries('./other.sql');

let client;
before(async () => {
  client = new pg.Client();
  await client.connect();
});
after(() => client.end());

// Corre una consulta con los SET indicados y devuelve el plan como texto.
async function plan(sql, settings = {}) {
  await client.query('RESET ALL');
  await client.query('SET max_parallel_workers_per_gather = 0');
  for (const [k, v] of Object.entries(settings)) await client.query(`SET ${k} = '${v}'`);
  const { rows } = await client.query(sql);
  await client.query('RESET ALL');
  return rows.map((r) => r['QUERY PLAN']).join('\n');
}

// Suma hit + read de la primera línea Buffers después de `node`.
function buffersOf(text, node) {
  const rest = text.slice(text.indexOf(node));
  const m = rest.match(/Buffers: shared ([^\n,]*)/);
  return [...m[1].matchAll(/(?:hit|read)=(\d+)/g)].reduce((s, x) => s + Number(x[1]), 0);
}

test('sizes y memory: 148 y 19 páginas, M = 512', async () => {
  const sizes = (await client.query(sel.sizes)).rows;
  assert.deepEqual(sizes.map((r) => [r.relname, r.b, r.n]), [['customer', 19, '1800'], ['shipment', 148, '12000']]);
  const m = (await client.query(sel.memory)).rows[0];
  assert.deepEqual(m, { work_mem: '4MB', block_size: '8192', m: '512' });
});

test('heights: los dos índices de shipment tienen altura 2', async () => {
  const rows = (await client.query(sel.heights)).rows;
  assert.deepEqual(rows.map((r) => [r.index, r.level, r.h]), [['shipment_pkey', '1', '2'], ['shipment_sender_idx', '1', '2']]);
});

test('A1: sequential scan de las 148 páginas, 319 devueltos', async () => {
  const p = await plan(sel.a1);
  assert.match(p, /^Seq Scan on shipment \(actual rows=319\.00/);
  assert.equal(buffersOf(p, 'Seq Scan'), 148);
});

test('pk_eq: index scan de 3 páginas (2 del índice y 1 del heap)', async () => {
  const p = await plan(sel.pk_eq);
  assert.match(p, /^Index Scan using shipment_pkey/);
  assert.equal(buffersOf(p, 'Index Scan'), 3);
});

test('A4: 10 envíos del remitente 53 en 10 páginas distintas, más 2 del índice', async () => {
  const p = await plan(sel.a4);
  assert.match(p, /^Bitmap Heap Scan on shipment \(actual rows=10\.00/);
  assert.match(p, /Heap Blocks: exact=10\n/);
  assert.equal(buffersOf(p, 'Bitmap Heap Scan'), 12);
  assert.equal(buffersOf(p, 'Bitmap Index Scan'), 2);
});

test('pk_range: 199 tuplas en 5 páginas', async () => {
  const p = await plan(sel.pk_range);
  assert.match(p, /^Index Scan using shipment_pkey on shipment \(actual rows=199\.00/);
  assert.equal(buffersOf(p, 'Index Scan'), 5);
});

test('correlation: shipment_id 1, created_at y sender_id cerca de 0', async () => {
  const rows = (await client.query(sel.correlation)).rows;
  const c = Object.fromEntries(rows.map((r) => [r.attname, Number(r.correlation)]));
  assert.equal(c.shipment_id, 1);
  assert.ok(Math.abs(c.created_at) < 0.05 && Math.abs(c.sender_id) < 0.05);
});

test('range_small: 507 tuplas en 147 de las 148 páginas; range_large: seq scan', async () => {
  const small = await plan(sel.range_small);
  assert.match(small, /^Bitmap Heap Scan on shipment \(actual rows=507\.00/);
  assert.match(small, /Heap Blocks: exact=147\n/);
  const large = await plan(sel.range_large);
  assert.match(large, /^Seq Scan on shipment \(actual rows=8249\.00/);
  assert.match(large, /Rows Removed by Filter: 3751/);
});

test('conjunciones y disyunciones: BitmapAnd, filtro, BitmapOr y seq scan', async () => {
  const conj = await plan(sel.conj);
  assert.match(conj, /BitmapAnd/);
  assert.match(conj, /^Bitmap Heap Scan on shipment \(actual rows=39\.00/);
  assert.match(conj, /Bitmap Index Scan on shipment_sender_idx \(actual rows=216\.00/);
  assert.match(conj, /Bitmap Index Scan on shipment_created_at_idx \(actual rows=1889\.00/);
  const one = await plan(sel.conj_one);
  assert.doesNotMatch(one, /BitmapAnd/);
  assert.match(one, /Filter: \(status = 'delivered'::text\)/);
  const disj = await plan(sel.disj);
  assert.match(disj, /BitmapOr/);
  assert.match(disj, /^Bitmap Heap Scan on shipment \(actual rows=151\.00/);
  const disjSeq = await plan(sel.disj_seq);
  assert.match(disjSeq, /^Seq Scan on shipment \(actual rows=329\.00/);
  const neg = await plan(sel.neg);
  assert.match(neg, /^Seq Scan on shipment \(actual rows=11990\.00/);
});

test('sort: quicksort con 4MB, external merge con 64kB, top-N heapsort con LIMIT', async () => {
  assert.match(await plan(srt.in_memory), /Sort Method: quicksort {2}Memory: 1413kB/);
  const ext = await plan(srt.external, { work_mem: '64kB' });
  assert.match(ext, /Sort Method: external merge {2}Disk: 1000kB/);
  assert.match(ext, /temp read=\d+ written=\d+/);
  assert.match(await plan(srt.top_n, { work_mem: '64kB' }), /Sort Method: top-N heapsort/);
});

test('trace_sort: 26 runs en 6 tapes, una pasada intermedia y la fusión final de 5', async () => {
  const logs = [];
  const onNotice = (n) => logs.push(n.message);
  client.on('notice', onNotice);
  await plan(srt.traced, { work_mem: '64kB', trace_sort: 'on', client_min_messages: 'log' });
  client.off('notice', onNotice);
  assert.ok(logs.some((l) => /switching to external sort with 6 tapes/.test(l)));
  assert.equal(logs.filter((l) => /finished writing run \d+ to tape/.test(l)).length, 26);
  assert.ok(logs.some((l) => /starting merge pass of 26 input runs on 6 tapes/.test(l)));
  assert.ok(logs.some((l) => /starting merge pass of 5 input runs on 5 tapes/.test(l)));
  assert.ok(logs.some((l) => /done \(except 5-way final merge\)/.test(l)));
});

test('joins: hash, merge, nested loop y Disabled: true', async () => {
  const hash = await plan(joi.hash);
  assert.match(hash, /Hash Join \(actual rows=12000\.00/);
  assert.match(hash, /Hash \(actual rows=1800\.00[\s\S]*Batches: 1 /);
  assert.match(hash, /Seq Scan on shipment s/);
  assert.equal(buffersOf(hash, 'Hash Join'), 167);
  const merge = await plan(joi.merge, { enable_hashjoin: 'off' });
  assert.match(merge, /Merge Join \(actual rows=12000\.00/);
  assert.match(merge, /Index Scan using shipment_sender_idx on shipment s/);
  assert.ok(buffersOf(merge, 'Merge Join') > 10000);
  const nl = await plan(joi.nestloop, { enable_hashjoin: 'off', enable_mergejoin: 'off' });
  assert.match(nl, /Nested Loop \(actual rows=12000\.00/);
  assert.match(nl, /Memoize[\s\S]*Hits: 10282 {2}Misses: 1718/);
  const dis = await plan(joi.disabled, { enable_hashjoin: 'off', enable_mergejoin: 'off', enable_nestloop: 'off' });
  assert.match(dis, /Nested Loop\n\s+Disabled: true/);
});

test('batches: con 64kB el hash de shipment se parte en 8 batches', async () => {
  const p = await plan(joi.batches, { work_mem: '64kB' });
  assert.match(p, /Hash Join \(actual rows=46421\.00/);
  assert.match(p, /Batches: 8 /);
  assert.match(p, /temp read=\d+ written=\d+/);
});

test('selective: hash join en chasqui_w11; one_customer: nested loop con índice', async () => {
  const p = await plan(joi.selective);
  assert.match(p, /^Hash Join \(actual rows=110\.00/);
  assert.match(p, /Seq Scan on customer c \(actual rows=25\.00/);
  const one = await plan(joi.one_customer);
  assert.match(one, /^Nested Loop \(actual rows=10\.00/);
  assert.match(one, /Bitmap Index Scan on shipment_sender_idx/);
});

test('equi_filter usa hash con Join Filter; theta solo puede usar nested loop', async () => {
  const eq = await plan(joi.equi_filter);
  assert.match(eq, /Hash Cond: \(a\.region = b\.region\)\n\s+Join Filter: \(a\.agency_id < b\.agency_id\)/);
  assert.match(eq, /^Hash Join \(actual rows=6\.00/);
  const theta = await plan(joi.theta);
  assert.match(theta, /^Nested Loop \(actual rows=962\.00/);
  assert.match(theta, /Materialize/);
  const off = await plan(joi.theta, { enable_nestloop: 'off' });
  assert.match(off, /Nested Loop/);
});

test('other: DISTINCT y GROUP BY con hash y con sort', async () => {
  assert.match(await plan(oth.distinct_hash), /^HashAggregate \(actual rows=26\.00/);
  const ds = await plan(oth.distinct_sort, { enable_hashagg: 'off' });
  assert.match(ds, /^Unique \(actual rows=26\.00[\s\S]*Sort \(actual rows=1800\.00/);
  assert.match(await plan(oth.group_sort, { enable_hashagg: 'off' }), /^GroupAggregate \(actual rows=4\.00/);
  assert.match(await plan(oth.group_hash), /^HashAggregate \(actual rows=4\.00/);
  const g = (await client.query(oth.group_result)).rows;
  assert.deepEqual(g.map((r) => [r.status, r.count, r.avg_price]), [
    ['delivered', '11115', '60.45'], ['in_transit', '66', '70.33'],
    ['registered', '500', '58.52'], ['returned', '319', '63.61'],
  ]);
});

test('other: INTERSECT, UNION y outer join', async () => {
  assert.match(await plan(oth.intersect), /^HashSetOp Intersect \(actual rows=26\.00/);
  assert.match(await plan(oth.union), /^HashAggregate \(actual rows=26\.00[\s\S]*Append \(actual rows=1829\.00/);
  assert.match(await plan(oth.outer_join), /Hash Right Join \(actual rows=12082\.00/);
});

test('pipelining: con LIMIT el seq scan lee 3 páginas; con ORDER BY, las 148', async () => {
  const pipe = await plan(oth.limit_pipeline);
  assert.match(pipe, /Seq Scan on shipment \(actual rows=5\.00[\s\S]*Rows Removed by Filter: 141/);
  assert.equal(buffersOf(pipe, 'Seq Scan'), 3);
  const block = await plan(oth.limit_blocking);
  assert.match(block, /Seq Scan on shipment \(actual rows=319\.00/);
  assert.equal(buffersOf(block, 'Seq Scan'), 148);
  const cte = await plan(oth.cte_materialized);
  assert.match(cte, /CTE Scan on r/);
  assert.equal(buffersOf(cte, 'Seq Scan'), 3);
});