Build your own storage engine

// pager.ts
import { openSync, readSync, writeSync, fstatSync, fsyncSync, ftruncateSync } from 'node:fs';

export const PAGE_SIZE = 16384;

// ── page header, mirroring InnoDB's FIL header (ch 18) ──
//   0..3    page number
//   4..7    page type  (1=leaf, 2=internal, 3=meta, 0=free)
//   8..15   LSN of last modification   (ch 61)
//  16..19   number of records
//  20..23   next page  (leaf chain, ch 24)
//  24..27   prev page
//  28..31   free-space offset
export const HDR = { NUM:0, TYPE:4, LSN:8, NRECS:16, NEXT:20, PREV:24, FREE:28 } as const;
export const HDR_SIZE = 32;

export enum PageType { Free=0, Leaf=1, Internal=2, Meta=3 }

export class Page {
  constructor(public no: number, public buf: Buffer, public dirty = false) {}

  get type()   { return this.buf.readUInt32LE(HDR.TYPE) as PageType; }
  set type(t)  { this.buf.writeUInt32LE(t, HDR.TYPE); this.dirty = true; }
  get lsn()    { return Number(this.buf.readBigUInt64LE(HDR.LSN)); }
  set lsn(v)   { this.buf.writeBigUInt64LE(BigInt(v), HDR.LSN); this.dirty = true; }
  get nrecs()  { return this.buf.readUInt32LE(HDR.NRECS); }
  set nrecs(v) { this.buf.writeUInt32LE(v, HDR.NRECS); this.dirty = true; }
  get next()   { return this.buf.readInt32LE(HDR.NEXT); }
  set next(v)  { this.buf.writeInt32LE(v, HDR.NEXT); this.dirty = true; }
  get free()   { return this.buf.readUInt32LE(HDR.FREE); }
  set free(v)  { this.buf.writeUInt32LE(v, HDR.FREE); this.dirty = true; }

  /** bytes left between the record heap and the end (ch 18) */
  get freeSpace() { return PAGE_SIZE - this.free; }
}

export class Pager {
  private fd: number;
  private cache = new Map<number, Page>();   // our buffer pool (ch 69)
private nPages: number;

  constructor(public path: string) {
    this.fd = openSync(path, 'a+');
    const size = fstatSync(this.fd).size;
    this.nPages = Math.floor(size / PAGE_SIZE);
    if (this.nPages === 0) this.allocate(PageType.Meta);   // page 0
  }

  get(no: number): Page {
    const hit = this.cache.get(no);
    if (hit) return hit;

    const buf = Buffer.alloc(PAGE_SIZE);
    if (no < this.nPages) readSync(this.fd, buf, 0, PAGE_SIZE, no * PAGE_SIZE);
    const pg = new Page(no, buf);
    this.cache.set(no, pg);
    return pg;
  }

  allocate(type: PageType): Page {
    const no = this.nPages++;
    const pg = new Page(no, Buffer.alloc(PAGE_SIZE), true);
    pg.buf.writeUInt32LE(no, HDR.NUM);
    pg.type  = type;
    pg.free  = HDR_SIZE;
    pg.next  = -1;
    pg.buf.writeInt32LE(-1, HDR.PREV);
    this.cache.set(no, pg);
    return pg;
  }

  /** flush dirty pages. WAL rule: caller must fsync the log FIRST (ch 61) */
  flush() {
    for (const pg of this.cache.values()) {
      if (!pg.dirty) continue;
      writeSync(this.fd, pg.buf, 0, PAGE_SIZE, pg.no * PAGE_SIZE);
      pg.dirty = false;
    }
    fsyncSync(this.fd);
  }

  get pageCount() { return this.nPages; }
}

// record.ts
export type Value = number | string | null;
export type Row    = Value[];

// ── varint: 7 bits of payload per byte, high bit = continue ──
export function putVarint(buf: Buffer, off: number, v: number): number {
  while (v >= 0x80) { buf[off++] = (v & 0x7f) | 0x80; v >>>= 7; }
  buf[off++] = v;
  return off;
}
export function getVarint(buf: Buffer, off: number): [number, number] {
  let v = 0, shift = 0, b: number;
  do { b = buf[off++]; v |= (b & 0x7f) << shift; shift += 7; } while (b & 0x80);
  return [v >>> 0, off];
}

// ── a record, mirroring InnoDB's DYNAMIC row format (ch 20) ──
//   varint  total length
//   u32     DB_TRX_ID      which transaction wrote this (ch 43)
//   u32     DB_ROLL_PTR    index into the undo array, -1 = none
//   u8      null bitmap    (up to 8 columns, enough for a toy)
//   ...     column data, varint-prefixed for TEXT
export interface Rec { key: number; row: Row; trxId: number; rollPtr: number; }

export function encode(r: Rec): Buffer {
  const parts: Buffer[] = [];
  const head = Buffer.alloc(9);
  head.writeUInt32LE(r.trxId, 0);
  head.writeInt32LE(r.rollPtr, 4);

  let nulls = 0;
  r.row.forEach((v, i) => { if (v === null) nulls |= (1 << i); });
  head.writeUInt8(nulls, 8);
  parts.push(head);

  for (const v of r.row) {
    if (v === null) continue;                      // null costs only its bit
if (typeof v === 'number') {
      const b = Buffer.alloc(4); b.writeInt32LE(v); parts.push(b);
    } else {
      const s = Buffer.from(v, 'utf8');
      const len = Buffer.alloc(5);
      const n = putVarint(len, 0, s.length);
      parts.push(len.subarray(0, n), s);
    }
  }
  const body = Buffer.concat(parts);
  const lenBuf = Buffer.alloc(5);
  const lenN = putVarint(lenBuf, 0, body.length);
  return Buffer.concat([lenBuf.subarray(0, lenN), body]);
}

export function decode(buf: Buffer, off: number, schema: ('int'|'text')[]) {
  const [, afterLen] = getVarint(buf, off);
  let p = afterLen;
  const trxId   = buf.readUInt32LE(p); p += 4;
  const rollPtr = buf.readInt32LE(p);  p += 4;
  const nulls   = buf.readUInt8(p);    p += 1;

  const row: Row = [];
  schema.forEach((ty, i) => {
    if (nulls & (1 << i)) { row.push(null); return; }
    if (ty === 'int') { row.push(buf.readInt32LE(p)); p += 4; }
    else {
      const [len, after] = getVarint(buf, p);
      row.push(buf.subarray(after, after + len).toString('utf8'));
      p = after + len;
    }
  });
  return { rec: { key: row[0] as number, row, trxId, rollPtr }, next: p };
}

// btree.ts
import { Pager, Page, PageType, PAGE_SIZE, HDR_SIZE } from './pager';
import { encode, decode, Rec } from './record';

const SLOT = 6;   // internal entry: 4-byte key + ... (see below)
export class BTree {
  constructor(
    private pager: Pager,
    private rootNo: number,
    private schema: ('int'|'text')[],
    private onWrite?: (pageNo: number) => number   // returns an LSN (ch 114)
  ) {}

  // ── read every record on a leaf page ──
private readLeaf(pg: Page): Rec[] {
    const out: Rec[] = [];
    let off = HDR_SIZE;
    for (let i = 0; i < pg.nrecs; i++) {
      const { rec, next } = decode(pg.buf, off, this.schema);
      out.push(rec); off = next;
    }
    return out;
  }

  // ── rewrite a leaf from a sorted record list ──
private writeLeaf(pg: Page, recs: Rec[]) {
    let off = HDR_SIZE;
    for (const r of recs) {
      const b = encode(r);
      b.copy(pg.buf, off); off += b.length;
    }
    pg.nrecs = recs.length;
    pg.free  = off;
    pg.dirty = true;
    if (this.onWrite) pg.lsn = this.onWrite(pg.no);
  }

  // ── internal node: [key,childPage] pairs, 8 bytes each ──
private readInternal(pg: Page): { key: number; child: number }[] {
    const out = [];
    for (let i = 0; i < pg.nrecs; i++) {
      const o = HDR_SIZE + i * 8;
      out.push({ key: pg.buf.readInt32LE(o), child: pg.buf.readInt32LE(o + 4) });
    }
    return out;
  }
  private writeInternal(pg: Page, es: { key: number; child: number }[]) {
    es.forEach((e, i) => {
      const o = HDR_SIZE + i * 8;
      pg.buf.writeInt32LE(e.key, o);
      pg.buf.writeInt32LE(e.child, o + 4);
    });
    pg.nrecs = es.length;
    pg.free  = HDR_SIZE + es.length * 8;
    pg.dirty = true;
    if (this.onWrite) pg.lsn = this.onWrite(pg.no);
  }

  // ── descend to the leaf that should hold `key`, recording the path ──
private descend(key: number): { leaf: Page; path: Page[] } {
    let pg = this.pager.get(this.rootNo);
    const path: Page[] = [];
    while (pg.type === PageType.Internal) {
      path.push(pg);
      const es = this.readInternal(pg);
      let child = es[0].child;
      for (const e of es) if (key >= e.key) child = e.child; else break;
      pg = this.pager.get(child);
    }
    return { leaf: pg, path };
  }

  search(key: number): Rec | null {
    const { leaf } = this.descend(key);
    return this.readLeaf(leaf).find(r => r.key === key) ?? null;
  }

  /** range scan — descend once, then walk the leaf chain (ch 24) */
  *range(lo: number, hi: number): Generator<Rec> {
    let { leaf } = this.descend(lo);
    while (true) {
      for (const r of this.readLeaf(leaf))
        if (r.key >= lo && r.key <= hi) yield r;
      if (leaf.next < 0) break;
      const nxt = this.pager.get(leaf.next);
      if (this.readLeaf(nxt)[0]?.key > hi) break;
      leaf = nxt;
    }
  }

  insert(rec: Rec) {
    const { leaf, path } = this.descend(rec.key);
    const recs = this.readLeaf(leaf);

    const at = recs.findIndex(r => r.key >= rec.key);
    if (at >= 0 && recs[at].key === rec.key) recs[at] = rec;      // update
else if (at < 0) recs.push(rec);
    else recs.splice(at, 0, rec);

    const need = recs.reduce((s, r) => s + encode(r).length, 0);
    if (HDR_SIZE + need <= PAGE_SIZE) { this.writeLeaf(leaf, recs); return; }

    // ── SPLIT (ch 28) ──
const isRightEdge = rec.key > recs[recs.length - 2]?.key;
    const mid = isRightEdge
      ? recs.length - 1 // right-edge: new page gets ONLY the new row
      : Math.floor(recs.length / 2);   // otherwise a 50/50 split
const left  = recs.slice(0, mid);
    const right = recs.slice(mid);

    const newPg = this.pager.allocate(PageType.Leaf);
    newPg.next = leaf.next;
    leaf.next  = newPg.no;                     // maintain the leaf chain
this.writeLeaf(leaf, left);
    this.writeLeaf(newPg, right);
    this.insertIntoParent(path, right[0].key, newPg.no, leaf.no);
  }

  private insertIntoParent(path: Page[], key: number, right: number, left: number) {
    if (path.length === 0) {
      // the root split — the tree grows one level taller
const oldRoot = this.pager.get(this.rootNo);
      const l = this.pager.allocate(oldRoot.type);
      oldRoot.buf.copy(l.buf); l.buf.writeUInt32LE(l.no, 0); l.dirty = true;

      oldRoot.type = PageType.Internal;
      this.writeInternal(oldRoot, [
        { key: -2147483648, child: l.no },
        { key, child: right }
      ]);
      return;
    }
    const parent = path[path.length - 1];
    const es = this.readInternal(parent);
    const at = es.findIndex(e => e.key > key);
    es.splice(at < 0 ? es.length : at, 0, { key, child: right });

    if (HDR_SIZE + es.length * 8 <= PAGE_SIZE) { this.writeInternal(parent, es); return; }

    // the parent is full too — split it, recursively, up to the root
const mid = Math.floor(es.length / 2);
    const up  = es[mid].key;
    const np  = this.pager.allocate(PageType.Internal);
    this.writeInternal(parent, es.slice(0, mid));
    this.writeInternal(np, es.slice(mid));
    this.insertIntoParent(path.slice(0, -1), up, np.no, parent.no);
  }
}

// wal.ts
import { openSync, writeSync, readFileSync, fsyncSync, existsSync, closeSync } from 'node:fs';

export enum LogType { Insert=1, Update=2, Delete=3, Commit=4, Abort=5 }

// record: u32 len | u64 lsn | u32 trxId | u8 type | u32 pageNo | payload
export class WAL {
  private fd: number;
  private _lsn = 0;

  constructor(public path: string) { this.fd = openSync(path, 'a+'); }

  get lsn() { return this._lsn; }

  append(trxId: number, type: LogType, pageNo: number, payload = Buffer.alloc(0)): number {
    const body = Buffer.alloc(17 + payload.length);
    this._lsn += body.length + 4;        // LSN = bytes ever written (ch 61)
    body.writeBigUInt64LE(BigInt(this._lsn), 0);
    body.writeUInt32LE(trxId, 8);
    body.writeUInt8(type, 12);
    body.writeUInt32LE(pageNo, 13);
    payload.copy(body, 17);

    const len = Buffer.alloc(4); len.writeUInt32LE(body.length);
    writeSync(this.fd, Buffer.concat([len, body]));
    return this._lsn;
  }

  /** the fsync. this is innodb_flush_log_at_trx_commit=1 (ch 63) */
sync() { fsyncSync(this.fd); }

  /** replay: collect committed transactions, then re-apply them (ch 67) */
recover(): { committed: Set<number>; records: any[] } {
    if (!existsSync(this.path)) return { committed: new Set(), records: [] };
    const buf = readFileSync(this.path);
    const records: any[] = [];
    const committed = new Set<number>();

    let off = 0;
    while (off + 4 <= buf.length) {
      const len = buf.readUInt32LE(off);
      if (off + 4 + len > buf.length) break;      // torn tail — stop here
const b = buf.subarray(off + 4, off + 4 + len);
      const rec = {
        lsn:     Number(b.readBigUInt64LE(0)),
        trxId:   b.readUInt32LE(8),
        type:    b.readUInt8(12) as LogType,
        pageNo:  b.readUInt32LE(13),
        payload: b.subarray(17)
      };
      if (rec.type === LogType.Commit) committed.add(rec.trxId);
      records.push(rec);
      this._lsn = rec.lsn;
      off += 4 + len;
    }
    // only records from COMMITTED transactions get re-applied.
// this is the undo phase, done by omission (ch 67).
return { committed, records: records.filter(r => committed.has(r.trxId)) };
  }

  close() { closeSync(this.fd); }
}

// mvcc.ts
import { Rec, Row } from './record';

export interface UndoEntry { trxId: number; row: Row; prev: number; }

export class TrxManager {
  private nextId = 1;
  private active = new Set<number>();
  public  undo: UndoEntry[] = [];       // our undo log, in memory (ch 44)
begin(): number { const id = this.nextId++; this.active.add(id); return id; }
  commit(id: number) { this.active.delete(id); }
  abort(id: number)  { this.active.delete(id); }

  /** the read view — a snapshot of who was running (ch 42) */
readView(creator: number): ReadView {
    return new ReadView(creator, new Set(this.active), this.nextId);
  }

  pushUndo(e: UndoEntry): number { this.undo.push(e); return this.undo.length - 1; }
}

export class ReadView {
  readonly upLimit: number;
  constructor(
    readonly creator: number,
    readonly mIds: Set<number>,
    readonly lowLimit: number
  ) {
    this.upLimit = mIds.size ? Math.min(...mIds) : lowLimit;
  }

  /** chapter 42's algorithm, line for line */
visible(trxId: number): boolean {
    if (trxId === this.creator)   return true;    // you wrote it
if (trxId <  this.upLimit)   return true;    // committed before us
if (trxId >= this.lowLimit)  return false;   // started after us
return !this.mIds.has(trxId);                  // in flight at snapshot time?
  }
}

/** walk the version chain until a visible version is found (ch 44) */
export function visibleVersion(
  rec: Rec, view: ReadView, undo: UndoEntry[]
): Row | null {
  if (view.visible(rec.trxId)) return rec.row;

  let ptr = rec.rollPtr;
  while (ptr >= 0) {
    const u = undo[ptr];
    if (view.visible(u.trxId)) return u.row;
    ptr = u.prev;
  }
  return null;   // no version this transaction is allowed to see
}

// sql.ts
import { BTree } from './btree';
import { Row, Rec } from './record';
import { ReadView, visibleVersion, UndoEntry } from './mvcc';

// ── tokenizer ──
const tokenize = (s: string) =>
  s.match(/\s*([A-Za-z_]\w*|\d+|'[^']*'|>=|<=|<>|.)/g)?.map(t => t.trim()) ?? [];

// ── AST ──
type Pred  = { col: number; op: string; val: any } | null;
type Query = { cols: number[]; pred: Pred; limit: number };

// ── parser: SELECT c,... FROM t [WHERE col op val] [LIMIT n] ──
export function parse(sql: string, schema: string[]): Query {
  const tk = tokenize(sql); let i = 0;
  const eat = (w?: string) => {
    const t = tk[i++];
    if (w && t?.toUpperCase() !== w) throw new Error(`expected ${w}, got ${t}`);
    return t;
  };

  eat('SELECT');
  const cols: number[] = [];
  if (tk[i] === '*') { i++; schema.forEach((_, n) => cols.push(n)); }
  else do { cols.push(schema.indexOf(eat()!)); } while (tk[i] === ',' && ++i);

  eat('FROM'); eat();                       // table name — one table only
let pred: Pred = null;
  if (tk[i]?.toUpperCase() === 'WHERE') {
    i++;
    const col = schema.indexOf(eat()!);
    const op  = eat()!;
    const raw = eat()!;
    const val = raw.startsWith("'") ? raw.slice(1, -1) : Number(raw);
    pred = { col, op, val };
  }

  let limit = Infinity;
  if (tk[i]?.toUpperCase() === 'LIMIT') { i++; limit = Number(eat()); }

  return { cols, pred, limit };
}

// ══ volcano iterators — each pulls from its child (ch 13) ══
/** leaf: scan the tree, applying MVCC visibility as we go */
function* scan(t: BTree, view: ReadView, undo: UndoEntry[]): Generator<Row> {
  for (const rec of t.range(-Infinity, Infinity)) {
    const row = visibleVersion(rec, view, undo);
    if (row) yield row;
  }
}

/** index range scan — the sargable path (ch 74) */
function* indexScan(t: BTree, lo: number, hi: number,
                      view: ReadView, undo: UndoEntry[]): Generator<Row> {
  for (const rec of t.range(lo, hi)) {
    const row = visibleVersion(rec, view, undo);
    if (row) yield row;
  }
}

function* filter(src: Generator<Row>, p: Pred): Generator<Row> {
  const test: Record<string, (a:any,b:any)=>boolean> = {
    '=':(a,b)=>a===b, '>':(a,b)=>a>b, '<':(a,b)=>a<b,
    '>=':(a,b)=>a>=b, '<=':(a,b)=>a<=b, '<>':(a,b)=>a!==b
  };
  for (const r of src) if (!p || test[p.op](r[p.col], p.val)) yield r;
}

function* project(src: Generator<Row>, cols: number[]): Generator<Row> {
  for (const r of src) yield cols.map(c => r[c]);
}

function* limitIter(src: Generator<Row>, n: number): Generator<Row> {
  let i = 0;
  for (const r of src) { if (i++ >= n) return; yield r; }
  // note: `return` stops pulling — the child is abandoned. ch 13.
}

/** the "optimizer": one rule — use the index when we can (ch 12) */
export function execute(
  q: Query, t: BTree, view: ReadView, undo: UndoEntry[]
): Row[] {
  const pkPred = q.pred && q.pred.col === 0 && typeof q.pred.val === 'number';

  let it: Generator<Row>;
  if (pkPred && q.pred!.op === '=') {
    it = indexScan(t, q.pred!.val, q.pred!.val, view, undo);          // const
  } else if (pkPred && ['>','>='].includes(q.pred!.op)) {
    it = indexScan(t, q.pred!.val, Infinity, view, undo);           // range
  } else {
    it = filter(scan(t, view, undo), q.pred);                         // full scan
  }

  return [...limitIter(project(it, q.cols), q.limit)];
}

// torture.ts — run with: node --test, or directly
import { spawnSync } from 'node:child_process';
import { unlinkSync, existsSync } from 'node:fs';
import { MiniDB } from './db';

// ── test 1: kill -9 during a write, then verify recovery ──
function crashTest() {
  for (const f of ['t.db', 't.wal']) if (existsSync(f)) unlinkSync(f);

  // child writes 10,000 rows, committing every 100, and is killed mid-flight
const child = spawnSync('node', ['-e', `
    const { MiniDB } = require('./db');
    const db = new MiniDB('t.db', 't.wal');
    for (let i = 0; i < 10000; i++) {
      const trx = db.begin();
      db.insert(trx, [i, 'row' + i]);
      if (i % 100 === 0) db.commit(trx);
      if (i === 5000) process.kill(process.pid, 'SIGKILL');
    }
  `], { timeout: 10000 });

  // now reopen. recovery runs in the constructor.
const db = new MiniDB('t.db', 't.wal');

  // INVARIANTS that must hold:
//  1. every COMMITTED row is present
//  2. no UNCOMMITTED row is present
//  3. the tree is still structurally valid
const rows = db.selectAll();
  const keys = rows.map(r => r[0] as number).sort((a,b)=>a-b);

  assert(keys.every((k,i) => i === 0 || k > keys[i-1]), 'keys must be unique and sorted');
  assert(db.verifyTree(), 'tree structure must be intact');
  assert(keys.length % 100 === 0 || keys.length === 0,
         'only whole committed batches should survive');
  console.log(`✓ recovered ${keys.length} committed rows after SIGKILL`);
}

// ── test 2: the chapter 31 experiment, as an assertion ──
function keyLocalityTest() {
  const seq  = new MiniDB('seq.db',  'seq.wal');
  const rand = new MiniDB('rand.db', 'rand.wal');

  for (let i = 0; i < 50000; i++) {
    const t1 = seq.begin();  seq.insert(t1, [i, 'x'.repeat(80)]);  seq.commit(t1);
    const t2 = rand.begin();
    rand.insert(t2, [Math.floor(Math.random() * 1e9), 'x'.repeat(80)]);
    rand.commit(t2);
  }
  console.log('sequential pages:', seq.pageCount);
  console.log('random pages:    ', rand.pageCount);
  console.log('ratio:           ', (rand.pageCount / seq.pageCount).toFixed(2) + 'x');
  // expect roughly 1.6-2.0x — the UUID penalty from chapter 31,
// measured in your own engine.
}

function assert(c: boolean, m: string) { if (!c) throw new Error('FAILED: ' + m); }

crashTest();
keyLocalityTest();

# Dump a page from YOUR engine:
node -e "
  const {Pager} = require('./pager');
  const p = new Pager('t.db');
  const pg = p.get(1);
  console.log(pg.buf.subarray(0,64).toString('hex').match(/.{1,8}/g).join(' '));
"

# Dump a page from a real InnoDB file (chapter 23):
sudo hexdump -C -s 49152 -n 64 /var/lib/mysql/yourdb/users.ibd

# They will not be identical — you designed your own header —
# but you will recognise every field: a page number, an LSN,
# a record count, a next-page pointer. That recognition is
# the point of this entire part.