Skip to main content

powdb_storage/
catalog.rs

1use crate::error::StorageError;
2use crate::heap::HeapFile;
3use crate::page::{OVERFLOW_CHAIN_END, OVERFLOW_PAYLOAD_CAP};
4use crate::row::{encode_row_into, encode_row_v2_into, plan_spill, OverflowStub, MAX_VALUE_SIZE};
5use crate::table::Table;
6use crate::types::*;
7use crate::wal::{Wal, WalDurabilityTicket, WalRecord, WalRecordType, WalSyncMode};
8use rustc_hash::FxHashMap;
9use std::fs;
10use std::io::{self, Read, Write};
11use std::path::{Path, PathBuf};
12use tracing::{info, warn};
13
14/// Reject an encoded row that exceeds the single-page capacity BEFORE it is
15/// appended to the WAL. The heap performs the same check at its own insert/
16/// update boundary, but the update paths log to the WAL first — a logged
17/// record whose row the heap then rejects would poison the next replay.
18fn check_encoded_row_size(encoded: &[u8]) -> io::Result<()> {
19    if encoded.len() > crate::page::MAX_ROW_DATA_SIZE {
20        return Err(crate::error::StorageError::RowTooLarge {
21            size: encoded.len(),
22            max: crate::page::MAX_ROW_DATA_SIZE,
23        }
24        .into());
25    }
26    Ok(())
27}
28
29/// Validate that a name (table or column) is safe for use in file paths and
30/// follows the identifier convention: starts with a letter or underscore,
31/// followed by letters, digits, or underscores.
32fn validate_identifier(kind: &str, name: &str) -> io::Result<()> {
33    if name.is_empty() {
34        return Err(io::Error::new(
35            io::ErrorKind::InvalidInput,
36            format!("invalid {kind} name: must not be empty"),
37        ));
38    }
39    let mut chars = name.chars();
40    // Infallible: we returned early if `name.is_empty()` above.
41    let first = chars.next().expect("non-empty name");
42    if !first.is_ascii_alphabetic() && first != '_' {
43        return Err(io::Error::new(
44            io::ErrorKind::InvalidInput,
45            format!("invalid {kind} name '{name}': must start with a letter or underscore"),
46        ));
47    }
48    for ch in chars {
49        if !ch.is_ascii_alphanumeric() && ch != '_' {
50            return Err(io::Error::new(
51                io::ErrorKind::InvalidInput,
52                format!(
53                    "invalid {kind} name '{name}': must contain only letters, digits, and underscores"
54                ),
55            ));
56        }
57    }
58    Ok(())
59}
60
61/// Validate a table name for path safety.
62fn validate_table_name(name: &str) -> io::Result<()> {
63    validate_identifier("table", name)
64}
65
66/// Validate a column name for path safety.
67fn validate_column_name(name: &str) -> io::Result<()> {
68    validate_identifier("column", name)
69}
70
71/// On-disk catalog file: lists every table's schema so we can reopen them
72/// after a restart. Format is a small custom binary blob (no serde dep).
73///
74/// Mission 3: version 2 appends a per-table list of indexed column names
75/// after the column list, so indexes can be rehydrated on `Catalog::open`.
76/// Version 1 files still load cleanly — they're treated as having zero
77/// indexed columns, and the next `create_index` (or implicit rebuild on
78/// first open, depending on the caller) will populate the list.
79const CATALOG_FILE: &str = "catalog.bin";
80pub const CATALOG_LSN_FILE: &str = "catalog.lsn";
81const CATALOG_MAGIC: &[u8; 4] = b"BCAT";
82/// Version 4 appends a per-table column-defaults section after the indexed
83/// column list; version 5 appends an auto-increment column section after that.
84/// Older files load cleanly (no defaults / no auto columns).
85pub const CATALOG_VERSION: u16 = 5;
86
87/// Mission 2 (durability): the single shared WAL file lives under the catalog's
88/// data directory with this name. One WAL covers every table in the catalog.
89const WAL_FILE: &str = "wal.log";
90const SYNC_STATE_DIR: &str = ".powdb-sync";
91const SYNC_IDENTITY_FILE: &str = "identity.json";
92
93/// WAL batch size: flush auto-triggers after this many records, in addition
94/// to the explicit `wal.flush()` each top-level mutation does. Kept small so
95/// the tests see a predictable amount of buffering.
96const WAL_BATCH_SIZE: usize = 64;
97type WalArchiveCallback<'a> = &'a mut dyn FnMut(&Path, &[WalRecord]) -> io::Result<()>;
98
99fn read_durable_lsn(data_dir: &Path) -> io::Result<u64> {
100    let path = data_dir.join(CATALOG_LSN_FILE);
101    let bytes = match fs::read(path) {
102        Ok(bytes) => bytes,
103        Err(err) if err.kind() == io::ErrorKind::NotFound => return Ok(0),
104        Err(err) => return Err(err),
105    };
106    if bytes.len() != 8 {
107        return Err(io::Error::new(
108            io::ErrorKind::InvalidData,
109            "catalog LSN sidecar has invalid length",
110        ));
111    }
112    let mut buf = [0u8; 8];
113    buf.copy_from_slice(&bytes);
114    Ok(u64::from_le_bytes(buf))
115}
116
117fn write_durable_lsn(data_dir: &Path, lsn: u64) -> io::Result<()> {
118    let path = data_dir.join(CATALOG_LSN_FILE);
119    let tmp_path = data_dir.join(format!("{CATALOG_LSN_FILE}.tmp"));
120    let mut file = fs::File::create(&tmp_path)?;
121    file.write_all(&lsn.to_le_bytes())?;
122    file.sync_all()?;
123    drop(file);
124    fs::rename(&tmp_path, &path)?;
125    sync_directory(data_dir)?;
126    Ok(())
127}
128
129#[cfg(unix)]
130fn sync_directory(path: &Path) -> io::Result<()> {
131    fs::File::open(path)?.sync_all()
132}
133
134#[cfg(not(unix))]
135fn sync_directory(path: &Path) -> io::Result<()> {
136    let _ = path;
137    Ok(())
138}
139
140fn max_record_lsn(records: &[WalRecord]) -> Option<u64> {
141    records.iter().map(|record| record.lsn).max()
142}
143
144/// System catalog: registry of all tables.
145///
146/// Mission C Phase 18: tables live in a `Vec<Table>` addressed by a
147/// stable `slot` index, with a parallel `FxHashMap<String, usize>` for
148/// name-based resolution. Append-only (PowDB has no DROP TABLE yet), so
149/// slots are stable for the lifetime of the `Catalog` — callers like
150/// `PreparedQuery::insert_fast` cache a slot at prepare time and skip
151/// the name probe on every subsequent `execute_prepared_take`.
152///
153/// Earlier design (pre-Phase 18) held tables in a `FxHashMap<String, Table>`
154/// directly. That meant the `insert_batch_1k` hot path paid an
155/// `FxHash("User")` + bucket walk per row just to dispatch into the
156/// table — about 20-40ns out of a 233ns budget.
157pub struct Catalog {
158    /// All tables, in insertion order. Indexed by `slot: usize`. A table's
159    /// slot is assigned by `create_table`/`open` and never reused.
160    tables: Vec<Table>,
161    /// Name → slot index. Populated in sync with `tables` on every
162    /// `create_table` / `open`.
163    name_to_slot: FxHashMap<String, usize>,
164    data_dir: PathBuf,
165    /// Mission 2: shared write-ahead log owned by the catalog. Every
166    /// mutation (insert/update/delete) records its intent here BEFORE
167    /// touching the heap so a mid-write crash can be recovered from on the
168    /// next open. Flushed to disk at the end of every top-level op.
169    wal: Wal,
170    /// Monotonic transaction-id counter. Autocommit statements may allocate
171    /// multiple ids (one per row-level primitive), while explicit transactions
172    /// reuse one id for the whole BEGIN..COMMIT scope.
173    next_tx_id: u64,
174    /// Active explicit transaction id, if any. Owned by the connection/session
175    /// driving this catalog through `Engine`.
176    active_tx_id: Option<u64>,
177    /// Durable WAL byte offset captured at BEGIN. ROLLBACK truncates back to
178    /// this boundary so auto-flushed uncommitted records cannot replay later.
179    tx_start_len: Option<u64>,
180    /// Autocommit row-mutation tx ids appended since the previous group commit.
181    /// `commit_autocommit` writes commit markers for these ids before fsync.
182    pending_autocommit_tx_ids: Vec<u64>,
183    /// Has this catalog been cleanly checkpointed at least once since it
184    /// was opened? Used by `Drop` to decide whether to treat its own flush
185    /// as fatal (it isn't — we still try best-effort).
186    checkpointed: bool,
187    /// Catalog-level durable LSN. Heap page LSNs cover row mutations, but
188    /// DDL-only changes can advance the WAL without touching a data page.
189    durable_lsn: u64,
190    /// Overflow-chain pages to return to their table's free list once the
191    /// current EXPLICIT transaction commits (design 3.6 pending-free list).
192    /// Populated only while `active_tx_id.is_some()`: a chain-replacing update
193    /// or a delete inside a transaction cannot free its old chain immediately,
194    /// because ROLLBACK resurrects the old row and its stub must still address a
195    /// live chain. Autocommit mutations free immediately (no rollback window).
196    /// Drained by `commit_transaction`; discarded (via reopen) by ROLLBACK.
197    /// Entries are `(table_slot, chain_pages)`.
198    pending_free_overflow: Vec<(usize, Vec<u32>)>,
199}
200
201impl Catalog {
202    /// Create a brand-new catalog. Wipes any existing catalog file in this directory.
203    ///
204    /// # Examples
205    ///
206    /// ```
207    /// use powdb_storage::catalog::Catalog;
208    /// use powdb_storage::types::{Schema, ColumnDef, TypeId};
209    ///
210    /// let dir = tempfile::tempdir().unwrap();
211    /// let mut catalog = Catalog::create(dir.path()).unwrap();
212    ///
213    /// let schema = Schema {
214    ///     table_name: "User".to_string(),
215    ///     columns: vec![
216    ///         ColumnDef { name: "name".to_string(), type_id: TypeId::Str, required: true, position: 0 },
217    ///         ColumnDef { name: "age".to_string(), type_id: TypeId::Int, required: false, position: 1 },
218    ///     ],
219    /// };
220    /// catalog.create_table(schema).unwrap();
221    /// ```
222    pub fn create(data_dir: &Path) -> io::Result<Self> {
223        crate::create_data_dir_secure(data_dir)?;
224        let wal_path = data_dir.join(WAL_FILE);
225        let wal = Wal::create(&wal_path, WAL_BATCH_SIZE)?;
226        let cat = Catalog {
227            tables: Vec::new(),
228            name_to_slot: FxHashMap::default(),
229            data_dir: data_dir.to_path_buf(),
230            wal,
231            next_tx_id: 1,
232            active_tx_id: None,
233            tx_start_len: None,
234            pending_autocommit_tx_ids: Vec::new(),
235            pending_free_overflow: Vec::new(),
236            checkpointed: false,
237            durable_lsn: 0,
238        };
239        cat.persist()?;
240        Ok(cat)
241    }
242
243    /// Open an existing catalog from disk, rehydrating every table. If no
244    /// catalog file is present this returns NotFound — callers can fall back
245    /// to `create` for a fresh data dir.
246    ///
247    /// Mission 2: after the per-table heap files are reopened, this replays
248    /// any records left in the WAL from a previous (crashed) session. The
249    /// WAL is then truncated once the replay lands cleanly on disk — that
250    /// re-establishes the "empty WAL = last shutdown was clean" invariant.
251    pub fn open(data_dir: &Path) -> io::Result<Self> {
252        Self::open_inner(data_dir, None)
253    }
254
255    /// Open an existing catalog and archive any replayed WAL records before
256    /// recovery truncates the WAL. This is for sync-aware callers that must
257    /// retain history needed by replicas.
258    ///
259    /// Replication boundary: this hook exists so `powdb-sync` can preserve WAL
260    /// history before storage recovery truncates it. Ordinary embedded/server
261    /// callers should use `open`; do not build application-level recovery flows
262    /// directly on this hook.
263    pub fn open_with_wal_archive<F>(data_dir: &Path, mut archive: F) -> io::Result<Self>
264    where
265        F: FnMut(&Path, &[WalRecord]) -> io::Result<()>,
266    {
267        let archive: WalArchiveCallback<'_> = &mut archive;
268        Self::open_inner(data_dir, Some(archive))
269    }
270
271    fn open_inner(data_dir: &Path, archive: Option<WalArchiveCallback<'_>>) -> io::Result<Self> {
272        let cat_path = data_dir.join(CATALOG_FILE);
273        if !cat_path.exists() {
274            return Err(io::Error::new(io::ErrorKind::NotFound, "no catalog file"));
275        }
276        let entries = read_catalog_file(&cat_path)?;
277        let durable_lsn = read_durable_lsn(data_dir)?;
278        let mut tables: Vec<Table> = Vec::with_capacity(entries.len());
279        let mut name_to_slot =
280            FxHashMap::with_capacity_and_hasher(entries.len(), Default::default());
281        for CatalogEntry {
282            schema,
283            indexed_cols,
284            defaults,
285            auto_cols,
286        } in entries
287        {
288            let name = schema.table_name.clone();
289            // Mission 3: rehydrate persisted indexes. `Table::open_with_indexes`
290            // tries to `BTree::load` each named index file; if a file is
291            // missing (e.g. first open after upgrade from catalog v1) it
292            // falls back to rebuilding from the heap scan and saving to
293            // disk so subsequent opens hit the fast path.
294            let mut table = Table::open_with_indexes(schema, data_dir, &indexed_cols)?;
295            table.set_defaults(defaults);
296            table.set_auto_cols(auto_cols);
297            name_to_slot.insert(name, tables.len());
298            tables.push(table);
299        }
300        let wal_path = data_dir.join(WAL_FILE);
301        let wal = Wal::open(&wal_path, WAL_BATCH_SIZE)?;
302        let mut cat = Catalog {
303            tables,
304            name_to_slot,
305            data_dir: data_dir.to_path_buf(),
306            wal,
307            next_tx_id: 1,
308            active_tx_id: None,
309            tx_start_len: None,
310            pending_autocommit_tx_ids: Vec::new(),
311            pending_free_overflow: Vec::new(),
312            checkpointed: false,
313            durable_lsn,
314        };
315        cat.replay_wal(archive)?;
316        // Restore WAL LSN monotonicity across the restart. Heap pages carry
317        // LSNs stamped by replay (catalog.rs set_page_lsn) and by DDL
318        // rewrites (stamp_all_pages_min_lsn), but `Wal::open` reset the
319        // counter to 1. If the next write reused an LSN <= a stamped page
320        // LSN, the following crash's replay would skip it as already-applied
321        // — the data-loss bug behind the v0.4.x yanks. This runs on every
322        // open (including the empty-WAL clean-shutdown path, where pages may
323        // still carry LSNs from an earlier recovery). LSNs must be monotonic
324        // across restarts.
325        let max_page_lsn = cat
326            .tables
327            .iter()
328            .map(|t| t.heap.max_page_lsn())
329            .max()
330            .unwrap_or(0);
331        let max_known_lsn = max_page_lsn.max(cat.durable_lsn);
332        cat.wal.set_next_lsn_at_least(max_known_lsn + 1);
333        // Auto-sweep overflow orphans after recovery: a crash is exactly when
334        // a chain page can end up flushed but referenced by no committed row
335        // (its Insert was uncommitted, or its Delete committed). Reclaim them
336        // now (design 3.6). Best-effort — a sweep failure must not block open.
337        if let Err(e) = cat.sweep_all() {
338            warn!(error = %e, "post-recovery overflow sweep failed (non-fatal)");
339        }
340        Ok(cat)
341    }
342
343    /// Replay every record currently buffered in the WAL file onto the open
344    /// tables. This is the recovery path: after a crash the heap files on
345    /// disk may be missing mutations that were logged to the WAL but never
346    /// written back to their pages. We re-apply every record unconditionally.
347    ///
348    /// **Idempotence:**
349    /// - `Delete`: idempotent — `HeapFile::delete` on an already-deleted or
350    ///   missing slot is a no-op.
351    /// - `Update`: idempotent — re-applies the same new row bytes to the
352    ///   same `RowId`, which either replaces the existing (already-updated)
353    ///   row with itself or lands the update for the first time.
354    /// - `Insert`: **NOT strictly idempotent**. `HeapFile::insert` allocates
355    ///   a fresh `RowId` on every call, so a row that was already flushed
356    ///   to disk will be re-inserted at a new location, producing a
357    ///   duplicate. See the mission report for the full caveat.
358    ///
359    /// The practical consequences are:
360    ///   1. On a "pure crash" (no heap pages ever flushed between open and
361    ///      crash), replay cleanly restores every logged row.
362    ///   2. On a crash where some heap pages were flushed by the hot-page
363    ///      eviction logic, replay may restore those rows a second time.
364    ///      A future mission can fix this with LSN-tagged pages.
365    ///
366    /// After a successful replay we truncate the WAL so the next shutdown
367    /// (crash or otherwise) replays only the NEW records.
368    fn replay_wal(&mut self, mut archive: Option<WalArchiveCallback<'_>>) -> io::Result<()> {
369        let records = self.wal.read_all()?;
370        if records.is_empty() {
371            return Ok(());
372        }
373        if archive.is_none() {
374            self.ensure_plain_wal_truncate_allowed(&records)?;
375        }
376        self.replay_records(&records)?;
377        if let Some(archive) = archive.as_mut() {
378            archive(&self.data_dir, &records)?;
379        }
380        self.wal.truncate()?;
381        Ok(())
382    }
383
384    /// Apply an LSN-preserving WAL record stream without appending it to the
385    /// local WAL. Sync callers must validate lineage and contiguity before
386    /// calling this method.
387    ///
388    /// Replication boundary: this is a storage adapter for `powdb-sync`, not a
389    /// general mutation API. Callers must reject unsupported record classes,
390    /// hold their own replica progress state, and pass only contiguous,
391    /// transaction-complete ranges or chunks.
392    pub fn apply_wal_records(&mut self, records: &[WalRecord]) -> io::Result<()> {
393        self.ensure_no_active_transaction_for_checkpoint()?;
394        self.ensure_no_pending_wal_records()?;
395        self.replay_records(records)
396    }
397
398    /// Sync callers use this before deciding an apply is a no-op. A replica with
399    /// local WAL history is divergent until a higher layer explicitly repairs it.
400    pub fn ensure_no_pending_wal_records(&self) -> io::Result<()> {
401        if self.wal.has_pending() || !self.wal.read_all()?.is_empty() {
402            return Err(io::Error::other(
403                "cannot apply replicated WAL records while local WAL records are pending",
404            ));
405        }
406        Ok(())
407    }
408
409    fn replay_records(&mut self, records: &[WalRecord]) -> io::Result<()> {
410        if records.is_empty() {
411            return Ok(());
412        }
413
414        info!(count = records.len(), "applying WAL records");
415
416        // Per-page LSN redo (ARIES-style). A record is already durable iff
417        // its *target page* carries an LSN >= the record's LSN. The previous
418        // implementation used a single per-table max LSN, which is unsafe:
419        // a low-LSN record on an unflushed page would be wrongly skipped
420        // because some other, flushed page of the same table advertised a
421        // higher LSN — silently dropping the record (one of the v0.4.x
422        // data-loss bugs). Every record now carries its real RowId (inserts
423        // included), so the target page is always known.
424        let has_boundaries = records.iter().any(|rec| {
425            matches!(
426                rec.record_type,
427                WalRecordType::Begin | WalRecordType::Commit | WalRecordType::Rollback
428            )
429        });
430        let mut committed_row_records = vec![true; records.len()];
431        if has_boundaries {
432            committed_row_records.fill(false);
433            let mut pending_tx_spans: Vec<(u64, Vec<usize>)> = Vec::new();
434            for (index, rec) in records.iter().enumerate() {
435                match rec.record_type {
436                    WalRecordType::Insert
437                    | WalRecordType::Update
438                    | WalRecordType::Delete
439                    | WalRecordType::OverflowWrite
440                    | WalRecordType::OverflowFree
441                        if rec.tx_id == 0 =>
442                    {
443                        committed_row_records[index] = true;
444                    }
445                    WalRecordType::Insert
446                    | WalRecordType::Update
447                    | WalRecordType::Delete
448                    | WalRecordType::OverflowWrite
449                    | WalRecordType::OverflowFree => {
450                        if let Some((_, rows)) = pending_tx_spans
451                            .iter_mut()
452                            .rev()
453                            .find(|(tx_id, _)| *tx_id == rec.tx_id)
454                        {
455                            rows.push(index);
456                        } else {
457                            pending_tx_spans.push((rec.tx_id, vec![index]));
458                        }
459                    }
460                    WalRecordType::Begin if rec.tx_id != 0 => {
461                        pending_tx_spans.push((rec.tx_id, Vec::new()));
462                    }
463                    WalRecordType::Commit if rec.tx_id != 0 => {
464                        if let Some(span_index) = pending_tx_spans
465                            .iter()
466                            .rposition(|(tx_id, _)| *tx_id == rec.tx_id)
467                        {
468                            let (_, rows) = pending_tx_spans.remove(span_index);
469                            for row_index in rows {
470                                committed_row_records[row_index] = true;
471                            }
472                        }
473                    }
474                    WalRecordType::Rollback if rec.tx_id != 0 => {
475                        if let Some(span_index) = pending_tx_spans
476                            .iter()
477                            .rposition(|(tx_id, _)| *tx_id == rec.tx_id)
478                        {
479                            pending_tx_spans.remove(span_index);
480                        }
481                    }
482                    _ => {}
483                }
484            }
485        }
486
487        let mut replayed_inserts = 0usize;
488        let mut replayed_updates = 0usize;
489        let mut replayed_deletes = 0usize;
490        let mut skipped = 0usize;
491        let mut skipped_uncommitted = 0usize;
492        let mut saw_ddl = false;
493        for (index, rec) in records.iter().enumerate() {
494            if has_boundaries
495                && !committed_row_records[index]
496                && matches!(
497                    rec.record_type,
498                    WalRecordType::Insert
499                        | WalRecordType::Update
500                        | WalRecordType::Delete
501                        | WalRecordType::OverflowWrite
502                        | WalRecordType::OverflowFree
503                )
504            {
505                skipped_uncommitted += 1;
506                continue;
507            }
508            match rec.record_type {
509                WalRecordType::Insert => {
510                    if let Some((table_name, rid, row_bytes)) = decode_wal_payload(&rec.data) {
511                        if let Some(slot) = self.name_to_slot.get(&table_name).copied() {
512                            let tbl = &mut self.tables[slot];
513                            // Already persisted on its page? Skip — re-running
514                            // the insert would allocate a fresh slot and
515                            // duplicate the row.
516                            if rec.lsn > 0 && tbl.heap.page_lsn(rid.page_id) >= rec.lsn {
517                                skipped += 1;
518                                continue;
519                            }
520                            // Not yet durable: place the row at its exact
521                            // logged RowId so later Update/Delete records
522                            // (which carry that RowId) stay correctly
523                            // targeted. A plain re-`insert` would self-assign
524                            // a fresh slot whose position can diverge from the
525                            // original after a partial-flush crash.
526                            tbl.heap.insert_at(rid, &row_bytes)?;
527                            tbl.heap.set_page_lsn(rid.page_id, rec.lsn)?;
528                            replayed_inserts += 1;
529                        }
530                    }
531                }
532                WalRecordType::Update => {
533                    if let Some((table_name, rid, row_bytes)) = decode_wal_payload(&rec.data) {
534                        if let Some(slot) = self.name_to_slot.get(&table_name).copied() {
535                            let tbl = &mut self.tables[slot];
536                            if rec.lsn > 0 && tbl.heap.page_lsn(rid.page_id) >= rec.lsn {
537                                skipped += 1;
538                                continue;
539                            }
540                            let new_rid = tbl.heap.update(rid, &row_bytes)?;
541                            tbl.heap.set_page_lsn(new_rid.page_id, rec.lsn)?;
542                            replayed_updates += 1;
543                        }
544                    }
545                }
546                WalRecordType::Delete => {
547                    if let Some((table_name, rid, _)) = decode_wal_payload(&rec.data) {
548                        if let Some(slot) = self.name_to_slot.get(&table_name).copied() {
549                            let tbl = &mut self.tables[slot];
550                            if rec.lsn > 0 && tbl.heap.page_lsn(rid.page_id) >= rec.lsn {
551                                skipped += 1;
552                                continue;
553                            }
554                            let _ = tbl.heap.delete(rid);
555                            tbl.heap.set_page_lsn(rid.page_id, rec.lsn)?;
556                            replayed_deletes += 1;
557                        }
558                    }
559                }
560                WalRecordType::OverflowWrite => {
561                    // Physical redo of one chain chunk. Applied by page id
562                    // under the per-page LSN skip, so double replay is a
563                    // no-op. Ordered before its Insert/Update in the log, so
564                    // the stub the row carries always points at live pages.
565                    if let Some((table_name, page_id, next_page, chunk)) =
566                        decode_overflow_write_payload(&rec.data)
567                    {
568                        if let Some(slot) = self.name_to_slot.get(&table_name).copied() {
569                            let tbl = &mut self.tables[slot];
570                            if rec.lsn > 0 && tbl.heap.overflow_page_lsn(page_id) >= rec.lsn {
571                                skipped += 1;
572                                continue;
573                            }
574                            tbl.heap
575                                .write_overflow_page(page_id, next_page, &chunk, rec.lsn)?;
576                        }
577                    }
578                }
579                WalRecordType::OverflowFree => {
580                    // Return a freed chain's pages to the in-memory free list.
581                    // Only reached for committed records (uncommitted frees
582                    // are skipped above), so a live row can never lose its
583                    // chain to a rolled-back free.
584                    if let Some((table_name, pages)) = decode_overflow_free_payload(&rec.data) {
585                        if let Some(slot) = self.name_to_slot.get(&table_name).copied() {
586                            self.tables[slot].heap.release_overflow_pages(&pages);
587                        }
588                    }
589                }
590                WalRecordType::Begin | WalRecordType::Commit | WalRecordType::Rollback => {
591                    // Boundary records were consumed in the first pass.
592                }
593                WalRecordType::DdlCreateTable => {
594                    saw_ddl = true;
595                    if let Some((schema, defaults, auto_cols)) = decode_ddl_create_table(&rec.data)
596                    {
597                        if !self.name_to_slot.contains_key(&schema.table_name) {
598                            if let Ok(mut table) = Table::create(schema, &self.data_dir) {
599                                table.set_defaults(defaults);
600                                table.set_auto_cols(auto_cols);
601                                let slot = self.tables.len();
602                                let name = table.schema.table_name.clone();
603                                self.tables.push(table);
604                                self.name_to_slot.insert(name, slot);
605                            }
606                        }
607                    }
608                }
609                WalRecordType::DdlDropTable => {
610                    saw_ddl = true;
611                    if let Some((table_name, _)) = decode_ddl_table_name(&rec.data) {
612                        if let Some(&slot) = self.name_to_slot.get(&table_name) {
613                            let heap_path = self.data_dir.join(format!("{table_name}.heap"));
614                            if heap_path.exists() {
615                                let _ = fs::remove_file(&heap_path);
616                            }
617                            for col_name in self.tables[slot].indexed_column_names() {
618                                let idx_path =
619                                    self.data_dir.join(format!("{table_name}_{col_name}.idx"));
620                                if idx_path.exists() {
621                                    let _ = fs::remove_file(&idx_path);
622                                }
623                            }
624                            self.name_to_slot.remove(&table_name);
625                            let last = self.tables.len() - 1;
626                            if slot != last {
627                                let moved_name = self.tables[last].schema.table_name.clone();
628                                self.tables.swap(slot, last);
629                                self.name_to_slot.insert(moved_name, slot);
630                            }
631                            self.tables.pop();
632                        }
633                    }
634                }
635                WalRecordType::DdlAddColumn => {
636                    saw_ddl = true;
637                    if let Some((table_name, col)) = decode_ddl_alter_add_column(&rec.data) {
638                        if let Some(&slot) = self.name_to_slot.get(&table_name) {
639                            let tbl = &mut self.tables[slot];
640                            if !tbl.schema.columns.iter().any(|c| c.name == col.name) {
641                                let old_schema = tbl.schema.clone();
642                                let has_rows = tbl.heap.scan().next().is_some();
643                                tbl.schema.columns.push(col);
644                                tbl.refresh_layout();
645                                if has_rows {
646                                    let fill = vec![Value::Empty; tbl.schema.columns.len()];
647                                    let data_dir = self.data_dir.clone();
648                                    let _ = tbl.rewrite_rows_for_schema_change(
649                                        &old_schema,
650                                        &fill,
651                                        &data_dir,
652                                    );
653                                }
654                            }
655                            // Stamp every page with the DDL's LSN so a
656                            // subsequent restart's per-page check skips the
657                            // pre-DDL Insert/Update/Delete records — they
658                            // have already been folded into the new layout
659                            // by the rewrite above. See
660                            // `stamp_all_pages_min_lsn` doc.
661                            if rec.lsn > 0 {
662                                let _ = tbl.heap.stamp_all_pages_min_lsn(rec.lsn);
663                            }
664                        }
665                    }
666                }
667                WalRecordType::DdlDropColumn => {
668                    saw_ddl = true;
669                    if let Some((table_name, col_name)) = decode_ddl_alter_drop_column(&rec.data) {
670                        if let Some(&slot) = self.name_to_slot.get(&table_name) {
671                            let tbl = &mut self.tables[slot];
672                            if let Some(idx) =
673                                tbl.schema.columns.iter().position(|c| c.name == col_name)
674                            {
675                                let old_schema = tbl.schema.clone();
676                                let has_rows = tbl.heap.scan().next().is_some();
677                                tbl.schema.columns.remove(idx);
678                                for (i, c) in tbl.schema.columns.iter_mut().enumerate() {
679                                    c.position = i as u16;
680                                }
681                                tbl.refresh_layout();
682                                if has_rows {
683                                    let fill = vec![Value::Empty; tbl.schema.columns.len()];
684                                    let data_dir = self.data_dir.clone();
685                                    let _ = tbl.rewrite_rows_for_schema_change(
686                                        &old_schema,
687                                        &fill,
688                                        &data_dir,
689                                    );
690                                }
691                            }
692                            if rec.lsn > 0 {
693                                let _ = tbl.heap.stamp_all_pages_min_lsn(rec.lsn);
694                            }
695                        }
696                    }
697                }
698            }
699        }
700        info!(
701            inserts = replayed_inserts,
702            updates = replayed_updates,
703            deletes = replayed_deletes,
704            skipped = skipped,
705            skipped_uncommitted = skipped_uncommitted,
706            "WAL record apply complete (commit-boundary + LSN idempotent)"
707        );
708        if saw_ddl {
709            self.persist()?;
710        }
711        // Persist the replayed changes to disk before truncating the WAL,
712        // otherwise a crash between here and the next checkpoint would lose
713        // the replayed records. `flush_all_dirty` on every heap moves every
714        // dirty page through the normal write path.
715        //
716        // Blocker B3: under the deferred-index-save model, the on-disk
717        // `.idx` files may lag the heap because the pre-crash session
718        // never got to its next `checkpoint`. Replay restored the
719        // heap rows above, but the btrees that loaded from those
720        // possibly-stale `.idx` files don't know about them. Rebuild
721        // every secondary index from the post-replay heap so the
722        // trees exactly match disk. The rebuild is O(heap) per
723        // indexed column, which is fine on a crash-recovery path.
724        for tbl in &mut self.tables {
725            tbl.heap.flush_all_dirty()?;
726            tbl.heap.flush()?;
727            tbl.rebuild_indexes_from_heap()?;
728            // Flush the rebuilt indexes now so a crash between here
729            // and the next mutation still leaves `.idx` files matching
730            // the heap. Without this, a second crash before any
731            // insert could leave us back where we started.
732            tbl.save_dirty_indexes()?;
733        }
734        if let Some(max_lsn) = max_record_lsn(records) {
735            self.record_durable_lsn_at_least(max_lsn)?;
736            self.wal.set_next_lsn_at_least(max_lsn.saturating_add(1));
737        }
738        Ok(())
739    }
740
741    /// Flush every dirty heap page and truncate the WAL. This is the
742    /// "clean shutdown" point — after this returns, the on-disk heap files
743    /// are fully consistent and the WAL is empty, so the next `open` will
744    /// skip replay entirely.
745    ///
746    /// Safe to call multiple times. Safe to call on a catalog that has
747    /// performed zero mutations since the last checkpoint (in which case
748    /// the flushes are no-ops and the truncate is a bounded syscall).
749    pub fn checkpoint(&mut self) -> io::Result<()> {
750        self.ensure_no_active_transaction_for_checkpoint()?;
751        self.ensure_plain_checkpoint_allowed_before_flush()?;
752        self.flush_checkpoint_state()?;
753        self.wal.flush()?;
754        self.record_durable_lsn_at_least(self.wal.last_appended_lsn())?;
755        self.wal.truncate()?;
756        self.checkpointed = true;
757        Ok(())
758    }
759
760    /// Flush every dirty heap page, archive retained WAL records, then
761    /// truncate the WAL. Sync-aware callers use this to make archive-before-
762    /// truncate explicit without making storage depend on the sync crate.
763    ///
764    /// Replication boundary: this hook is for retained-history publication.
765    /// It should stay behind sync-aware lifecycle helpers rather than becoming
766    /// an ordinary checkpoint surface for application code.
767    pub fn checkpoint_with_wal_archive<F>(&mut self, mut archive: F) -> io::Result<()>
768    where
769        F: FnMut(&Path, &[WalRecord]) -> io::Result<()>,
770    {
771        self.ensure_no_active_transaction_for_checkpoint()?;
772        self.commit_autocommit()?;
773        self.flush_checkpoint_state()?;
774        self.wal.flush()?;
775        let records = self.wal.read_all()?;
776        let archive: WalArchiveCallback<'_> = &mut archive;
777        archive(&self.data_dir, &records)?;
778        if let Some(max_lsn) = max_record_lsn(&records) {
779            self.record_durable_lsn_at_least(max_lsn)?;
780        } else {
781            self.record_durable_lsn_at_least(self.wal.last_appended_lsn())?;
782        }
783        self.wal.truncate()?;
784        self.checkpointed = true;
785        Ok(())
786    }
787
788    fn ensure_no_active_transaction_for_checkpoint(&self) -> io::Result<()> {
789        if self.active_tx_id.is_some() {
790            return Err(io::Error::other(
791                "cannot checkpoint while an explicit transaction is active",
792            ));
793        }
794        Ok(())
795    }
796
797    fn flush_checkpoint_state(&mut self) -> io::Result<()> {
798        for tbl in &mut self.tables {
799            tbl.heap.flush_all_dirty()?;
800            tbl.heap.flush()?;
801            // Blocker B3: the hot insert/update/delete paths no longer
802            // fsync index files per row — they only mark the in-memory
803            // btree dirty. Checkpoint is where those deferred saves
804            // actually hit disk. Clean (non-dirty) indexes are free.
805            tbl.save_dirty_indexes()?;
806        }
807        Ok(())
808    }
809
810    fn ensure_plain_checkpoint_allowed_before_flush(&self) -> io::Result<()> {
811        if !self.sync_identity_file_exists() {
812            return Ok(());
813        }
814        if self.wal.has_pending() {
815            return Err(io::Error::other(
816                "sync identity exists but checkpoint/recovery was called without a WAL archive hook; refusing to truncate retained history",
817            ));
818        }
819        let records = self.wal.read_all()?;
820        self.ensure_plain_wal_truncate_allowed(&records)
821    }
822
823    fn ensure_plain_wal_truncate_allowed(&self, records: &[WalRecord]) -> io::Result<()> {
824        if records.is_empty() {
825            return Ok(());
826        }
827        if self.sync_identity_file_exists() {
828            return Err(io::Error::other(
829                "sync identity exists but checkpoint/recovery was called without a WAL archive hook; refusing to truncate retained history",
830            ));
831        }
832        Ok(())
833    }
834
835    fn sync_identity_file_exists(&self) -> bool {
836        self.data_dir
837            .join(SYNC_STATE_DIR)
838            .join(SYNC_IDENTITY_FILE)
839            .exists()
840    }
841
842    fn record_durable_lsn_at_least(&mut self, lsn: u64) -> io::Result<()> {
843        if lsn <= self.durable_lsn {
844            return Ok(());
845        }
846        self.durable_lsn = lsn;
847        write_durable_lsn(&self.data_dir, lsn)
848    }
849
850    /// Allocate or return the transaction id for the current mutation.
851    #[inline]
852    /// Free (or defer freeing) the overflow-chain pages a mutation just
853    /// orphaned. In autocommit there is no rollback window, so the pages return
854    /// to the table's free list immediately and the next spill reuses them
855    /// (bounding steady-state churn). Inside an explicit transaction the free is
856    /// held on `pending_free_overflow` until COMMIT: a ROLLBACK reopens the
857    /// catalog from disk, discarding this list, so the resurrected old row still
858    /// points at a live chain. Reuse is crash-safe without an `OverflowFree`
859    /// record because a later spill that overwrites a reused page logs its own
860    /// per-page `OverflowWrite` (LSN-idempotent), and post-recovery `sweep`
861    /// reclaims anything the in-memory list lost.
862    fn free_overflow_chain(&mut self, slot: usize, pages: Vec<u32>) {
863        if pages.is_empty() {
864            return;
865        }
866        if self.active_tx_id.is_some() {
867            self.pending_free_overflow.push((slot, pages));
868        } else {
869            self.tables[slot].release_overflow_pages(&pages);
870        }
871    }
872
873    fn next_tx(&mut self) -> u64 {
874        if let Some(id) = self.active_tx_id {
875            return id;
876        }
877        let id = self.next_tx_id;
878        self.next_tx_id = self.next_tx_id.wrapping_add(1);
879        id
880    }
881
882    /// Begin a connection/session-scoped explicit transaction.
883    pub fn begin_transaction(&mut self) -> io::Result<()> {
884        if self.active_tx_id.is_some() {
885            return Err(io::Error::new(
886                io::ErrorKind::InvalidInput,
887                "explicit transaction is already active",
888            ));
889        }
890        let start_len = self.wal.synced_len()?;
891        let id = self.next_tx_id;
892        self.next_tx_id = self.next_tx_id.wrapping_add(1);
893        self.active_tx_id = Some(id);
894        self.tx_start_len = Some(start_len);
895        self.pending_autocommit_tx_ids.clear();
896        if !self.wal.is_off() {
897            self.wal.append(id, WalRecordType::Begin, &[])?;
898            self.wal.flush()?;
899        }
900        Ok(())
901    }
902
903    /// Commit the active explicit transaction by appending a durable boundary
904    /// marker after its row records.
905    pub fn commit_transaction(&mut self) -> io::Result<()> {
906        if let Some(id) = self.active_tx_id.take() {
907            if !self.wal.is_off() {
908                self.wal.append(id, WalRecordType::Commit, &[])?;
909                self.wal.flush()?;
910            }
911        }
912        self.tx_start_len = None;
913        // The transaction committed: its rows are durable and can no longer be
914        // resurrected by ROLLBACK, so the old chains they replaced/removed are
915        // safe to reclaim. (Populated only while a tx was active.)
916        for (slot, pages) in std::mem::take(&mut self.pending_free_overflow) {
917            self.tables[slot].release_overflow_pages(&pages);
918        }
919        Ok(())
920    }
921
922    /// Commit any autocommit row mutations accumulated by the current
923    /// statement. Pure reads/DDL have no pending tx ids and fall through to a
924    /// cheap WAL flush/no-op.
925    pub fn commit_autocommit(&mut self) -> io::Result<()> {
926        if !self.wal.is_off() && !self.pending_autocommit_tx_ids.is_empty() {
927            self.pending_autocommit_tx_ids.sort_unstable();
928            self.pending_autocommit_tx_ids.dedup();
929            for id in self.pending_autocommit_tx_ids.drain(..) {
930                self.wal.append(id, WalRecordType::Commit, &[])?;
931            }
932        }
933        self.wal.flush()
934    }
935
936    /// Append a mutation record to the WAL buffer. **Does not flush.**
937    ///
938    /// Mission B (post-review): per-row `wal.flush()` was a ~1ms fsync on
939    /// every mutation, turning `update_by_filter` into a ~19s workload.
940    /// The flush is now deferred to [`Self::sync_wal`], which the executor
941    /// calls exactly once at the end of every mutating statement. This
942    /// gives us statement-level group commit: N-row updates pay one fsync,
943    /// not N.
944    ///
945    /// Durability contract: any path that observes `Ok(...)` back from
946    /// the executor must have called `sync_wal` before returning that
947    /// Ok. Replay is still correct because WAL records are appended in
948    /// order and only records that reached `fdatasync`ed bytes are
949    /// replayed.
950    fn wal_log(
951        &mut self,
952        tx_id: u64,
953        record_type: WalRecordType,
954        table: &str,
955        rid: RowId,
956        row_bytes: &[u8],
957    ) -> io::Result<()> {
958        // Mission B (post-review, second pass): when the WAL is in Off
959        // mode the `append` call below is a no-op, so building the
960        // payload first wastes a `Vec` allocation + ~3 extends per
961        // mutation. The catalog hot paths check `wal.is_off()` before
962        // calling here, but this guard is the belt-and-braces version
963        // for any internal caller that doesn't.
964        if self.wal.is_off() {
965            return Ok(());
966        }
967        let payload = encode_wal_payload(table, rid, row_bytes);
968        self.wal.append(tx_id, record_type, &payload)?;
969        if self.active_tx_id.is_none() {
970            self.pending_autocommit_tx_ids.push(tx_id);
971        }
972        Ok(())
973    }
974
975    /// Flush any buffered WAL records to disk. Called by the executor
976    /// at the end of every mutating statement so the group-commit
977    /// window is exactly one statement.
978    ///
979    /// See `Self::wal_log` for the durability contract.
980    #[inline]
981    pub fn sync_wal(&mut self) -> io::Result<()> {
982        self.wal.flush()
983    }
984
985    /// Set the WAL sync mode. Production code should leave this at the
986    /// default ([`WalSyncMode::Full`]). Benchmarks set it to
987    /// [`WalSyncMode::Off`] to compare apples-to-apples against
988    /// `:memory:` SQLite (which has zero fsync cost).
989    ///
990    /// **Never** call this with `Off` in production — a machine crash
991    /// can lose any record written since the last `sync_wal` returned.
992    pub fn set_wal_sync_mode(&mut self, mode: WalSyncMode) {
993        self.wal.set_sync_mode(mode);
994    }
995
996    /// Defer Full-mode commit fsyncs (WAL group commit). While enabled, the
997    /// commit paths register the WAL generation they need durable instead of
998    /// fsyncing inline; the pending claim is retrieved with
999    /// [`Self::take_wal_durability_ticket`] and the caller must wait on it
1000    /// before acknowledging the statement. This lets the fsync leave the
1001    /// engine's exclusive-lock hold so overlapping committers can share one
1002    /// fsync. `Normal`/`Off` modes are unaffected.
1003    pub fn set_wal_sync_deferred(&mut self, defer: bool) {
1004        self.wal.set_defer_sync(defer);
1005    }
1006
1007    /// Take the durability claim registered by deferred commit flushes since
1008    /// the last take, if any. See [`Self::set_wal_sync_deferred`].
1009    pub fn take_wal_durability_ticket(&mut self) -> Option<WalDurabilityTicket> {
1010        self.wal.take_durability_ticket()
1011    }
1012
1013    /// Number of fsyncs issued against the WAL (test/metrics hook).
1014    pub fn wal_fsync_count(&self) -> u64 {
1015        self.wal.fsync_count()
1016    }
1017
1018    /// Discard in-memory mutations made since the last `sync_wal()` and
1019    /// restore the catalog to its on-disk state. Used by ROLLBACK to
1020    /// undo an in-progress transaction's changes.
1021    ///
1022    /// This re-opens the catalog from the checkpoint file and replays
1023    /// only the durable (already flushed) WAL records. Any WAL records
1024    /// that were appended but not yet flushed are lost.
1025    ///
1026    /// **Critical**: before replacing `*self` we must discard every
1027    /// dirty in-memory page across all heaps. Otherwise the old
1028    /// `Catalog`'s `Drop` impl calls `checkpoint()` which flushes those
1029    /// dirty pages to disk — and the freshly-opened replacement catalog
1030    /// would then read the flushed (uncommitted) rows back, defeating
1031    /// the entire rollback.
1032    pub fn rollback_to_last_sync(&mut self) -> io::Result<()> {
1033        self.rollback_to_last_sync_inner(None)
1034    }
1035
1036    /// Roll back the active transaction, then reopen/replay any remaining WAL
1037    /// through an archive hook before recovery truncates it. Sync-aware callers
1038    /// use this when committed pre-transaction records must remain available to
1039    /// replicas after rollback.
1040    pub fn rollback_to_last_sync_with_wal_archive<F>(&mut self, mut archive: F) -> io::Result<()>
1041    where
1042        F: FnMut(&Path, &[WalRecord]) -> io::Result<()>,
1043    {
1044        let archive: WalArchiveCallback<'_> = &mut archive;
1045        self.rollback_to_last_sync_inner(Some(archive))
1046    }
1047
1048    fn rollback_to_last_sync_inner(
1049        &mut self,
1050        mut archive: Option<WalArchiveCallback<'_>>,
1051    ) -> io::Result<()> {
1052        let start_len = self.tx_start_len.unwrap_or(0);
1053        let prearchived = if let Some(archive) = archive.as_mut() {
1054            let records = self.wal.read_through_len(start_len)?;
1055            if !records.is_empty() {
1056                archive(&self.data_dir, &records)?;
1057            }
1058            true
1059        } else {
1060            false
1061        };
1062
1063        let start_len = self.tx_start_len.take().unwrap_or(0);
1064        if let Some(id) = self.active_tx_id.take() {
1065            if !self.wal.is_off() {
1066                let _ = self.wal.append(id, WalRecordType::Rollback, &[]);
1067            }
1068        }
1069        self.wal.discard_and_truncate_to(start_len)?;
1070
1071        // Step 1: throw away every uncommitted in-memory write so the
1072        // upcoming Drop of `*self` has nothing dirty to flush. This covers
1073        // both the heap pages AND the btree index mutations: the Drop below
1074        // runs `checkpoint()` (active_tx_id was already taken above), whose
1075        // `save_dirty_indexes` would otherwise flush the rolled-back index
1076        // writes to the `.idx` files — poisoning the unique index. The
1077        // freshly-opened replacement catalog reloads clean trees from the
1078        // untouched on-disk `.idx`, so discarding the dirty flags here is
1079        // what actually reverts the transaction's index writes.
1080        for tbl in &mut self.tables {
1081            tbl.heap.discard_dirty();
1082            tbl.discard_dirty_indexes();
1083        }
1084        // Step 2: discard WAL records appended since the last explicit
1085        // sync point. Large pending records can spill through BufWriter and
1086        // become file-visible before `sync_wal()`; truncating to the last
1087        // synced boundary prevents `open()` below from replaying rolled-back
1088        // transaction records.
1089        self.wal.discard_pending()?;
1090        // Step 3: re-open the catalog from disk. The heap files on disk
1091        // still reflect the last checkpoint (pre-transaction state)
1092        // because we never flushed the transaction's dirty pages.
1093        let data_dir = self.data_dir.clone();
1094        let sync_mode = self.wal.sync_mode();
1095        let restored = if prearchived {
1096            let mut already_archived = |_dir: &Path, _records: &[WalRecord]| Ok(());
1097            let archive: WalArchiveCallback<'_> = &mut already_archived;
1098            Self::open_inner(&data_dir, Some(archive))?
1099        } else {
1100            Self::open_inner(&data_dir, archive)?
1101        };
1102        *self = restored;
1103        self.wal.set_sync_mode(sync_mode);
1104        Ok(())
1105    }
1106
1107    fn abandon_active_transaction_for_drop(&mut self) -> io::Result<()> {
1108        for tbl in &mut self.tables {
1109            tbl.heap.discard_dirty();
1110        }
1111        self.pending_autocommit_tx_ids.clear();
1112        let truncate_result = match self.tx_start_len.take() {
1113            Some(start_len) => self.wal.discard_and_truncate_to(start_len),
1114            None => self.wal.discard_pending(),
1115        };
1116        self.active_tx_id = None;
1117        truncate_result
1118    }
1119
1120    /// Returns a reference to the data directory.
1121    pub fn data_dir(&self) -> &Path {
1122        &self.data_dir
1123    }
1124
1125    /// Highest page LSN across all tables (0 if nothing has been written).
1126    /// This is the durability high-water mark — the LSN a backup taken now
1127    /// corresponds to, and the value `Catalog::open` uses to restore
1128    /// `next_lsn` after a reopen/restore.
1129    pub fn max_lsn(&self) -> u64 {
1130        let max_page_lsn = self
1131            .tables
1132            .iter()
1133            .map(|t| t.heap.max_page_lsn())
1134            .max()
1135            .unwrap_or(0);
1136        max_page_lsn
1137            .max(self.durable_lsn)
1138            .max(self.wal.last_appended_lsn())
1139    }
1140
1141    pub fn create_table(&mut self, schema: Schema) -> io::Result<()> {
1142        self.create_table_full(schema, Vec::new(), Vec::new())
1143    }
1144
1145    /// Create a table whose columns carry literal defaults. `defaults` is
1146    /// aligned to `schema.columns` by position (and may be shorter / empty for
1147    /// columns without a default).
1148    pub fn create_table_with_defaults(
1149        &mut self,
1150        schema: Schema,
1151        defaults: Vec<Option<Value>>,
1152    ) -> io::Result<()> {
1153        self.create_table_full(schema, defaults, Vec::new())
1154    }
1155
1156    /// Create a table with per-column literal defaults and auto-increment
1157    /// flags. Both vecs are aligned to `schema.columns` by position (and may be
1158    /// empty). Defaults and auto flags are WAL-logged and persisted in the
1159    /// catalog so they survive a restart.
1160    pub fn create_table_full(
1161        &mut self,
1162        schema: Schema,
1163        defaults: Vec<Option<Value>>,
1164        auto_cols: Vec<bool>,
1165    ) -> io::Result<()> {
1166        validate_table_name(&schema.table_name)?;
1167        for col in &schema.columns {
1168            validate_column_name(&col.name)?;
1169        }
1170        let name = schema.table_name.clone();
1171        if self.name_to_slot.contains_key(&name) {
1172            return Err(io::Error::new(
1173                io::ErrorKind::AlreadyExists,
1174                format!("table '{name}' already exists"),
1175            ));
1176        }
1177        if !self.wal.is_off() {
1178            let payload = encode_ddl_create_table(&schema, &defaults, &auto_cols);
1179            self.wal
1180                .append(0, WalRecordType::DdlCreateTable, &payload)?;
1181            self.wal.flush()?;
1182        }
1183        let mut table = Table::create(schema, &self.data_dir)?;
1184        table.set_defaults(defaults);
1185        table.set_auto_cols(auto_cols);
1186        let slot = self.tables.len();
1187        self.tables.push(table);
1188        self.name_to_slot.insert(name, slot);
1189        self.persist()?;
1190        Ok(())
1191    }
1192
1193    /// Per-column literal defaults for a table, aligned to its columns by
1194    /// position. `None` when the table is unknown; an empty slice when no
1195    /// column has a default.
1196    pub fn column_defaults(&self, table: &str) -> Option<&[Option<Value>]> {
1197        let slot = *self.name_to_slot.get(table)?;
1198        Some(self.tables[slot].defaults())
1199    }
1200
1201    /// Which columns of a table are `auto`, aligned to its columns by position.
1202    /// `None` when the table is unknown; an empty slice when none are auto.
1203    pub fn auto_columns(&self, table: &str) -> Option<&[bool]> {
1204        let slot = *self.name_to_slot.get(table)?;
1205        Some(self.tables[slot].auto_cols())
1206    }
1207
1208    /// Fill any omitted (`Empty`) auto column in `values` from the table's
1209    /// sequence and advance it. No-op when the table is unknown or has no auto
1210    /// columns.
1211    pub fn assign_auto_columns(&mut self, table: &str, values: &mut [Value]) {
1212        if let Some(&slot) = self.name_to_slot.get(table) {
1213            self.tables[slot].assign_auto(values);
1214        }
1215    }
1216
1217    /// Write the current set of schemas to disk atomically (write-then-rename).
1218    ///
1219    /// Mission 3: also writes the per-table list of indexed column names so
1220    /// `Catalog::open` can rehydrate b-tree indexes on restart.
1221    fn persist(&self) -> io::Result<()> {
1222        let cat_path = self.data_dir.join(CATALOG_FILE);
1223        let tmp_path = self.data_dir.join(format!("{CATALOG_FILE}.tmp"));
1224        let entries: Vec<CatalogEntryRef<'_>> = self
1225            .tables
1226            .iter()
1227            .map(|t| CatalogEntryRef {
1228                schema: &t.schema,
1229                indexed_cols: t.indexed_column_metas(),
1230                defaults: t.defaults(),
1231                auto_cols: t.auto_cols(),
1232            })
1233            .collect();
1234        write_catalog_file(&tmp_path, &entries)?;
1235        fs::rename(&tmp_path, &cat_path)?;
1236        Ok(())
1237    }
1238
1239    /// Resolve a table name to its stable slot index. Prepared-query
1240    /// fast paths cache this once and skip the hash probe on every
1241    /// subsequent execution. Slots never shift once assigned.
1242    #[inline]
1243    pub fn table_slot(&self, name: &str) -> Option<usize> {
1244        self.name_to_slot.get(name).copied()
1245    }
1246
1247    /// O(1) slot-indexed table access. Panics on an out-of-range slot
1248    /// — callers must have obtained the slot via `table_slot()`.
1249    #[inline]
1250    pub fn table_by_slot(&self, slot: usize) -> &Table {
1251        &self.tables[slot]
1252    }
1253
1254    /// Mutable counterpart to [`Self::table_by_slot`].
1255    #[inline]
1256    pub fn table_by_slot_mut(&mut self, slot: usize) -> &mut Table {
1257        &mut self.tables[slot]
1258    }
1259
1260    pub fn get_table(&self, name: &str) -> Option<&Table> {
1261        let slot = *self.name_to_slot.get(name)?;
1262        Some(&self.tables[slot])
1263    }
1264
1265    pub fn get_table_mut(&mut self, name: &str) -> Option<&mut Table> {
1266        let slot = *self.name_to_slot.get(name)?;
1267        Some(&mut self.tables[slot])
1268    }
1269
1270    /// Whether `table` may hold v2 (spilled) rows (see
1271    /// [`Table::has_overflow_rows`]). Unknown table ⇒ false. The executor gates
1272    /// its v1-only raw-byte fast paths on this.
1273    #[inline]
1274    pub fn table_has_overflow(&self, table: &str) -> bool {
1275        self.get_table(table)
1276            .map(|t| t.has_overflow_rows())
1277            .unwrap_or(false)
1278    }
1279
1280    /// Private helper: resolve a table name to `&Table`, or return an
1281    /// `io::Error` with the same "table '<name>' not found" message the
1282    /// older `get_mut().ok_or_else(...)` callers produced. Phase 18
1283    /// consolidates ~14 copies of that idiom into this one place.
1284    #[inline]
1285    fn by_name(&self, table: &str) -> io::Result<&Table> {
1286        let slot = *self.name_to_slot.get(table).ok_or_else(|| {
1287            io::Error::new(
1288                io::ErrorKind::NotFound,
1289                format!("table '{table}' not found"),
1290            )
1291        })?;
1292        Ok(&self.tables[slot])
1293    }
1294
1295    /// Mutable counterpart to [`Self::by_name`].
1296    #[inline]
1297    fn by_name_mut(&mut self, table: &str) -> io::Result<&mut Table> {
1298        let slot = self.slot_of(table)?;
1299        Ok(&mut self.tables[slot])
1300    }
1301
1302    /// Mark-and-sweep one table's overflow pages, returning the number of pages
1303    /// reclaimed (design 3.6, door D12). Reclaimed pages are logged as a single
1304    /// `OverflowFree` record so the reclamation is crash-safe, then returned to
1305    /// the free list for reuse. Intended to run under the table write lock.
1306    pub fn sweep(&mut self, table: &str) -> io::Result<usize> {
1307        let slot = self.slot_of(table)?;
1308        let reclaimed = self.tables[slot].sweep_overflow()?;
1309        if !reclaimed.is_empty() && !self.wal.is_off() {
1310            let payload = encode_overflow_free_payload(table, &reclaimed);
1311            self.wal.append(0, WalRecordType::OverflowFree, &payload)?;
1312            self.wal.flush()?;
1313        }
1314        Ok(reclaimed.len())
1315    }
1316
1317    /// Sweep overflow pages across every table. Returns the total reclaimed.
1318    pub fn sweep_all(&mut self) -> io::Result<usize> {
1319        let names: Vec<String> = self
1320            .tables
1321            .iter()
1322            .map(|t| t.schema.table_name.clone())
1323            .collect();
1324        let mut total = 0;
1325        for name in names {
1326            total += self.sweep(&name)?;
1327        }
1328        Ok(total)
1329    }
1330
1331    fn slot_of(&self, table: &str) -> io::Result<usize> {
1332        self.name_to_slot.get(table).copied().ok_or_else(|| {
1333            io::Error::new(
1334                io::ErrorKind::NotFound,
1335                format!("table '{table}' not found"),
1336            )
1337        })
1338    }
1339
1340    pub fn insert(&mut self, table: &str, values: &Row) -> io::Result<RowId> {
1341        // Mission 2: encode the row into a scratch buffer first so we can
1342        // log it to the WAL before touching the heap. We re-encode inside
1343        // `Table::insert`, which keeps the insert hot path untouched — the
1344        // WAL encode here is additive.
1345        //
1346        // Mission B (post-review, second pass): in `WalSyncMode::Off` the
1347        // entire WAL pipeline is a no-op, so skip the per-row
1348        // `encode_row_into` allocation and `wal_log` call entirely.
1349        if self.wal.is_off() {
1350            return self.by_name_mut(table)?.insert(values);
1351        }
1352        // Allocate the tx id up front: any overflow chains for a spilled row
1353        // must be logged under the SAME tx (and before the Insert record) so
1354        // an uncommitted big row's chain writes are skipped on replay.
1355        let tx_id = self.next_tx();
1356        let slot = self.slot_of(table)?;
1357        let row_bytes = {
1358            let Catalog { tables, wal, .. } = self;
1359            encode_row_with_spill_logged(&mut tables[slot], wal, tx_id, values)?
1360        };
1361        // Insert the (v1 or v2) row bytes into the heap FIRST so the Insert
1362        // record carries the real RowId. Index maintenance uses the logical
1363        // `values`, so a spilled column is indexed by its full value, never
1364        // the stub. See the v0.4.x idempotency rationale in the git history.
1365        let new_rid = self.tables[slot].insert_encoded(values, &row_bytes)?;
1366        self.wal_log(tx_id, WalRecordType::Insert, table, new_rid, &row_bytes)?;
1367        let lsn = self.wal.last_appended_lsn();
1368        if lsn > 0 {
1369            self.tables[slot].heap.set_page_lsn(new_rid.page_id, lsn)?;
1370        }
1371        Ok(new_rid)
1372    }
1373
1374    /// WAL-logged insert addressed by table slot index instead of name.
1375    /// Backs the executor's prepared-insert fast path, which resolves the
1376    /// slot at prepare time to skip the name→slot hash probe. Behaves exactly
1377    /// like [`Self::insert`] (logs the record with the real RowId, stamps the
1378    /// landing page's LSN) — the prepared path previously called the raw
1379    /// `Table::insert` and bypassed the WAL entirely, silently losing every
1380    /// prepared insert on a crash.
1381    pub fn insert_by_slot(&mut self, slot: usize, values: &Row) -> io::Result<RowId> {
1382        if self.wal.is_off() {
1383            return self.tables[slot].insert(values);
1384        }
1385        let tx_id = self.next_tx();
1386        let autocommit = self.active_tx_id.is_none();
1387        let Catalog { tables, wal, .. } = self;
1388        let tbl = &mut tables[slot];
1389        // Spill-aware encode (logs any overflow chains under `tx_id`, before
1390        // the Insert record). Returns v1 bytes for rows that fit inline.
1391        let row_bytes = encode_row_with_spill_logged(tbl, wal, tx_id, values)?;
1392        // Insert first so the WAL record carries the real RowId (see
1393        // `insert` for the ordering/durability argument).
1394        let new_rid = tbl.insert_encoded(values, &row_bytes)?;
1395        let payload = encode_wal_payload(&tbl.schema.table_name, new_rid, &row_bytes);
1396        wal.append(tx_id, WalRecordType::Insert, &payload)?;
1397        if autocommit {
1398            self.pending_autocommit_tx_ids.push(tx_id);
1399        }
1400        let lsn = wal.last_appended_lsn();
1401        if lsn > 0 {
1402            tbl.heap.set_page_lsn(new_rid.page_id, lsn)?;
1403        }
1404        Ok(new_rid)
1405    }
1406
1407    pub fn get(&self, table: &str, rid: RowId) -> Option<Row> {
1408        self.get_table(table)?.get(rid)
1409    }
1410
1411    pub fn delete(&mut self, table: &str, rid: RowId) -> io::Result<()> {
1412        let slot = self.slot_of(table)?;
1413        // Capture the deleted row's overflow chain BEFORE the heap slot is
1414        // cleared, so it can be freed once safe (design 3.6). Empty for
1415        // inline-only tables (cheap `has_overflow_rows` check).
1416        let old_pages = self.tables[slot].overflow_chain_pages_at(rid)?;
1417        // Mission B (post-review, second pass): WAL Off → no payload
1418        // construction.
1419        if self.wal.is_off() {
1420            self.tables[slot].delete(rid)?;
1421            self.free_overflow_chain(slot, old_pages);
1422            return Ok(());
1423        }
1424        let tx_id = self.next_tx();
1425        // Delete records carry only the rid — no row payload.
1426        self.wal_log(tx_id, WalRecordType::Delete, table, rid, &[])?;
1427        self.tables[slot].delete(rid)?;
1428        self.free_overflow_chain(slot, old_pages);
1429        Ok(())
1430    }
1431
1432    /// Mission C Phase 12: bulk delete a list of rids, batching btree
1433    /// maintenance. See [`Table::delete_many`] for the full explanation
1434    /// and fall-through rules. Returns the number of rows removed.
1435    pub fn delete_many(&mut self, table: &str, rids: &[RowId]) -> io::Result<u64> {
1436        // Mission 2: log every rid as an individual Delete record. The
1437        // WAL flush is deferred to the executor's statement-end
1438        // `sync_wal` — see [`Self::wal_log`] for the group-commit rules.
1439        //
1440        // Mission B (post-review, second pass): in Off mode skip the
1441        // entire per-row payload loop — `wal.append` would no-op every
1442        // call but the `encode_wal_payload` Vec alloc would still run.
1443        let slot = self.slot_of(table)?;
1444        // Gather every deleted row's overflow chain up front (empty and cheap
1445        // for inline-only tables) so the pages can be freed once safe.
1446        let old_pages = self.collect_overflow_pages(slot, rids)?;
1447        if self.wal.is_off() {
1448            let count = self.tables[slot].delete_many(rids)?;
1449            self.free_overflow_chain(slot, old_pages);
1450            return Ok(count);
1451        }
1452        let tx_id = self.next_tx();
1453        for &rid in rids {
1454            let payload = encode_wal_payload(table, rid, &[]);
1455            self.wal.append(tx_id, WalRecordType::Delete, &payload)?;
1456        }
1457        if self.active_tx_id.is_none() && !rids.is_empty() {
1458            self.pending_autocommit_tx_ids.push(tx_id);
1459        }
1460        let count = self.tables[slot].delete_many(rids)?;
1461        self.free_overflow_chain(slot, old_pages);
1462        Ok(count)
1463    }
1464
1465    /// Collect all overflow-chain pages referenced by `rids` in one table.
1466    /// Returns empty for inline-only tables without touching any row.
1467    fn collect_overflow_pages(&self, slot: usize, rids: &[RowId]) -> io::Result<Vec<u32>> {
1468        if !self.tables[slot].has_overflow_rows() {
1469            return Ok(Vec::new());
1470        }
1471        let mut pages = Vec::new();
1472        for &rid in rids {
1473            pages.extend(self.tables[slot].overflow_chain_pages_at(rid)?);
1474        }
1475        Ok(pages)
1476    }
1477
1478    /// Single-pass scan-and-delete driven by a raw-bytes predicate. See
1479    /// [`Table::scan_delete_matching`] and `HeapFile::scan_delete_matching`
1480    /// for the fusion rationale.
1481    ///
1482    /// Prefer [`Self::scan_delete_matching_logged`] from any
1483    /// caller that needs crash durability. This variant writes no WAL
1484    /// records, so a crash between the scan and the next checkpoint
1485    /// would lose the deletes. Kept here for internal paths (e.g.
1486    /// `drop_table`) where the whole heap is about to be removed anyway.
1487    pub fn scan_delete_matching<P>(&mut self, table: &str, pred: P) -> io::Result<u64>
1488    where
1489        P: FnMut(&[u8]) -> bool,
1490    {
1491        self.by_name_mut(table)?.scan_delete_matching(pred)
1492    }
1493
1494    /// WAL-logged variant of [`Self::scan_delete_matching`].
1495    /// Every matched row emits one `WalRecordType::Delete` record in the
1496    /// same single-pass scan (via the table's `_with_hook` variant), so
1497    /// crash recovery sees every deletion. Used by the executor's
1498    /// `Delete(Filter(SeqScan))` and bare `Delete(SeqScan)` fast paths.
1499    ///
1500    /// Performance cost vs the non-logged primitive is one per-row WAL
1501    /// append into the in-memory buffer plus one `fsync` at the end —
1502    /// the heap scan itself still runs as a single pass with one
1503    /// `ensure_hot` per page.
1504    pub fn scan_delete_matching_logged<P>(&mut self, table: &str, pred: P) -> io::Result<u64>
1505    where
1506        P: FnMut(&[u8]) -> bool,
1507    {
1508        // Mission B (post-review, second pass): in Off mode the per-row
1509        // hook would build a Vec, do five extends, and then `append`
1510        // would no-op. Skip the WAL hook entirely and route through
1511        // the no-WAL primitive — same single-pass scan, zero per-row
1512        // payload work.
1513        if self.wal.is_off() {
1514            return self.by_name_mut(table)?.scan_delete_matching(pred);
1515        }
1516        // Resolve slot up front so we can split the borrow — the user
1517        // hook closes over `&mut self.wal`, which can't coexist with a
1518        // `by_name_mut` borrow of `self.tables`.
1519        let slot = *self.name_to_slot.get(table).ok_or_else(|| {
1520            io::Error::new(
1521                io::ErrorKind::NotFound,
1522                format!("table '{table}' not found"),
1523            )
1524        })?;
1525        let tx_id = self.next_tx();
1526        let autocommit = self.active_tx_id.is_none();
1527        // Split-borrow the catalog fields so the hook can write into
1528        // `wal` while the scan pins `tables[slot]` mutably.
1529        let Catalog { tables, wal, .. } = self;
1530        let tbl = &mut tables[slot];
1531        // Pre-encode the table-name prefix of every WAL payload once —
1532        // it doesn't vary row-to-row, and the per-row rid+row bytes are
1533        // the only things we append inside the hook.
1534        let name_bytes = table.as_bytes();
1535        let count = tbl.scan_delete_matching_with_hook(pred, |rid, row_bytes| {
1536            let mut payload: Vec<u8> =
1537                Vec::with_capacity(4 + name_bytes.len() + 10 + row_bytes.len());
1538            payload.extend_from_slice(&(name_bytes.len() as u32).to_le_bytes());
1539            payload.extend_from_slice(name_bytes);
1540            payload.extend_from_slice(&rid.page_id.to_le_bytes());
1541            payload.extend_from_slice(&rid.slot_index.to_le_bytes());
1542            // Delete records carry no row payload on replay, but we
1543            // match the `encode_wal_payload` layout so `decode_wal_payload`
1544            // (which is type-agnostic) parses them cleanly.
1545            payload.extend_from_slice(&0u32.to_le_bytes());
1546            // Best-effort append — if it errors we have no way to
1547            // propagate from inside the hook; we swallow it here and
1548            // the outer scan's `io::Result` will still succeed. In
1549            // practice the `BufWriter`-backed `Wal::append` only errors
1550            // on allocation failure or a disk-full fsync, both of
1551            // which would fail the outer flush below as well.
1552            let _ = wal.append(tx_id, WalRecordType::Delete, &payload);
1553        })?;
1554        if autocommit && count > 0 {
1555            self.pending_autocommit_tx_ids.push(tx_id);
1556        }
1557        // Flush is deferred to the executor's statement-end `sync_wal`.
1558        Ok(count)
1559    }
1560
1561    /// Single-pass fused scan + in-place patch with WAL logging.
1562    /// Evaluates `pred` on raw row bytes and applies `try_mutate` to each
1563    /// match on the same hot page — no second pass. Returns
1564    /// `(patched_count, fallback_rids)`.
1565    ///
1566    /// Perf sprint: update analogue of `scan_delete_matching_logged`.
1567    /// Eliminates the two-pass collect-then-patch pattern.
1568    pub fn scan_patch_matching_logged<P, M>(
1569        &mut self,
1570        table: &str,
1571        pred: P,
1572        try_mutate: M,
1573    ) -> io::Result<(u64, Vec<RowId>)>
1574    where
1575        P: FnMut(&[u8]) -> bool,
1576        M: FnMut(&mut [u8]) -> Option<u16>,
1577    {
1578        if self.wal.is_off() {
1579            return self.by_name_mut(table)?.scan_patch_matching_with_hook(
1580                pred,
1581                try_mutate,
1582                |_, _| {},
1583            );
1584        }
1585        let slot = *self.name_to_slot.get(table).ok_or_else(|| {
1586            io::Error::new(
1587                io::ErrorKind::NotFound,
1588                format!("table '{table}' not found"),
1589            )
1590        })?;
1591        let tx_id = self.next_tx();
1592        let autocommit = self.active_tx_id.is_none();
1593        let Catalog { tables, wal, .. } = self;
1594        let tbl = &mut tables[slot];
1595        let name_bytes = table.as_bytes();
1596        let result = tbl.scan_patch_matching_with_hook(pred, try_mutate, |rid, row_bytes| {
1597            let mut payload: Vec<u8> =
1598                Vec::with_capacity(4 + name_bytes.len() + 10 + row_bytes.len());
1599            payload.extend_from_slice(&(name_bytes.len() as u32).to_le_bytes());
1600            payload.extend_from_slice(name_bytes);
1601            payload.extend_from_slice(&rid.page_id.to_le_bytes());
1602            payload.extend_from_slice(&rid.slot_index.to_le_bytes());
1603            payload.extend_from_slice(&(row_bytes.len() as u32).to_le_bytes());
1604            payload.extend_from_slice(row_bytes);
1605            let _ = wal.append(tx_id, WalRecordType::Update, &payload);
1606        })?;
1607        if autocommit && result.0 > 0 {
1608            self.pending_autocommit_tx_ids.push(tx_id);
1609        }
1610        Ok(result)
1611    }
1612
1613    pub fn update(&mut self, table: &str, rid: RowId, values: &Row) -> io::Result<RowId> {
1614        // Mission B (post-review, second pass): WAL Off → no payload
1615        // construction.
1616        if self.wal.is_off() {
1617            let slot = self.slot_of(table)?;
1618            let old_pages = self.tables[slot].overflow_chain_pages_at(rid)?;
1619            let new_rid = self.tables[slot].update(rid, values)?;
1620            self.free_overflow_chain(slot, old_pages);
1621            return Ok(new_rid);
1622        }
1623        let tx_id = self.next_tx();
1624        let slot = self.slot_of(table)?;
1625        // Capture the old row's overflow chain (empty for inline-only tables)
1626        // BEFORE the update replaces it, so it can be freed once safe (design
1627        // 3.6). A chain-replacing update always orphans the old chain.
1628        let old_pages = self.tables[slot].overflow_chain_pages_at(rid)?;
1629        // Spill-aware encode: logs any overflow chains under `tx_id` (before
1630        // the Update record) and returns the v1/v2 row bytes. An overflow
1631        // transition relocates the row via heap delete+insert inside
1632        // `update_encoded`; the old row's chain (if any) is left for `sweep`.
1633        let row_bytes = {
1634            let Catalog { tables, wal, .. } = self;
1635            encode_row_with_spill_logged(&mut tables[slot], wal, tx_id, values)?
1636        };
1637        // Reject oversized rows BEFORE appending the Update record: a logged
1638        // Update the heap then rejects would poison the next replay. (A v2
1639        // stub row is always small; only a non-spilled v1 row can trip this.)
1640        check_encoded_row_size(&row_bytes)?;
1641        self.wal_log(tx_id, WalRecordType::Update, table, rid, &row_bytes)?;
1642        let new_rid = self.tables[slot].update_encoded(rid, values, &row_bytes, None)?;
1643        self.free_overflow_chain(slot, old_pages);
1644        Ok(new_rid)
1645    }
1646
1647    /// Mission C Phase 2: update with a hint about which columns actually
1648    /// changed. Lets [`Table::update_hinted`] skip the old-row read when
1649    /// the hint shows no indexed column is in the changed set.
1650    pub fn update_hinted(
1651        &mut self,
1652        table: &str,
1653        rid: RowId,
1654        values: &Row,
1655        changed_col_indices: Option<&[usize]>,
1656    ) -> io::Result<RowId> {
1657        // Mission B (post-review, second pass): WAL Off → no payload
1658        // construction. The `update_by_filter` powql bench drives this
1659        // path tens of thousands of times per iteration.
1660        if self.wal.is_off() {
1661            let slot = self.slot_of(table)?;
1662            let old_pages = self.tables[slot].overflow_chain_pages_at(rid)?;
1663            let new_rid = self.tables[slot].update_hinted(rid, values, changed_col_indices)?;
1664            self.free_overflow_chain(slot, old_pages);
1665            return Ok(new_rid);
1666        }
1667        let tx_id = self.next_tx();
1668        let slot = self.slot_of(table)?;
1669        let old_pages = self.tables[slot].overflow_chain_pages_at(rid)?;
1670        let row_bytes = {
1671            let Catalog { tables, wal, .. } = self;
1672            encode_row_with_spill_logged(&mut tables[slot], wal, tx_id, values)?
1673        };
1674        // Same pre-WAL size gate as [`Self::update`].
1675        check_encoded_row_size(&row_bytes)?;
1676        self.wal_log(tx_id, WalRecordType::Update, table, rid, &row_bytes)?;
1677        let new_rid =
1678            self.tables[slot].update_encoded(rid, values, &row_bytes, changed_col_indices)?;
1679        self.free_overflow_chain(slot, old_pages);
1680        Ok(new_rid)
1681    }
1682
1683    /// Mission C Phase 4: fast-path update that patches a row's raw bytes
1684    /// in place, skipping decode/encode. Caller guarantees the mutation
1685    /// preserves the row length and touches no indexed column. Returns
1686    /// `Ok(true)` if the patch landed, `Ok(false)` if the row is gone.
1687    ///
1688    /// This primitive does NOT log to the WAL. Executor
1689    /// callers must route through [`Self::update_row_bytes_logged`] (or
1690    /// [`Self::update_row_bytes_logged_by_slot`]) so crash recovery
1691    /// sees the patched bytes. This raw form is retained for replay
1692    /// itself and any future callers that can tolerate the non-durable
1693    /// contract.
1694    #[inline]
1695    pub fn with_row_bytes_mut<F>(&mut self, table: &str, rid: RowId, f: F) -> io::Result<bool>
1696    where
1697        F: FnOnce(&mut [u8]),
1698    {
1699        self.by_name_mut(table)?.with_row_bytes_mut(rid, f)
1700    }
1701
1702    /// WAL-logged variant of [`Self::with_row_bytes_mut`].
1703    /// Applies `f` to the live row bytes on the hot page, then reads
1704    /// the mutated bytes back and emits a `WalRecordType::Update`
1705    /// record so replay will re-apply the same patch after a crash.
1706    ///
1707    /// Ordering: the hot-page mutation happens first (in-memory only,
1708    /// no disk I/O), then the WAL record is appended and flushed. A
1709    /// crash after the mutation but before the WAL flush loses the
1710    /// update, but the caller never saw success in that case, so the
1711    /// contract holds: any `Ok(true)` return is durable.
1712    ///
1713    /// No hot-page eviction can happen between steps because this
1714    /// method holds the catalog's `&mut self` exclusively.
1715    #[inline]
1716    pub fn update_row_bytes_logged<F>(&mut self, table: &str, rid: RowId, f: F) -> io::Result<bool>
1717    where
1718        F: FnOnce(&mut [u8]),
1719    {
1720        let slot = *self.name_to_slot.get(table).ok_or_else(|| {
1721            io::Error::new(
1722                io::ErrorKind::NotFound,
1723                format!("table '{table}' not found"),
1724            )
1725        })?;
1726        self.update_row_bytes_logged_by_slot(slot, rid, f)
1727    }
1728
1729    /// Slot-indexed counterpart to [`Self::update_row_bytes_logged`].
1730    /// Used by prepared-query fast paths that already cached the table
1731    /// slot at prepare time and want to skip the name->slot probe on
1732    /// every execution.
1733    #[inline]
1734    pub fn update_row_bytes_logged_by_slot<F>(
1735        &mut self,
1736        slot: usize,
1737        rid: RowId,
1738        f: F,
1739    ) -> io::Result<bool>
1740    where
1741        F: FnOnce(&mut [u8]),
1742    {
1743        // Step 1: apply the mutation on the hot page. Failure here
1744        // (slot gone) short-circuits with Ok(false) — no WAL record.
1745        let tbl = &mut self.tables[slot];
1746        let ok = tbl.with_row_bytes_mut(rid, f)?;
1747        if !ok {
1748            return Ok(false);
1749        }
1750        // Mission B (post-review, second pass): in Off mode the per-row
1751        // get + clone + table-name clone + wal_log call are all wasted
1752        // — `wal.append` would no-op. Skip the snapshot path entirely.
1753        if self.wal.is_off() {
1754            return Ok(true);
1755        }
1756        // Step 2: snapshot the now-mutated bytes. `HeapFile::get`
1757        // observes the pinned hot page, so it returns the fresh row.
1758        let new_bytes = match tbl.heap.get(rid) {
1759            Some(b) => b,
1760            // Shouldn't happen — we just patched it — but be defensive.
1761            None => return Ok(false),
1762        };
1763        // Step 3: log + flush. Clone the table name out of the schema
1764        // so we can drop the `&mut tbl` borrow before touching `self.wal`.
1765        let table_name = tbl.schema.table_name.clone();
1766        let tx_id = self.next_tx();
1767        self.wal_log(tx_id, WalRecordType::Update, &table_name, rid, &new_bytes)?;
1768        Ok(true)
1769    }
1770
1771    /// Mission C Phase 10: var-column in-place update fast path. Patches
1772    /// a single variable-length column's bytes directly into the row's
1773    /// slot, shrinking the row if the new value is smaller. Returns
1774    /// `Ok(false)` if the new value would grow the row (caller must fall
1775    /// back to the full encode path) or the row is gone.
1776    ///
1777    /// Caller guarantees no indexed column is touched — indexes are NOT
1778    /// maintained by this primitive.
1779    ///
1780    /// Not WAL-logged. Executor callers should use
1781    /// [`Self::patch_var_col_logged`] instead.
1782    #[inline]
1783    pub fn patch_var_col_in_place(
1784        &mut self,
1785        table: &str,
1786        rid: RowId,
1787        col_idx: usize,
1788        new_value: Option<&[u8]>,
1789    ) -> io::Result<bool> {
1790        self.by_name_mut(table)?
1791            .patch_var_col_in_place(rid, col_idx, new_value)
1792    }
1793
1794    /// WAL-logged variant of [`Self::patch_var_col_in_place`].
1795    /// Runs the in-place shrink on the hot page, then reads the mutated
1796    /// row bytes back and logs a `WalRecordType::Update` record. On a
1797    /// `false` return (grow-case bail) nothing is logged — the caller's
1798    /// fall-through to `update_hinted` handles the WAL itself.
1799    pub fn patch_var_col_logged(
1800        &mut self,
1801        table: &str,
1802        rid: RowId,
1803        col_idx: usize,
1804        new_value: Option<&[u8]>,
1805    ) -> io::Result<bool> {
1806        let slot = *self.name_to_slot.get(table).ok_or_else(|| {
1807            io::Error::new(
1808                io::ErrorKind::NotFound,
1809                format!("table '{table}' not found"),
1810            )
1811        })?;
1812        let tbl = &mut self.tables[slot];
1813        let ok = tbl.patch_var_col_in_place(rid, col_idx, new_value)?;
1814        if !ok {
1815            return Ok(false);
1816        }
1817        // Mission B (post-review, second pass): WAL Off → skip the
1818        // snapshot + clone + log entirely.
1819        if self.wal.is_off() {
1820            return Ok(true);
1821        }
1822        let new_bytes = match tbl.heap.get(rid) {
1823            Some(b) => b,
1824            None => return Ok(false),
1825        };
1826        let table_name = tbl.schema.table_name.clone();
1827        let tx_id = self.next_tx();
1828        self.wal_log(tx_id, WalRecordType::Update, &table_name, rid, &new_bytes)?;
1829        Ok(true)
1830    }
1831
1832    pub fn scan(&self, table: &str) -> io::Result<impl Iterator<Item = (RowId, Row)> + '_> {
1833        Ok(self.by_name(table)?.scan())
1834    }
1835
1836    /// Zero-copy scan: passes raw row bytes to the callback without any
1837    /// per-row allocation. Used by the executor's fast paths.
1838    pub fn for_each_row_raw<F>(&self, table: &str, f: F) -> io::Result<()>
1839    where
1840        F: FnMut(RowId, &[u8]),
1841    {
1842        self.by_name(table)?.for_each_row_raw(f);
1843        Ok(())
1844    }
1845
1846    /// Zero-copy scan with early termination. The callback returns
1847    /// `ControlFlow::Break(())` to stop. Used by `Limit` fast paths so a
1848    /// `limit 100` query doesn't pay decode/predicate cost for every row
1849    /// in the table after the limit is reached.
1850    pub fn try_for_each_row_raw<F>(&self, table: &str, f: F) -> io::Result<()>
1851    where
1852        F: FnMut(RowId, &[u8]) -> std::ops::ControlFlow<()>,
1853    {
1854        self.by_name(table)?.try_for_each_row_raw(f);
1855        Ok(())
1856    }
1857
1858    pub fn create_index(&mut self, table: &str, column: &str) -> io::Result<()> {
1859        self.create_index_unique(table, column, false)
1860    }
1861
1862    /// Create an index with an explicit uniqueness flag. `unique = true`
1863    /// for primary-key-like columns where duplicate values should
1864    /// overwrite. `unique = false` for secondary indexes that allow
1865    /// duplicate column values (the default via `create_index`).
1866    pub fn create_index_unique(
1867        &mut self,
1868        table: &str,
1869        column: &str,
1870        unique: bool,
1871    ) -> io::Result<()> {
1872        let data_dir = self.data_dir.clone();
1873        self.by_name_mut(table)?
1874            .create_index_with_unique(column, &data_dir, unique)?;
1875        // Mission 3: persist the updated catalog so the indexed column
1876        // list survives a restart. `Table::create_index` already saved
1877        // the btree file itself.
1878        self.persist()
1879    }
1880
1881    /// Whether `table.column` has a UNIQUE index. Returns `Some(true)` for
1882    /// a unique index, `Some(false)` for a non-unique index, and `None`
1883    /// when the column is not indexed or the table is unknown.
1884    pub fn is_index_unique(&self, table: &str, column: &str) -> Option<bool> {
1885        self.get_table(table)?.is_index_unique(column)
1886    }
1887
1888    /// Whether `table.column` has any index (unique or non-unique).
1889    pub fn has_index(&self, table: &str, column: &str) -> bool {
1890        self.get_table(table)
1891            .map(|t| t.has_index(column))
1892            .unwrap_or(false)
1893    }
1894
1895    pub fn index_lookup(&self, table: &str, column: &str, key: &Value) -> io::Result<Option<Row>> {
1896        Ok(self
1897            .by_name(table)?
1898            .index_lookup(column, key)
1899            .map(|(_, row)| row))
1900    }
1901
1902    pub fn list_tables(&self) -> Vec<&str> {
1903        // Phase 18: iterate the Vec directly — schema.table_name is
1904        // the source of truth, and Vec order is insertion order (more
1905        // deterministic than the old FxHashMap keys).
1906        self.tables
1907            .iter()
1908            .map(|t| t.schema.table_name.as_str())
1909            .collect()
1910    }
1911
1912    pub fn schema(&self, table: &str) -> Option<&Schema> {
1913        let slot = *self.name_to_slot.get(table)?;
1914        Some(&self.tables[slot].schema)
1915    }
1916
1917    /// Drop a table: remove from the catalog and delete its data files.
1918    /// Returns `Err` if the table doesn't exist.
1919    pub fn drop_table(&mut self, name: &str) -> io::Result<()> {
1920        validate_table_name(name)?;
1921        let slot = *self.name_to_slot.get(name).ok_or_else(|| {
1922            io::Error::new(io::ErrorKind::NotFound, format!("table '{name}' not found"))
1923        })?;
1924        if !self.wal.is_off() {
1925            let payload = encode_ddl_drop_table(name);
1926            self.wal.append(0, WalRecordType::DdlDropTable, &payload)?;
1927            self.wal.flush()?;
1928        }
1929        // Remove the data file.
1930        let table = &self.tables[slot];
1931        let heap_path = self
1932            .data_dir
1933            .join(format!("{}.heap", table.schema.table_name));
1934        if heap_path.exists() {
1935            fs::remove_file(&heap_path)?;
1936        }
1937        // Mission 3: remove only the .idx files that actually exist
1938        // (i.e. the columns the table currently has indexed). The pre-
1939        // Mission-3 code iterated every schema column blindly — harmless
1940        // but noisy. Now that we persist a real list of indexed columns,
1941        // we can be precise.
1942        for col_name in table.indexed_column_names() {
1943            let idx_path = self.data_dir.join(format!("{name}_{col_name}.idx"));
1944            if idx_path.exists() {
1945                let _ = fs::remove_file(&idx_path);
1946            }
1947        }
1948        // Swap-remove from the Vec and fix up name_to_slot.
1949        self.name_to_slot.remove(name);
1950        let last = self.tables.len() - 1;
1951        if slot != last {
1952            let moved_name = self.tables[last].schema.table_name.clone();
1953            self.tables.swap(slot, last);
1954            self.name_to_slot.insert(moved_name, slot);
1955        }
1956        self.tables.pop();
1957        self.persist()?;
1958        Ok(())
1959    }
1960
1961    /// Add a column to an existing table's schema and backfill all
1962    /// existing rows to match the new shape.
1963    ///
1964    /// Older versions of this method only mutated the in-memory schema
1965    /// and relied on a (false) claim that "the heap format already
1966    /// handles short rows gracefully". It doesn't: `decode_row` reads
1967    /// exactly `n_var + 1` variable-column offsets from the row bytes
1968    /// using the CURRENT schema. Any row encoded with the old schema's
1969    /// (smaller) offset table would walk off the end of its buffer and
1970    /// panic with "range end index X out of range for slice of length Y"
1971    /// — which is exactly what a bare `Type` scan triggered right after
1972    /// an ALTER ADD COLUMN.
1973    ///
1974    /// The fix: rewrite every existing row through
1975    /// `Table::rewrite_rows_for_schema_change` so the on-disk
1976    /// encoding matches the new schema layout. Existing rows get
1977    /// `Value::Empty` for the new column.
1978    ///
1979    /// If the new column is `required` we refuse to add it to a
1980    /// non-empty table — there is no default value to backfill with,
1981    /// and silently storing `Empty` in a required slot would just
1982    /// shift the invariant violation to the next query.
1983    pub fn alter_table_add_column(&mut self, table: &str, col: ColumnDef) -> io::Result<()> {
1984        let data_dir = self.data_dir.clone();
1985        {
1986            let tbl = self.by_name_mut(table)?;
1987            if tbl.schema.columns.iter().any(|c| c.name == col.name) {
1988                return Err(io::Error::new(
1989                    io::ErrorKind::AlreadyExists,
1990                    format!("column '{}' already exists in table '{table}'", col.name),
1991                ));
1992            }
1993        }
1994        let barrier_lsn = if !self.wal.is_off() {
1995            let payload = encode_ddl_alter_add_column(table, &col);
1996            self.wal.append(0, WalRecordType::DdlAddColumn, &payload)?;
1997            self.wal.flush()?;
1998            self.wal.last_appended_lsn()
1999        } else {
2000            0
2001        };
2002        let tbl = self.by_name_mut(table)?;
2003
2004        let old_schema = tbl.schema.clone();
2005
2006        // Peek at the heap to learn whether there are any existing
2007        // rows at all. An empty table is always safe to alter — no
2008        // rewrite needed, required columns are fine, etc.
2009        let has_rows = tbl.heap.scan().next().is_some();
2010
2011        if has_rows && col.required {
2012            return Err(io::Error::new(
2013                io::ErrorKind::InvalidInput,
2014                format!(
2015                    "cannot add required column '{}' to non-empty table '{table}': \
2016                     no default value to backfill existing rows with",
2017                    col.name
2018                ),
2019            ));
2020        }
2021
2022        // Commit the new column into the schema and refresh the
2023        // cached layout so the rewrite below encodes with the new
2024        // shape.
2025        tbl.schema.columns.push(col);
2026        tbl.refresh_layout();
2027
2028        if has_rows {
2029            // Build the "fill" template: all Empty, matching the new
2030            // schema width. `rewrite_rows_for_schema_change` will
2031            // overwrite old-column slots from each live row and leave
2032            // the new slot as Empty.
2033            let fill: Vec<Value> = vec![Value::Empty; tbl.schema.columns.len()];
2034            tbl.rewrite_rows_for_schema_change(&old_schema, &fill, &data_dir)?;
2035        }
2036        // P0 fix (v0.4.3): stamp every heap page with the DDL record's
2037        // LSN so any pre-DDL Insert/Update/Delete WAL record gets
2038        // skipped on replay. Without this barrier, a restart after
2039        // `alter add column` would replay pre-alter inserts (encoded in
2040        // the OLD layout) onto a heap that's already in the NEW layout,
2041        // producing a mixed-version heap that panics on the next
2042        // projection. Regression: see `restart_after_alter_add_column_then_index`.
2043        if barrier_lsn > 0 {
2044            tbl.heap.stamp_all_pages_min_lsn(barrier_lsn)?;
2045            tbl.heap.flush()?;
2046        }
2047
2048        self.persist()?;
2049        Ok(())
2050    }
2051
2052    /// Remove a column from an existing table's schema and rewrite
2053    /// every live row to match the new shape.
2054    ///
2055    /// Older versions of this method only mutated the in-memory schema
2056    /// and claimed that "reads simply won't decode the dropped column".
2057    /// That was wrong in several ways:
2058    ///
2059    ///   1. The null bitmap is indexed by column position. Dropping a
2060    ///      column shifts every later column's bit left, but old rows
2061    ///      still have bits in the original positions — so `is_null`
2062    ///      checks silently lie for every column after the dropped one.
2063    ///   2. The bitmap's byte width (`ceil(n_cols/8)`) can shrink when
2064    ///      `n_cols` crosses an 8-boundary, shifting every subsequent
2065    ///      byte of the row against the decoder's cursor.
2066    ///   3. Fixed-region size and the variable-offset-table width both
2067    ///      depend on the column set, so dropping any fixed or variable
2068    ///      column slides every following byte.
2069    ///
2070    /// The fix mirrors `alter_table_add_column`: snapshot the old
2071    /// schema, mutate to the new schema, then rewrite every row
2072    /// through `Table::rewrite_rows_for_schema_change`. Dropping a
2073    /// column from an empty table skips the rewrite.
2074    pub fn alter_table_drop_column(&mut self, table: &str, col_name: &str) -> io::Result<()> {
2075        let data_dir = self.data_dir.clone();
2076        {
2077            let tbl = self.by_name_mut(table)?;
2078            tbl.schema
2079                .columns
2080                .iter()
2081                .position(|c| c.name == col_name)
2082                .ok_or_else(|| {
2083                    io::Error::new(
2084                        io::ErrorKind::NotFound,
2085                        format!("column '{col_name}' not found in table '{table}'"),
2086                    )
2087                })?;
2088        }
2089        let barrier_lsn = if !self.wal.is_off() {
2090            let payload = encode_ddl_alter_drop_column(table, col_name);
2091            self.wal.append(0, WalRecordType::DdlDropColumn, &payload)?;
2092            self.wal.flush()?;
2093            self.wal.last_appended_lsn()
2094        } else {
2095            0
2096        };
2097        let tbl = self.by_name_mut(table)?;
2098        let idx = tbl
2099            .schema
2100            .columns
2101            .iter()
2102            .position(|c| c.name == col_name)
2103            .ok_or_else(|| {
2104                io::Error::new(
2105                    io::ErrorKind::NotFound,
2106                    format!("column '{col_name}' not found in table '{table}'"),
2107                )
2108            })?;
2109
2110        // Snapshot for decoding old rows.
2111        let old_schema = tbl.schema.clone();
2112        let has_rows = tbl.heap.scan().next().is_some();
2113
2114        // Commit the schema change.
2115        tbl.schema.columns.remove(idx);
2116        for (i, col) in tbl.schema.columns.iter_mut().enumerate() {
2117            col.position = i as u16;
2118        }
2119        tbl.refresh_layout();
2120
2121        if has_rows {
2122            // Build a filler matching the new (smaller) shape. The
2123            // rewrite path overwrites each new-column slot from the
2124            // matching old-column value by name, so the filler only
2125            // matters for brand-new columns — drop has none, so
2126            // `Empty` is a safe placeholder that never gets read.
2127            let fill: Vec<Value> = vec![Value::Empty; tbl.schema.columns.len()];
2128            tbl.rewrite_rows_for_schema_change(&old_schema, &fill, &data_dir)?;
2129        }
2130        // P0 fix: see matching comment in alter_table_add_column.
2131        if barrier_lsn > 0 {
2132            tbl.heap.stamp_all_pages_min_lsn(barrier_lsn)?;
2133            tbl.heap.flush()?;
2134        }
2135
2136        self.persist()?;
2137        Ok(())
2138    }
2139}
2140
2141impl Drop for Catalog {
2142    fn drop(&mut self) {
2143        if self.active_tx_id.is_some() {
2144            if let Err(e) = self.abandon_active_transaction_for_drop() {
2145                warn!(error = %e, "catalog drop active transaction cleanup failed");
2146            }
2147            return;
2148        }
2149        // Mission 2: best-effort clean shutdown. `checkpoint` flushes
2150        // every heap and truncates the WAL, which is what
2151        // [`Catalog::open`] relies on to know that no replay is needed.
2152        //
2153        // We swallow errors here because Rust's `Drop` can't propagate
2154        // them and panicking during unwind is always a bigger problem
2155        // than a failed flush. The worst case on a failed drop-time
2156        // checkpoint is that the next open sees a non-empty WAL and
2157        // replays it (potentially producing duplicates — see the
2158        // [`Self::replay_wal`] caveat). That's strictly better than
2159        // losing committed writes.
2160        if let Err(e) = self.checkpoint() {
2161            warn!(error = %e, "catalog drop checkpoint failed");
2162        }
2163    }
2164}
2165
2166// ─── WAL payload codec ─────────────────────────────────────────────────────
2167//
2168// Per-record payload layout (little-endian):
2169//
2170//   table_name_len : u32
2171//   table_name     : utf-8 bytes
2172//   page_id        : u32   (for insert: 0, ignored on replay)
2173//   slot_index     : u16   (for insert: 0, ignored on replay)
2174//   row_len        : u32
2175//   row_bytes      : raw encoded row (length = row_len)
2176//
2177// Lives next to `Catalog` because this is the only code that produces or
2178// consumes these records — the `Wal` itself is payload-agnostic.
2179
2180fn encode_wal_payload(table: &str, rid: RowId, row_bytes: &[u8]) -> Vec<u8> {
2181    let name = table.as_bytes();
2182    let mut out = Vec::with_capacity(4 + name.len() + 4 + 2 + 4 + row_bytes.len());
2183    out.extend_from_slice(&(name.len() as u32).to_le_bytes());
2184    out.extend_from_slice(name);
2185    out.extend_from_slice(&rid.page_id.to_le_bytes());
2186    out.extend_from_slice(&rid.slot_index.to_le_bytes());
2187    out.extend_from_slice(&(row_bytes.len() as u32).to_le_bytes());
2188    out.extend_from_slice(row_bytes);
2189    out
2190}
2191
2192fn decode_wal_payload(data: &[u8]) -> Option<(String, RowId, Vec<u8>)> {
2193    let mut pos = 0usize;
2194    if data.len() < 4 {
2195        return None;
2196    }
2197    let name_len = u32::from_le_bytes(data[pos..pos + 4].try_into().ok()?) as usize;
2198    pos += 4;
2199    if pos + name_len > data.len() {
2200        return None;
2201    }
2202    let name = std::str::from_utf8(&data[pos..pos + name_len])
2203        .ok()?
2204        .to_string();
2205    pos += name_len;
2206    if pos + 4 + 2 + 4 > data.len() {
2207        return None;
2208    }
2209    let page_id = u32::from_le_bytes(data[pos..pos + 4].try_into().ok()?);
2210    pos += 4;
2211    let slot_index = u16::from_le_bytes(data[pos..pos + 2].try_into().ok()?);
2212    pos += 2;
2213    let row_len = u32::from_le_bytes(data[pos..pos + 4].try_into().ok()?) as usize;
2214    pos += 4;
2215    if pos + row_len > data.len() {
2216        return None;
2217    }
2218    let row_bytes = data[pos..pos + row_len].to_vec();
2219    Some((
2220        name,
2221        RowId {
2222            page_id,
2223            slot_index,
2224        },
2225        row_bytes,
2226    ))
2227}
2228
2229/// Write one out-of-line value's overflow chain to the heap (head-first,
2230/// singly linked) and log each chunk as a `WalRecordType::OverflowWrite`
2231/// record under `tx_id`, ordered BEFORE the row's Insert/Update record so the
2232/// stub the row carries always points at logged, replayable pages. Returns the
2233/// stub (u64 length, head page, whole-value CRC32). Enforces `MAX_VALUE_SIZE`.
2234fn write_overflow_chain_logged(
2235    heap: &mut HeapFile,
2236    wal: &mut Wal,
2237    table: &str,
2238    tx_id: u64,
2239    value: &[u8],
2240) -> io::Result<OverflowStub> {
2241    if value.len() > MAX_VALUE_SIZE {
2242        return Err(StorageError::ValueTooLarge {
2243            size: value.len(),
2244            max: MAX_VALUE_SIZE,
2245        }
2246        .into());
2247    }
2248    let n = value.len().div_ceil(OVERFLOW_PAYLOAD_CAP).max(1);
2249    let mut pages = Vec::with_capacity(n);
2250    for _ in 0..n {
2251        pages.push(heap.allocate_overflow_page()?);
2252    }
2253    for i in 0..n {
2254        let start = i * OVERFLOW_PAYLOAD_CAP;
2255        let end = (start + OVERFLOW_PAYLOAD_CAP).min(value.len());
2256        let chunk = &value[start..end];
2257        let next = if i + 1 < n {
2258            pages[i + 1]
2259        } else {
2260            OVERFLOW_CHAIN_END
2261        };
2262        let payload = encode_overflow_write_payload(table, pages[i], next, chunk);
2263        wal.append(tx_id, WalRecordType::OverflowWrite, &payload)?;
2264        let lsn = wal.last_appended_lsn();
2265        heap.write_overflow_page(pages[i], next, chunk, lsn)?;
2266    }
2267    Ok(OverflowStub::new(
2268        value.len() as u64,
2269        pages[0],
2270        crc32fast::hash(value),
2271    ))
2272}
2273
2274/// Spill-aware encode for the WAL path. If the row fits inline, returns its v1
2275/// bytes untouched. Otherwise writes each spilled value's chain (with WAL
2276/// logging under `tx_id`) and returns the v2 stub-row bytes to be inserted and
2277/// logged in the row's Insert/Update record.
2278fn encode_row_with_spill_logged(
2279    tbl: &mut Table,
2280    wal: &mut Wal,
2281    tx_id: u64,
2282    values: &Row,
2283) -> io::Result<Vec<u8>> {
2284    // Size the v1 encoding WITHOUT encoding it (a >64KB value would panic the
2285    // debug-mode v1 encoder). Only actually encode v1 when the row fits inline.
2286    let v1_len = crate::row::v1_encoded_len(tbl.row_layout(), values);
2287    let is_indexed = tbl.indexed_col_mask();
2288    let chosen = plan_spill(tbl.row_layout(), values, v1_len, &is_indexed);
2289    if chosen.is_empty() {
2290        let mut v1 = Vec::new();
2291        encode_row_into(&tbl.schema, values, &mut v1);
2292        return Ok(v1);
2293    }
2294    let table_name = tbl.schema.table_name.clone();
2295    let n_var = tbl.row_layout().n_var();
2296    let mut spilled: Vec<Option<OverflowStub>> = vec![None; n_var];
2297    for col_idx in chosen {
2298        let var_idx = tbl
2299            .row_layout()
2300            .var_index(col_idx)
2301            .expect("plan_spill only returns var columns");
2302        let bytes: Vec<u8> = match &values[col_idx] {
2303            Value::Str(s) => s.as_bytes().to_vec(),
2304            Value::Bytes(b) => b.clone(),
2305            _ => continue,
2306        };
2307        let stub = write_overflow_chain_logged(&mut tbl.heap, wal, &table_name, tx_id, &bytes)?;
2308        spilled[var_idx] = Some(stub);
2309    }
2310    let mut out = Vec::new();
2311    encode_row_v2_into(&tbl.schema, tbl.row_layout(), values, &spilled, &mut out);
2312    Ok(out)
2313}
2314
2315/// `OverflowWrite` payload: `table_len u16 | table | page_id u32 |
2316/// next_page u32 | chunk_len u16 | chunk bytes`.
2317///
2318/// NOTE (deviation from design 3.5): the design lists the payload as
2319/// `page_id | next_page | chunk_len | chunk`, but overflow pages live in
2320/// per-table heap files with independent page-id spaces, so replay needs the
2321/// table identity to route the write. The table name is length-prefixed
2322/// exactly like [`encode_wal_payload`]. The chunk-level fields are unchanged.
2323fn encode_overflow_write_payload(
2324    table: &str,
2325    page_id: u32,
2326    next_page: u32,
2327    chunk: &[u8],
2328) -> Vec<u8> {
2329    let name = table.as_bytes();
2330    let mut out = Vec::with_capacity(2 + name.len() + 4 + 4 + 2 + chunk.len());
2331    out.extend_from_slice(&(name.len() as u16).to_le_bytes());
2332    out.extend_from_slice(name);
2333    out.extend_from_slice(&page_id.to_le_bytes());
2334    out.extend_from_slice(&next_page.to_le_bytes());
2335    out.extend_from_slice(&(chunk.len() as u16).to_le_bytes());
2336    out.extend_from_slice(chunk);
2337    out
2338}
2339
2340fn decode_overflow_write_payload(data: &[u8]) -> Option<(String, u32, u32, Vec<u8>)> {
2341    let mut pos = 0usize;
2342    if data.len() < 2 {
2343        return None;
2344    }
2345    let name_len = u16::from_le_bytes(data[pos..pos + 2].try_into().ok()?) as usize;
2346    pos += 2;
2347    if pos + name_len + 4 + 4 + 2 > data.len() {
2348        return None;
2349    }
2350    let name = std::str::from_utf8(&data[pos..pos + name_len])
2351        .ok()?
2352        .to_string();
2353    pos += name_len;
2354    let page_id = u32::from_le_bytes(data[pos..pos + 4].try_into().ok()?);
2355    pos += 4;
2356    let next_page = u32::from_le_bytes(data[pos..pos + 4].try_into().ok()?);
2357    pos += 4;
2358    let chunk_len = u16::from_le_bytes(data[pos..pos + 2].try_into().ok()?) as usize;
2359    pos += 2;
2360    if pos + chunk_len > data.len() {
2361        return None;
2362    }
2363    Some((
2364        name,
2365        page_id,
2366        next_page,
2367        data[pos..pos + chunk_len].to_vec(),
2368    ))
2369}
2370
2371/// `OverflowFree` payload: `table_len u16 | table | count u32 |
2372/// page_id u32 x count`. Table name added for the same routing reason as
2373/// [`encode_overflow_write_payload`].
2374fn encode_overflow_free_payload(table: &str, pages: &[u32]) -> Vec<u8> {
2375    let name = table.as_bytes();
2376    let mut out = Vec::with_capacity(2 + name.len() + 4 + pages.len() * 4);
2377    out.extend_from_slice(&(name.len() as u16).to_le_bytes());
2378    out.extend_from_slice(name);
2379    out.extend_from_slice(&(pages.len() as u32).to_le_bytes());
2380    for p in pages {
2381        out.extend_from_slice(&p.to_le_bytes());
2382    }
2383    out
2384}
2385
2386fn decode_overflow_free_payload(data: &[u8]) -> Option<(String, Vec<u32>)> {
2387    let mut pos = 0usize;
2388    if data.len() < 2 {
2389        return None;
2390    }
2391    let name_len = u16::from_le_bytes(data[pos..pos + 2].try_into().ok()?) as usize;
2392    pos += 2;
2393    if pos + name_len + 4 > data.len() {
2394        return None;
2395    }
2396    let name = std::str::from_utf8(&data[pos..pos + name_len])
2397        .ok()?
2398        .to_string();
2399    pos += name_len;
2400    let count = u32::from_le_bytes(data[pos..pos + 4].try_into().ok()?) as usize;
2401    pos += 4;
2402    if pos + count * 4 > data.len() {
2403        return None;
2404    }
2405    let mut pages = Vec::with_capacity(count);
2406    for _ in 0..count {
2407        pages.push(u32::from_le_bytes(data[pos..pos + 4].try_into().ok()?));
2408        pos += 4;
2409    }
2410    Some((name, pages))
2411}
2412
2413// ─── DDL WAL payload codecs ─────────────────────────────────────────────────
2414
2415fn encode_ddl_create_table(
2416    schema: &Schema,
2417    defaults: &[Option<Value>],
2418    auto_cols: &[bool],
2419) -> Vec<u8> {
2420    let name = schema.table_name.as_bytes();
2421    let mut out = Vec::new();
2422    out.extend_from_slice(&(name.len() as u32).to_le_bytes());
2423    out.extend_from_slice(name);
2424    out.extend_from_slice(&(schema.columns.len() as u16).to_le_bytes());
2425    for col in &schema.columns {
2426        let cn = col.name.as_bytes();
2427        out.extend_from_slice(&(cn.len() as u32).to_le_bytes());
2428        out.extend_from_slice(cn);
2429        out.push(col.type_id as u8);
2430        out.push(col.required as u8);
2431        out.extend_from_slice(&col.position.to_le_bytes());
2432    }
2433    // Trailing sections. Records written before each feature existed simply
2434    // lack the corresponding trailing bytes, so the decoder treats their
2435    // absence as "none" (length-detected, append-only).
2436    encode_defaults_section(&mut out, defaults);
2437    encode_auto_section(&mut out, auto_cols);
2438    out
2439}
2440
2441fn decode_ddl_create_table(data: &[u8]) -> Option<(Schema, Vec<Option<Value>>, Vec<bool>)> {
2442    let mut pos = 0usize;
2443    if data.len() < 4 {
2444        return None;
2445    }
2446    let name_len = u32::from_le_bytes(data[pos..pos + 4].try_into().ok()?) as usize;
2447    pos += 4;
2448    if pos + name_len > data.len() {
2449        return None;
2450    }
2451    let table_name = std::str::from_utf8(&data[pos..pos + name_len])
2452        .ok()?
2453        .to_string();
2454    pos += name_len;
2455    if pos + 2 > data.len() {
2456        return None;
2457    }
2458    let n_cols = u16::from_le_bytes(data[pos..pos + 2].try_into().ok()?) as usize;
2459    pos += 2;
2460    let mut columns = Vec::with_capacity(n_cols);
2461    for _ in 0..n_cols {
2462        if pos + 4 > data.len() {
2463            return None;
2464        }
2465        let cn_len = u32::from_le_bytes(data[pos..pos + 4].try_into().ok()?) as usize;
2466        pos += 4;
2467        if pos + cn_len + 4 > data.len() {
2468            return None;
2469        }
2470        let col_name = std::str::from_utf8(&data[pos..pos + cn_len])
2471            .ok()?
2472            .to_string();
2473        pos += cn_len;
2474        let type_id = TypeId::from_u8(data[pos])?;
2475        pos += 1;
2476        let required = data[pos] != 0;
2477        pos += 1;
2478        if pos + 2 > data.len() {
2479            return None;
2480        }
2481        let position = u16::from_le_bytes(data[pos..pos + 2].try_into().ok()?);
2482        pos += 2;
2483        columns.push(ColumnDef {
2484            name: col_name,
2485            type_id,
2486            required,
2487            position,
2488        });
2489    }
2490    // Trailing sections are present on records written after each feature
2491    // landed; older records end early, decoding to "none".
2492    let defaults = if pos < data.len() {
2493        decode_defaults_section(data, &mut pos, columns.len())?
2494    } else {
2495        Vec::new()
2496    };
2497    let auto_cols = if pos < data.len() {
2498        decode_auto_section(data, &mut pos, columns.len())?
2499    } else {
2500        Vec::new()
2501    };
2502    Some((
2503        Schema {
2504            table_name,
2505            columns,
2506        },
2507        defaults,
2508        auto_cols,
2509    ))
2510}
2511
2512fn encode_ddl_drop_table(table_name: &str) -> Vec<u8> {
2513    let name = table_name.as_bytes();
2514    let mut out = Vec::with_capacity(4 + name.len());
2515    out.extend_from_slice(&(name.len() as u32).to_le_bytes());
2516    out.extend_from_slice(name);
2517    out
2518}
2519
2520fn encode_ddl_alter_add_column(table_name: &str, col: &ColumnDef) -> Vec<u8> {
2521    let name = table_name.as_bytes();
2522    let cn = col.name.as_bytes();
2523    let mut out = Vec::with_capacity(4 + name.len() + 4 + cn.len() + 4);
2524    out.extend_from_slice(&(name.len() as u32).to_le_bytes());
2525    out.extend_from_slice(name);
2526    out.extend_from_slice(&(cn.len() as u32).to_le_bytes());
2527    out.extend_from_slice(cn);
2528    out.push(col.type_id as u8);
2529    out.push(col.required as u8);
2530    out.extend_from_slice(&col.position.to_le_bytes());
2531    out
2532}
2533
2534fn encode_ddl_alter_drop_column(table_name: &str, col_name: &str) -> Vec<u8> {
2535    let name = table_name.as_bytes();
2536    let cn = col_name.as_bytes();
2537    let mut out = Vec::with_capacity(4 + name.len() + 4 + cn.len());
2538    out.extend_from_slice(&(name.len() as u32).to_le_bytes());
2539    out.extend_from_slice(name);
2540    out.extend_from_slice(&(cn.len() as u32).to_le_bytes());
2541    out.extend_from_slice(cn);
2542    out
2543}
2544
2545fn decode_ddl_table_name(data: &[u8]) -> Option<(String, usize)> {
2546    if data.len() < 4 {
2547        return None;
2548    }
2549    let name_len = u32::from_le_bytes(data[0..4].try_into().ok()?) as usize;
2550    if 4 + name_len > data.len() {
2551        return None;
2552    }
2553    let name = std::str::from_utf8(&data[4..4 + name_len])
2554        .ok()?
2555        .to_string();
2556    Some((name, 4 + name_len))
2557}
2558
2559fn decode_ddl_alter_add_column(data: &[u8]) -> Option<(String, ColumnDef)> {
2560    let (table_name, mut pos) = decode_ddl_table_name(data)?;
2561    if pos + 4 > data.len() {
2562        return None;
2563    }
2564    let cn_len = u32::from_le_bytes(data[pos..pos + 4].try_into().ok()?) as usize;
2565    pos += 4;
2566    if pos + cn_len + 4 > data.len() {
2567        return None;
2568    }
2569    let col_name = std::str::from_utf8(&data[pos..pos + cn_len])
2570        .ok()?
2571        .to_string();
2572    pos += cn_len;
2573    let type_id = TypeId::from_u8(data[pos])?;
2574    pos += 1;
2575    let required = data[pos] != 0;
2576    pos += 1;
2577    if pos + 2 > data.len() {
2578        return None;
2579    }
2580    let position = u16::from_le_bytes(data[pos..pos + 2].try_into().ok()?);
2581    Some((
2582        table_name,
2583        ColumnDef {
2584            name: col_name,
2585            type_id,
2586            required,
2587            position,
2588        },
2589    ))
2590}
2591
2592fn decode_ddl_alter_drop_column(data: &[u8]) -> Option<(String, String)> {
2593    let (table_name, pos) = decode_ddl_table_name(data)?;
2594    if pos + 4 > data.len() {
2595        return None;
2596    }
2597    let cn_len = u32::from_le_bytes(data[pos..pos + 4].try_into().ok()?) as usize;
2598    if pos + 4 + cn_len > data.len() {
2599        return None;
2600    }
2601    let col_name = std::str::from_utf8(&data[pos + 4..pos + 4 + cn_len])
2602        .ok()?
2603        .to_string();
2604    Some((table_name, col_name))
2605}
2606
2607// ─── Catalog file format ────────────────────────────────────────────────────
2608//
2609// Layout (version 2):
2610//   magic     [4]      = "BCAT"
2611//   version   u16
2612//   n_tables  u32
2613//   for each table:
2614//     table_name_len  u32
2615//     table_name      utf8 bytes
2616//     n_columns       u16
2617//     for each column:
2618//       name_len      u32
2619//       name          utf8 bytes
2620//       type_id       u8
2621//       required      u8
2622//       position      u16
2623//     ── version 2 appends: ──
2624//     n_indexed_cols  u16
2625//     for each indexed column:
2626//       name_len      u32
2627//       name          utf8 bytes
2628//
2629// Version 1 files are accepted by the reader (same shape minus the
2630// trailing indexed-column block) and treated as having zero indexed
2631// columns. Writers always emit version 2 from Mission 3 onwards.
2632
2633/// Per-indexed-column metadata persisted in the catalog file.
2634pub(crate) struct IndexedColMeta {
2635    pub name: String,
2636    pub unique: bool,
2637}
2638
2639/// In-memory catalog entry pairing a schema with its indexed column list.
2640/// Produced by the reader; the writer takes the borrowed counterpart below.
2641pub(crate) struct CatalogEntry {
2642    pub schema: Schema,
2643    pub indexed_cols: Vec<IndexedColMeta>,
2644    /// Per-column defaults aligned to `schema.columns` by position. Empty when
2645    /// no column has a default (v1–v3 files always decode to empty).
2646    pub defaults: Vec<Option<Value>>,
2647    /// Which columns are `auto`, aligned to `schema.columns`. Empty when none
2648    /// (v1–v4 files always decode to empty).
2649    pub auto_cols: Vec<bool>,
2650}
2651
2652/// Borrowed view passed to the writer.
2653pub(crate) struct CatalogEntryRef<'a> {
2654    pub schema: &'a Schema,
2655    pub indexed_cols: Vec<IndexedColMeta>,
2656    pub defaults: &'a [Option<Value>],
2657    pub auto_cols: &'a [bool],
2658}
2659
2660// ─── Column-default codecs (shared by catalog.bin and the WAL DDL record) ────
2661
2662/// Encode a single scalar value: a `type_id` tag byte followed by a
2663/// type-specific, length-prefixed (for variable-width types) payload. Lossless
2664/// — used to persist literal column defaults.
2665fn encode_value_blob(out: &mut Vec<u8>, v: &Value) {
2666    out.push(v.type_id() as u8);
2667    match v {
2668        Value::Int(n) => out.extend_from_slice(&n.to_le_bytes()),
2669        Value::Float(f) => out.extend_from_slice(&f.to_bits().to_le_bytes()),
2670        Value::Bool(b) => out.push(*b as u8),
2671        Value::Str(s) => {
2672            out.extend_from_slice(&(s.len() as u32).to_le_bytes());
2673            out.extend_from_slice(s.as_bytes());
2674        }
2675        Value::DateTime(n) => out.extend_from_slice(&n.to_le_bytes()),
2676        Value::Uuid(u) => out.extend_from_slice(u),
2677        Value::Bytes(b) => {
2678            out.extend_from_slice(&(b.len() as u32).to_le_bytes());
2679            out.extend_from_slice(b);
2680        }
2681        Value::Empty => {}
2682    }
2683}
2684
2685/// Inverse of [`encode_value_blob`]. Returns `None` on any malformed/truncated
2686/// input so a corrupt record fails closed rather than panicking.
2687fn decode_value_blob(data: &[u8], pos: &mut usize) -> Option<Value> {
2688    let tag = *data.get(*pos)?;
2689    *pos += 1;
2690    let type_id = TypeId::from_u8(tag)?;
2691    let take_fixed = |pos: &mut usize, n: usize| -> Option<Vec<u8>> {
2692        if *pos + n > data.len() {
2693            return None;
2694        }
2695        let slice = data[*pos..*pos + n].to_vec();
2696        *pos += n;
2697        Some(slice)
2698    };
2699    match type_id {
2700        TypeId::Empty => Some(Value::Empty),
2701        TypeId::Int => Some(Value::Int(i64::from_le_bytes(
2702            take_fixed(pos, 8)?.try_into().ok()?,
2703        ))),
2704        TypeId::Float => Some(Value::Float(f64::from_bits(u64::from_le_bytes(
2705            take_fixed(pos, 8)?.try_into().ok()?,
2706        )))),
2707        TypeId::Bool => Some(Value::Bool(take_fixed(pos, 1)?[0] != 0)),
2708        TypeId::DateTime => Some(Value::DateTime(i64::from_le_bytes(
2709            take_fixed(pos, 8)?.try_into().ok()?,
2710        ))),
2711        TypeId::Uuid => Some(Value::Uuid(take_fixed(pos, 16)?.try_into().ok()?)),
2712        TypeId::Str => {
2713            let len = u32::from_le_bytes(take_fixed(pos, 4)?.try_into().ok()?) as usize;
2714            Some(Value::Str(String::from_utf8(take_fixed(pos, len)?).ok()?))
2715        }
2716        TypeId::Bytes => {
2717            let len = u32::from_le_bytes(take_fixed(pos, 4)?.try_into().ok()?) as usize;
2718            Some(Value::Bytes(take_fixed(pos, len)?))
2719        }
2720    }
2721}
2722
2723/// Encode the per-table defaults as a sparse list: a `u16` count of columns
2724/// that have a default, then `(position: u16, value blob)` pairs. The common
2725/// "no defaults" case costs two bytes.
2726fn encode_defaults_section(out: &mut Vec<u8>, defaults: &[Option<Value>]) {
2727    let present: Vec<(u16, &Value)> = defaults
2728        .iter()
2729        .enumerate()
2730        .filter_map(|(i, d)| d.as_ref().map(|v| (i as u16, v)))
2731        .collect();
2732    out.extend_from_slice(&(present.len() as u16).to_le_bytes());
2733    for (pos, v) in present {
2734        out.extend_from_slice(&pos.to_le_bytes());
2735        encode_value_blob(out, v);
2736    }
2737}
2738
2739/// Inverse of [`encode_defaults_section`]. Builds a `Vec` of length `n_cols`
2740/// with `None` for columns without a default. Returns `None` on truncation.
2741fn decode_defaults_section(
2742    data: &[u8],
2743    pos: &mut usize,
2744    n_cols: usize,
2745) -> Option<Vec<Option<Value>>> {
2746    if *pos + 2 > data.len() {
2747        return None;
2748    }
2749    let count = u16::from_le_bytes(data[*pos..*pos + 2].try_into().ok()?) as usize;
2750    *pos += 2;
2751    let mut out = vec![None; n_cols];
2752    for _ in 0..count {
2753        if *pos + 2 > data.len() {
2754            return None;
2755        }
2756        let col = u16::from_le_bytes(data[*pos..*pos + 2].try_into().ok()?) as usize;
2757        *pos += 2;
2758        let value = decode_value_blob(data, pos)?;
2759        if col < n_cols {
2760            out[col] = Some(value);
2761        }
2762    }
2763    Some(out)
2764}
2765
2766/// Encode the per-table `auto` columns as a sparse list: a `u16` count of auto
2767/// columns, then their positions (`u16` each). "No auto columns" costs two
2768/// bytes.
2769fn encode_auto_section(out: &mut Vec<u8>, auto_cols: &[bool]) {
2770    let present: Vec<u16> = auto_cols
2771        .iter()
2772        .enumerate()
2773        .filter_map(|(i, &a)| if a { Some(i as u16) } else { None })
2774        .collect();
2775    out.extend_from_slice(&(present.len() as u16).to_le_bytes());
2776    for pos in present {
2777        out.extend_from_slice(&pos.to_le_bytes());
2778    }
2779}
2780
2781/// Inverse of [`encode_auto_section`]. Builds a `bool` vec of length `n_cols`.
2782/// Returns `None` on truncation.
2783fn decode_auto_section(data: &[u8], pos: &mut usize, n_cols: usize) -> Option<Vec<bool>> {
2784    if *pos + 2 > data.len() {
2785        return None;
2786    }
2787    let count = u16::from_le_bytes(data[*pos..*pos + 2].try_into().ok()?) as usize;
2788    *pos += 2;
2789    let mut out = vec![false; n_cols];
2790    for _ in 0..count {
2791        if *pos + 2 > data.len() {
2792            return None;
2793        }
2794        let col = u16::from_le_bytes(data[*pos..*pos + 2].try_into().ok()?) as usize;
2795        *pos += 2;
2796        if col < n_cols {
2797            out[col] = true;
2798        }
2799    }
2800    Some(out)
2801}
2802
2803fn write_catalog_file(path: &Path, entries: &[CatalogEntryRef<'_>]) -> io::Result<()> {
2804    let mut buf: Vec<u8> = Vec::with_capacity(64);
2805    buf.extend_from_slice(CATALOG_MAGIC);
2806    buf.extend_from_slice(&CATALOG_VERSION.to_le_bytes());
2807    buf.extend_from_slice(&(entries.len() as u32).to_le_bytes());
2808
2809    for entry in entries {
2810        let schema = entry.schema;
2811        let name = schema.table_name.as_bytes();
2812        buf.extend_from_slice(&(name.len() as u32).to_le_bytes());
2813        buf.extend_from_slice(name);
2814        buf.extend_from_slice(&(schema.columns.len() as u16).to_le_bytes());
2815        for col in &schema.columns {
2816            let cn = col.name.as_bytes();
2817            buf.extend_from_slice(&(cn.len() as u32).to_le_bytes());
2818            buf.extend_from_slice(cn);
2819            buf.push(col.type_id as u8);
2820            buf.push(if col.required { 1 } else { 0 });
2821            buf.extend_from_slice(&col.position.to_le_bytes());
2822        }
2823        // Per-table indexed column list with uniqueness flags (version 3).
2824        buf.extend_from_slice(&(entry.indexed_cols.len() as u16).to_le_bytes());
2825        for meta in &entry.indexed_cols {
2826            let cn = meta.name.as_bytes();
2827            buf.extend_from_slice(&(cn.len() as u32).to_le_bytes());
2828            buf.extend_from_slice(cn);
2829            buf.push(if meta.unique { 1 } else { 0 });
2830        }
2831        // Per-table column defaults (version 4).
2832        encode_defaults_section(&mut buf, entry.defaults);
2833        // Per-table auto-increment columns (version 5).
2834        encode_auto_section(&mut buf, entry.auto_cols);
2835    }
2836
2837    // Append a CRC32 checksum of the entire payload so the reader can
2838    // detect corruption (the WAL and btree .idx files already do this;
2839    // catalog.bin was the one file missing a checksum).
2840    let crc = crc32fast::hash(&buf);
2841    buf.extend_from_slice(&crc.to_le_bytes());
2842
2843    let mut f = fs::OpenOptions::new()
2844        .create(true)
2845        .write(true)
2846        .truncate(true)
2847        .open(path)?;
2848    f.write_all(&buf)?;
2849    f.sync_data()?;
2850    Ok(())
2851}
2852
2853fn read_catalog_file(path: &Path) -> io::Result<Vec<CatalogEntry>> {
2854    let mut f = fs::File::open(path)?;
2855    let mut buf = Vec::new();
2856    f.read_to_end(&mut buf)?;
2857
2858    let mut pos = 0usize;
2859    // Minimum: 4 (magic) + 2 (version) + 4 (n_tables) + 4 (crc) = 14
2860    if buf.len() < 14 || &buf[0..4] != CATALOG_MAGIC {
2861        return Err(io::Error::new(
2862            io::ErrorKind::InvalidData,
2863            "bad catalog magic",
2864        ));
2865    }
2866
2867    // Verify the trailing CRC32 checksum.
2868    let payload = &buf[..buf.len() - 4];
2869    let stored_crc = u32::from_le_bytes(
2870        buf[buf.len() - 4..]
2871            .try_into()
2872            .map_err(|_| io::Error::new(io::ErrorKind::InvalidData, "truncated catalog CRC"))?,
2873    );
2874    let computed_crc = crc32fast::hash(payload);
2875    if stored_crc != computed_crc {
2876        return Err(io::Error::new(
2877            io::ErrorKind::InvalidData,
2878            format!(
2879                "catalog CRC32 mismatch: expected {stored_crc:#010x}, got {computed_crc:#010x}"
2880            ),
2881        ));
2882    }
2883    // Strip the CRC suffix so the parsing loop below doesn't walk into it.
2884    let buf = &buf[..buf.len() - 4];
2885    pos += 4;
2886    let version = u16::from_le_bytes(
2887        buf[pos..pos + 2]
2888            .try_into()
2889            .map_err(|_| io::Error::new(io::ErrorKind::InvalidData, "truncated catalog header"))?,
2890    );
2891    pos += 2;
2892    // Accept every version from 1 up to the current CATALOG_VERSION: the
2893    // field-reading staircase below fills in fields a newer version added
2894    // (indexed-col uniqueness at v3, defaults at v4, auto columns at v5) and
2895    // defaults them for older files, so any 1..=CATALOG_VERSION file loads.
2896    // A range check (not an enumerated list) is what makes this back-compat
2897    // hold automatically on the next bump — the previous `version != 1 &&
2898    // version != 2 && version != CATALOG_VERSION` form silently rejected the
2899    // intermediate v3/v4 files when the constant moved to 5, which would have
2900    // failed to open a v0.6.x database on upgrade (data loss).
2901    if version == 0 || version > CATALOG_VERSION {
2902        return Err(io::Error::new(
2903            io::ErrorKind::InvalidData,
2904            format!("unsupported catalog version: {version}"),
2905        ));
2906    }
2907    let n_tables = u32::from_le_bytes(
2908        buf[pos..pos + 4]
2909            .try_into()
2910            .map_err(|_| io::Error::new(io::ErrorKind::InvalidData, "truncated catalog header"))?,
2911    ) as usize;
2912    pos += 4;
2913
2914    // Don't size an allocation from an unvalidated count: a corrupt or hostile
2915    // catalog could claim billions of tables and make the `Vec::with_capacity`
2916    // below attempt a huge allocation (host abort — fatal in embedded mode). A
2917    // file of `buf.len()` bytes can describe at most that many tables (each
2918    // needs several header bytes), so a larger count is corrupt. Mirrors the
2919    // btree's node-count guard.
2920    if n_tables > buf.len() {
2921        return Err(io::Error::new(
2922            io::ErrorKind::InvalidData,
2923            format!("catalog file corrupt: implausible table count {n_tables}"),
2924        ));
2925    }
2926
2927    let mut entries = Vec::with_capacity(n_tables);
2928    for _ in 0..n_tables {
2929        let name_len = read_u32(buf, &mut pos)? as usize;
2930        let table_name = read_string(buf, &mut pos, name_len)?;
2931        let n_cols = read_u16(buf, &mut pos)? as usize;
2932
2933        let mut columns = Vec::with_capacity(n_cols);
2934        for _ in 0..n_cols {
2935            let cname_len = read_u32(buf, &mut pos)? as usize;
2936            let name = read_string(buf, &mut pos, cname_len)?;
2937            let type_id_raw = read_u8(buf, &mut pos)?;
2938            let type_id = type_id_from_u8(type_id_raw)?;
2939            let required = read_u8(buf, &mut pos)? != 0;
2940            let position = read_u16(buf, &mut pos)?;
2941            columns.push(ColumnDef {
2942                name,
2943                type_id,
2944                required,
2945                position,
2946            });
2947        }
2948
2949        // Version 3 appends indexed column list with uniqueness flag.
2950        // Version 2 has indexed column names without uniqueness (default
2951        // to non-unique). Version 1 has no index info at all.
2952        let indexed_cols: Vec<IndexedColMeta> = if version >= 3 {
2953            let n = read_u16(buf, &mut pos)? as usize;
2954            let mut v = Vec::with_capacity(n);
2955            for _ in 0..n {
2956                let l = read_u32(buf, &mut pos)? as usize;
2957                let name = read_string(buf, &mut pos, l)?;
2958                let unique = read_u8(buf, &mut pos)? != 0;
2959                v.push(IndexedColMeta { name, unique });
2960            }
2961            v
2962        } else if version >= 2 {
2963            let n = read_u16(buf, &mut pos)? as usize;
2964            let mut v = Vec::with_capacity(n);
2965            for _ in 0..n {
2966                let l = read_u32(buf, &mut pos)? as usize;
2967                let name = read_string(buf, &mut pos, l)?;
2968                v.push(IndexedColMeta {
2969                    name,
2970                    unique: false,
2971                });
2972            }
2973            v
2974        } else {
2975            Vec::new()
2976        };
2977
2978        // Version 4 appends a column-defaults section after the index list.
2979        let defaults = if version >= 4 {
2980            decode_defaults_section(buf, &mut pos, columns.len()).ok_or_else(|| {
2981                io::Error::new(io::ErrorKind::InvalidData, "truncated catalog defaults")
2982            })?
2983        } else {
2984            Vec::new()
2985        };
2986
2987        // Version 5 appends an auto-increment column section after that.
2988        let auto_cols = if version >= 5 {
2989            decode_auto_section(buf, &mut pos, columns.len()).ok_or_else(|| {
2990                io::Error::new(io::ErrorKind::InvalidData, "truncated catalog auto columns")
2991            })?
2992        } else {
2993            Vec::new()
2994        };
2995
2996        entries.push(CatalogEntry {
2997            schema: Schema {
2998                table_name,
2999                columns,
3000            },
3001            indexed_cols,
3002            defaults,
3003            auto_cols,
3004        });
3005    }
3006
3007    Ok(entries)
3008}
3009
3010fn read_u8(buf: &[u8], pos: &mut usize) -> io::Result<u8> {
3011    if *pos >= buf.len() {
3012        return Err(io::Error::new(
3013            io::ErrorKind::UnexpectedEof,
3014            "truncated catalog",
3015        ));
3016    }
3017    let v = buf[*pos];
3018    *pos += 1;
3019    Ok(v)
3020}
3021fn read_u16(buf: &[u8], pos: &mut usize) -> io::Result<u16> {
3022    if *pos + 2 > buf.len() {
3023        return Err(io::Error::new(
3024            io::ErrorKind::UnexpectedEof,
3025            "truncated catalog",
3026        ));
3027    }
3028    let v = u16::from_le_bytes(
3029        buf[*pos..*pos + 2]
3030            .try_into()
3031            .expect("bounds checked above"),
3032    );
3033    *pos += 2;
3034    Ok(v)
3035}
3036fn read_u32(buf: &[u8], pos: &mut usize) -> io::Result<u32> {
3037    if *pos + 4 > buf.len() {
3038        return Err(io::Error::new(
3039            io::ErrorKind::UnexpectedEof,
3040            "truncated catalog",
3041        ));
3042    }
3043    let v = u32::from_le_bytes(
3044        buf[*pos..*pos + 4]
3045            .try_into()
3046            .expect("bounds checked above"),
3047    );
3048    *pos += 4;
3049    Ok(v)
3050}
3051fn read_string(buf: &[u8], pos: &mut usize, len: usize) -> io::Result<String> {
3052    if *pos + len > buf.len() {
3053        return Err(io::Error::new(
3054            io::ErrorKind::UnexpectedEof,
3055            "truncated catalog string",
3056        ));
3057    }
3058    let s = std::str::from_utf8(&buf[*pos..*pos + len])
3059        .map_err(|_| io::Error::new(io::ErrorKind::InvalidData, "non-utf8 in catalog"))?
3060        .to_string();
3061    *pos += len;
3062    Ok(s)
3063}
3064fn type_id_from_u8(v: u8) -> io::Result<TypeId> {
3065    match v {
3066        0 => Ok(TypeId::Empty),
3067        1 => Ok(TypeId::Int),
3068        2 => Ok(TypeId::Float),
3069        3 => Ok(TypeId::Bool),
3070        4 => Ok(TypeId::Str),
3071        5 => Ok(TypeId::DateTime),
3072        6 => Ok(TypeId::Uuid),
3073        7 => Ok(TypeId::Bytes),
3074        _ => Err(io::Error::new(
3075            io::ErrorKind::InvalidData,
3076            format!("unknown type id: {v}"),
3077        )),
3078    }
3079}
3080
3081#[cfg(test)]
3082mod tests {
3083    use super::*;
3084    fn temp_catalog(name: &str) -> Catalog {
3085        let dir = std::env::temp_dir().join(format!("powdb_cat_{name}_{}", std::process::id()));
3086        Catalog::create(&dir).unwrap()
3087    }
3088
3089    fn schema_two_cols() -> Schema {
3090        Schema {
3091            table_name: "T".into(),
3092            columns: vec![
3093                ColumnDef {
3094                    name: "id".into(),
3095                    type_id: TypeId::Int,
3096                    required: true,
3097                    position: 0,
3098                },
3099                ColumnDef {
3100                    name: "status".into(),
3101                    type_id: TypeId::Str,
3102                    required: false,
3103                    position: 1,
3104                },
3105            ],
3106        }
3107    }
3108
3109    #[test]
3110    fn replay_records_treats_reused_tx_ids_as_ordered_spans() {
3111        let mut cat = temp_catalog("reused_tx_ids");
3112        let schema = schema_two_cols();
3113        cat.create_table(schema.clone()).unwrap();
3114        cat.checkpoint().unwrap();
3115
3116        let mut committed_row = Vec::new();
3117        encode_row_into(
3118            &schema,
3119            &[Value::Int(1), Value::Str("committed".into())],
3120            &mut committed_row,
3121        );
3122        let mut incomplete_row = Vec::new();
3123        encode_row_into(
3124            &schema,
3125            &[Value::Int(2), Value::Str("incomplete".into())],
3126            &mut incomplete_row,
3127        );
3128
3129        let records = vec![
3130            WalRecord {
3131                tx_id: 1,
3132                record_type: WalRecordType::Begin,
3133                lsn: 1,
3134                data: Vec::new(),
3135            },
3136            WalRecord {
3137                tx_id: 1,
3138                record_type: WalRecordType::Insert,
3139                lsn: 2,
3140                data: encode_wal_payload(
3141                    "T",
3142                    RowId {
3143                        page_id: 1,
3144                        slot_index: 0,
3145                    },
3146                    &committed_row,
3147                ),
3148            },
3149            WalRecord {
3150                tx_id: 1,
3151                record_type: WalRecordType::Commit,
3152                lsn: 3,
3153                data: Vec::new(),
3154            },
3155            WalRecord {
3156                tx_id: 1,
3157                record_type: WalRecordType::Begin,
3158                lsn: 4,
3159                data: Vec::new(),
3160            },
3161            WalRecord {
3162                tx_id: 1,
3163                record_type: WalRecordType::Insert,
3164                lsn: 5,
3165                data: encode_wal_payload(
3166                    "T",
3167                    RowId {
3168                        page_id: 1,
3169                        slot_index: 1,
3170                    },
3171                    &incomplete_row,
3172                ),
3173            },
3174        ];
3175
3176        cat.apply_wal_records(&records).unwrap();
3177        let rows: Vec<_> = cat.scan("T").unwrap().collect();
3178        assert_eq!(rows.len(), 1);
3179        assert_eq!(rows[0].1[0], Value::Int(1));
3180        assert_eq!(rows[0].1[1], Value::Str("committed".into()));
3181    }
3182
3183    #[test]
3184    fn ddl_create_table_codec_roundtrips_defaults_and_auto() {
3185        let schema = schema_two_cols();
3186        let defaults = vec![None, Some(Value::Str("active".into()))];
3187        let auto_cols = vec![true, false];
3188        let encoded = encode_ddl_create_table(&schema, &defaults, &auto_cols);
3189        let (decoded_schema, decoded_defaults, decoded_auto) =
3190            decode_ddl_create_table(&encoded).unwrap();
3191        assert_eq!(decoded_schema.columns.len(), 2);
3192        assert_eq!(decoded_defaults, defaults);
3193        assert_eq!(decoded_auto, auto_cols);
3194    }
3195
3196    #[test]
3197    fn ddl_create_table_codec_back_compat_without_trailing_sections() {
3198        // Simulate a record written before column defaults / auto existed: the
3199        // old encoder stopped right after the columns, with no trailing
3200        // sections. The new decoder must read those as "none".
3201        let schema = schema_two_cols();
3202        let full = encode_ddl_create_table(&schema, &[], &[]);
3203        // Each empty trailing section is a u16 count of 0 (two bytes); chop
3204        // both off to mimic the pre-feature on-disk shape.
3205        let legacy = &full[..full.len() - 4];
3206        let (decoded_schema, decoded_defaults, decoded_auto) =
3207            decode_ddl_create_table(legacy).unwrap();
3208        assert_eq!(decoded_schema.columns.len(), 2);
3209        assert!(decoded_defaults.is_empty(), "no defaults section -> empty");
3210        assert!(decoded_auto.is_empty(), "no auto section -> empty");
3211    }
3212
3213    #[test]
3214    fn ddl_create_table_codec_back_compat_defaults_but_no_auto() {
3215        // A record from the column-defaults release (#129) has a defaults
3216        // section but no auto section; the auto-aware decoder must still read it.
3217        let schema = schema_two_cols();
3218        let defaults = vec![None, Some(Value::Str("active".into()))];
3219        let full = encode_ddl_create_table(&schema, &defaults, &[]);
3220        // Drop only the trailing auto section (its empty u16 count).
3221        let legacy = &full[..full.len() - 2];
3222        let (_schema, decoded_defaults, decoded_auto) = decode_ddl_create_table(legacy).unwrap();
3223        assert_eq!(decoded_defaults, defaults);
3224        assert!(decoded_auto.is_empty());
3225    }
3226
3227    #[test]
3228    fn read_catalog_file_accepts_intermediate_versions_3_and_4() {
3229        // Regression: the version gate accepted only {1, 2, CATALOG_VERSION}, so
3230        // a catalog written at version 3 (v0.6.x) or 4 (the column-defaults
3231        // release) was rejected with "unsupported catalog version" — the
3232        // database would fail to open on upgrade from those releases = data
3233        // loss. The field-reading staircase already handles v3/v4; only the gate
3234        // was stale. Build faithful v3/v4 catalog files by hand and confirm they
3235        // load (defaults/auto default to empty for the versions that lack them).
3236        use std::io::Write as _;
3237        fn write_legacy_catalog(path: &std::path::Path, version: u16) {
3238            let mut buf: Vec<u8> = Vec::new();
3239            buf.extend_from_slice(CATALOG_MAGIC);
3240            buf.extend_from_slice(&version.to_le_bytes());
3241            buf.extend_from_slice(&1u32.to_le_bytes()); // n_tables
3242                                                        // table "T"
3243            buf.extend_from_slice(&1u32.to_le_bytes());
3244            buf.extend_from_slice(b"T");
3245            buf.extend_from_slice(&2u16.to_le_bytes()); // n_cols
3246                                                        // col id: Int, required, pos 0
3247            buf.extend_from_slice(&2u32.to_le_bytes());
3248            buf.extend_from_slice(b"id");
3249            buf.push(TypeId::Int as u8);
3250            buf.push(1);
3251            buf.extend_from_slice(&0u16.to_le_bytes());
3252            // col status: Str, not required, pos 1
3253            buf.extend_from_slice(&6u32.to_le_bytes());
3254            buf.extend_from_slice(b"status");
3255            buf.push(TypeId::Str as u8);
3256            buf.push(0);
3257            buf.extend_from_slice(&1u16.to_le_bytes());
3258            // version >= 3: indexed-column section (count 0).
3259            buf.extend_from_slice(&0u16.to_le_bytes());
3260            // version >= 4: column-defaults section (none here). v3 omits it.
3261            if version >= 4 {
3262                encode_defaults_section(&mut buf, &[None, None]);
3263            }
3264            // v3/v4 never wrote the v5 auto section.
3265            let crc = crc32fast::hash(&buf);
3266            buf.extend_from_slice(&crc.to_le_bytes());
3267            let mut f = fs::File::create(path).unwrap();
3268            f.write_all(&buf).unwrap();
3269        }
3270
3271        for version in [3u16, 4u16] {
3272            let path = std::env::temp_dir().join(format!(
3273                "powdb_cat_v{version}_compat_{}.bin",
3274                std::process::id()
3275            ));
3276            write_legacy_catalog(&path, version);
3277            let entries = read_catalog_file(&path)
3278                .unwrap_or_else(|e| panic!("version {version} catalog must load, got: {e}"));
3279            assert_eq!(entries.len(), 1);
3280            assert_eq!(entries[0].schema.table_name, "T");
3281            assert_eq!(entries[0].schema.columns.len(), 2);
3282            assert!(
3283                entries[0].auto_cols.is_empty(),
3284                "v{version} has no auto cols"
3285            );
3286            fs::remove_file(&path).ok();
3287        }
3288    }
3289
3290    #[test]
3291    fn read_catalog_file_rejects_implausible_table_count() {
3292        // A corrupt/hostile catalog must not be trusted to size an allocation:
3293        // `Vec::with_capacity(n_tables)` on an unvalidated u32 would attempt a
3294        // huge allocation and abort the host. A file can describe at most as
3295        // many tables as it has bytes, so a count exceeding the payload length
3296        // is rejected with a clear error before any allocation. (We use a small
3297        // implausible count over a tiny buffer; a genuinely huge count would
3298        // abort the test runner pre-fix, but it hits the very same guard.)
3299        use std::io::Write as _;
3300        let mut buf: Vec<u8> = Vec::new();
3301        buf.extend_from_slice(CATALOG_MAGIC);
3302        buf.extend_from_slice(&CATALOG_VERSION.to_le_bytes());
3303        buf.extend_from_slice(&1000u32.to_le_bytes()); // claims 1000 tables…
3304                                                       // …but no table data follows (payload is only 10 bytes).
3305        let crc = crc32fast::hash(&buf);
3306        buf.extend_from_slice(&crc.to_le_bytes());
3307        let path =
3308            std::env::temp_dir().join(format!("powdb_cat_badcount_{}.bin", std::process::id()));
3309        fs::File::create(&path).unwrap().write_all(&buf).unwrap();
3310
3311        let msg = match read_catalog_file(&path) {
3312            Ok(_) => panic!("implausible table count must be rejected, got Ok"),
3313            Err(e) => e.to_string(),
3314        };
3315        assert!(
3316            msg.contains("implausible table count"),
3317            "expected an implausible-table-count error, got: {msg}"
3318        );
3319        fs::remove_file(&path).ok();
3320    }
3321
3322    #[test]
3323    fn data_dir_and_max_lsn_accessors() {
3324        let dir = std::env::temp_dir().join(format!("powdb_cat_maxlsn_{}", std::process::id()));
3325        let mut cat = Catalog::create(&dir).unwrap();
3326
3327        // data_dir() reflects the directory the catalog was created in.
3328        assert_eq!(cat.data_dir(), dir.as_path());
3329
3330        // A fresh catalog has stamped no page LSNs yet.
3331        assert_eq!(cat.max_lsn(), 0);
3332
3333        let schema = Schema {
3334            table_name: "users".into(),
3335            columns: vec![ColumnDef {
3336                name: "name".into(),
3337                type_id: TypeId::Str,
3338                required: true,
3339                position: 0,
3340            }],
3341        };
3342        cat.create_table(schema).unwrap();
3343
3344        cat.insert("users", &vec![Value::Str("Alice".into())])
3345            .unwrap();
3346        cat.sync_wal().unwrap();
3347
3348        // An inserted (and synced) row stamps a page LSN, raising the
3349        // durability high-water mark above zero.
3350        assert!(cat.max_lsn() > 0);
3351    }
3352
3353    #[test]
3354    fn test_create_table_and_insert() {
3355        let mut cat = temp_catalog("basic");
3356        let schema = Schema {
3357            table_name: "users".into(),
3358            columns: vec![
3359                ColumnDef {
3360                    name: "name".into(),
3361                    type_id: TypeId::Str,
3362                    required: true,
3363                    position: 0,
3364                },
3365                ColumnDef {
3366                    name: "age".into(),
3367                    type_id: TypeId::Int,
3368                    required: false,
3369                    position: 1,
3370                },
3371            ],
3372        };
3373        cat.create_table(schema).unwrap();
3374
3375        let row = vec![Value::Str("Alice".into()), Value::Int(30)];
3376        let rid = cat.insert("users", &row).unwrap();
3377
3378        let result = cat.get("users", rid).unwrap();
3379        assert_eq!(result[0], Value::Str("Alice".into()));
3380        assert_eq!(result[1], Value::Int(30));
3381    }
3382
3383    #[test]
3384    fn test_scan_table() {
3385        let mut cat = temp_catalog("scan");
3386        let schema = Schema {
3387            table_name: "items".into(),
3388            columns: vec![
3389                ColumnDef {
3390                    name: "name".into(),
3391                    type_id: TypeId::Str,
3392                    required: true,
3393                    position: 0,
3394                },
3395                ColumnDef {
3396                    name: "price".into(),
3397                    type_id: TypeId::Float,
3398                    required: true,
3399                    position: 1,
3400                },
3401            ],
3402        };
3403        cat.create_table(schema).unwrap();
3404
3405        for i in 0..50 {
3406            cat.insert(
3407                "items",
3408                &vec![
3409                    Value::Str(format!("item_{i}")),
3410                    Value::Float(i as f64 * 1.5),
3411                ],
3412            )
3413            .unwrap();
3414        }
3415
3416        let rows: Vec<_> = cat.scan("items").unwrap().collect();
3417        assert_eq!(rows.len(), 50);
3418    }
3419
3420    #[test]
3421    fn test_index_lookup() {
3422        let mut cat = temp_catalog("idx");
3423        let schema = Schema {
3424            table_name: "users".into(),
3425            columns: vec![
3426                ColumnDef {
3427                    name: "email".into(),
3428                    type_id: TypeId::Str,
3429                    required: true,
3430                    position: 0,
3431                },
3432                ColumnDef {
3433                    name: "name".into(),
3434                    type_id: TypeId::Str,
3435                    required: true,
3436                    position: 1,
3437                },
3438            ],
3439        };
3440        cat.create_table(schema).unwrap();
3441        cat.create_index("users", "email").unwrap();
3442
3443        cat.insert(
3444            "users",
3445            &vec![
3446                Value::Str("alice@example.com".into()),
3447                Value::Str("Alice".into()),
3448            ],
3449        )
3450        .unwrap();
3451        cat.insert(
3452            "users",
3453            &vec![
3454                Value::Str("bob@example.com".into()),
3455                Value::Str("Bob".into()),
3456            ],
3457        )
3458        .unwrap();
3459
3460        let result = cat
3461            .index_lookup("users", "email", &Value::Str("bob@example.com".into()))
3462            .unwrap();
3463        assert!(result.is_some());
3464        let row = result.unwrap();
3465        assert_eq!(row[1], Value::Str("Bob".into()));
3466    }
3467
3468    #[test]
3469    fn test_delete_row() {
3470        let mut cat = temp_catalog("delete");
3471        let schema = Schema {
3472            table_name: "t".into(),
3473            columns: vec![ColumnDef {
3474                name: "v".into(),
3475                type_id: TypeId::Int,
3476                required: true,
3477                position: 0,
3478            }],
3479        };
3480        cat.create_table(schema).unwrap();
3481        let r1 = cat.insert("t", &vec![Value::Int(1)]).unwrap();
3482        let r2 = cat.insert("t", &vec![Value::Int(2)]).unwrap();
3483        cat.delete("t", r1).unwrap();
3484        assert!(cat.get("t", r1).is_none());
3485        assert!(cat.get("t", r2).is_some());
3486    }
3487
3488    #[test]
3489    fn test_update_row() {
3490        let mut cat = temp_catalog("update");
3491        let schema = Schema {
3492            table_name: "t".into(),
3493            columns: vec![ColumnDef {
3494                name: "v".into(),
3495                type_id: TypeId::Int,
3496                required: true,
3497                position: 0,
3498            }],
3499        };
3500        cat.create_table(schema).unwrap();
3501        let rid = cat.insert("t", &vec![Value::Int(1)]).unwrap();
3502        let new_rid = cat.update("t", rid, &vec![Value::Int(99)]).unwrap();
3503        let row = cat.get("t", new_rid).unwrap();
3504        assert_eq!(row[0], Value::Int(99));
3505    }
3506
3507    #[test]
3508    fn test_persist_and_reopen() {
3509        let dir = std::env::temp_dir().join(format!("powdb_cat_persist_{}", std::process::id()));
3510        // Fresh dir
3511        let _ = std::fs::remove_dir_all(&dir);
3512
3513        {
3514            let mut cat = Catalog::create(&dir).unwrap();
3515            cat.create_table(Schema {
3516                table_name: "users".into(),
3517                columns: vec![
3518                    ColumnDef {
3519                        name: "name".into(),
3520                        type_id: TypeId::Str,
3521                        required: true,
3522                        position: 0,
3523                    },
3524                    ColumnDef {
3525                        name: "age".into(),
3526                        type_id: TypeId::Int,
3527                        required: false,
3528                        position: 1,
3529                    },
3530                ],
3531            })
3532            .unwrap();
3533            cat.insert("users", &vec![Value::Str("Alice".into()), Value::Int(30)])
3534                .unwrap();
3535            cat.insert("users", &vec![Value::Str("Bob".into()), Value::Int(25)])
3536                .unwrap();
3537        }
3538
3539        // Reopen — schema and rows should both still be there
3540        let cat = Catalog::open(&dir).unwrap();
3541        let schema = cat.schema("users").unwrap();
3542        assert_eq!(schema.columns.len(), 2);
3543        assert_eq!(schema.columns[0].name, "name");
3544        assert_eq!(schema.columns[0].type_id, TypeId::Str);
3545        assert_eq!(schema.columns[1].type_id, TypeId::Int);
3546
3547        let rows: Vec<_> = cat.scan("users").unwrap().collect();
3548        assert_eq!(rows.len(), 2);
3549
3550        std::fs::remove_dir_all(&dir).ok();
3551    }
3552
3553    #[test]
3554    fn test_open_missing_dir_errors() {
3555        let dir = std::env::temp_dir().join(format!("powdb_cat_missing_{}", std::process::id()));
3556        let _ = std::fs::remove_dir_all(&dir);
3557        std::fs::create_dir_all(&dir).unwrap();
3558        // No catalog.bin yet
3559        assert!(Catalog::open(&dir).is_err());
3560        std::fs::remove_dir_all(&dir).ok();
3561    }
3562
3563    #[test]
3564    fn test_list_tables() {
3565        let mut cat = temp_catalog("list");
3566        cat.create_table(Schema {
3567            table_name: "a".into(),
3568            columns: vec![ColumnDef {
3569                name: "x".into(),
3570                type_id: TypeId::Int,
3571                required: true,
3572                position: 0,
3573            }],
3574        })
3575        .unwrap();
3576        cat.create_table(Schema {
3577            table_name: "b".into(),
3578            columns: vec![ColumnDef {
3579                name: "y".into(),
3580                type_id: TypeId::Int,
3581                required: true,
3582                position: 0,
3583            }],
3584        })
3585        .unwrap();
3586        let mut tables = cat.list_tables();
3587        tables.sort();
3588        assert_eq!(tables, vec!["a", "b"]);
3589    }
3590
3591    #[test]
3592    fn test_path_traversal_table_name_rejected() {
3593        let mut cat = temp_catalog("path_trav");
3594        // Names with path separators must be rejected.
3595        let bad_names = vec![
3596            "../etc/passwd",
3597            "foo/bar",
3598            "table\0name",
3599            "",
3600            "123starts_with_digit",
3601            "has-dashes",
3602            "has spaces",
3603            "has.dots",
3604        ];
3605        for name in bad_names {
3606            let schema = Schema {
3607                table_name: name.into(),
3608                columns: vec![ColumnDef {
3609                    name: "x".into(),
3610                    type_id: TypeId::Int,
3611                    required: true,
3612                    position: 0,
3613                }],
3614            };
3615            let result = cat.create_table(schema);
3616            assert!(result.is_err(), "expected error for table name '{name}'");
3617            assert_eq!(result.unwrap_err().kind(), io::ErrorKind::InvalidInput);
3618        }
3619        // Valid names must still work.
3620        let good_names = vec!["users", "_private", "Table_123", "_"];
3621        for name in good_names {
3622            let schema = Schema {
3623                table_name: name.into(),
3624                columns: vec![ColumnDef {
3625                    name: "x".into(),
3626                    type_id: TypeId::Int,
3627                    required: true,
3628                    position: 0,
3629                }],
3630            };
3631            assert!(
3632                cat.create_table(schema).is_ok(),
3633                "expected ok for table name '{name}'"
3634            );
3635        }
3636    }
3637
3638    #[test]
3639    fn test_path_traversal_column_name_rejected() {
3640        let mut cat = temp_catalog("col_path_trav");
3641        let schema = Schema {
3642            table_name: "valid_table".into(),
3643            columns: vec![ColumnDef {
3644                name: "../bad".into(),
3645                type_id: TypeId::Int,
3646                required: true,
3647                position: 0,
3648            }],
3649        };
3650        let result = cat.create_table(schema);
3651        assert!(result.is_err());
3652        assert_eq!(result.unwrap_err().kind(), io::ErrorKind::InvalidInput);
3653    }
3654
3655    #[test]
3656    fn test_drop_table_validates_name() {
3657        let mut cat = temp_catalog("drop_trav");
3658        let result = cat.drop_table("../etc/passwd");
3659        assert!(result.is_err());
3660        // Should fail with InvalidInput (validation), not NotFound.
3661        assert_eq!(result.unwrap_err().kind(), io::ErrorKind::InvalidInput);
3662    }
3663}