Skip to main content

powdb_storage/
catalog.rs

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