-- Semana 10: índices sobre 2 millones de envíos. Crea índices y tablas de
-- prueba, así que se corre sobre una copia de chasqui_big (make big la crea:
-- chasqui_w10_big) o sobre su propio chasqui_big:
--   psql -d chasqui_w10_big -f big.sql
-- Al final borra todo lo que creó, para poder correrlo otra vez. Los tiempos
-- cambian de una corrida a otra y de una máquina a otra; las páginas, no.
\timing on
SET max_parallel_workers_per_gather = 0;
SET max_parallel_maintenance_workers = 0;

-- 1. Tamaños: la tabla y su clave primaria.
-- name: sizes
SELECT relname, relpages, reltuples::bigint AS reltuples,
       pg_size_pretty(pg_relation_size(oid)) AS size
FROM pg_class
WHERE relname IN ('shipment', 'shipment_pkey')
ORDER BY relpages DESC;

-- 2. Forma real del B+ tree de shipment_pkey: páginas y entradas por nivel.
-- name: pk_meta
SELECT root, level FROM bt_metap('shipment_pkey');

-- name: pk_levels
SELECT type, btpo_level AS level, count(*) AS pages,
       sum(live_items) AS items, round(avg(live_items)) AS avg_items
FROM bt_multi_page_stats('shipment_pkey', 1, -1)
GROUP BY type, btpo_level
ORDER BY btpo_level DESC;

-- name: pk_lookup
EXPLAIN (ANALYZE, BUFFERS, COSTS OFF, TIMING OFF, SUMMARY OFF)
SELECT * FROM shipment WHERE shipment_id = 1500000;

-- 3. Un índice secundario sobre una clave no única: sender_id.
-- name: sender_seq
EXPLAIN (ANALYZE, BUFFERS, COSTS OFF)
SELECT * FROM shipment WHERE sender_id = 777;

-- name: create_sender
CREATE INDEX shipment_sender_idx ON shipment (sender_id);

-- name: sender_idx
EXPLAIN (ANALYZE, BUFFERS, COSTS OFF)
SELECT * FROM shipment WHERE sender_id = 777;

-- name: create_sender_nodedup
CREATE INDEX shipment_sender_nodedup_idx ON shipment (sender_id)
  WITH (deduplicate_items = off);

-- name: sender_sizes
SELECT relname, relpages, pg_size_pretty(pg_relation_size(oid)) AS size
FROM pg_class
WHERE relname IN ('shipment_sender_idx', 'shipment_sender_nodedup_idx')
ORDER BY relname;

DROP INDEX shipment_sender_nodedup_idx;
DROP INDEX shipment_sender_idx;

-- 4. Bulk loading frente a inserciones una por una. Las dos tablas tienen
--    las mismas filas; en la primera el índice se crea al final (CREATE
--    INDEX ordena y arma el árbol de abajo hacia arriba), en la segunda
--    existe desde antes y recibe 2 millones de inserciones en orden aleatorio
--    de sender_id. Sin deduplicación, para comparar solo el llenado.
CREATE TABLE shipment_bulk (LIKE shipment);
CREATE TABLE shipment_incr (LIKE shipment);
CREATE INDEX shipment_incr_sender_idx ON shipment_incr (sender_id)
  WITH (deduplicate_items = off);

-- name: load_bulk
INSERT INTO shipment_bulk SELECT * FROM shipment;

-- name: create_bulk
CREATE INDEX shipment_bulk_sender_idx ON shipment_bulk (sender_id)
  WITH (deduplicate_items = off);

-- name: load_incr
INSERT INTO shipment_incr SELECT * FROM shipment;

-- name: bulk_vs_incr
SELECT 'bulk' AS build, count(*) AS leaves, round(avg(live_items)) AS avg_items,
       round(avg(free_size)) AS avg_free
FROM bt_multi_page_stats('shipment_bulk_sender_idx', 1, -1) WHERE type = 'l'
UNION ALL
SELECT 'incremental', count(*), round(avg(live_items)), round(avg(free_size))
FROM bt_multi_page_stats('shipment_incr_sender_idx', 1, -1) WHERE type = 'l';

DROP TABLE shipment_bulk;
DROP TABLE shipment_incr;

-- 5. Selectividad: un índice en created_at y rangos cada vez más anchos.
--    Recuerden que created_at no sigue el orden físico de la tabla.
-- name: create_created
CREATE INDEX shipment_created_idx ON shipment (created_at);

-- name: range_10min
EXPLAIN (ANALYZE, BUFFERS, COSTS OFF)
SELECT count(*), sum(price) FROM shipment
WHERE created_at >= '2026-03-02' AND created_at < '2026-03-02 00:10';

-- name: range_day
EXPLAIN (ANALYZE, BUFFERS, COSTS OFF)
SELECT count(*), sum(price) FROM shipment
WHERE created_at >= '2026-03-02' AND created_at < '2026-03-03';

-- name: range_month
EXPLAIN (ANALYZE, BUFFERS, COSTS OFF)
SELECT count(*), sum(price) FROM shipment
WHERE created_at >= '2026-03-01' AND created_at < '2026-04-01';

-- name: range_3months
EXPLAIN (COSTS OFF)
SELECT count(*), sum(price) FROM shipment
WHERE created_at >= '2026-02-01' AND created_at < '2026-05-01';

-- El mismo mes, obligando a cada estrategia.
SET enable_bitmapscan = off;
SET enable_seqscan = off;
-- name: month_index_scan
EXPLAIN (ANALYZE, BUFFERS, COSTS OFF)
SELECT count(*), sum(price) FROM shipment
WHERE created_at >= '2026-03-01' AND created_at < '2026-04-01';

RESET enable_seqscan;
SET enable_indexscan = off;
-- name: month_seq_scan
EXPLAIN (ANALYZE, BUFFERS, COSTS OFF)
SELECT count(*), sum(price) FROM shipment
WHERE created_at >= '2026-03-01' AND created_at < '2026-04-01';

RESET enable_bitmapscan;
RESET enable_indexscan;

-- 6. Dos índices de una columna: el planner los combina con un BitmapAnd.
-- name: create_origin
CREATE INDEX shipment_origin_idx ON shipment (origin_agency_id);

-- name: bitmap_and
EXPLAIN (ANALYZE, BUFFERS, COSTS OFF, TIMING OFF)
SELECT count(*) FROM shipment
WHERE origin_agency_id = 12
  AND created_at >= '2026-04-01' AND created_at < '2026-04-08';

DROP INDEX shipment_origin_idx;
DROP INDEX shipment_created_idx;

-- 7. Un índice compuesto y el orden de sus columnas.
-- name: create_origin_created
CREATE INDEX shipment_origin_created_idx ON shipment (origin_agency_id, created_at);

-- name: origin_created
EXPLAIN (ANALYZE, BUFFERS, COSTS OFF, TIMING OFF)
SELECT count(*) FROM shipment
WHERE origin_agency_id = 12
  AND created_at >= '2026-04-01' AND created_at < '2026-05-01';

-- Solo la segunda columna: skip scan.
-- name: skip_scan
EXPLAIN (ANALYZE, BUFFERS, COSTS OFF, TIMING OFF)
SELECT count(*) FROM shipment
WHERE created_at >= '2026-04-01' AND created_at < '2026-04-01 01:00';

DROP INDEX shipment_origin_created_idx;

-- name: create_created_origin
CREATE INDEX shipment_created_origin_idx ON shipment (created_at, origin_agency_id);

-- name: created_origin
EXPLAIN (ANALYZE, BUFFERS, COSTS OFF, TIMING OFF)
SELECT count(*) FROM shipment
WHERE origin_agency_id = 12
  AND created_at >= '2026-04-01' AND created_at < '2026-05-01';

DROP INDEX shipment_created_origin_idx;

-- 8. BRIN: un mínimo y un máximo por cada 128 páginas.
-- name: create_brin
CREATE INDEX shipment_id_brin ON shipment USING brin (shipment_id);

-- name: create_brin_created
CREATE INDEX shipment_created_brin ON shipment USING brin (created_at);

-- name: brin_sizes
SELECT relname, relpages, pg_size_pretty(pg_relation_size(oid)) AS size
FROM pg_class
WHERE relname IN ('shipment_pkey', 'shipment_id_brin', 'shipment_created_brin')
ORDER BY relpages DESC, relname;

-- name: brin_items
SELECT blknum, value
FROM brin_page_items(get_raw_page('shipment_id_brin', 2), 'shipment_id_brin')
ORDER BY blknum
LIMIT 4;

-- name: brin_items_created
SELECT blknum, value
FROM brin_page_items(get_raw_page('shipment_created_brin', 2), 'shipment_created_brin')
ORDER BY blknum
LIMIT 4;

-- Sin el B-tree de la clave primaria, para ver al BRIN trabajar.
SET enable_indexscan = off;
-- name: brin_id
EXPLAIN (ANALYZE, BUFFERS, COSTS OFF, TIMING OFF)
SELECT count(*) FROM shipment WHERE shipment_id BETWEEN 1000000 AND 1009999;
RESET enable_indexscan;

DROP INDEX shipment_id_brin;
DROP INDEX shipment_created_brin;

-- 9. Hash frente a B-tree en customer.dni (201 800 clientes).
-- name: create_dni_hash
CREATE INDEX customer_dni_hash ON customer USING hash (dni);

-- name: dni_sizes
SELECT relname, relpages, pg_size_pretty(pg_relation_size(oid)) AS size
FROM pg_class
WHERE relname IN ('customer_dni_key', 'customer_dni_hash')
ORDER BY relname;

-- name: dni_hash
EXPLAIN (ANALYZE, BUFFERS, COSTS OFF, TIMING OFF, SUMMARY OFF)
SELECT first_name, last_name FROM customer WHERE dni = '80123456';

-- La metapágina del índice quedó guardada en la sesión: otra búsqueda lee
-- solo el bucket y la página del heap.
-- name: dni_hash_again
EXPLAIN (ANALYZE, BUFFERS, COSTS OFF, TIMING OFF, SUMMARY OFF)
SELECT first_name, last_name FROM customer WHERE dni = '80123457';

DROP INDEX customer_dni_hash;

-- name: dni_btree
EXPLAIN (ANALYZE, BUFFERS, COSTS OFF, TIMING OFF, SUMMARY OFF)
SELECT first_name, last_name FROM customer WHERE dni = '80123456';

-- 10. Lo que cuesta mantener índices: insertar 512 000 envíos en una tabla
--     sin índices, con la clave primaria y con la clave primaria y tres
--     índices más.
CREATE TABLE shipment_w0 (LIKE shipment);
CREATE TABLE shipment_w1 (LIKE shipment INCLUDING INDEXES);
CREATE TABLE shipment_w4 (LIKE shipment INCLUDING INDEXES);
CREATE INDEX ON shipment_w4 (sender_id);
CREATE INDEX ON shipment_w4 (recipient_id);
CREATE INDEX ON shipment_w4 (created_at);

-- name: write_0
EXPLAIN (ANALYZE, WAL, COSTS OFF)
INSERT INTO shipment_w0 SELECT * FROM shipment WHERE shipment_id <= 600000;

-- name: write_1
EXPLAIN (ANALYZE, WAL, COSTS OFF)
INSERT INTO shipment_w1 SELECT * FROM shipment WHERE shipment_id <= 600000;

-- name: write_4
EXPLAIN (ANALYZE, WAL, COSTS OFF)
INSERT INTO shipment_w4 SELECT * FROM shipment WHERE shipment_id <= 600000;

DROP TABLE shipment_w0;
DROP TABLE shipment_w1;
DROP TABLE shipment_w4;

// B+ tree en memoria: claves numéricas únicas y un valor por clave.
//
// Representación (Silberschatz, sec. 14.3):
// - Un nodo tiene `keys` ordenadas. Un nodo interno con k claves tiene k + 1
//   hijos en `children`: children[i] guarda las claves < keys[i] y
//   children[i + 1] las claves >= keys[i].
// - Una hoja tiene `values` (values[i] es el valor de keys[i]) y `next`, la
//   hoja siguiente, para recorrer rangos sin volver a la raíz.
// - `order` es n, el máximo de hijos de un nodo interno. Una hoja guarda a lo
//   más n - 1 claves; al llenarse, se divide.

export class Node {
  constructor(leaf) {
    this.leaf = leaf;
    this.keys = [];
    this.children = leaf ? null : [];
    this.values = leaf ? [] : null;
    this.next = null;
  }
}

// Primera posición i tal que keys[i] > key (búsqueda binaria).
export function upperBound(keys, key) {
  let lo = 0;
  let hi = keys.length;
  while (lo < hi) {
    const mid = (lo + hi) >> 1;
    if (keys[mid] <= key) lo = mid + 1;
    else hi = mid;
  }
  return lo;
}

export class BPlusTree {
  constructor(order = 4) {
    if (order < 3) throw new RangeError('el orden de un B+ tree es al menos 3');
    this.order = order;
    this.root = new Node(true);
    this.size = 0;
  }

  // Devuelve el valor asociado a key, o undefined si no está.
  search(key) {
    // TODO (paso 1): bajar desde la raíz hasta la hoja que podría tener la
    // clave, usando upperBound en cada nodo interno, y buscarla en la hoja.
    throw new Error('search: por implementar');
  }

  // Inserta (key, value). Si la clave ya existe, reemplaza el valor.
  insert(key, value) {
    // TODO (paso 2): bajar recordando el camino; insertar en la hoja; si la
    // hoja queda con `order` claves, dividirla y copiar al padre la primera
    // clave de la hoja nueva; si un nodo interno queda con `order` claves,
    // dividirlo y subir la clave del medio. Si se divide la raíz, el árbol
    // crece un nivel.
    throw new Error('insert: por implementar');
  }

  // Generador de pares [key, value] con lo <= key <= hi, en orden.
  *range(lo, hi) {
    // TODO (paso 3): buscar la hoja de lo y avanzar por `next`.
    throw new Error('range: por implementar');
  }

  // Número de niveles: 1 si la raíz es una hoja.
  height() {
    let h = 1;
    for (let n = this.root; !n.leaf; n = n.children[0]) h++;
    return h;
  }

  // Dibujo del árbol por niveles, por ejemplo:
  //   [20]
  //   [10] [20 30]
  toString() {
    const lines = [];
    let level = [this.root];
    while (level.length) {
      lines.push(level.map((n) => `[${n.keys.join(' ')}]`).join(' '));
      level = level[0].leaf ? [] : level.flatMap((n) => n.children);
    }
    return lines.join('\n');
  }
}

// Ejemplo de extendable hashing del apunte: los DNI de los clientes 1 a 10 de
// Chasqui, con el hash que usa PostgreSQL para character(8) (hashbpchar) y
// buckets de 2 entradas. Muestra el directorio después de cada inserción.
//   PGDATABASE=chasqui_w10 node exthash-demo.mjs [cuántos clientes]
import pg from 'pg';
import { ExtendibleHash } from './exthash.mjs';

export async function dniHashes(client, n) {
  const { rows } = await client.query(
    `SELECT dni, hashbpchar(dni) AS h, hashbpchar(dni)::bit(32)::text AS bits
     FROM customer WHERE customer_id <= $1 ORDER BY customer_id`,
    [n],
  );
  return rows;
}

if (import.meta.main) {
  const n = Number(process.argv[2] ?? 9);
  const client = new pg.Client();
  await client.connect();
  const rows = await dniHashes(client, n);
  await client.end();

  const hashes = new Map(rows.map((r) => [r.dni, r.h]));
  const table = new ExtendibleHash(2, (dni) => hashes.get(dni));
  for (const r of rows) {
    table.insert(r.dni, r.dni);
    console.log(`+ ${r.dni} (bits ${r.bits.slice(0, 4)}...)`);
    console.log(table.toString() + '\n');
  }
}

// Extendable hashing en memoria (Silberschatz, sec. 24.5), para seguir paso
// a paso el ejemplo del apunte. No es un índice en disco: cada bucket es un
// arreglo de a lo más `capacity` entradas, y el directorio es un arreglo de
// 2^globalDepth referencias a buckets.
//
// - Se usan los primeros `globalDepth` bits del hash (de 32 bits) para elegir
//   la entrada del directorio.
// - Cada bucket tiene su profundidad local `depth` <= globalDepth: todas sus
//   claves comparten los primeros `depth` bits, y lo apuntan
//   2^(globalDepth - depth) entradas del directorio.
// - Si un bucket lleno tiene depth < globalDepth, se divide y solo se
//   reasignan punteros. Si depth = globalDepth, primero se duplica el
//   directorio.

const BITS = 32;

// Un hash de 32 bits sencillo (FNV-1a) para claves de texto, por si no se
// pasa otro. El ejemplo del apunte usa los hashes de PostgreSQL.
export function fnv1a(key) {
  let h = 0x811c9dc5;
  for (const ch of String(key)) {
    h ^= ch.codePointAt(0);
    h = Math.imul(h, 0x01000193);
  }
  return h >>> 0;
}

// Los primeros `n` bits de un hash de 32 bits, como número.
export function prefix(h, n) {
  return n === 0 ? 0 : (h >>> 0) >>> (BITS - n);
}

export class ExtendibleHash {
  constructor(capacity = 2, hash = fnv1a) {
    if (capacity < 1) throw new RangeError('un bucket guarda al menos una entrada');
    this.capacity = capacity;
    this.hash = (key) => hash(key) >>> 0;
    this.globalDepth = 0;
    this.directory = [{ depth: 0, entries: [] }];
    this.size = 0;
    this.splits = 0;
    this.doublings = 0;
  }

  bucketFor(key) {
    return this.directory[prefix(this.hash(key), this.globalDepth)];
  }

  search(key) {
    const e = this.bucketFor(key).entries.find((x) => x.key === key);
    return e ? e.value : undefined;
  }

  insert(key, value) {
    const h = this.hash(key);
    for (;;) {
      const bucket = this.directory[prefix(h, this.globalDepth)];
      const old = bucket.entries.find((x) => x.key === key);
      if (old) {
        old.value = value;
        return;
      }
      if (bucket.entries.length < this.capacity) {
        bucket.entries.push({ key, value, h });
        this.size++;
        return;
      }
      // Si todas tienen el mismo hash que la clave nueva, dividir no sirve
      // de nada: haría falta un bucket de overflow (el libro lo menciona).
      if (bucket.entries.every((e) => e.h === h)) {
        throw new RangeError('todas las claves del bucket tienen el mismo hash');
      }
      if (bucket.depth === this.globalDepth) this.double();
      this.split(bucket);
      // Se vuelve a intentar: el bucket de la clave puede seguir lleno.
    }
  }

  // Duplica el directorio: la entrada j pasa a ser las entradas 2j y 2j + 1.
  double() {
    this.directory = this.directory.flatMap((b) => [b, b]);
    this.globalDepth++;
    this.doublings++;
  }

  // Divide un bucket en dos de profundidad depth + 1 según el bit siguiente.
  split(bucket) {
    const d = bucket.depth + 1;
    const zero = { depth: d, entries: [] };
    const one = { depth: d, entries: [] };
    for (const e of bucket.entries) {
      (prefix(e.h, d) & 1 ? one : zero).entries.push(e);
    }
    for (let i = 0; i < this.directory.length; i++) {
      if (this.directory[i] === bucket) {
        this.directory[i] = (i >> (this.globalDepth - d)) & 1 ? one : zero;
      }
    }
    this.splits++;
  }

  // Cuántos buckets distintos hay.
  buckets() {
    return new Set(this.directory).size;
  }

  // Una línea por entrada del directorio, por ejemplo:
  //   00 -> [0] 61830975 15433383
  //   01 -> [0] (igual que 00)
  //   10 -> [10] 17232205 23448138
  // Entre corchetes, los bits que comparten las claves del bucket (tantos
  // como su profundidad local).
  toString() {
    const bin = (x, n) => (n === 0 ? '' : x.toString(2).padStart(n, '0'));
    const seen = new Map();
    const lines = [`profundidad global ${this.globalDepth}`];
    this.directory.forEach((b, i) => {
      const name = bin(i, this.globalDepth) || '-';
      const label = `[${bin(i >> (this.globalDepth - b.depth), b.depth)}]`;
      if (seen.has(b)) {
        lines.push(`${name} -> ${label} (igual que ${seen.get(b)})`);
      } else {
        seen.set(b, name);
        lines.push(`${name} -> ${label} ${b.entries.map((e) => e.key).join(' ')}`.trimEnd());
      }
    });
    return lines.join('\n');
  }
}

-- Semana 10: índices de PostgreSQL sobre chasqui_w10 (12 000 envíos).
-- Se corre después de make, que deja la base recién creada y con VACUUM:
--   psql -d chasqui_w10 -f indexes.sql
-- Cada índice nuevo se crea dentro de BEGIN ... ROLLBACK: el archivo no deja
-- nada creado. Los UPDATE de las secciones 10 y 11 se deshacen, pero dejan
-- páginas que ya no son "todas visibles"; para repetir los números del
-- apunte, corran make antes. Sin workers en paralelo, para que los planes
-- sean más cortos.
SET max_parallel_workers_per_gather = 0;

-- 1. La clave primaria ya tiene un B-tree: buscar un envío lee 3 páginas
--    (raíz, hoja y la página del heap).
-- name: pk_lookup
EXPLAIN (ANALYZE, BUFFERS, COSTS OFF, TIMING OFF, SUMMARY OFF)
SELECT * FROM shipment WHERE shipment_id = 5000;

-- 2. La forma de shipment_pkey: level cuenta desde 0 en las hojas.
-- name: pk_meta
SELECT root, level FROM bt_metap('shipment_pkey');

-- name: pk_pages
SELECT type, btpo_level AS level, count(*) AS pages,
       sum(live_items) AS items, round(avg(live_items)) AS avg_items,
       round(avg(free_size)) AS avg_free
FROM bt_multi_page_stats('shipment_pkey', 1, -1)
GROUP BY type, btpo_level
ORDER BY btpo_level DESC;

-- 3. Por dentro: la raíz guarda (clave, página hija); la primera entrada no
--    tiene clave (es "menos infinito"). Una hoja guarda (clave, ctid) y su
--    primera entrada es la high key: la cota superior de la hoja.
-- name: root_items
SELECT itemoffset, ctid, itemlen, data
FROM bt_page_items('shipment_pkey', 3)
LIMIT 4;

-- name: leaf_items
SELECT itemoffset, ctid, itemlen, data, htid
FROM bt_page_items('shipment_pkey', 1)
LIMIT 4;

-- 4. Sin índice en sender_id: sequential scan de las 148 páginas.
-- name: sender_seq
EXPLAIN (ANALYZE, BUFFERS, COSTS OFF, TIMING OFF, SUMMARY OFF)
SELECT * FROM shipment WHERE sender_id = 42;

BEGIN;
-- name: create_sender
CREATE INDEX shipment_sender_idx ON shipment (sender_id);

-- Con el índice: 2 páginas del índice y una página del heap por cada página
-- distinta donde hay un envío del cliente 42.
-- name: sender_idx
EXPLAIN (ANALYZE, BUFFERS, COSTS OFF, TIMING OFF, SUMMARY OFF)
SELECT * FROM shipment WHERE sender_id = 42;

-- 5. Clave no única: una entrada por valor con la lista de ctid
--    (deduplicación, desde PostgreSQL 13).
-- name: sender_pages
SELECT type, count(*) AS pages, sum(live_items) AS items
FROM bt_multi_page_stats('shipment_sender_idx', 1, -1)
GROUP BY type
ORDER BY type DESC;

-- name: posting_list
SELECT itemoffset, itemlen, htid, tids
FROM bt_page_items('shipment_sender_idx', 1)
LIMIT 3;

-- name: sender_1
SELECT ctid, shipment_id FROM shipment WHERE sender_id = 1 ORDER BY ctid;
ROLLBACK;

-- 6. Orden de las columnas en un índice compuesto: igualdad primero,
--    rango después.
BEGIN;
-- name: create_origin_created
CREATE INDEX shipment_origin_created_idx ON shipment (origin_agency_id, created_at);

-- name: origin_created
EXPLAIN (ANALYZE, BUFFERS, COSTS OFF, TIMING OFF, SUMMARY OFF)
SELECT count(*) FROM shipment
WHERE origin_agency_id = 12
  AND created_at >= '2026-04-01' AND created_at < '2026-05-01';

-- Sin condición sobre la primera columna: skip scan (PostgreSQL 18), una
-- búsqueda por cada valor de origin_agency_id.
-- name: skip_scan
EXPLAIN (ANALYZE, BUFFERS, COSTS OFF, TIMING OFF, SUMMARY OFF)
SELECT count(*) FROM shipment
WHERE created_at >= '2026-04-01' AND created_at < '2026-04-02';
ROLLBACK;

BEGIN;
-- name: create_created_origin
CREATE INDEX shipment_created_origin_idx ON shipment (created_at, origin_agency_id);

-- name: created_origin
EXPLAIN (ANALYZE, BUFFERS, COSTS OFF, TIMING OFF, SUMMARY OFF)
SELECT count(*) FROM shipment
WHERE origin_agency_id = 12
  AND created_at >= '2026-04-01' AND created_at < '2026-05-01';
ROLLBACK;

-- 7. Índice parcial: solo los envíos sin entregar (885 de 12 000).
BEGIN;
-- name: create_pending
CREATE INDEX shipment_pending_idx ON shipment (origin_agency_id)
  WHERE delivered_at IS NULL;

-- name: create_origin
CREATE INDEX shipment_origin_idx ON shipment (origin_agency_id);

-- name: partial_sizes
SELECT relname, relpages, reltuples::integer AS entries
FROM pg_class
WHERE relname IN ('shipment_pending_idx', 'shipment_origin_idx')
ORDER BY relname;

-- name: pending
EXPLAIN (ANALYZE, BUFFERS, COSTS OFF, TIMING OFF, SUMMARY OFF)
SELECT shipment_id, status FROM shipment
WHERE origin_agency_id = 1 AND delivered_at IS NULL;
ROLLBACK;

-- 8. Índice sobre una expresión: un índice en last_name no sirve para
--    lower(last_name); uno en lower(last_name), sí.
BEGIN;
-- name: create_last_name
CREATE INDEX customer_last_name_idx ON customer (last_name);

-- name: lower_without
EXPLAIN (COSTS OFF)
SELECT customer_id FROM customer WHERE lower(last_name) = 'villanueva ramírez';

-- name: create_lower
CREATE INDEX customer_lower_last_name_idx ON customer (lower(last_name));

-- ANALYZE junta estadísticas de la expresión indexada.
ANALYZE customer;

-- name: lower_with
EXPLAIN (ANALYZE, COSTS OFF, TIMING OFF, SUMMARY OFF)
SELECT customer_id FROM customer WHERE lower(last_name) = 'villanueva ramírez';
ROLLBACK;

-- 9. Índice hash: solo igualdad, una columna y sin UNIQUE.
BEGIN;
-- name: create_dni_hash
CREATE INDEX customer_dni_hash ON customer USING hash (dni);

-- name: dni_hash
EXPLAIN (ANALYZE, BUFFERS, COSTS OFF, TIMING OFF, SUMMARY OFF)
SELECT first_name, last_name FROM customer WHERE dni = '61830975';

-- Para un rango, el planner usa el B-tree de la restricción UNIQUE.
-- name: dni_range
EXPLAIN (COSTS OFF)
SELECT first_name, last_name FROM customer
WHERE dni BETWEEN '61830900' AND '61830999';
ROLLBACK;

-- Los bits del hash que usa PostgreSQL para un character(8): el ejemplo de
-- extendable hashing del apunte usa los primeros bits.
-- name: dni_bits
SELECT customer_id, dni, hashbpchar(dni) AS h, hashbpchar(dni)::bit(32) AS bits
FROM customer
WHERE customer_id <= 10
ORDER BY customer_id;

-- 10. Covering index: con INCLUDE (status) la consulta se responde solo con
--     el índice (index-only scan). Necesita el visibility map al día.
BEGIN;
-- name: create_covering
CREATE INDEX shipment_sender_status_idx ON shipment (sender_id) INCLUDE (status);

-- name: covering
EXPLAIN (ANALYZE, BUFFERS, COSTS OFF, TIMING OFF, SUMMARY OFF)
SELECT sender_id, status FROM shipment WHERE sender_id = 42;

-- Un UPDATE deja páginas que ya no son "todas visibles": el index-only scan
-- tiene que ir al heap a mirar (Heap Fetches).
-- name: covering_update
UPDATE shipment SET price = price + 1 WHERE sender_id = 42;

-- name: covering_after
EXPLAIN (ANALYZE, BUFFERS, COSTS OFF, TIMING OFF, SUMMARY OFF)
SELECT sender_id, status FROM shipment WHERE sender_id = 42;
ROLLBACK;

-- 11. CLUSTER ordena la tabla una vez; correlation mide cuánto dura el
--     orden. Y un BRIN solo sirve si la columna sigue el orden físico.
-- name: correlation
SELECT attname, correlation
FROM pg_stats
WHERE tablename = 'shipment' AND attname IN ('shipment_id', 'created_at')
ORDER BY attname;

BEGIN;
-- name: create_created_brin
CREATE INDEX shipment_created_brin ON shipment USING brin (created_at)
  WITH (pages_per_range = 4);

SET LOCAL enable_seqscan = off;

-- name: brin_before
EXPLAIN (ANALYZE, BUFFERS, COSTS OFF, TIMING OFF, SUMMARY OFF)
SELECT count(*) FROM shipment
WHERE created_at >= '2026-04-01' AND created_at < '2026-04-02';

-- name: create_created
CREATE INDEX shipment_created_idx ON shipment (created_at);

-- name: cluster
CLUSTER shipment USING shipment_created_idx;

ANALYZE shipment;

-- name: correlation_cluster
SELECT attname, correlation
FROM pg_stats
WHERE tablename = 'shipment' AND attname IN ('shipment_id', 'created_at')
ORDER BY attname;

DROP INDEX shipment_created_idx;

-- name: brin_after
EXPLAIN (ANALYZE, BUFFERS, COSTS OFF, TIMING OFF, SUMMARY OFF)
SELECT count(*) FROM shipment
WHERE created_at >= '2026-04-01' AND created_at < '2026-04-02';

-- Un UPDATE de uno de cada diez envíos escribe versiones nuevas donde haya
-- espacio, no en orden: el orden de CLUSTER se empieza a perder.
-- name: cluster_update
UPDATE shipment SET price = price WHERE shipment_id % 10 = 0;

ANALYZE shipment;

-- name: correlation_update
SELECT attname, correlation
FROM pg_stats
WHERE tablename = 'shipment' AND attname IN ('shipment_id', 'created_at')
ORDER BY attname;
ROLLBACK;

-- 12. Dos errores a propósito, fuera de una transacción (no cambian nada):
--     un índice hash no admite UNIQUE ni varias columnas.
-- name: hash_unique
CREATE UNIQUE INDEX customer_dni_hash ON customer USING hash (dni);

-- name: hash_multicolumn
CREATE INDEX customer_dni_city_hash ON customer USING hash (dni, city);

// 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 del B+ tree de la semana 10. Agreguen las suyas al final (pasos 3 y 4).
import { test } from 'node:test';
import assert from 'node:assert/strict';
import { BPlusTree } from './bptree.mjs';

test('un árbol vacío no encuentra nada', () => {
  const t = new BPlusTree(4);
  assert.equal(t.search(10), undefined);
  assert.equal(t.height(), 1);
});

test('inserta y encuentra sin dividir', () => {
  const t = new BPlusTree(4);
  t.insert(20, 'b');
  t.insert(10, 'a');
  t.insert(30, 'c');
  assert.equal(t.search(10), 'a');
  assert.equal(t.search(30), 'c');
  assert.equal(t.search(25), undefined);
  assert.equal(t.toString(), '[10 20 30]');
});

test('el ejemplo de la guía: orden 4 y las claves 10, 20, 5, 15, 25, 30, 35', () => {
  const t = new BPlusTree(4);
  for (const k of [10, 20, 5, 15, 25, 30, 35]) t.insert(k, `v${k}`);
  for (const k of [10, 20, 5, 15, 25, 30, 35]) assert.equal(t.search(k), `v${k}`);
  assert.equal(t.height(), 2);
  assert.equal(t.toString(), '[15 25]\n[5 10] [15 20] [25 30 35]');
});

test('una clave repetida reemplaza el valor', () => {
  const t = new BPlusTree(4);
  t.insert(7, 'viejo');
  t.insert(7, 'nuevo');
  assert.equal(t.search(7), 'nuevo');
  assert.equal(t.size, 1);
});

test('10 000 claves en orden y al revés: todas se encuentran y el árbol es bajo', () => {
  for (const keys of [[...Array(10000).keys()], [...Array(10000).keys()].reverse()]) {
    const t = new BPlusTree(32);
    for (const k of keys) t.insert(k, k * 2);
    for (const k of keys) assert.equal(t.search(k), k * 2);
    assert.equal(t.size, 10000);
    assert.ok(t.height() <= 4, `altura ${t.height()}`);
  }
});

// Pruebas de exthash.mjs: el ejemplo del apunte con los hashes de PostgreSQL
// y los invariantes de extendable hashing con muchas claves.
import { test } from 'node:test';
import assert from 'node:assert/strict';
import pg from 'pg';
import { ExtendibleHash, prefix } from './exthash.mjs';
import { dniHashes } from './exthash-demo.mjs';

// Revisa la estructura: cada bucket con depth <= globalDepth, lo apuntan
// exactamente 2^(globalDepth - depth) entradas seguidas, y todas sus claves
// comparten los primeros depth bits con esas entradas.
function checkInvariants(t) {
  const count = new Map();
  t.directory.forEach((b, i) => {
    assert.ok(b.depth <= t.globalDepth);
    assert.ok(b.entries.length <= t.capacity);
    for (const e of b.entries) {
      assert.equal(prefix(e.h, b.depth), i >> (t.globalDepth - b.depth));
    }
    count.set(b, (count.get(b) ?? 0) + 1);
  });
  for (const [b, n] of count) assert.equal(n, 2 ** (t.globalDepth - b.depth));
}

test('el ejemplo del apunte: DNI de los clientes 1 a 10, buckets de 2', async () => {
  const client = new pg.Client();
  await client.connect();
  const rows = await dniHashes(client, 10);
  await client.end();
  assert.deepEqual(rows.slice(0, 3).map((r) => r.bits.slice(0, 3)), ['000', '100', '111']);

  const hashes = new Map(rows.map((r) => [r.dni, r.h]));
  const t = new ExtendibleHash(2, (dni) => hashes.get(dni));
  const depths = [];
  for (const r of rows) {
    t.insert(r.dni, r.dni);
    depths.push(t.globalDepth);
    checkInvariants(t);
  }
  assert.deepEqual(depths, [0, 0, 1, 1, 2, 2, 2, 2, 3, 4]);

  // Después del cliente 9: el estado que dibuja el apunte.
  const nine = new ExtendibleHash(2, (dni) => hashes.get(dni));
  for (const r of rows.slice(0, 9)) nine.insert(r.dni, r.dni);
  assert.equal(nine.toString(), [
    'profundidad global 3',
    '000 -> [00] 61830975 15433383',
    '001 -> [00] (igual que 000)',
    '010 -> [01] 75503630 36581411',
    '011 -> [01] (igual que 010)',
    '100 -> [100] 17232205 23448138',
    '101 -> [101] 58728634',
    '110 -> [11] 36948162 24986841',
    '111 -> [11] (igual que 110)',
  ].join('\n'));
  assert.equal(nine.buckets(), 5);
  assert.equal(nine.directory.length, 8);
  for (const r of rows.slice(0, 9)) assert.equal(nine.search(r.dni), r.dni);
  assert.equal(nine.search('00000000'), undefined);
});

test('10 000 claves: todas se encuentran y se cumplen los invariantes', () => {
  const t = new ExtendibleHash(8);
  for (let i = 0; i < 10000; i++) t.insert(`k${i}`, i);
  checkInvariants(t);
  assert.equal(t.size, 10000);
  for (let i = 0; i < 10000; i++) assert.equal(t.search(`k${i}`), i);
  // Cada split crea un bucket nuevo; el directorio crece mucho menos seguido.
  assert.equal(t.buckets(), t.splits + 1);
  assert.ok(t.doublings < t.splits);
});

test('una clave repetida reemplaza el valor', () => {
  const t = new ExtendibleHash(2);
  t.insert('a', 1);
  t.insert('a', 2);
  assert.equal(t.search('a'), 2);
  assert.equal(t.size, 1);
});

test('más claves con el mismo hash que la capacidad de un bucket: error claro', () => {
  const t = new ExtendibleHash(2, () => 7);
  t.insert('a', 1);
  t.insert('b', 2);
  assert.throws(() => t.insert('c', 3), /mismo hash/);
});

// Pruebas de indexes.sql sobre chasqui_w10 recién creada (make test la vuelve
// a crear). Corren el mismo texto del archivo y revisan las cifras que cita
// el apunte: tipo de nodo, índice usado, páginas leídas y filas. No miran
// tiempos. Cada índice se crea dentro de BEGIN ... ROLLBACK.
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('./indexes.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());

// El plan como texto, una línea por fila.
async function plan(sql) {
  const { rows } = await client.query(sql);
  return rows.map((r) => r['QUERY PLAN']).join('\n');
}

// Páginas que tocó el nodo de más arriba (hit + read), de su primera línea
// "Buffers:".
function pages(text) {
  const line = text.split('\n').find((l) => l.includes('Buffers:'));
  return [...line.matchAll(/(?:hit|read)=(\d+)/g)].reduce((s, m) => s + Number(m[1]), 0);
}

async function inTransaction(fn) {
  await client.query('BEGIN');
  try {
    await fn();
  } finally {
    await client.query('ROLLBACK');
  }
}

test('la clave primaria: Index Scan de 3 páginas, altura 2', async () => {
  const p = await plan(q.pk_lookup);
  assert.match(p, /^Index Scan using shipment_pkey on shipment/);
  assert.equal(pages(p), 3);
  const meta = (await client.query(q.pk_meta)).rows[0];
  assert.deepEqual(meta, { root: '3', level: '1' });
  const levels = (await client.query(q.pk_pages)).rows;
  assert.deepEqual(levels.map((r) => [r.type, r.level, r.pages, r.items]),
    [['r', '1', '1', '33'], ['l', '0', '33', '12032']]);
});

test('la raíz apunta a las hojas; la hoja 1 empieza con su high key, 367', async () => {
  const root = (await client.query(q.root_items)).rows;
  assert.deepEqual(root.map((r) => r.ctid), ['(1,0)', '(2,1)', '(4,1)', '(5,1)']);
  assert.equal(root[0].data, '');
  assert.equal(root[1].data, '6f 01 00 00 00 00 00 00');
  const leaf = (await client.query(q.leaf_items)).rows;
  assert.equal(leaf[0].data, '6f 01 00 00 00 00 00 00');
  assert.equal(leaf[0].htid, null);
  assert.equal(parseInt('016f', 16), 367);
  assert.deepEqual(leaf.slice(1).map((r) => r.htid), ['(0,1)', '(0,2)', '(0,3)']);
});

test('sender_id sin índice lee 148 páginas; con índice, 2 + 45', async () => {
  const seq = await plan(q.sender_seq);
  assert.match(seq, /^Seq Scan on shipment \(actual rows=51\.00/);
  assert.match(seq, /Rows Removed by Filter: 11949/);
  assert.equal(pages(seq), 148);
  await inTransaction(async () => {
    await client.query(q.create_sender);
    const idx = await plan(q.sender_idx);
    assert.match(idx, /^Bitmap Heap Scan on shipment/);
    assert.match(idx, /Heap Blocks: exact=45\n/);
    assert.equal(pages(idx), 47);
    assert.match(idx, /Bitmap Index Scan on shipment_sender_idx[^\n]*\n[^\n]*\n[^\n]*\n\s+Buffers: shared (hit|read)=2\n/);

    // Clave no única: 1718 valores distintos en 15 hojas, más 14 high keys.
    const sp = (await client.query(q.sender_pages)).rows;
    assert.deepEqual(sp, [{ type: 'r', pages: '1', items: '15' }, { type: 'l', pages: '15', items: '1732' }]);
    const { rows: [{ n }] } = await client.query('SELECT count(DISTINCT sender_id) AS n FROM shipment');
    assert.equal(n, '1718');

    // La posting list del cliente 1 son los ctid de sus 8 envíos, en orden.
    const pl = (await client.query(q.posting_list)).rows;
    const ctids = (await client.query(q.sender_1)).rows.map((r) => r.ctid);
    assert.equal(ctids.length, 8);
    assert.deepEqual(pl[1].tids.replace(/[{}"]/g, '').match(/\(\d+,\d+\)/g), ctids);
    assert.equal(pl[1].itemlen, 64);
  });
});

test('índice compuesto: (origen, fecha) lee 3 páginas; (fecha, origen), 26', async () => {
  await inTransaction(async () => {
    await client.query(q.create_origin_created);
    const good = await plan(q.origin_created);
    assert.match(good, /Index Only Scan using shipment_origin_created_idx/);
    assert.match(good, /actual rows=80\.00/);
    assert.match(good, /Heap Fetches: 0/);
    assert.equal(pages(good), 3);
    // Skip scan: cerca de una búsqueda por cada una de las 28 agencias de origen.
    const skip = await plan(q.skip_scan);
    assert.match(skip, /Index Only Scan using shipment_origin_created_idx/);
    assert.match(skip, /actual rows=72\.00/);
    assert.match(skip, /Index Searches: 26\n/);
    assert.equal(pages(skip), 56);
  });
  const { rows: [{ n }] } = await client.query('SELECT count(DISTINCT origin_agency_id) AS n FROM shipment');
  assert.equal(n, '28');
  await inTransaction(async () => {
    await client.query(q.create_created_origin);
    const bad = await plan(q.created_origin);
    assert.match(bad, /Index Only Scan using shipment_created_origin_idx/);
    assert.match(bad, /actual rows=80\.00/);
    assert.equal(pages(bad), 26);
  });
});

test('índice parcial: 885 entradas en 2 páginas, frente a 13', async () => {
  await inTransaction(async () => {
    await client.query(q.create_pending);
    await client.query(q.create_origin);
    const sizes = (await client.query(q.partial_sizes)).rows;
    assert.deepEqual(sizes, [
      { relname: 'shipment_origin_idx', relpages: 13, entries: 12000 },
      { relname: 'shipment_pending_idx', relpages: 2, entries: 885 },
    ]);
    const p = await plan(q.pending);
    assert.match(p, /Bitmap Index Scan on shipment_pending_idx/);
    assert.match(p, /actual rows=108\.00/);
  });
});

test('índice sobre lower(last_name): sin él, sequential scan', async () => {
  await inTransaction(async () => {
    await client.query(q.create_last_name);
    assert.match(await plan(q.lower_without), /^Seq Scan on customer/);
    await client.query(q.create_lower);
    await client.query('ANALYZE customer');
    const p = await plan(q.lower_with);
    assert.match(p, /Bitmap Index Scan on customer_lower_last_name_idx/);
    assert.match(p, /actual rows=6\.00/);
  });
});

test('índice hash: igualdad en 3 páginas; un rango usa el B-tree', async () => {
  await inTransaction(async () => {
    await client.query(q.create_dni_hash);
    const p = await plan(q.dni_hash);
    assert.match(p, /^Index Scan using customer_dni_hash on customer/);
    assert.equal(pages(p), 3);
    assert.match(await plan(q.dni_range), /^Index Scan using customer_dni_key on customer/);
  });
  await assert.rejects(client.query(q.hash_unique),
    { message: 'access method "hash" does not support unique indexes' });
  await assert.rejects(client.query(q.hash_multicolumn),
    { message: 'access method "hash" does not support multicolumn indexes' });
  const bits = (await client.query(q.dni_bits)).rows;
  assert.equal(bits.length, 10);
  assert.deepEqual(bits[0], { customer_id: 1, dni: '61830975', h: 292428428, bits: '00010001011011100001101010001100' });
});

test('BRIN: inútil en created_at hasta que CLUSTER ordena la tabla', async () => {
  const corr = (await client.query(q.correlation)).rows;
  assert.equal(corr.find((r) => r.attname === 'shipment_id').correlation, 1);
  assert.ok(Math.abs(corr.find((r) => r.attname === 'created_at').correlation) < 0.01);
  await inTransaction(async () => {
    await client.query(q.create_created_brin);
    await client.query('SET LOCAL enable_seqscan = off');
    const before = await plan(q.brin_before);
    assert.match(before, /Bitmap Index Scan on shipment_created_brin/);
    assert.match(before, /Heap Blocks: lossy=148\n/);
    assert.match(before, /Rows Removed by Index Recheck: 11928\n/);
    await client.query(q.create_created);
    await client.query(q.cluster);
    await client.query('ANALYZE shipment');
    const clustered = (await client.query(q.correlation_cluster)).rows;
    assert.equal(clustered.find((r) => r.attname === 'created_at').correlation, 1);
    await client.query('DROP INDEX shipment_created_idx');
    const after = await plan(q.brin_after);
    assert.match(after, /Bitmap Index Scan on shipment_created_brin/);
    assert.match(after, /actual rows=72\.00/);
    assert.match(after, /Heap Blocks: lossy=4\n/);
    assert.equal(pages(after), 6);
    const upd = await client.query(q.cluster_update);
    assert.equal(upd.rowCount, 1200);
    await client.query('ANALYZE shipment');
    const later = (await client.query(q.correlation_update)).rows;
    const c = later.find((r) => r.attname === 'created_at').correlation;
    assert.ok(c > 0.7 && c < 0.9, `correlation ${c}`);
  });
});

// Al final: el UPDATE (aunque se deshaga) deja páginas que ya no son "todas
// visibles" y cambiaría los index-only scans de las pruebas anteriores.
test('covering index: index-only scan sin ir al heap, hasta que hay un UPDATE', async () => {
  await inTransaction(async () => {
    await client.query(q.create_covering);
    const p = await plan(q.covering);
    assert.match(p, /^Index Only Scan using shipment_sender_status_idx/);
    assert.match(p, /Heap Fetches: 0\n/);
    assert.equal(pages(p), 3);
    const upd = await client.query(q.covering_update);
    assert.equal(upd.rowCount, 51);
    const later = await plan(q.covering_after);
    assert.match(later, /actual rows=51\.00/);
    const fetches = Number(later.match(/Heap Fetches: (\d+)/)[1]);
    assert.ok(fetches > 51, `Heap Fetches: ${fetches}`);
  });
});

# Semana 10: índices. Un B+ tree en memoria en Node.js (el laboratorio) y los
# índices de PostgreSQL vistos con EXPLAIN y pageinspect. Se usa dentro del
# contenedor del curso.
#
#   make          crea la base de datos chasqui_w10 con los datos de ejemplo,
#                 instala pageinspect y hace VACUUM (ANALYZE): así el
#                 visibility map y las estadísticas quedan en un estado
#                 conocido y los planes del apunte se repiten
#   make test     la vuelve a crear y corre todas las pruebas (*.test.mjs).
#                 Las de bptree.test.mjs fallan hasta que completen el
#                 laboratorio: search, insert y range todavía lanzan
#                 "por implementar"
#   make lab      solo las pruebas del B+ tree (no necesita la base de datos)
#   make big      crea chasqui_w10_big, una copia de chasqui_big (que tiene
#                 que existir: make big en datasets/chasqui), y corre big.sql
#                 en ella. Tarda menos de un minuto; la copia ocupa unos
#                 275 MB, y más mientras big.sql tiene sus tablas de prueba
#   make clean    borra chasqui_w10 y chasqui_w10_big
#
# indexes.sql se corre sobre chasqui_w10: psql -d chasqui_w10 -f indexes.sql
# Cada ejemplo que crea un índice va dentro de BEGIN ... ROLLBACK, así que el
# archivo no deja nada creado. Pero un UPDATE deshecho igual deja páginas que
# ya no son "todas visibles": para ver los mismos números del apunte, corran
# make antes.

DB = chasqui_w10
BIG = chasqui_w10_big
PSQL = psql -X -q -v ON_ERROR_STOP=1

.PHONY: all test lab big clean

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

# Las pruebas crean índices dentro de transacciones: una después de otra.
test: all
	PGDATABASE=$(DB) node --test --test-concurrency=1 --test-reporter=spec

lab:
	node --test --test-reporter=spec bptree.test.mjs

big:
	dropdb --if-exists $(BIG)
	createdb -T chasqui_big $(BIG)
	$(PSQL) -d $(BIG) -c "ALTER DATABASE $(BIG) SET timezone TO 'America/Lima'"
	$(PSQL) -d $(BIG) -c "CREATE EXTENSION IF NOT EXISTS pageinspect" \
	        -c "VACUUM (ANALYZE)"
	psql -X -v ON_ERROR_STOP=1 -d $(BIG) -f big.sql

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