Scenarios: Pagination, Export & Large Data

EXPLAIN (ANALYZE, BUFFERS)
SELECT id, created_at, total FROM orders
ORDER BY created_at DESC, id DESC
LIMIT 20 OFFSET 100000;
-- Limit (actual rows=20) -> Index Scan Backward ... (actual rows=100020)
--   Buffers: shared hit=81234 read=20311

// BUG
const rows = await sql`
  SELECT id, created_at FROM orders
  WHERE tenant_id = ${tenantId}
    AND (${cursor}::timestamptz IS NULL OR created_at < ${cursor})
  ORDER BY created_at DESC
  LIMIT ${limit}`;
const next = rows.at(-1)?.created_at;

// FIX
const rows = await sql`
  SELECT id, created_at FROM orders
  WHERE tenant_id = ${tenantId}
    AND (${c?.ts ?? null}::timestamptz IS NULL
         OR (created_at, id) < (${c?.ts}, ${c?.id}))
  ORDER BY created_at DESC, id DESC
  LIMIT ${limit + 1}`;
// index: CREATE INDEX CONCURRENTLY ON orders (tenant_id, created_at, id);

import { createHmac, timingSafeEqual } from "node:crypto";
type Cursor = { v: 1; k: [string, number]; f: string };
const sign = (p: string) => createHmac("sha256", KEY).update(p).digest("base64url");
export const encode = (c: Cursor) => {
  const p = Buffer.from(JSON.stringify(c)).toString("base64url");
  return `${p}.${sign(p)}`;
};
export const decode = (s: string, filterHash: string): Cursor => {
  const [p, sig] = s.split(".");
  if (!p || !sig || !timingSafeEqual(Buffer.from(sig), Buffer.from(sign(p)))) throw new BadRequest("invalid_cursor");
  const c = JSON.parse(Buffer.from(p, "base64url").toString()) as Cursor;
  if (c.v !== 1 || c.f !== filterHash) throw new BadRequest("invalid_cursor");
  return c;
};

-- BUG
WHERE (created_at, id) < ($1, $2)
ORDER BY created_at DESC, id ASC

-- FIX (PostgreSQL / SQL Server)
WHERE created_at < $1 OR (created_at = $1 AND id > $2)
ORDER BY created_at DESC, id ASC
-- CREATE INDEX CONCURRENTLY orders_created_desc_id_asc ON orders (created_at DESC, id ASC);

-- capped count
SELECT count(*) AS n FROM (
  SELECT 1 FROM orders WHERE tenant_id = $1 AND status = 'paid' LIMIT 10001
) s;                                  -- n = 10001  → UI hiển thị "10,000+"

-- estimate for a filtered query (parse the JSON plan)
EXPLAIN (FORMAT JSON) SELECT 1 FROM orders WHERE tenant_id = $1 AND status = 'paid';
-- -> [0].Plan["Plan Rows"]

-- deferred join
SELECT o.* FROM orders o
JOIN (
  SELECT id FROM orders
  WHERE tenant_id = $1
  ORDER BY created_at DESC, id DESC
  LIMIT 50 OFFSET 24950            -- page 500
) p USING (id)
ORDER BY o.created_at DESC, o.id DESC;
-- cần index (tenant_id, created_at, id) để subquery là index-only scan

async function page(dir: "next" | "prev", c: [Date, number] | null, n: number) {
  const cmp = dir === "next" ? sql`<` : sql`>`;
  const ord = dir === "next" ? sql`DESC` : sql`ASC`;
  const rows = await sql`
    SELECT id, created_at FROM orders
    WHERE ${c ? sql`(created_at, id) ${cmp} (${c[0]}, ${c[1]})` : sql`TRUE`}
    ORDER BY created_at ${ord}, id ${ord}
    LIMIT ${n + 1}`;
  const more = rows.length > n;
  const items = rows.slice(0, n);
  if (dir === "prev") items.reverse();
  return { items, hasNext: dir === "next" ? more : true, hasPrev: dir === "prev" ? more : c !== null };
}

-- phase 1 (non-null)
WHERE shipped_at IS NOT NULL AND (shipped_at, id) < ($1, $2)
ORDER BY shipped_at DESC, id DESC LIMIT $3;
-- phase 2 (nulls), khi phase 1 hết
WHERE shipped_at IS NULL AND id < $2
ORDER BY id DESC LIMIT $3;

const pit = await es.openPointInTime({ index: "products", keep_alive: "2m" });
let after: unknown[] | undefined;
for (;;) {
  const r = await es.search({
    size: 5000, pit: { id: pit.id, keep_alive: "2m" },
    sort: [{ updated_at: "desc" }, { product_id: "asc" }],
    ...(after && { search_after: after }),
  });
  if (r.hits.hits.length === 0) break;
  await sink(r.hits.hits);
  after = r.hits.hits.at(-1)!.sort;
}
await es.closePointInTime({ id: pit.id });

// BUG
const rows = await prisma.order.findMany({ where: { tenantId } });
const csv = ["id,total,email", ...rows.map(r => `${r.id},${r.total},${r.email}`)].join("\n");
await s3.send(new PutObjectCommand({ Bucket, Key, Body: csv }));

// FIX
import { pipeline } from "node:stream/promises";
import { createGzip } from "node:zlib";
import { PassThrough } from "node:stream";
import QueryStream from "pg-query-stream";
import { stringify } from "csv-stringify";
import { Upload } from "@aws-sdk/lib-storage";

const client = await pool.connect();
try {
  const rows = client.query(new QueryStream(
    "SELECT id, total, email FROM orders WHERE tenant_id = $1 ORDER BY id", [tenantId], { batchSize: 1000 }));
  const body = new PassThrough();
  const upload = new Upload({ client: s3, params: { Bucket, Key, Body: body, ContentEncoding: "gzip" },
                              partSize: 8 * 1024 * 1024, queueSize: 4 });
  await Promise.all([
    pipeline(rows, stringify({ header: true, columns: ["id", "total", "email"] }), createGzip(), body),
    upload.done(),
  ]);
} finally {
  client.release();
}

// BUG
dbStream.on("data", (row) => gzip.write(toCsvLine(row)));
dbStream.on("end", () => gzip.end());
gzip.pipe(uploadBody);

// FIX
import { pipeline } from "node:stream/promises";
import { Transform } from "node:stream";
const toCsv = new Transform({
  writableObjectMode: true,
  transform(row, _enc, cb) { cb(null, toCsvLine(row)); },
});
await pipeline(dbStream, toCsv, createGzip(), uploadBody);

import { to as copyTo } from "pg-copy-streams";
const client = await pool.connect();
try {
  const src = client.query(copyTo(
    "COPY (SELECT id, total, created_at FROM orders WHERE tenant_id = 42) TO STDOUT WITH (FORMAT csv, HEADER)"));
  await pipeline(src, createGzip(), uploadBody);
} finally { client.release(); }

-- coordinator
BEGIN ISOLATION LEVEL REPEATABLE READ READ ONLY;
SELECT pg_export_snapshot();          -- '00000003-0000001B-1'
-- worker N (while coordinator txn stays open)
BEGIN ISOLATION LEVEL REPEATABLE READ READ ONLY;
SET TRANSACTION SNAPSHOT '00000003-0000001B-1';
COPY (SELECT ... WHERE id BETWEEN 1 AND 10000000) TO STDOUT WITH (FORMAT csv);

import ExcelJS from "exceljs";
const wb = new ExcelJS.stream.xlsx.WorkbookWriter({ stream: uploadBody, useSharedStrings: false });
let ws = wb.addWorksheet("orders-1"); let n = 0; let sheet = 1;
for await (const row of rows) {
  if (++n > 1_048_575) { await ws.commit(); ws = wb.addWorksheet(`orders-${++sheet}`); n = 1; }
  ws.addRow([row.id, row.total, safeCell(row.note)]).commit();
}
await ws.commit(); await wb.commit();

const DANGER = /^[=+\-@\t\r]/;
export const safeCell = (v: unknown) =>
  typeof v === "string" && DANGER.test(v) ? `'${v}` : v;
// then let the CSV library quote/escape:  "'=HYPERLINK(""http://..."")"

CREATE TABLE export_jobs (
  id uuid PRIMARY KEY, tenant_id bigint NOT NULL, idempotency_key text NOT NULL,
  status text NOT NULL, as_of timestamptz NOT NULL,
  s3_upload_id text, last_key bigint, parts jsonb NOT NULL DEFAULT '[]',
  rows_done bigint NOT NULL DEFAULT 0, lease_until timestamptz,
  UNIQUE (tenant_id, idempotency_key)
);

import { from as copyFrom } from "pg-copy-streams";
const client = await pool.connect();
try {
  await client.query("CREATE TEMP TABLE staging (LIKE supplier_items_raw) ON COMMIT DROP");
  await client.query("BEGIN");
  await pipeline(fs.createReadStream(path), client.query(copyFrom(
    "COPY staging (sku, name, price, qty) FROM STDIN WITH (FORMAT csv, HEADER)")));
  await client.query(`
    INSERT INTO supplier_items (sku, name, price, qty)
    SELECT sku, name, price::numeric, qty::int FROM staging
    ON CONFLICT (sku) DO UPDATE SET name = EXCLUDED.name, price = EXCLUDED.price, qty = EXCLUDED.qty
    WHERE (supplier_items.name, supplier_items.price, supplier_items.qty)
          IS DISTINCT FROM (EXCLUDED.name, EXCLUDED.price, EXCLUDED.qty)`);
  await client.query("COMMIT");
} finally { client.release(); }

// BUG: 10_000 rows × 8 cols = 80_000 parameters
await db.insertInto("items").values(rows).execute();

// FIX: 8 parameters regardless of row count
await sql`
  INSERT INTO items (sku, name, price, qty, a, b, c, d)
  SELECT * FROM unnest(
    ${skus}::text[], ${names}::text[], ${prices}::numeric[], ${qtys}::int[],
    ${as}::text[], ${bs}::text[], ${cs}::text[], ${ds}::text[])
  ON CONFLICT (sku) DO NOTHING`;

-- PostgreSQL
DELETE FROM orders
WHERE id IN (
  SELECT id FROM orders
  WHERE created_at < '2023-01-01'
  ORDER BY id
  LIMIT 10000
);
-- SQL Server
DELETE TOP (10000) FROM orders WHERE created_at < '2023-01-01';
-- MySQL
DELETE FROM orders WHERE created_at < '2023-01-01' ORDER BY id LIMIT 10000;

EXPLAIN (ANALYZE) DELETE FROM orders
WHERE id IN (SELECT id FROM orders WHERE created_at < '2023-01-01' ORDER BY id LIMIT 5000);
--  Delete on orders (actual time=412.3..412.3 rows=0)
--  ...
--  Trigger for constraint order_items_order_id_fkey: time=39871.2 calls=5000
--  Trigger audit_orders_delete: time=911.4 calls=5000

async function waitForReplicas(maxLagMs = 5000) {
  for (;;) {
    const { rows } = await primary.query(
      "SELECT coalesce(max(extract(epoch FROM replay_lag)) * 1000, 0) AS lag_ms FROM pg_stat_replication");
    if (Number(rows[0].lag_ms) < maxLagMs) return;
    await sleep(2000);
  }
}

SET lock_timeout = '3s';
ALTER TABLE orders DETACH PARTITION orders_2022 CONCURRENTLY;
-- archive: COPY orders_2022 TO STDOUT ... → S3, rồi:
DROP TABLE orders_2022;

let { last_id: lastId } = await getCheckpoint("backfill_region");
let size = 5000;
for (;;) {
  const t0 = Date.now();
  const { rows } = await db.query(`
    WITH b AS (SELECT id FROM orders WHERE id > $1 ORDER BY id LIMIT $2)
    UPDATE orders o SET region_code = r.code
    FROM b JOIN addresses a ON a.order_id = b.id JOIN regions r ON r.zip = a.zip
    WHERE o.id = b.id AND o.region_code IS NULL
    RETURNING (SELECT max(id) FROM b) AS max_id`, [lastId, size]);
  const maxId = rows[0]?.max_id ?? (await maxIdAfter(lastId, size));
  if (maxId == null) break;               // no more rows
  lastId = maxId;
  await saveCheckpoint("backfill_region", lastId);
  const ms = Date.now() - t0;
  size = ms > 500 ? Math.max(500, size / 2) : ms < 100 ? Math.min(20000, size * 2) : size;
  await waitForReplicas(5000);
  await sleep(100);
}

-- BUG
UPDATE orders SET region_code = compute_region(zip)
WHERE id IN (SELECT id FROM orders WHERE region_code IS NULL LIMIT 5000);

-- FIX
UPDATE orders SET region_code = compute_region(zip)
WHERE id > $1 AND id <= $1 + 5000 AND region_code IS NULL;

// pg: a `name` makes it a server-side prepared statement cached per connection
await client.query({
  name: "orders-by-tenant",
  text: "SELECT * FROM orders WHERE tenant_id = $1 AND status = $2 ORDER BY created_at DESC LIMIT 50",
  values: [tenantId, status],
});

PREPARE q(bigint, text) AS SELECT ... WHERE tenant_id = $1 AND status = $2 ...;
EXECUTE q(7, 'open');  -- ×6
EXPLAIN EXECUTE q(42, 'open');  -- "Index Cond: (tenant_id = $1)" → generic plan

SELECT relname, n_mod_since_analyze, last_autoanalyze, last_analyze
FROM pg_stat_user_tables WHERE relname = 'orders';

ALTER TABLE orders SET (autovacuum_analyze_scale_factor = 0.005, autovacuum_analyze_threshold = 10000);
CREATE STATISTICS orders_country_city (dependencies, mcv) ON country, city FROM orders;
ANALYZE orders;

SELECT pid, usename, application_name, state, xact_start, backend_xmin,
       now() - xact_start AS age, left(query, 80) AS last_query
FROM pg_stat_activity
WHERE state LIKE 'idle in transaction%'
ORDER BY xact_start;
ALTER ROLE app SET idle_in_transaction_session_timeout = '60s';

Limit (actual time=5210.4..5210.5 rows=50 loops=1)
  Buffers: shared hit=1033 read=412871
  -> Sort (actual rows=50) Sort Key: created_at DESC  Sort Method: top-N heapsort  Memory: 39kB
    -> Bitmap Heap Scan on orders (cost=.. rows=50 ..) (actual rows=1902114 loops=1)
          Recheck Cond: (tenant_id = 42)
          Filter: (status = 'open')
          Rows Removed by Filter: 1899870
          -> Bitmap Index Scan on orders_tenant_id_idx (actual rows=3801984)
Planning Time: 0.3 ms
Execution Time: 5211.0 ms

-- migration (BUG: no lock_timeout)
ALTER TABLE orders ADD COLUMN region_code text;

-- FIX
SET lock_timeout = '3s';
ALTER TABLE orders ADD COLUMN region_code text;   -- retry on lock_not_available (55P03)

-- who blocks whom
SELECT pid, pg_blocking_pids(pid) AS blocked_by, wait_event_type, state, left(query, 60)
FROM pg_stat_activity WHERE datname = current_database() AND wait_event_type = 'Lock';

CREATE INDEX idx_orders ON orders (status, created_at);
SELECT id, total FROM orders
WHERE created_at >= now() - interval '7 days'
ORDER BY created_at DESC
LIMIT 100;
-- FIX
CREATE INDEX CONCURRENTLY orders_created_at_idx ON orders (created_at);

CREATE INDEX CONCURRENTLY orders_tenant_created_incl
  ON orders (tenant_id, created_at DESC) INCLUDE (status, total);
VACUUM (ANALYZE) orders;
EXPLAIN (ANALYZE, BUFFERS)
SELECT created_at, status, total FROM orders WHERE tenant_id = 42 ORDER BY created_at DESC LIMIT 50;
-- Index Only Scan ... Heap Fetches: 0

Nested Loop (cost=1.1..950.2 rows=12) (actual time=0.1..184220.7 rows=2210034 loops=1)
  -> Index Scan using customers_city_idx on customers (rows=3) (actual rows=48210)
        Index Cond: (city = 'Hanoi')  Filter: (country = 'VN')
  -> Index Scan using orders_customer_id_idx on orders (rows=4) (actual rows=46 loops=48210)

CREATE STATISTICS customers_country_city (dependencies, mcv) ON country, city FROM customers;
ANALYZE customers;

CREATE UNIQUE INDEX ON sales_mv (tenant_id, day);
REFRESH MATERIALIZED VIEW CONCURRENTLY sales_mv;

-- incremental alternative
INSERT INTO daily_sales (tenant_id, day, revenue, orders)
SELECT tenant_id, created_at::date, sum(total), count(*)
FROM orders WHERE created_at >= current_date - 1
GROUP BY 1, 2
ON CONFLICT (tenant_id, day) DO UPDATE SET revenue = EXCLUDED.revenue, orders = EXCLUDED.orders;

-- BUG
ALTER TABLE users ADD CONSTRAINT users_email_key UNIQUE (email);
-- FIX
ALTER TABLE users DROP CONSTRAINT users_email_key;
CREATE UNIQUE INDEX CONCURRENTLY users_email_active_uq ON users (lower(email)) WHERE deleted_at IS NULL;
CREATE VIEW active_users AS SELECT * FROM users WHERE deleted_at IS NULL;

-- scans all partitions
SELECT * FROM events WHERE user_id = 42 ORDER BY created_at DESC LIMIT 20;
SELECT * FROM events WHERE created_at::date = '2026-09-01';
-- prunes
SELECT * FROM events
WHERE user_id = 42 AND created_at >= now() - interval '30 days'
ORDER BY created_at DESC LIMIT 20;

// BUG 1: N+1
const orders = await db.query("SELECT * FROM orders WHERE tenant_id=$1 ORDER BY id DESC LIMIT 20", [t]);
for (const o of orders.rows) o.items = (await db.query("SELECT * FROM order_items WHERE order_id=$1", [o.id])).rows;

// BUG 2: LIMIT counts joined rows
SELECT o.*, i.* FROM orders o JOIN order_items i ON i.order_id = o.id
WHERE o.tenant_id = $1 ORDER BY o.id DESC LIMIT 20;

// FIX
SELECT o.*, coalesce(it.items, '[]') AS items
FROM (SELECT * FROM orders WHERE tenant_id = $1 AND id < $2 ORDER BY id DESC LIMIT 20) o
LEFT JOIN LATERAL (
  SELECT json_agg(i ORDER BY i.id) AS items FROM order_items i WHERE i.order_id = o.id
) it ON true
ORDER BY o.id DESC;