Phase 7: secondary indexes — entry contract, DDL, atomic maintenance, base↔index validation; writer reclaim fix

Author: Thanga-Ganapathy

CHANGELOG.md

@@ -8,6 +8,47 @@ under a category (`Added` / `Changed` / `Fixed` / `Removed` / `Security`).
 
 ## [Unreleased]
 
+### Phase 7 — Indexing
+
+#### Added
+- `index`: the secondary-index entry contract — keys are the
+  order-preserving encoding of the indexed columns (+ encoded-PK suffix for
+  non-unique indexes); entry values are the encoded PK; unique indexes skip
+  rows with NULL indexed columns, non-unique include them (`DECISIONS.md`
+  D18) — plus probe/prefix-scan bounds and thin maintenance ops over
+  `txn::WriteCtx`.
+- `catalog`: `IndexDef` (single-column, composite, unique) persisted in the
+  table definition (codec v2; v1 records still decode); per-index
+  `("iroot", table, index)` root entries committed atomically with the base
+  root.
+- Index DDL: `create_index` (validates, **backfills from existing rows**,
+  rejects unique violations found in the data) and `drop_index` (frees the
+  index tree; the implicit backing of a `unique` column cannot be dropped).
+- Automatic maintenance in the same write transaction as every insert /
+  update / delete — updates touch only indexes whose keys changed; multi-row
+  atomicity covers index entries (in-batch unique collisions included).
+- `unique` columns now create **implicit unique indexes**
+  (`uniq_<table>_<column>`) and are enforced by index probes — D16's
+  provisional scan probe is deleted.
+- `CatSnapshot::validate()`: structural validation of every tree plus an
+  entry-for-entry **brute-force cross-check of every index against its base
+  table**, over the snapshot's pinned version.
+- Exit-criteria tests: seeded random-DML property test (insert/update/delete
+  against a model; every index brute-force-checked along the way), unique
+  violations at insert/update/backfill, atomicity, page reclamation for
+  dropped indexes/tables, pinned-snapshot consistency.
+
+#### Changed
+- `txn`: `Snapshot::validate_tree` exposes the B+tree structural validator
+  for trees under the pinned root.
+
+#### Fixed
+- `txn`: the writer reclaimed watermark-cleared pages at the start of
+  **every** batch, including batches that end up committing nothing (all
+  transactions rejected) — leaving the in-memory freelist ahead of the disk
+  until the next commit, which `validate()` reported as freelist corruption.
+  Reclamation now runs only inside committing batches (`DECISIONS.md` D19).
+
 ### Phase 6 — Schema, catalog & constraints
 
 #### Added

Cargo.lock

@@ -5,6 +5,47 @@ Per `PLAN.md` §1 rule 6, every resolution of an ambiguity or deviation from
 
 ---
 
+## D19 — Reclamation only runs inside a committing batch
+
+**Phase:** 7 · **Status:** accepted (bug fix of Phase 4 behavior)
+
+The Phase 4 writer reclaimed watermark-cleared pages at the **start of every
+batch**. A batch whose transactions are all rejected never commits — but
+reclamation had already mutated the pager's in-memory freelist, leaving it
+ahead of the disk (surfaced by `validate()` as freelist corruption; on a
+crash in that window the parked pages would simply leak, as they always do
+when the writer's in-memory park list dies — no disk state was ever wrong).
+
+**Decision:** the writer reclaims **after applying a batch's jobs and only
+when the batch will commit**, so the freelist changes ride the same fsync
+pair. Pages a batch frees become reusable one batch later (instead of within
+the same batch) — a negligible cost. Found by Phase 7's `drop index` tests;
+regression-tested at the txn layer.
+
+## D18 — Index entry contract: PK as value, NULL rows skip unique indexes
+
+**Phase:** 7 · **Status:** accepted
+
+`ARCHITECTURE.md` §3.6 fixes the key shape (encoded indexed columns, PK
+suffix when non-unique) but leaves the entry value and NULL semantics open.
+
+**Decision:**
+- **Entry value = the encoded PK** (the base-tree key) for both unique and
+  non-unique indexes: uniform, and resolving an entry to its row is one base
+  lookup without re-deriving key suffixes.
+- **Unique indexes skip rows with any NULL indexed column** — NULLs never
+  conflict (D17), and since unique entries carry no PK suffix, storing NULL
+  rows would falsely collide. Non-unique indexes include NULL rows (nulls
+  sort first, queryable).
+- **`unique` columns are enforced by an implicit single-column unique
+  index** named `uniq_<table>_<column>` (replacing D16's provisional scan
+  probe). The implicit backing cannot be dropped; `drop table` frees index
+  trees with the base.
+- Index names are per-table; the catalog stores one `("iroot", table, index)`
+  root entry per index, updated in the same commit as the base root. The
+  table-definition codec is now **version 2** (v1 records still decode,
+  index-less).
+
 ## D17 — Write-path semantics the SPEC leaves open
 
 **Phase:** 6 · **Status:** accepted

DECISIONS.md

@@ -14,6 +14,7 @@ common.workspace = true
 pager.workspace = true
 btree.workspace = true
 types.workspace = true
+index.workspace = true
 txn.workspace = true
 thiserror.workspace = true
 

crates/catalog/Cargo.toml

@@ -8,11 +8,14 @@
 use types::{decode_row, encode_row, TypeKind, Value};
 
 use crate::schema::{
-    CheckExpr, CmpOp, ColumnDef, DefaultSpec, TableDef, UpdatePolicy, MAX_CHECK_DEPTH,
+    CheckExpr, CmpOp, ColumnDef, DefaultSpec, IndexDef, TableDef, UpdatePolicy, MAX_CHECK_DEPTH,
 };
 use crate::{CatalogCorruption, CatalogError, Result};
 
-const VERSION: u8 = 1;
+/// Version 2 added the secondary-index section (Phase 7); version-1 records
+/// (no indexes) still decode.
+const VERSION: u8 = 2;
+const VERSION_NO_INDEXES: u8 = 1;
 
 const FLAG_NULLABLE: u8 = 1 << 0;
 const FLAG_UNIQUE: u8 = 1 << 1;
@@ -79,14 +82,23 @@ pub(crate) fn encode_table(def: &TableDef) -> Result<Vec<u8>> {
     for check in &def.checks {
         push_check(&mut out, check)?;
     }
+    push_count(&mut out, def.indexes.len())?;
+    for index in &def.indexes {
+        push_str(&mut out, &index.name)?;
+        out.push(u8::from(index.unique));
+        push_count(&mut out, index.columns.len())?;
+        for col in &index.columns {
+            push_str(&mut out, col)?;
+        }
+    }
     Ok(out)
 }
 
 /// Decode a stored table definition.
 pub(crate) fn decode_table(bytes: &[u8]) -> Result<TableDef> {
     let mut r = Reader { rest: bytes };
     let version = r.byte()?;
-    if version != VERSION {
+    if version != VERSION && version != VERSION_NO_INDEXES {
         return Err(corrupt(CatalogCorruption::BadVersion { version }));
     }
     let name = r.string()?;
@@ -129,6 +141,29 @@ pub(crate) fn decode_table(bytes: &[u8]) -> Result<TableDef> {
     for _ in 0..check_count {
         checks.push(r.check(0)?);
     }
+    let mut indexes = Vec::new();
+    if version >= VERSION {
+        let index_count = r.count()?;
+        indexes.reserve(index_count.min(64));
+        for _ in 0..index_count {
+            let name = r.string()?;
+            let unique = match r.byte()? {
+                0 => false,
+                1 => true,
+                tag => return Err(corrupt(CatalogCorruption::BadTag { tag })),
+            };
+            let col_count = r.count()?;
+            let mut cols = Vec::with_capacity(col_count.min(16));
+            for _ in 0..col_count {
+                cols.push(r.string()?);
+            }
+            indexes.push(IndexDef {
+                name,
+                columns: cols,
+                unique,
+            });
+        }
+    }
     if !r.rest.is_empty() {
         return Err(corrupt(CatalogCorruption::TrailingBytes));
     }
@@ -137,6 +172,7 @@ pub(crate) fn decode_table(bytes: &[u8]) -> Result<TableDef> {
         columns,
         pk,
         checks,
+        indexes,
     })
 }
 

crates/catalog/src/codec.rs

@@ -9,7 +9,7 @@ use types::{UuidV7Gen, Value};
 
 use crate::codec::decode_table;
 use crate::job::{decode_padded, CatalogJob, Env, JobOp, JobOut};
-use crate::schema::{ColumnDef, TableDef};
+use crate::schema::{ColumnDef, IndexDef, TableDef};
 use crate::store;
 use crate::{CatalogCorruption, CatalogError, Result};
 
@@ -86,6 +86,26 @@ impl<B: IoBackend + 'static> Catalog<B> {
         .map(|_| ())
     }
 
+    /// `create index` — validate, **backfill from existing rows** (unique
+    /// violations reject the DDL), and persist.
+    pub fn create_index(&self, table: &str, index: IndexDef) -> Result<()> {
+        self.submit(JobOp::CreateIndex {
+            table: table.to_string(),
+            index,
+        })
+        .map(|_| ())
+    }
+
+    /// `drop index` — remove the definition and free the index's pages. The
+    /// implicit index backing a `unique` column cannot be dropped.
+    pub fn drop_index(&self, table: &str, index: &str) -> Result<()> {
+        self.submit(JobOp::DropIndex {
+            table: table.to_string(),
+            index: index.to_string(),
+        })
+        .map(|_| ())
+    }
+
     /// Insert one row (named columns; omitted columns take defaults/generated
     /// values). Returns the full materialized row in schema column order.
     pub fn insert(&self, table: &str, row: Vec<(String, Value)>) -> Result<Vec<Value>> {
@@ -213,13 +233,69 @@ impl<B: IoBackend> CatSnapshot<B> {
         Ok(rows)
     }
 
+    /// Cross-check the whole database: every tree is structurally valid and
+    /// **every index matches its base table entry-for-entry** (the Phase 7
+    /// exit criterion). Read-only, over this snapshot's pinned version.
+    pub fn validate(&self) -> Result<()> {
+        if let Some(cat_root) = self.snap.root() {
+            self.snap.validate_tree(cat_root)?;
+        }
+        for table in self.tables()? {
+            let def = self.table(&table)?;
+            let data_root = self.data_root(&table)?;
+            self.snap.validate_tree(data_root)?;
+            let base = self.snap.scan_in(data_root)?;
+
+            for idx in &def.indexes {
+                let iroot = self.index_root(&table, &idx.name)?;
+                self.snap.validate_tree(iroot)?;
+
+                // Brute-force expectation from the base rows.
+                let mut expected: Vec<(Vec<u8>, Vec<u8>)> = Vec::new();
+                for (pk_key, bytes) in &base {
+                    let row = decode_padded(&def, bytes)?;
+                    let cols: Vec<types::Value> = idx
+                        .columns
+                        .iter()
+                        .map(|name| {
+                            def.col_index(name)
+                                .and_then(|ci| row.get(ci).cloned())
+                                .unwrap_or(types::Value::Null)
+                        })
+                        .collect();
+                    if let Some(e) = index::entry(&cols, pk_key, idx.unique)? {
+                        expected.push((e.key, e.value));
+                    }
+                }
+                expected.sort();
+
+                let actual = self.snap.scan_in(iroot)?;
+                if expected != actual {
+                    return Err(CatalogError::IndexOutOfSync {
+                        table: table.clone(),
+                        index: idx.name.clone(),
+                    });
+                }
+            }
+        }
+        Ok(())
+    }
+
     fn data_root(&self, table: &str) -> Result<pager::PageId> {
         let key = store::root_key(table)?;
         match self.snap.get(&key)? {
             Some(bytes) => store::decode_root(&bytes),
             None => Err(CatalogError::Corrupt(CatalogCorruption::MissingEntry)),
         }
     }
+
+    fn index_root(&self, table: &str, index: &str) -> Result<pager::PageId> {
+        let key = store::iroot_key(table, index)?;
+        match self.snap.get(&key)? {
+            Some(bytes) => store::decode_root(&bytes),
+            None => Err(CatalogError::Corrupt(CatalogCorruption::MissingEntry)),
+        }
+    }
 }
 
 /// Map a transaction-layer error back to the catalog's typed error: a