spg_storage/table.rs
1//! The `Table` storage object: row insert/update/delete, index
2//! construction + rebuild (BTree / BRIN / GIN / GIN-trgm /
3//! GIN-fulltext / NSW), cold-locator registration, and schema
4//! mutation (add/drop/rename column). Split out of lib.rs (monster
5//! tier-3 cut 4). The `Table` struct itself stays in lib.rs as
6//! storage vocabulary; this module is the inherent `impl` over it.
7//! `Table`'s private fields are reachable here because `table` is a
8//! descendant module of the crate root where the struct is declared.
9
10use super::*;
11
12impl Table {
13 pub fn new(schema: TableSchema) -> Self {
14 Self {
15 schema,
16 rel_id: crate::row_header::RelId::UNASSIGNED,
17 rows: PersistentVec::new(),
18 headers: PersistentVec::new(),
19 rowids: PersistentVec::new(),
20 next_rowid: 1,
21 dead_rows: 0,
22 stat_tup_ins: 0,
23 stat_tup_upd: 0,
24 stat_tup_del: 0,
25 scan_stats: crate::ScanStats::default(),
26 last_autovacuum_us: None,
27 last_analyze_us: None,
28 indices: Vec::new(),
29 hot_bytes: 0,
30 cold_row_count: 0,
31 cold_row_count_stale: false,
32 redo_log: None,
33 excl_indexes: Vec::new(),
34 prune_horizon: 0,
35 }
36 }
37
38 /// v7.37.15 (Phase C.1) — allocate the next stable [`RowId`] for
39 /// this relation. Monotonic, never reused. Callers push the
40 /// returned id onto `rowids` in lock-step with the `rows` /
41 /// `headers` append so `rowids[i]` names the row at slot `i`.
42 fn alloc_rowid(&mut self) -> crate::row_header::RowId {
43 let id = crate::row_header::RowId(self.next_rowid);
44 self.next_rowid += 1;
45 id
46 }
47
48 /// v7.37.15 (Phase C.1) — read-only access to the stable row ids
49 /// parallel to `rows()`. `rowids().len() == rows().len()` is the
50 /// load-bearing lock-step invariant (asserted in debug builds at
51 /// every mutation boundary alongside `headers`).
52 #[must_use]
53 pub fn rowids(&self) -> &PersistentVec<crate::row_header::RowId> {
54 &self.rowids
55 }
56
57 /// v7.37.15 (Phase C.1) — this relation's stable identity.
58 /// [`RelId::UNASSIGNED`](crate::row_header::RelId::UNASSIGNED) for
59 /// a bare `Table::new`; a real id once the catalog stamps it.
60 #[must_use]
61 pub fn rel_id(&self) -> crate::row_header::RelId {
62 self.rel_id
63 }
64
65 /// v7.37.15 (Phase C.1) — stamp this relation's stable identity.
66 /// Called by `Catalog::create_table` and the deserialize
67 /// dense-assign pass; idempotent overwrite.
68 pub(crate) fn set_rel_id(&mut self, id: crate::row_header::RelId) {
69 self.rel_id = id;
70 }
71
72 /// v7.37.15 (Phase C.1) — rebuild the `rowids` vec so it is dense
73 /// `1..=rows.len()` and reset the allocator above it. Used on the
74 /// load / snapshot-restore path where rows arrive without ids
75 /// (pre-V6 envelope): every row gets a fresh id, sufficient while
76 /// ids are process-local bookkeeping. Keeps the lock-step
77 /// invariant against the freshly-loaded `rows`.
78 pub fn assign_dense_rowids(&mut self) {
79 let n = self.rows.len();
80 let mut fresh: PersistentVec<crate::row_header::RowId> = PersistentVec::new();
81 for i in 0..n {
82 fresh.push_mut(crate::row_header::RowId((i + 1) as u64));
83 }
84 self.rowids = fresh;
85 self.next_rowid = (n as u64) + 1;
86 debug_assert_eq!(
87 self.rows.len(),
88 self.rowids.len(),
89 "rowids must stay in lock-step with rows after assign_dense_rowids"
90 );
91 }
92
93 /// v7.37.16 (autovacuum) — number of tombstoned-but-present hot rows.
94 /// Incrementally maintained; drives the engine's autovacuum threshold.
95 #[must_use]
96 pub fn dead_rows(&self) -> u64 {
97 self.dead_rows
98 }
99
100 /// v7.37.16 (autovacuum) — loader-side rebase of the dead-row
101 /// counter (the v53 MVCC appendix restores headers verbatim).
102 pub(crate) fn set_dead_rows_on_load(&mut self, dead: u64) {
103 self.dead_rows = dead;
104 }
105
106 /// v7.39 (pg_stat knife A) — bump the volatile write counters the
107 /// engine's DML dispatcher reports per statement.
108 pub fn bump_write_stats(&mut self, ins: u64, upd: u64, del: u64) {
109 self.stat_tup_ins = self.stat_tup_ins.saturating_add(ins);
110 self.stat_tup_upd = self.stat_tup_upd.saturating_add(upd);
111 self.stat_tup_del = self.stat_tup_del.saturating_add(del);
112 }
113
114 /// `(n_tup_ins, n_tup_upd, n_tup_del)` for pg_stat_user_tables.
115 #[must_use]
116 pub fn write_stats(&self) -> (u64, u64, u64) {
117 (self.stat_tup_ins, self.stat_tup_upd, self.stat_tup_del)
118 }
119
120 /// v7.39 (pg_stat knife C) — maintenance stamps for
121 /// pg_stat_user_tables (`(last_autovacuum_us, last_analyze_us)`).
122 #[must_use]
123 pub fn maintenance_stamps(&self) -> (Option<i64>, Option<i64>) {
124 (self.last_autovacuum_us, self.last_analyze_us)
125 }
126
127 pub fn stamp_autovacuum(&mut self, unix_us: i64) {
128 self.last_autovacuum_us = Some(unix_us);
129 }
130
131 pub fn stamp_analyze(&mut self, unix_us: i64) {
132 self.last_analyze_us = Some(unix_us);
133 }
134
135 /// v7.39 (pg_stat knife B) — the scan counters (read side of
136 /// pg_stat_user_tables).
137 #[must_use]
138 pub fn scan_stats(&self) -> &crate::ScanStats {
139 &self.scan_stats
140 }
141
142 /// v7.39 (pg_stat knife B) — one sequential scan over the visible
143 /// rows, reported by engine scan loops that walk headers directly
144 /// (parallel shards, the aggregate full scan) instead of
145 /// `scan_visible`.
146 pub fn note_seq_scan(&self) {
147 use core::sync::atomic::Ordering;
148 self.scan_stats.seq_scan.fetch_add(1, Ordering::Relaxed);
149 let visible = (self.rows.len() as u64).saturating_sub(self.dead_rows);
150 self.scan_stats
151 .seq_tup_read
152 .fetch_add(visible, Ordering::Relaxed);
153 }
154
155 /// v7.39 (pg_stat knife B) — one index scan returning `fetched`
156 /// rows (the engine's index-seek paths report here).
157 pub fn note_index_scan(&self, fetched: u64) {
158 use core::sync::atomic::Ordering;
159 self.scan_stats.idx_scan.fetch_add(1, Ordering::Relaxed);
160 self.scan_stats
161 .idx_tup_fetch
162 .fetch_add(fetched, Ordering::Relaxed);
163 }
164
165 /// v7.37.15 (Phase A.2) — read-only access to the per-row
166 /// MVCC visibility headers. `headers().len() == rows().len()`
167 /// is the load-bearing invariant; Phase B scan paths consult
168 /// `headers()[idx]` to decide visibility.
169 #[must_use]
170 pub fn headers(&self) -> &PersistentVec<crate::row_header::RowHeader> {
171 &self.headers
172 }
173
174 /// v7.37.15 (Phase B TDD) — `#[cfg(test)]`-only mutable header
175 /// access for tests that need to simulate Phase C semantics
176 /// (writer-side xmin/xmax stamping) before the real stamping
177 /// API lands. Phase C will provide a writer-aware setter that
178 /// keeps headers + xact bookkeeping consistent.
179 #[cfg(test)]
180 pub(crate) fn headers_mut_for_test(
181 &mut self,
182 ) -> &mut PersistentVec<crate::row_header::RowHeader> {
183 &mut self.headers
184 }
185
186 /// v7.37.16 (Epic W) — `#[cfg(test)]`-only read of the relation's
187 /// next-RowId allocator cursor, so the snapshot round-trip tests can
188 /// assert it is restored correctly (strictly above every persisted
189 /// id) without a public accessor on the hot path.
190 #[cfg(test)]
191 pub(crate) fn next_rowid_for_test(&self) -> u64 {
192 self.next_rowid
193 }
194
195 /// v7.37.15 (Phase C) — engine writer path. Same as [`insert`]
196 /// but stamps `xmin` on the new row's header with the writing
197 /// transaction's id (caller-supplied; obtained from the engine's
198 /// monotonic version counter). The fresh insert is alive
199 /// (`xmax = XMAX_ALIVE`); a later UPDATE / DELETE will set
200 /// `xmax` to a later version, leaving the row physically
201 /// present until vacuum reclaims it (Phase D).
202 ///
203 /// Callers in [`crate::row_header::next_version`] order:
204 /// 1. allocate version V via `next_version()`
205 /// 2. call `insert_with_xmin(row, V)`
206 /// 3. update any indexes (as `insert` does)
207 ///
208 /// `xmin = XMIN_FROZEN` short-circuits to plain `insert`
209 /// behaviour so the legacy in-memory / WAL-replay paths keep
210 /// returning identical results when they end up here.
211 pub fn insert_with_xmin(&mut self, row: Row<'static>, xmin: u64) -> Result<(), StorageError> {
212 if xmin == crate::row_header::XMIN_FROZEN {
213 return self.insert(row);
214 }
215 self.insert(row)?;
216 // Insert appended `RowHeader::frozen()`; overwrite with the
217 // alive-xmin header so visibility scans against snapshots
218 // taken before the writer's commit hide this row. Subsequent
219 // commit is recorded by the WAL; replay re-applies via the
220 // plain `insert_no_index` path and stamps frozen — but a
221 // snapshot taken AFTER commit sees `xmin = V <= snapshot.version`
222 // and the in_progress bitset no longer contains V, so the
223 // row passes the visibility predicate identically.
224 let last = self
225 .headers
226 .len()
227 .checked_sub(1)
228 .expect("insert appended a header");
229 if let Some(new_headers) = self
230 .headers
231 .set(last, crate::row_header::RowHeader::alive(xmin))
232 {
233 self.headers = new_headers;
234 }
235 debug_assert_eq!(
236 self.rows.len(),
237 self.headers.len(),
238 "headers must stay in lock-step with rows after insert_with_xmin"
239 );
240 Ok(())
241 }
242
243 /// v7.39 (round 493) — publish the snapshot floor the insert path may
244 /// prune dead index entries under. See `prune_horizon`.
245 ///
246 /// The engine sets this from `vacuum_oldest_active()` before a
247 /// statement's inserts. `0` disables pruning, which is the default and
248 /// is always safe.
249 pub fn set_prune_horizon(&mut self, horizon: u64) {
250 self.prune_horizon = horizon;
251 }
252
253 /// v7.37.15 (Phase D) — single-table vacuum pass. Walks the
254 /// header vec and physically removes any row whose delete
255 /// commit is older than `oldest_active_snapshot`. Returns the
256 /// number of reclaimable rows (with `dry_run == true`) or the
257 /// number actually reclaimed.
258 ///
259 /// `oldest_active_snapshot` is the floor of every live
260 /// snapshot's `version` — the engine maintains this; hosts
261 /// pass it through.
262 ///
263 /// Phase D ships the storage primitive. Hosts (spg-embedded /
264 /// spg-server) schedule the pass on their own thread.
265 pub fn vacuum(
266 &mut self,
267 oldest_active_snapshot: u64,
268 dry_run: bool,
269 ) -> crate::vacuum::VacuumReport {
270 let examined = self.headers.len() as u64;
271 // Collect the reclaimable positions in a first pass so the
272 // mutation can rebuild both the rows and the headers vec
273 // together (their lock-step invariant survives).
274 let to_reclaim: alloc::vec::Vec<usize> = (0..self.headers.len())
275 .filter(|&i| {
276 self.headers
277 .get(i)
278 .map(|h| crate::vacuum::is_reclaimable(h.xmax, oldest_active_snapshot))
279 .unwrap_or(false)
280 })
281 .collect();
282 if to_reclaim.is_empty() || dry_run {
283 return crate::vacuum::VacuumReport {
284 rows_reclaimed: to_reclaim.len() as u64,
285 rows_examined: examined,
286 per_table: alloc::vec::Vec::new(),
287 };
288 }
289 // Drive the existing per-position delete path so both rows
290 // and headers shrink together (it's the only mutator that
291 // already maintains the lock-step invariant).
292 let removed = self.delete_rows_no_index(&to_reclaim);
293 self.rebuild_indices();
294 crate::vacuum::VacuumReport {
295 rows_reclaimed: removed as u64,
296 rows_examined: examined,
297 per_table: alloc::vec::Vec::new(),
298 }
299 }
300
301 /// v7.37.15 (Phase C) — mark the row at `position` as deleted
302 /// by version `xmax`. The row stays physically present; later
303 /// vacuum (Phase D) reclaims it once no live snapshot can
304 /// still see it.
305 ///
306 /// Returns `Err(Corrupt)` on out-of-bounds and silently no-ops
307 /// when the row is already tombstoned (a later DELETE on an
308 /// already-deleted row should not change xmax — the original
309 /// deletion wins).
310 pub fn mark_row_deleted(&mut self, position: usize, xmax: u64) -> Result<(), StorageError> {
311 if position >= self.headers.len() {
312 return Err(StorageError::Corrupt(alloc::format!(
313 "mark_row_deleted: position {position} out of bounds (headers={})",
314 self.headers.len()
315 )));
316 }
317 let mut h = *self.headers.get(position).expect("position bounds-checked");
318 if h.xmax != crate::row_header::XMAX_ALIVE {
319 // Already tombstoned by an earlier delete. Keep the
320 // original xmax — first-deleter-wins.
321 return Ok(());
322 }
323 h.xmax = xmax;
324 if let Some(new_headers) = self.headers.set(position, h) {
325 self.headers = new_headers;
326 }
327 self.dead_rows += 1;
328 // v7.37.15 (Epic W durable-tombstone slice) — capture the
329 // in-place tombstone as row-level redo so a gate-on
330 // (`SPG_MVCC_INPLACE`) DELETE / UPDATE-old-version /
331 // ON-CONFLICT survives crash recovery. Unlike `delete_rows`
332 // (which records `RowChange::Delete` with physical positions),
333 // the tombstone keeps the slot, so it is named by the row's
334 // stable `RowId` — read from `self.rowids()[position]` here,
335 // before any later compaction shifts the slot. `xmax` is the
336 // deleting statement's writer version (the engine passes
337 // `writer_version_for_current_stmt`), so no post-drain stamp is
338 // needed. Only paid for when redo capture is on; a no-op
339 // (already-tombstoned / out-of-bounds) returned above and
340 // records nothing.
341 if self.redo_log.is_some() {
342 let rowid = self
343 .rowids()
344 .get(position)
345 .copied()
346 .unwrap_or(crate::row_header::RowId::UNASSIGNED);
347 self.record_redo(move |table| RowChange::Tombstone {
348 table,
349 rowids: alloc::vec![rowid],
350 xmax,
351 });
352 }
353 Ok(())
354 }
355
356 /// v7.37.16 — batch form of [`Table::mark_row_deleted`]: stamp `xmax`
357 /// on every alive, in-bounds position and record ONE
358 /// `RowChange::Tombstone` carrying all affected `RowId`s (the codec
359 /// and replay already handle multi-rowid records). The per-row form
360 /// paid one redo record — a Vec alloc plus a log push — PER ROW,
361 /// ~800 ns/row on a 10k-row gate-on DELETE (heavy_write del_10k).
362 /// Semantics match the single-row form: already-tombstoned keeps its
363 /// original xmax (first-deleter-wins), out-of-bounds is skipped.
364 /// Returns the number of rows NEWLY tombstoned.
365 pub fn mark_rows_deleted(&mut self, positions: &[usize], xmax: u64) -> usize {
366 let mut rowids: alloc::vec::Vec<crate::row_header::RowId> = alloc::vec::Vec::new();
367 let capture = self.redo_log.is_some();
368 let mut newly = 0usize;
369 for &position in positions {
370 // v7.37.16 — `get_mut` (transient in-place edit when the
371 // headers trie is uniquely owned) instead of the `set`
372 // path-copy: a 10k-row tombstone pass was spending ~3 ms in
373 // per-row spine copies.
374 match self.headers.get_mut(position) {
375 Some(h) if h.xmax == crate::row_header::XMAX_ALIVE => {
376 h.xmax = xmax;
377 }
378 _ => continue, // out-of-bounds or already tombstoned
379 }
380 self.dead_rows += 1;
381 newly += 1;
382 if capture {
383 rowids.push(
384 self.rowids()
385 .get(position)
386 .copied()
387 .unwrap_or(crate::row_header::RowId::UNASSIGNED),
388 );
389 }
390 }
391 if capture && !rowids.is_empty() {
392 self.record_redo(move |table| RowChange::Tombstone {
393 table,
394 rowids,
395 xmax,
396 });
397 }
398 newly
399 }
400
401 /// v7.37.17 (Phase E RC rebase) — extract the write-set one writer
402 /// version left on this table, expressed against stable [`RowId`]s
403 /// so it can be replayed onto a FRESHER catalog clone whose
404 /// physical slots differ. `inserted` carries INSERT rows and the
405 /// new versions of UPDATEs (`xmin == v`); `tombstoned` carries the
406 /// ids DELETE / UPDATE-old-version stamped (`xmax == v`). A row
407 /// both inserted and tombstoned by the same version appears in
408 /// both lists; replay applies inserts first, tombstones second —
409 /// net effect identical.
410 #[must_use]
411 pub fn extract_tx_writeset(&self, v: u64) -> crate::TxWriteSet {
412 let mut inserted: alloc::vec::Vec<(crate::row_header::RowId, Row<'static>)> =
413 alloc::vec::Vec::new();
414 let mut tombstoned: alloc::vec::Vec<crate::row_header::RowId> = alloc::vec::Vec::new();
415 for (i, h) in self.headers.iter().enumerate() {
416 let rid = self
417 .rowids
418 .get(i)
419 .copied()
420 .unwrap_or(crate::row_header::RowId::UNASSIGNED);
421 if h.xmin == v
422 && let Some(row) = self.rows.get(i)
423 {
424 inserted.push((rid, row.clone()));
425 }
426 if h.xmax == v {
427 tombstoned.push(rid);
428 }
429 }
430 crate::TxWriteSet {
431 inserted,
432 tombstoned,
433 }
434 }
435
436 /// v7.37.17 (Phase E4 fix) — read-only conflict probe for a
437 /// write-set's tombstones against THIS (fresher) relation: a target
438 /// RowId that is gone, or already tombstoned by a DIFFERENT
439 /// version, is a write-write conflict. Callers use this BEFORE
440 /// `replay_tx_writeset` so a conflicting UPDATE can drop its
441 /// paired insert too (atomicity of tombstone+insert pairs).
442 #[must_use]
443 pub fn tombstone_conflicts(
444 &self,
445 rids: &[crate::row_header::RowId],
446 v: u64,
447 ) -> alloc::vec::Vec<crate::row_header::RowId> {
448 rids.iter()
449 .filter(
450 |rid| match (0..self.rowids.len()).find(|&i| self.rowids.get(i) == Some(rid)) {
451 Some(i) => self
452 .headers
453 .get(i)
454 .is_some_and(|h| h.xmax != crate::row_header::XMAX_ALIVE && h.xmax != v),
455 None => true,
456 },
457 )
458 .copied()
459 .collect()
460 }
461
462 /// v7.37.17 (Phase E RC rebase) — replay a write-set extracted from
463 /// an OLDER clone of this relation onto this (fresher) one, keeping
464 /// the original RowIds. Deliberately does NOT capture redo: a
465 /// replay re-expresses writes the transaction already made, it is
466 /// not a new mutation (the redo story rides the eventual COMMIT).
467 /// Returns the ids whose tombstone could not be applied because the
468 /// row is gone or already tombstoned by a DIFFERENT version — the
469 /// write-write conflict surface (RC skips them per PG semantics;
470 /// RR/SER turn them into serialization_failure — Phase E3).
471 pub fn replay_tx_writeset(
472 &mut self,
473 ws: &crate::TxWriteSet,
474 v: u64,
475 ) -> alloc::vec::Vec<crate::row_header::RowId> {
476 for (rid, row) in &ws.inserted {
477 // Full insert (validation + index maintenance + fresh
478 // header/rowid), then re-stamp the header's xmin and put
479 // the ORIGINAL RowId back. The allocator id the insert
480 // burned is simply never used — ids are never recycled, so
481 // a gap is harmless. Insert can only fail on schema
482 // mismatch, impossible for a row this same relation
483 // already accepted; a debug_assert documents that.
484 let res = self.insert(row.clone());
485 debug_assert!(res.is_ok(), "writeset replay re-inserts a validated row");
486 if res.is_err() {
487 continue;
488 }
489 let last = self.rows.len() - 1;
490 if let Some(h) = self.headers.get_mut(last) {
491 h.xmin = v;
492 }
493 if let Some(slot) = self.rowids.get_mut(last) {
494 *slot = *rid;
495 }
496 }
497 let mut conflicts: alloc::vec::Vec<crate::row_header::RowId> = alloc::vec::Vec::new();
498 for rid in &ws.tombstoned {
499 let pos = (0..self.rowids.len()).find(|&i| self.rowids.get(i) == Some(rid));
500 match pos {
501 Some(i) => match self.headers.get_mut(i) {
502 Some(h) if h.xmax == crate::row_header::XMAX_ALIVE => {
503 h.xmax = v;
504 self.dead_rows += 1;
505 }
506 Some(h) if h.xmax == v => {} // already ours (idempotent)
507 _ => conflicts.push(*rid),
508 },
509 None => conflicts.push(*rid),
510 }
511 }
512 conflicts
513 }
514
515 /// v7.34 (crash-recovery P0 #2) — start capturing row-level redo into
516 /// this table (engine call before a mutating statement when
517 /// persistence is on). Idempotent; existing captured changes are kept.
518 pub fn enable_redo(&mut self) {
519 if self.redo_log.is_none() {
520 self.redo_log = Some(Vec::new());
521 }
522 }
523
524 /// v7.34 — drain the captured redo changes and stop capturing.
525 /// Returns the physical [`RowChange`]s applied since `enable_redo`,
526 /// in apply order (empty when capture was off or nothing changed).
527 pub fn take_redo(&mut self) -> Vec<RowChange> {
528 self.redo_log.take().unwrap_or_default()
529 }
530
531 /// Record one captured change when redo capture is on. The table name
532 /// rides on the change (taken from the schema) so a drained log is
533 /// self-describing against the whole catalog.
534 fn record_redo(&mut self, make: impl FnOnce(String) -> RowChange) {
535 if self.redo_log.is_some() {
536 let change = make(self.schema.name.clone());
537 if let Some(log) = self.redo_log.as_mut() {
538 log.push(change);
539 }
540 }
541 }
542
543 /// Total encoded byte size of every row currently in the hot tier
544 /// (`self.rows`). See struct docs for the maintenance contract.
545 /// Returns 0 for an empty table.
546 #[must_use]
547 pub const fn hot_bytes(&self) -> u64 {
548 self.hot_bytes
549 }
550
551 /// v6.7.0 — cached count of cold-tier rows. See struct field
552 /// docs for the staleness contract.
553 #[must_use]
554 pub const fn cold_row_count(&self) -> u64 {
555 self.cold_row_count
556 }
557
558 /// v6.7.0 — overwrite the cached count. Called by the engine's
559 /// `analyze_one_table` after walking the indices.
560 pub fn set_cold_row_count(&mut self, n: u64) {
561 self.cold_row_count = n;
562 self.cold_row_count_stale = false;
563 }
564
565 /// v6.7.0 — mark the cached count as potentially out of date.
566 /// Called by freezer / promote / DELETE paths so a subsequent
567 /// `spg_statistic` read knows the number may not reflect the
568 /// current state.
569 pub fn mark_cold_row_count_stale(&mut self) {
570 self.cold_row_count_stale = true;
571 }
572
573 /// v6.7.0 — report whether the cached count is known to be out
574 /// of date. Exposed for completeness; the virtual table surface
575 /// returns the cached value regardless.
576 #[must_use]
577 pub const fn cold_row_count_stale(&self) -> bool {
578 self.cold_row_count_stale
579 }
580
581 /// v7.36 — O(1) "could this table possibly have cold rows?"
582 /// predicate, intended for perf-critical executor hot paths
583 /// that just need to skip the cold-tier branch when there's
584 /// definitely nothing there. Reads the cached `cold_row_count`:
585 /// - cache fresh + cache == 0 → return false (fast path)
586 /// - cache stale → return true (conservative; the executor
587 /// pays the cold-aware path's `iter_cold_rows_*` cost but
588 /// stays correct)
589 /// - cache fresh + cache > 0 → return true
590 /// `count_cold_locators` remains the right call for the EXACT
591 /// count (ANALYZE etc.) — its O(N) walk is unsuitable per join
592 /// stage.
593 #[must_use]
594 pub const fn has_cold_rows_fast(&self) -> bool {
595 self.cold_row_count_stale || self.cold_row_count > 0
596 }
597
598 /// r944 — every BTree index a cold row could have been filed under.
599 ///
600 /// The freeze writes a row's locator into exactly ONE index
601 /// (`register_cold_locators` takes a single index name) and the
602 /// freezer picks that index by its own rule, so a reader that guesses
603 /// a different one finds nothing. Round 943 is that bug: the freezer
604 /// chose the first BTree index over any integer column, the scan
605 /// looked at the first index on the primary key's column, and 15
606 /// frozen rows of 40 vanished from a plain `SELECT`.
607 ///
608 /// Union over all of them rather than guessing one. Because each
609 /// row's locator exists in exactly one index, the union yields every
610 /// row once and needs no visited-set.
611 ///
612 /// Deliberately NOT filtered to declared-unique indices. Freezing
613 /// through an index whose keys repeat is a real limitation —
614 /// `resolve_cold_locator` resolves BY KEY and cannot say which of two
615 /// rows sharing one was meant — but that limit belongs to the freeze,
616 /// which builds the segment keyed that way. Filtering it here only
617 /// hides rows that were frozen anyway, which is the bug rather than a
618 /// guard against it; the freezer's own tests freeze tables whose
619 /// integer index carries no uniqueness constraint.
620 pub fn cold_capable_indices(&self) -> impl Iterator<Item = &Index> {
621 self.indices
622 .iter()
623 .filter(|i| matches!(i.kind, IndexKind::BTree(_)))
624 }
625
626 /// v6.7.0 — walk every BTree index and count `RowLocator::Cold`
627 /// entries; return the MAX across indices. The freeze path
628 /// (`freeze_oldest_to_cold`) writes cold locators to ONE
629 /// designated index — that index ends up with the full per-row
630 /// count. MAX-across-indices yields the precise count when a
631 /// PK-style index exists; for multi-index tables without a
632 /// covering index it's a lower bound (rare in practice).
633 /// Caller responsibility: only invoke under `engine.write()`
634 /// or after taking ownership; the walk is O(N) over every
635 /// (key, locator) pair.
636 #[must_use]
637 pub fn count_cold_locators(&self) -> u64 {
638 let mut best: u64 = 0;
639 for idx in &self.indices {
640 if let IndexKind::BTree(map) = &idx.kind {
641 let n: u64 = map
642 .iter()
643 .map(|(_, locs)| locs.iter().filter(|l| l.is_cold()).count() as u64)
644 .sum();
645 if n > best {
646 best = n;
647 }
648 }
649 }
650 best
651 }
652
653 pub const fn schema(&self) -> &TableSchema {
654 &self.schema
655 }
656
657 /// v6.7.2 — mutable schema accessor for ALTER TABLE paths.
658 /// Used by `Engine::exec_alter_table` to flip per-table
659 /// settings like `hot_tier_bytes`.
660 pub const fn schema_mut(&mut self) -> &mut TableSchema {
661 &mut self.schema
662 }
663
664 /// v4.39: returns the persistent row vector by reference. Callers that
665 /// used to take `&[Row]` should switch to `.iter()` (via
666 /// `IntoIterator for &PersistentVec`) or `.get(i)` for indexing.
667 pub const fn rows(&self) -> &PersistentVec<Row<'static>> {
668 &self.rows
669 }
670
671 pub const fn row_count(&self) -> usize {
672 self.rows.len()
673 }
674
675 /// v7.37.15 (Phase B) — answer "is row at `idx` visible under
676 /// `snapshot`?" without exposing the header internals to the
677 /// engine. Callers in scan paths consult this BEFORE yielding
678 /// the row.
679 ///
680 /// Defensive: out-of-bounds `idx` and the (impossible, asserted)
681 /// length mismatch return `false`, mirroring "row is not there
682 /// so it's not visible." Production scans never see either.
683 ///
684 /// Phase A always returns `true` because every header is
685 /// `RowHeader::frozen()` and `Snapshot::unbounded()` accepts
686 /// every header. The full visibility behaviour engages once
687 /// Phase C writers start stamping real `xmin`/`xmax`.
688 #[must_use]
689 pub fn is_row_visible(&self, idx: usize, snapshot: &crate::snapshot::Snapshot) -> bool {
690 match self.headers.get(idx) {
691 Some(h) => self.header_visible(idx, h, snapshot),
692 None => false,
693 }
694 }
695
696 /// v7.39 (round 486) — the visibility decision once the header is
697 /// already in hand. `scan_visible` walks rows and headers in
698 /// lockstep, so it has the header without paying a second trie
699 /// descent to look it up by index.
700 fn header_visible(
701 &self,
702 idx: usize,
703 h: &crate::row_header::RowHeader,
704 snapshot: &crate::snapshot::Snapshot,
705 ) -> bool {
706 // v7.39 (round 297, E3 Phase 1b) — `SKIP LOCKED` rides here so
707 // that every row source honours it; see `Snapshot::locked_out`.
708 if let Some((rel, set)) = &snapshot.locked_out
709 && *rel == self.rel_id
710 && set.contains(&idx)
711 {
712 return false;
713 }
714 snapshot.visible(h)
715 }
716
717 /// v7.37.15 (Phase D) — true iff every row in this table is
718 /// known-all-visible to every snapshot (frozen xmin + alive
719 /// xmax). When true, `scan_visible` skips the per-row check
720 /// entirely — the scan degenerates to a plain `rows().iter()`.
721 ///
722 /// Maintained lazily: any insert/update that stamps a non-
723 /// frozen xmin / xmax clears the cached flag; the next call to
724 /// this method recomputes by walking the header vec. The walk
725 /// is O(n) in the rare case (only when an MVCC writer ran on
726 /// this table); steady-state legacy workloads hit the cached
727 /// `true` and scan at pre-v7.37.15 speed.
728 ///
729 /// Phase D wires this into the engine's hot-tier scan
730 /// optimisation; the bit also serves the per-segment all-
731 /// visible bitmap (each cold segment is a separately tracked
732 /// `all_visible` bit, but cold segments are frozen wholesale
733 /// so they're trivially `true`).
734 #[must_use]
735 pub fn is_all_visible(&self) -> bool {
736 // Compute on the fly. Caching is a follow-up optimisation
737 // (would require &mut self or a Cell); the v7.37.15
738 // initial ship favours correctness + simplicity over the
739 // amortised constant.
740 self.headers
741 .iter()
742 .all(crate::row_header::RowHeader::is_all_visible_fast)
743 }
744
745 /// v7.37.15 (Phase B / D) — iterate over `(idx, row)` pairs whose
746 /// header is visible under `snapshot`. This is the engine-side
747 /// drop-in replacement for `for (i, r) in t.rows().iter().enumerate()`
748 /// at scan sites. The check is a single branch + atomic
749 /// register read inside the snapshot path; with `Snapshot::unbounded`
750 /// the optimiser folds the gate away.
751 ///
752 /// `'a` lifetime on `snapshot` keeps the helper zero-cost in
753 /// the hot loop — no Arc bump, no allocation.
754 /// v7.39 (round 560) — is the row at this position visible to the
755 /// snapshot? Exposed so an index-only walk can decide without
756 /// fetching the row it is deciding about.
757 #[must_use]
758 pub fn position_visible(&self, idx: usize, snapshot: &crate::snapshot::Snapshot) -> bool {
759 self.headers
760 .get(idx)
761 .is_some_and(|h| self.header_visible(idx, h, snapshot))
762 }
763
764 /// v7.39 (round 562) — the same question asked many times over
765 /// ascending positions, without descending the header trie for each
766 /// one.
767 ///
768 /// A profile of the server serving a 100k-row index-only range put
769 /// 27% of the connection thread's CPU on the per-row visibility test.
770 /// The headers are a `PersistentVec` — a 32-way trie — so
771 /// `position_visible` is four dependent pointer loads per row. A
772 /// sequential scan never pays that: it walks rows and headers in
773 /// lockstep. An index walk cannot, but its positions arrive in
774 /// ascending order and a leaf holds 32 of them, so keeping the run
775 /// between calls turns 32 descents into one.
776 ///
777 /// A position outside the held run just descends, so an index whose
778 /// order is uncorrelated with position costs what it costs today.
779 #[must_use]
780 pub fn header_runs(&self) -> HeaderRuns<'_> {
781 HeaderRuns {
782 table: self,
783 run: None,
784 }
785 }
786
787 /// v7.39 (round 559) — how many rows a snapshot sees, without
788 /// touching a single one of them.
789 ///
790 /// `count(*)` already short-circuits to `rows.len()` in the
791 /// aggregate layer, so the O(1) part was never the problem: the cost
792 /// is UPSTREAM, materialising every visible row so that layer can
793 /// take its length. `scan_visible` zips the row trie with the
794 /// headers, and a count needs only the headers.
795 ///
796 /// Measured over pgwire on 500k rows, `SELECT count(*)`:
797 ///
798 /// ```text
799 /// PG18 (2 parallel workers) 8.2 ms
800 /// PG18 (parallelism off) 10.3 ms
801 /// SPG 16.5 ms = 33 ns/row
802 /// ```
803 ///
804 /// — 1.6x slower than a single-threaded PG on the commonest
805 /// aggregate there is, which no ledger entry recorded.
806 pub fn count_visible(&self, snapshot: &crate::snapshot::Snapshot) -> usize {
807 self.note_seq_scan();
808 self.headers
809 .iter()
810 .enumerate()
811 .filter(|(i, h)| self.header_visible(*i, h, snapshot))
812 .count()
813 }
814
815 pub fn scan_visible<'a, 'b>(
816 &'a self,
817 snapshot: &'b crate::snapshot::Snapshot,
818 ) -> impl Iterator<Item = (usize, &'a Row<'static>)> + 'b
819 where
820 'a: 'b,
821 {
822 // v7.39 (pg_stat knife B) — one sequential scan; tup_read is
823 // the visible-row estimate (an early-terminating consumer —
824 // LIMIT — reads fewer; the lazy iterator can't report back).
825 // Two relaxed atomic adds per SCAN (not per row).
826 self.note_seq_scan();
827 // v7.39 (round 486) — headers ride alongside the rows instead of
828 // being looked up by index. `headers.len() == rows.len()` is an
829 // asserted invariant, so the zip drops nothing; an index lookup
830 // costs a trie descent per row, and the walk costs one per leaf.
831 self.rows
832 .iter()
833 .zip(self.headers.iter())
834 .enumerate()
835 .filter(move |(i, (_, h))| self.header_visible(*i, h, snapshot))
836 .map(|(i, (r, _))| (i, r))
837 }
838
839 /// The hot-tier slot a scan should resume at, given the last
840 /// [`RowId`](crate::row_header::RowId) it consumed and where that row
841 /// used to sit.
842 ///
843 /// Slots move. `vacuum` reclaims tombstones by rebuilding the row
844 /// vector, so every position after the first reclaimed one shifts
845 /// down — a reader that remembered a bare index would silently skip
846 /// or repeat rows. Row ids do not move: they are allocated
847 /// monotonically and never reused, which makes them the only stable
848 /// way to say "carry on after this row".
849 ///
850 /// `hint` is the position that row occupied when it was read. It is
851 /// still right whenever nothing was reclaimed under the reader, so
852 /// the check costs one lookup; the binary search is the fallback for
853 /// when it is not, and it works because appends only ever push
854 /// larger ids and reclaiming preserves their order.
855 pub fn resume_slot_after(&self, last: crate::row_header::RowId, hint: usize) -> usize {
856 if hint > 0 && self.rowids.get(hint - 1).is_some_and(|&r| r == last) {
857 return hint;
858 }
859 let (mut lo, mut hi) = (0usize, self.rowids.len());
860 while lo < hi {
861 let mid = lo + (hi - lo) / 2;
862 match self.rowids.get(mid) {
863 Some(&r) if r <= last => lo = mid + 1,
864 _ => hi = mid,
865 }
866 }
867 lo
868 }
869
870 /// The same visibility-gated walk as [`Table::scan_visible`], resuming
871 /// at hot-tier index `start`.
872 ///
873 /// A server-side cursor hands out its result in batches and has to
874 /// continue where the previous batch stopped. Restarting the walk per
875 /// batch and discarding a growing prefix would make an N-batch drain
876 /// quadratic in the row count, so the resume point is a parameter
877 /// rather than something the caller skips over.
878 ///
879 /// `start` is a hot-tier position, not a [`RowId`](crate::row_header::RowId):
880 /// callers that resume across a compaction must re-derive it, which is
881 /// why the cursor path only resumes tables with no cold segments.
882 ///
883 /// `note_seq_scan` fires only for `start == 0`. One cursor drained in
884 /// 300 batches is one sequential scan of the table, and counting it
885 /// 300 times would misreport `pg_stat_user_tables.seq_scan`.
886 pub fn scan_visible_from<'a, 'b>(
887 &'a self,
888 start: usize,
889 snapshot: &'b crate::snapshot::Snapshot,
890 ) -> impl Iterator<Item = (usize, &'a Row<'static>)> + 'b
891 where
892 'a: 'b,
893 {
894 if start == 0 {
895 self.note_seq_scan();
896 }
897 self.rows
898 .iter()
899 .zip(self.headers.iter())
900 .enumerate()
901 .skip(start)
902 .filter(move |(i, (_, h))| self.header_visible(*i, h, snapshot))
903 .map(|(i, (r, _))| (i, r))
904 }
905
906 /// v6.8.0 — exposed for the engine layer to patch
907 /// `Index::included_columns` post-creation. Could fold into
908 /// `add_index` once the engine's IF-NOT-EXISTS guard moves up,
909 /// but the patch shape is the minimal change for v6.8.0.
910 pub fn indices_mut(&mut self) -> &mut [Index] {
911 &mut self.indices
912 }
913
914 pub fn indices(&self) -> &[Index] {
915 &self.indices
916 }
917
918 /// Compute the next `AUTO_INCREMENT` value for the column at
919 /// `col_pos`. Defined as `max(existing) + 1`, falling back to `1`
920 /// when the column currently holds no integer values. NULL / non-
921 /// integer cells are skipped. Returns `None` when the column isn't
922 /// an integer type.
923 pub fn next_auto_value(&self, col_pos: usize) -> Option<i64> {
924 let ty = self.schema.columns.get(col_pos)?.ty;
925 if !matches!(ty, DataType::SmallInt | DataType::Int | DataType::BigInt) {
926 return None;
927 }
928 let mut max: Option<i64> = None;
929 for row in &self.rows {
930 match row.values.get(col_pos) {
931 Some(Value::SmallInt(n)) => {
932 let v = i64::from(*n);
933 max = Some(max.map_or(v, |m| m.max(v)));
934 }
935 Some(Value::Int(n)) => {
936 let v = i64::from(*n);
937 max = Some(max.map_or(v, |m| m.max(v)));
938 }
939 Some(Value::BigInt(n)) => {
940 max = Some(max.map_or(*n, |m| m.max(*n)));
941 }
942 _ => {}
943 }
944 }
945 // v7.39 (round 220) — `ALTER … ALTER COLUMN … RESTART [WITH n]`
946 // lifts the next allocated value to at least n (a floor over the
947 // max+1 scan). A dump-restore RESTART lands exactly on n; a
948 // backward RESTART is safely ignored (no duplicate-key landmine,
949 // unlike PG).
950 let base = max.map_or(1, |m| m + 1);
951 let floor = self
952 .schema
953 .columns
954 .get(col_pos)
955 .and_then(|c| c.auto_restart)
956 .unwrap_or(i64::MIN);
957 Some(base.max(floor))
958 }
959
960 /// Return the first index defined over `column_position`, if any.
961 /// (`v0.8` supports at most one index per column logically; the search
962 /// just picks the first match.)
963 pub fn index_on(&self, column_position: usize) -> Option<&Index> {
964 // v6.7.1 — prefer BTree (has the key→locator map needed
965 // for `lookup_eq`) over BRIN (metadata-only). When only a
966 // BRIN exists on the column, return None so the executor
967 // falls back to the hot-tier row scan instead of trying
968 // to use BRIN for an equality lookup (which would always
969 // return an empty slice and look like "no rows matched").
970 self.indices
971 .iter()
972 .find(|i| i.column_position == column_position && matches!(i.kind, IndexKind::BTree(_)))
973 .or_else(|| {
974 self.indices.iter().find(|i| {
975 i.column_position == column_position && matches!(i.kind, IndexKind::Nsw(_))
976 })
977 })
978 }
979
980 /// Insert one row after validating it matches the schema (length + type).
981 /// Returns `StorageError` on mismatch — the table is left unchanged.
982 /// Updates every defined index with the new row's key.
983 pub fn insert(&mut self, row: Row<'static>) -> Result<(), StorageError> {
984 if row.len() != self.schema.columns.len() {
985 return Err(StorageError::ArityMismatch {
986 expected: self.schema.columns.len(),
987 actual: row.len(),
988 });
989 }
990 for (i, (val, col)) in row.values.iter().zip(&self.schema.columns).enumerate() {
991 if val.is_null() {
992 if !col.nullable {
993 return Err(StorageError::NullInNotNull {
994 column: col.name.clone(),
995 });
996 }
997 continue;
998 }
999 // v7.39 (read01 round 54) — `data_type()` is None for the
1000 // eval-only variants that carry no DataType (RegClass, Composite).
1001 // They are NOT NULL, so `.expect("non-null")` PANICKED on them —
1002 // materialising a CTE like `WITH w AS (SELECT 't'::regclass)` blew
1003 // up the query with an "internal error". Report a clean type
1004 // mismatch instead; the engine coerces these before they get here
1005 // on every path that knows how.
1006 let Some(actual) = val.data_type() else {
1007 // An eval-only value (RegClass carries oid + name, Composite a
1008 // field tuple) has no DataType in the storage lattice. It is
1009 // NOT NULL, so the old `.expect("non-null")` PANICKED — which
1010 // is how `WITH w AS (SELECT 't'::regclass)` blew up with an
1011 // "internal error". Accept it: the value keeps its dual shape
1012 // and downstream comparisons (RegClass vs BigInt oid) handle it.
1013 continue;
1014 };
1015 // A Vector column needs the variant AND the dimension to
1016 // agree, which the equality inside `column_accepts` already
1017 // encodes because DataType::Vector carries the dim.
1018 let compatible = column_accepts(actual, col.ty);
1019 if !compatible {
1020 return Err(StorageError::TypeMismatch {
1021 column: col.name.clone(),
1022 expected: col.ty,
1023 actual,
1024 position: i,
1025 });
1026 }
1027 }
1028 let new_row_idx = self.rows.len();
1029 // v7.39 (round 493) — disjoint borrows: the BTree arm below reads
1030 // headers to decide which of this key's locators are dead while
1031 // holding `indices` mutably.
1032 let horizon = self.prune_horizon;
1033 let headers = &self.headers;
1034 // Pre-validate before mutating: ensure indices receive an IndexKey.
1035 // For NSW we defer the graph update to *after* the row is pushed
1036 // so the kNN search can see it in `self.rows`.
1037 for idx in &mut self.indices {
1038 match &mut idx.kind {
1039 IndexKind::BTree(map) => {
1040 if let Some(key) = IndexKey::from_value(&row.values[idx.column_position]) {
1041 // v4.40: PersistentBTreeMap has no in-place entry-or-default.
1042 // Clone-then-insert keeps the same semantics — for typical
1043 // unique-key schemas the Vec is 1-element so the clone is
1044 // O(1). For dup-heavy columns it's O(M) per insert, traded
1045 // for the structural-sharing win at clone time.
1046 //
1047 // v7.39 (round 558) — TAKE the list instead of cloning it.
1048 // `insert_mut` returns the previous value by MOVE, so the
1049 // O(M) copy the note above accepted is avoidable, and the
1050 // retain below still gets the list in hand. What that
1051 // trade cost, measured on a 50k table:
1052 //
1053 // UPDATE h SET v = 1 WHERE v <= 10000 (10k -> ONE key)
1054 // v indexed 150.6 ms v unindexed 31.2 ms
1055 // UPDATE h SET v = v + 1 WHERE v <= 10000 (distinct keys)
1056 // v indexed 34.7 ms v unindexed 32.0 ms
1057 //
1058 // 11.9 µs/row when the new keys collide against 0.27 when
1059 // they do not — 44x for the same row count, because the
1060 // k-th insert under one key copied a k-element list. Under
1061 // in-place MVCC an UPDATE appends a new row VERSION, so an
1062 // ordinary `SET flag = 'done'` over a batch lands every
1063 // one of them on the same key.
1064 let mut entries =
1065 map.insert_mut(key.clone(), Vec::new()).unwrap_or_default();
1066 // v7.39 (round 493) — drop this key's dead versions while
1067 // the list is already in hand.
1068 //
1069 // "The Vec is 1-element for unique-key schemas" is what
1070 // churn breaks: a posting list carries one locator per row
1071 // VERSION, so deleting and re-inserting the same id grows
1072 // it without bound between vacuums. Round 492 counted 61
1073 // locators under one PK by cycle 60, each costing the
1074 // uniqueness probe a header lookup, and round 490 found the
1075 // range seek walking the same versions.
1076 //
1077 // Vacuum already prunes them — by rebuilding every index,
1078 // which is why it runs rarely enough for this to matter.
1079 // Here the work is free: the list is cloned on this path
1080 // anyway and is about to be written back.
1081 //
1082 // Safety is vacuum's own argument: `prune_horizon` is the
1083 // floor of every live snapshot, so a version reclaimable
1084 // under it is invisible to every reader that exists or can
1085 // yet begin (a later snapshot's version is >= the floor).
1086 // A horizon of 0 keeps everything.
1087 // v7.39 (round 558) — AMORTISE it.
1088 //
1089 // The retain walks the whole list, so running it on
1090 // every insert is O(M) per insert and O(n²) over a
1091 // statement that puts n row versions under one key.
1092 // Measured on a 50k table, 10k rows updated:
1093 //
1094 // retain every insert off
1095 // SET v = 1 (dupes) 135.9 ms 11.7
1096 // SET v = v+1 (distinct) 31.4 ms 13.4
1097 //
1098 // and the second line has no colliding key at all —
1099 // the OTHER index (g, 100 distinct values over 50k
1100 // rows) supplies lists long enough on its own. Every
1101 // insert on every index was paying it.
1102 //
1103 // Pruning only when the list has DOUBLED keeps round
1104 // 493's bound — the list stays within 2x its pruned
1105 // size, so the seek still never walks an unbounded
1106 // version chain — while the total work over n inserts
1107 // becomes n + n/2 + n/4 + … = O(n). Skipping a prune
1108 // can only delay reclamation; it never drops a live
1109 // locator, so the safety argument in the note above is
1110 // untouched.
1111 if horizon > 0 && entries.len() > 1 && entries.len().is_power_of_two() {
1112 entries.retain(|loc| match loc {
1113 RowLocator::Hot(i) => headers.get(*i).is_none_or(|h| {
1114 !crate::vacuum::is_reclaimable(h.xmax, horizon)
1115 }),
1116 RowLocator::Cold { .. } => true,
1117 });
1118 }
1119 entries.push(RowLocator::Hot(new_row_idx));
1120 map.insert_mut(key, entries);
1121 }
1122 }
1123 IndexKind::Gin(map) => {
1124 // v7.12.3 — extend posting list per lexeme word.
1125 // NULL or non-TsVector cell → no-op (cell carries
1126 // no lexemes to index).
1127 if let Value::TsVector(lexemes) = &row.values[idx.column_position] {
1128 for lex in lexemes {
1129 if let Some(entries) = map.get_mut(&lex.word) {
1130 entries.push(RowLocator::Hot(new_row_idx));
1131 } else {
1132 map.insert_mut(
1133 lex.word.clone(),
1134 alloc::vec![RowLocator::Hot(new_row_idx)],
1135 );
1136 }
1137 }
1138 }
1139 }
1140 IndexKind::GinTrgm(map) => {
1141 // v7.15.0 — trigram GIN. Shingle the TEXT cell
1142 // into PG-compatible 3-byte trigrams and extend
1143 // each trigram's posting list.
1144 if let Value::Text(s) = &row.values[idx.column_position] {
1145 for tri in trgm::extract_trigrams(s) {
1146 // r1019 — address the String-keyed map with the borrowed
1147 // trigram; allocate one only for a key the map has never
1148 // seen, which after the first rows is rare.
1149 let key = trgm::trigram_str(&tri);
1150 if let Some(entries) = map.get_mut_by(key) {
1151 entries.push(RowLocator::Hot(new_row_idx));
1152 } else {
1153 map.insert_mut(
1154 alloc::string::ToString::to_string(key),
1155 alloc::vec![RowLocator::Hot(new_row_idx)],
1156 );
1157 }
1158 }
1159 }
1160 }
1161 IndexKind::GinFulltext(map) => {
1162 // v7.17.0 Phase 2.2 — MySQL FULLTEXT-shape
1163 // GIN over a TEXT / VARCHAR cell. Tokenise
1164 // via the storage-local `simple_lex` (same
1165 // rule as `to_tsvector('simple', text)`) and
1166 // extend each lexeme's posting list.
1167 let text_cell = match &row.values[idx.column_position] {
1168 Value::Text(s) => Some(s.as_ref()),
1169 // mysqldump-style mediumtext / longtext
1170 // land as Value::Text on insert; varchar
1171 // cells likewise. Anything else (NULL,
1172 // integer, …) contributes no lexemes.
1173 _ => None,
1174 };
1175 if let Some(s) = text_cell {
1176 for lex in fts_simple::simple_lex(s) {
1177 if let Some(entries) = map.get_mut(&lex) {
1178 entries.push(RowLocator::Hot(new_row_idx));
1179 } else {
1180 map.insert_mut(lex, alloc::vec![RowLocator::Hot(new_row_idx)]);
1181 }
1182 }
1183 }
1184 }
1185 IndexKind::GinJsonb(map) => {
1186 // v7.37.8(sentori Epic 5 P2)— real JSONB-GIN.
1187 // Extract canonical `(path, leaf)` tokens from
1188 // the cell text and extend each token's posting
1189 // list. NULL or non-Json cell contributes no
1190 // tokens(`labels @> '...'` against a NULL row
1191 // is always false so absence here is correct).
1192 let json_cell = match &row.values[idx.column_position] {
1193 Value::Json(s) => Some(s.as_ref()),
1194 _ => None,
1195 };
1196 if let Some(s) = json_cell {
1197 for tok in jsonb_gin::extract_tokens(s) {
1198 if let Some(entries) = map.get_mut(&tok) {
1199 entries.push(RowLocator::Hot(new_row_idx));
1200 } else {
1201 map.insert_mut(tok, alloc::vec![RowLocator::Hot(new_row_idx)]);
1202 }
1203 }
1204 }
1205 }
1206 // NSW handled below after the row push (so the new row
1207 // is visible to the kNN-graph connect step). BRIN
1208 // carries no per-row state.
1209 IndexKind::Nsw(_) | IndexKind::Brin { .. } => {}
1210 }
1211 }
1212 // v7.39 (round 215) — maintain the range-exclusion indexes for the
1213 // freshly-inserted row (before the move; `new_row_idx` is the slot it
1214 // will occupy). Mirrors the BTree maintenance above.
1215 if !self.excl_indexes.is_empty() {
1216 self.excl_indexes_on_insert(&row, new_row_idx);
1217 }
1218 // v5.2.1: maintain incremental hot-tier byte counter. Computed
1219 // before the move so we don't need to borrow `row` after push.
1220 self.hot_bytes = self
1221 .hot_bytes
1222 .saturating_add(row_body_encoded_len(&row, &self.schema) as u64);
1223 // v7.34 — capture the row-level redo before the row is moved in.
1224 // v7.37.15 (Epic W slice 1) — carry the stable RowId this insert
1225 // will receive. `alloc_rowid` below hands out `RowId(next_rowid)`
1226 // and bumps the counter unconditionally, so the id read here is
1227 // exactly the one the row ends up with. `writer_version` (xmin)
1228 // is 0: the writing TxId is not threaded to this layer yet (the
1229 // header pushed below is `RowHeader::frozen()`).
1230 let redo_rowid = crate::row_header::RowId(self.next_rowid);
1231 self.record_redo(|table| RowChange::Insert {
1232 table,
1233 row: row.clone(),
1234 rowid: redo_rowid,
1235 writer_version: 0,
1236 });
1237 // v4.39.1: push_mut keeps streaming inserts at Vec::push speed when
1238 // the table is uniquely owned (the spg-embedded path); inside a TX
1239 // wrap where a Catalog snapshot exists, push_mut path-copies the
1240 // tail just like push() and the snapshot stays valid.
1241 self.rows.push_mut(row);
1242 // v7.37.15 (Phase A.2) — keep `headers` lock-step with `rows`.
1243 // Phase A defaults every new insert to RowHeader::frozen() so
1244 // visibility checks against any snapshot return true; Phase C
1245 // upgrades the inserter to stamp the writing tx's xmin.
1246 self.headers
1247 .push_mut(crate::row_header::RowHeader::frozen());
1248 // v7.37.15 (Phase C.1) — allocate + push the stable RowId in
1249 // lock-step with rows/headers. Index locators still address
1250 // by physical slot at this commit; the id is additive
1251 // bookkeeping the lock table / HOT chains / WAL migrate to.
1252 let rid = self.alloc_rowid();
1253 self.rowids.push_mut(rid);
1254 // v7.37.15 (Epic W slice 1) — the id captured for the redo log
1255 // above must be the one actually assigned to the row.
1256 debug_assert_eq!(
1257 rid, redo_rowid,
1258 "redo-captured RowId must match the allocated RowId"
1259 );
1260 debug_assert_eq!(
1261 self.rows.len(),
1262 self.headers.len(),
1263 "headers must stay in lock-step with rows after insert"
1264 );
1265 debug_assert_eq!(
1266 self.rows.len(),
1267 self.rowids.len(),
1268 "rowids must stay in lock-step with rows after insert"
1269 );
1270 // NSW updates after the push so the new row is visible to the
1271 // greedy search used during connect.
1272 let new_row_idx = self.rows.len() - 1;
1273 let nsw_targets: Vec<usize> = self
1274 .indices
1275 .iter()
1276 .enumerate()
1277 .filter_map(|(i, idx)| {
1278 if matches!(idx.kind, IndexKind::Nsw(_)) {
1279 Some(i)
1280 } else {
1281 None
1282 }
1283 })
1284 .collect();
1285 for idx_pos in nsw_targets {
1286 nsw_insert_at(self, idx_pos, new_row_idx);
1287 }
1288 Ok(())
1289 }
1290
1291 /// Build a new B-tree index over the named column. Rebuilds from
1292 /// existing rows. Errors if `column_name` doesn't exist or the index
1293 /// name is taken.
1294 pub fn add_index(&mut self, name: String, column_name: &str) -> Result<(), StorageError> {
1295 if self.indices.iter().any(|i| i.name == name) {
1296 return Err(StorageError::DuplicateIndex { name });
1297 }
1298 let column_position = self.schema.column_position(column_name).ok_or_else(|| {
1299 StorageError::ColumnNotFound {
1300 column: column_name.into(),
1301 }
1302 })?;
1303 let mut idx = Index::new_btree(name, column_position);
1304 if let IndexKind::BTree(map) = &mut idx.kind {
1305 for (i, row) in self.rows.iter().enumerate() {
1306 if let Some(key) = IndexKey::from_value(&row.values[column_position]) {
1307 if let Some(entries) = map.get_mut(&key) {
1308 entries.push(RowLocator::Hot(i));
1309 } else {
1310 map.insert_mut(key, alloc::vec![RowLocator::Hot(i)]);
1311 }
1312 }
1313 }
1314 }
1315 self.indices.push(idx);
1316 Ok(())
1317 }
1318
1319 /// v7.39 (round 215) — ensure a range-exclusion index exists on
1320 /// `column_position`, building it from the current rows. Idempotent: a
1321 /// second call for the same column is a no-op. Called at CREATE TABLE /
1322 /// ALTER ADD EXCLUDE and on catalog load (rebuild-from-constraints).
1323 /// Tombstoned rows are indexed too (they are filtered by the consumer via
1324 /// `is_deleted()` at query time — the established index pattern).
1325 pub fn ensure_excl_range_index(&mut self, column_position: usize) {
1326 if self
1327 .excl_indexes
1328 .iter()
1329 .any(|e| e.column_position == column_position)
1330 {
1331 return;
1332 }
1333 let mut map: crate::PersistentBTreeMap<(i128, u8), Vec<RowLocator>> =
1334 crate::PersistentBTreeMap::new();
1335 for (i, row) in self.rows.iter().enumerate() {
1336 if let Some(v) = row.values.get(column_position)
1337 && let Some(key) = crate::range_excl_index_key(v)
1338 {
1339 if let Some(entries) = map.get_mut(&key) {
1340 entries.push(RowLocator::Hot(i));
1341 } else {
1342 map.insert_mut(key, alloc::vec![RowLocator::Hot(i)]);
1343 }
1344 }
1345 }
1346 self.excl_indexes.push(crate::ExclRangeIndex {
1347 column_position,
1348 map,
1349 });
1350 }
1351
1352 /// v7.39 (round 215) — the range-exclusion index on `column_position`, if
1353 /// one was built. The EXCLUDE enforcement path probes its
1354 /// [`predecessor`](crate::PersistentBTreeMap::predecessor) + successors to
1355 /// find candidate overlaps in O(log n).
1356 #[must_use]
1357 pub fn excl_range_index(
1358 &self,
1359 column_position: usize,
1360 ) -> Option<&crate::PersistentBTreeMap<(i128, u8), Vec<RowLocator>>> {
1361 self.excl_indexes
1362 .iter()
1363 .find(|e| e.column_position == column_position)
1364 .map(|e| &e.map)
1365 }
1366
1367 /// v7.39 (round 215) — add a freshly-appended row at `row_idx` to every
1368 /// range-exclusion index. Called from `insert` after the row is pushed,
1369 /// mirroring the BTree secondary-index maintenance.
1370 fn excl_indexes_on_insert(&mut self, row: &Row<'static>, row_idx: usize) {
1371 for ex in &mut self.excl_indexes {
1372 if let Some(v) = row.values.get(ex.column_position)
1373 && let Some(key) = crate::range_excl_index_key(v)
1374 {
1375 if let Some(entries) = ex.map.get_mut(&key) {
1376 entries.push(RowLocator::Hot(row_idx));
1377 } else {
1378 ex.map
1379 .insert_mut(key, alloc::vec![RowLocator::Hot(row_idx)]);
1380 }
1381 }
1382 }
1383 }
1384
1385 /// v7.39 (round 215) — rebuild every range-exclusion index from the
1386 /// current rows (called from `rebuild_indices`, i.e. after a physical
1387 /// compaction/delete that shifted slots). Preserves which columns are
1388 /// indexed; re-emits all `Hot` locators.
1389 fn rebuild_excl_indexes(&mut self) {
1390 let cols: Vec<usize> = self
1391 .excl_indexes
1392 .iter()
1393 .map(|e| e.column_position)
1394 .collect();
1395 self.excl_indexes.clear();
1396 for c in cols {
1397 self.ensure_excl_range_index(c);
1398 }
1399 }
1400
1401 /// Build a new NSW (HNSW-flavoured) index over the named column.
1402 /// Required for `ORDER BY col <-> literal LIMIT k` to plan as a
1403 /// graph traversal instead of a full scan. Column must be a Vector
1404 /// type. `m` is the maximum number of neighbours per node.
1405 pub fn add_nsw_index(
1406 &mut self,
1407 name: String,
1408 column_name: &str,
1409 m: usize,
1410 ) -> Result<(), StorageError> {
1411 self.add_nsw_index_inner(name, column_name, m, None)
1412 }
1413
1414 /// v6.0.4 — synchronous rebuild of the named NSW index. If
1415 /// `new_encoding` is `Some(target)` and differs from the column's
1416 /// current encoding, every stored cell at the indexed column is
1417 /// re-coded into the target encoding before the new graph
1418 /// builds. Returns `IndexNotFound` if no index by that name exists
1419 /// and `Unsupported` for non-NSW indexes (`BTree` REBUILD is a no-op
1420 /// the engine layer rejects, not a storage-level concept).
1421 ///
1422 /// Holds the caller's `&mut self` for the duration — no
1423 /// concurrency / staging / WAL-replay machinery in v6.0.4. The
1424 /// "live" optimisation lands as v6.0.4.1.
1425 pub fn rebuild_nsw_index(
1426 &mut self,
1427 name: &str,
1428 new_encoding: Option<VecEncoding>,
1429 ) -> Result<(), StorageError> {
1430 let idx_pos = self
1431 .indices
1432 .iter()
1433 .position(|i| i.name == name)
1434 .ok_or_else(|| StorageError::IndexNotFound {
1435 name: String::from(name),
1436 })?;
1437 let col_pos = self.indices[idx_pos].column_position;
1438 let m = match &self.indices[idx_pos].kind {
1439 IndexKind::Nsw(g) => g.m,
1440 IndexKind::BTree(_)
1441 | IndexKind::Brin { .. }
1442 | IndexKind::Gin(_)
1443 | IndexKind::GinTrgm(_)
1444 | IndexKind::GinFulltext(_)
1445 | IndexKind::GinJsonb(_) => {
1446 return Err(StorageError::Unsupported(format!(
1447 "ALTER INDEX REBUILD on non-NSW index {name:?} — only NSW indexes can rebuild"
1448 )));
1449 }
1450 };
1451 let col_name = self.schema.columns[col_pos].name.clone();
1452 // 1. Optional re-encoding pass. Done first so the cells
1453 // match the schema before the graph rebuild walks them.
1454 if let Some(target) = new_encoding {
1455 let current = match self.schema.columns[col_pos].ty {
1456 DataType::Vector { encoding, .. } => encoding,
1457 ref other => {
1458 return Err(StorageError::Unsupported(format!(
1459 "ALTER INDEX REBUILD WITH (encoding=…) on non-vector column type {other:?}"
1460 )));
1461 }
1462 };
1463 if target != current {
1464 let DataType::Vector { dim, .. } = self.schema.columns[col_pos].ty else {
1465 unreachable!("checked above")
1466 };
1467 let n = self.rows.len();
1468 for i in 0..n {
1469 let row = self
1470 .rows
1471 .get_mut(i)
1472 .expect("row index in bounds (we iterated up to len())");
1473 let cell = core::mem::replace(&mut row.values[col_pos], Value::Null);
1474 let recoded = recode_vector_cell(cell, target)?;
1475 row.values[col_pos] = recoded;
1476 }
1477 self.schema.columns[col_pos].ty = DataType::Vector {
1478 dim,
1479 encoding: target,
1480 };
1481 }
1482 }
1483 // 2. Drop the existing index slot + rebuild from row payload.
1484 self.indices.remove(idx_pos);
1485 self.add_nsw_index_inner(String::from(name), &col_name, m, None)?;
1486 Ok(())
1487 }
1488
1489 /// Restore an NSW index from a pre-built graph (used on
1490 /// deserialize). Skips the bulk-build pass since the topology is
1491 /// already known. Returns `DuplicateIndex` or `ColumnNotFound` on
1492 /// schema mismatch as usual.
1493 pub fn restore_nsw_index(
1494 &mut self,
1495 name: String,
1496 column_name: &str,
1497 graph: NswGraph,
1498 ) -> Result<(), StorageError> {
1499 self.add_nsw_index_inner(name, column_name, graph.m, Some(graph))
1500 }
1501
1502 /// Restore a `BTree` index from a pre-built `(IndexKey, Vec<RowLocator>)`
1503 /// map. Used by [`Catalog::deserialize`] when reading a v9 (or later)
1504 /// catalog snapshot — the map travels on disk so cold-tier locators
1505 /// survive a round-trip, instead of being rebuilt from `self.rows`
1506 /// (which would lose every Cold entry). Same error contract as
1507 /// [`Table::add_index`].
1508 pub fn restore_btree_index(
1509 &mut self,
1510 name: String,
1511 column_name: &str,
1512 map: PersistentBTreeMap<IndexKey, Vec<RowLocator>>,
1513 ) -> Result<(), StorageError> {
1514 if self.indices.iter().any(|i| i.name == name) {
1515 return Err(StorageError::DuplicateIndex { name });
1516 }
1517 let column_position = self.schema.column_position(column_name).ok_or_else(|| {
1518 StorageError::ColumnNotFound {
1519 column: column_name.into(),
1520 }
1521 })?;
1522 self.indices.push(Index {
1523 name,
1524 column_position,
1525 kind: IndexKind::BTree(map),
1526 included_columns: Vec::new(),
1527 partial_predicate: None,
1528 expression: None,
1529 is_unique: false,
1530 nulls_not_distinct: false,
1531 descending: false,
1532 nulls_first: None,
1533 collation: None,
1534 extra_column_positions: Vec::new(),
1535 });
1536 Ok(())
1537 }
1538
1539 /// v6.7.1 — public restore counterpart for BRIN indices. Used
1540 /// by `Catalog::deserialize` when a v10 snapshot carries a
1541 /// BRIN index entry. BRIN carries no in-memory data — only the
1542 /// `column_type` snapshot is restored.
1543 pub fn restore_brin_index(
1544 &mut self,
1545 name: String,
1546 column_name: &str,
1547 column_type: DataType,
1548 ) -> Result<(), StorageError> {
1549 if self.indices.iter().any(|i| i.name == name) {
1550 return Err(StorageError::DuplicateIndex { name });
1551 }
1552 let column_position = self.schema.column_position(column_name).ok_or_else(|| {
1553 StorageError::ColumnNotFound {
1554 column: column_name.into(),
1555 }
1556 })?;
1557 self.indices
1558 .push(Index::new_brin(name, column_position, column_type));
1559 Ok(())
1560 }
1561
1562 /// v6.7.1 — public CREATE INDEX counterpart for BRIN. Creates
1563 /// the index entry with a snapshot of the indexed column's
1564 /// current `DataType`.
1565 pub fn add_brin_index(&mut self, name: String, column_name: &str) -> Result<(), StorageError> {
1566 if self.indices.iter().any(|i| i.name == name) {
1567 return Err(StorageError::DuplicateIndex { name });
1568 }
1569 let column_position = self.schema.column_position(column_name).ok_or_else(|| {
1570 StorageError::ColumnNotFound {
1571 column: column_name.into(),
1572 }
1573 })?;
1574 let column_type = self.schema.columns[column_position].ty;
1575 self.indices
1576 .push(Index::new_brin(name, column_position, column_type));
1577 Ok(())
1578 }
1579
1580 /// v7.12.3 — Build a new GIN inverted index over a `tsvector`
1581 /// column. Populates posting lists from existing rows. Errors
1582 /// if the column doesn't exist, isn't `TsVector`, or the index
1583 /// name is taken.
1584 pub fn add_gin_index(&mut self, name: String, column_name: &str) -> Result<(), StorageError> {
1585 if self.indices.iter().any(|i| i.name == name) {
1586 return Err(StorageError::DuplicateIndex { name });
1587 }
1588 let column_position = self.schema.column_position(column_name).ok_or_else(|| {
1589 StorageError::ColumnNotFound {
1590 column: column_name.into(),
1591 }
1592 })?;
1593 if self.schema.columns[column_position].ty != DataType::TsVector {
1594 return Err(StorageError::Corrupt(format!(
1595 "GIN index {name:?} requires a tsvector column; \
1596 {column_name:?} is {:?}",
1597 self.schema.columns[column_position].ty
1598 )));
1599 }
1600 let mut idx = Index::new_gin(name, column_position);
1601 if let IndexKind::Gin(map) = &mut idx.kind {
1602 for (i, row) in self.rows.iter().enumerate() {
1603 if let Value::TsVector(lexemes) = &row.values[column_position] {
1604 for lex in lexemes {
1605 if let Some(entries) = map.get_mut(&lex.word) {
1606 entries.push(RowLocator::Hot(i));
1607 } else {
1608 map.insert_mut(lex.word.clone(), alloc::vec![RowLocator::Hot(i)]);
1609 }
1610 }
1611 }
1612 }
1613 }
1614 self.indices.push(idx);
1615 Ok(())
1616 }
1617
1618 /// v7.12.3 — Restore a GIN index from a deserialised snapshot.
1619 /// Mirrors [`Self::restore_btree_index`] but takes the GIN's
1620 /// `word → Vec<RowLocator>` posting-list map (already populated
1621 /// from the catalog stream) instead of an `IndexKey` map.
1622 pub fn restore_gin_index(
1623 &mut self,
1624 name: String,
1625 column_name: &str,
1626 map: PersistentBTreeMap<String, Vec<RowLocator>>,
1627 ) -> Result<(), StorageError> {
1628 if self.indices.iter().any(|i| i.name == name) {
1629 return Err(StorageError::DuplicateIndex { name });
1630 }
1631 let column_position = self.schema.column_position(column_name).ok_or_else(|| {
1632 StorageError::ColumnNotFound {
1633 column: column_name.into(),
1634 }
1635 })?;
1636 let mut idx = Index::new_gin(name, column_position);
1637 idx.kind = IndexKind::Gin(map);
1638 self.indices.push(idx);
1639 Ok(())
1640 }
1641
1642 /// v7.15.0 — `gin_trgm_ops` GIN over a TEXT column. Walks
1643 /// every row, shingles the cell into PG-compatible trigrams,
1644 /// and builds the posting-list map. NULL / non-TEXT cells
1645 /// contribute nothing (no trigrams).
1646 pub fn add_gin_trgm_index(
1647 &mut self,
1648 name: String,
1649 column_name: &str,
1650 ) -> Result<(), StorageError> {
1651 if self.indices.iter().any(|i| i.name == name) {
1652 return Err(StorageError::DuplicateIndex { name });
1653 }
1654 let column_position = self.schema.column_position(column_name).ok_or_else(|| {
1655 StorageError::ColumnNotFound {
1656 column: column_name.into(),
1657 }
1658 })?;
1659 if !matches!(
1660 self.schema.columns[column_position].ty,
1661 DataType::Text | DataType::Varchar(_)
1662 ) {
1663 return Err(StorageError::Corrupt(format!(
1664 "trigram-GIN index {name:?} requires a TEXT/VARCHAR column; \
1665 {column_name:?} is {:?}",
1666 self.schema.columns[column_position].ty
1667 )));
1668 }
1669 let mut idx = Index::new_gin_trgm(name, column_position);
1670 if let IndexKind::GinTrgm(map) = &mut idx.kind {
1671 for (i, row) in self.rows.iter().enumerate() {
1672 if let Value::Text(s) = &row.values[column_position] {
1673 for tri in trgm::extract_trigrams(s) {
1674 // r1019 — address the String-keyed map with the borrowed
1675 // trigram; allocate one only for a key the map has never
1676 // seen, which after the first rows is rare.
1677 let key = trgm::trigram_str(&tri);
1678 if let Some(entries) = map.get_mut_by(key) {
1679 entries.push(RowLocator::Hot(i));
1680 } else {
1681 map.insert_mut(
1682 alloc::string::ToString::to_string(key),
1683 alloc::vec![RowLocator::Hot(i)],
1684 );
1685 }
1686 }
1687 }
1688 }
1689 }
1690 self.indices.push(idx);
1691 Ok(())
1692 }
1693
1694 /// v7.15.0 — restore a trigram-GIN from its catalog snapshot
1695 /// payload. Mirrors [`Self::restore_gin_index`].
1696 pub fn restore_gin_trgm_index(
1697 &mut self,
1698 name: String,
1699 column_name: &str,
1700 map: PersistentBTreeMap<String, Vec<RowLocator>>,
1701 ) -> Result<(), StorageError> {
1702 if self.indices.iter().any(|i| i.name == name) {
1703 return Err(StorageError::DuplicateIndex { name });
1704 }
1705 let column_position = self.schema.column_position(column_name).ok_or_else(|| {
1706 StorageError::ColumnNotFound {
1707 column: column_name.into(),
1708 }
1709 })?;
1710 let mut idx = Index::new_gin_trgm(name, column_position);
1711 idx.kind = IndexKind::GinTrgm(map);
1712 self.indices.push(idx);
1713 Ok(())
1714 }
1715
1716 /// v7.17.0 Phase 2.2 — MySQL `FULLTEXT KEY` GIN over a TEXT
1717 /// column. Walks every row, tokenises the cell into lower-
1718 /// cased word lexemes (`fts_simple::simple_lex` — same rule
1719 /// as `to_tsvector('simple', text)`), and builds the
1720 /// posting-list map. NULL / non-TEXT cells contribute
1721 /// nothing (no lexemes).
1722 pub fn add_gin_fulltext_index(
1723 &mut self,
1724 name: String,
1725 column_name: &str,
1726 ) -> Result<(), StorageError> {
1727 if self.indices.iter().any(|i| i.name == name) {
1728 return Err(StorageError::DuplicateIndex { name });
1729 }
1730 let column_position = self.schema.column_position(column_name).ok_or_else(|| {
1731 StorageError::ColumnNotFound {
1732 column: column_name.into(),
1733 }
1734 })?;
1735 if !matches!(
1736 self.schema.columns[column_position].ty,
1737 DataType::Text | DataType::Varchar(_)
1738 ) {
1739 return Err(StorageError::Corrupt(format!(
1740 "fulltext-GIN index {name:?} requires a TEXT/VARCHAR column; \
1741 {column_name:?} is {:?}",
1742 self.schema.columns[column_position].ty
1743 )));
1744 }
1745 let mut idx = Index::new_gin_fulltext(name, column_position);
1746 if let IndexKind::GinFulltext(map) = &mut idx.kind {
1747 for (i, row) in self.rows.iter().enumerate() {
1748 if let Value::Text(s) = &row.values[column_position] {
1749 for lex in fts_simple::simple_lex(s) {
1750 if let Some(entries) = map.get_mut(&lex) {
1751 entries.push(RowLocator::Hot(i));
1752 } else {
1753 map.insert_mut(lex, alloc::vec![RowLocator::Hot(i)]);
1754 }
1755 }
1756 }
1757 }
1758 }
1759 self.indices.push(idx);
1760 Ok(())
1761 }
1762
1763 /// v7.17.0 Phase 2.2 — restore a fulltext-GIN from its
1764 /// catalog snapshot payload. Mirrors
1765 /// [`Self::restore_gin_trgm_index`].
1766 pub fn restore_gin_fulltext_index(
1767 &mut self,
1768 name: String,
1769 column_name: &str,
1770 map: PersistentBTreeMap<String, Vec<RowLocator>>,
1771 ) -> Result<(), StorageError> {
1772 if self.indices.iter().any(|i| i.name == name) {
1773 return Err(StorageError::DuplicateIndex { name });
1774 }
1775 let column_position = self.schema.column_position(column_name).ok_or_else(|| {
1776 StorageError::ColumnNotFound {
1777 column: column_name.into(),
1778 }
1779 })?;
1780 let mut idx = Index::new_gin_fulltext(name, column_position);
1781 idx.kind = IndexKind::GinFulltext(map);
1782 self.indices.push(idx);
1783 Ok(())
1784 }
1785
1786 /// v7.37.8(sentori Epic 5 P2)— JSONB-GIN over a `Json` /
1787 /// `Jsonb` column. Walks every row, extracts canonical
1788 /// `(path, leaf)` tokens via
1789 /// [`crate::jsonb_gin::extract_tokens`], and builds the
1790 /// posting-list map. NULL or non-Json cells contribute no
1791 /// tokens(`<col> @> <jsonb>` against a NULL row is always
1792 /// false so absence here is correct).
1793 pub fn add_gin_jsonb_index(
1794 &mut self,
1795 name: String,
1796 column_name: &str,
1797 ) -> Result<(), StorageError> {
1798 if self.indices.iter().any(|i| i.name == name) {
1799 return Err(StorageError::DuplicateIndex { name });
1800 }
1801 let column_position = self.schema.column_position(column_name).ok_or_else(|| {
1802 StorageError::ColumnNotFound {
1803 column: column_name.into(),
1804 }
1805 })?;
1806 if !matches!(
1807 self.schema.columns[column_position].ty,
1808 DataType::Json | DataType::Jsonb
1809 ) {
1810 return Err(StorageError::Corrupt(format!(
1811 "JSONB-GIN index {name:?} requires a JSON/JSONB column; \
1812 {column_name:?} is {:?}",
1813 self.schema.columns[column_position].ty
1814 )));
1815 }
1816 let mut idx = Index::new_gin_jsonb(name, column_position);
1817 if let IndexKind::GinJsonb(map) = &mut idx.kind {
1818 for (i, row) in self.rows.iter().enumerate() {
1819 if let Value::Json(s) = &row.values[column_position] {
1820 for tok in jsonb_gin::extract_tokens(s) {
1821 if let Some(entries) = map.get_mut(&tok) {
1822 entries.push(RowLocator::Hot(i));
1823 } else {
1824 map.insert_mut(tok, alloc::vec![RowLocator::Hot(i)]);
1825 }
1826 }
1827 }
1828 }
1829 }
1830 self.indices.push(idx);
1831 Ok(())
1832 }
1833
1834 /// v7.37.8 — restore a JSONB-GIN from its catalog snapshot
1835 /// payload. Mirrors [`Self::restore_gin_fulltext_index`].
1836 pub fn restore_gin_jsonb_index(
1837 &mut self,
1838 name: String,
1839 column_name: &str,
1840 map: PersistentBTreeMap<String, Vec<RowLocator>>,
1841 ) -> Result<(), StorageError> {
1842 if self.indices.iter().any(|i| i.name == name) {
1843 return Err(StorageError::DuplicateIndex { name });
1844 }
1845 let column_position = self.schema.column_position(column_name).ok_or_else(|| {
1846 StorageError::ColumnNotFound {
1847 column: column_name.into(),
1848 }
1849 })?;
1850 let mut idx = Index::new_gin_jsonb(name, column_position);
1851 idx.kind = IndexKind::GinJsonb(map);
1852 self.indices.push(idx);
1853 Ok(())
1854 }
1855
1856 /// v5.1: register cold-tier locators on a `BTree` index. Used
1857 /// after [`Catalog::load_segment_bytes`] to wire every cold-
1858 /// tier row's PK back to its segment so
1859 /// [`Catalog::lookup_by_pk`] can resolve it. Each call
1860 /// appends to the index — keys that already have hot or cold
1861 /// locators keep them. Returns the number of locators
1862 /// registered.
1863 ///
1864 /// Pre-v5.2 (freezer) this is the only path that adds Cold
1865 /// variants to a PB; post-freezer the background freezer
1866 /// thread produces these as a batch under the engine write
1867 /// lock and this API becomes its in-memory primitive.
1868 ///
1869 /// Errors if `index_name` doesn't exist or names an NSW graph
1870 /// (NSW indices don't carry per-key row locators — they're
1871 /// vector-search structures).
1872 pub fn register_cold_locators<I>(
1873 &mut self,
1874 index_name: &str,
1875 locators: I,
1876 ) -> Result<usize, StorageError>
1877 where
1878 I: IntoIterator<Item = (IndexKey, RowLocator)>,
1879 {
1880 let idx = self
1881 .indices
1882 .iter_mut()
1883 .find(|i| i.name == index_name)
1884 .ok_or_else(|| StorageError::Corrupt(format!("index {index_name:?} not found")))?;
1885 let map = match &mut idx.kind {
1886 IndexKind::BTree(map) => map,
1887 IndexKind::Nsw(_)
1888 | IndexKind::Brin { .. }
1889 | IndexKind::Gin(_)
1890 | IndexKind::GinTrgm(_)
1891 | IndexKind::GinFulltext(_)
1892 | IndexKind::GinJsonb(_) => {
1893 return Err(StorageError::Corrupt(format!(
1894 "index {index_name:?} is not BTree; cold locators apply only to BTree indices"
1895 )));
1896 }
1897 };
1898 let mut count = 0usize;
1899 for (key, locator) in locators {
1900 if let Some(entries) = map.get_mut(&key) {
1901 entries.push(locator);
1902 } else {
1903 map.insert_mut(key, alloc::vec![locator]);
1904 }
1905 count += 1;
1906 }
1907 Ok(count)
1908 }
1909
1910 /// v7.12.3 — GIN-side parallel to [`Self::register_cold_locators`].
1911 /// Re-attaches `word → cold RowLocator` posting-list entries after
1912 /// the from-rows rebuild loop. Errors when the index doesn't
1913 /// exist or isn't a GIN. Both tsvector-GIN and trigram-GIN
1914 /// variants share posting-list shape (`String → Vec<RowLocator>`),
1915 /// so this helper accepts either.
1916 pub fn register_gin_cold_locators<I>(
1917 &mut self,
1918 index_name: &str,
1919 locators: I,
1920 ) -> Result<usize, StorageError>
1921 where
1922 I: IntoIterator<Item = (String, RowLocator)>,
1923 {
1924 let idx = self
1925 .indices
1926 .iter_mut()
1927 .find(|i| i.name == index_name)
1928 .ok_or_else(|| StorageError::Corrupt(format!("index {index_name:?} not found")))?;
1929 let map = match &mut idx.kind {
1930 // v7.17.0 Phase 2.2 — fulltext-GIN posting lists are
1931 // shape-compatible with tsvector / trigram GINs, so
1932 // cold-locator re-attach handles all three.
1933 // v7.37.8 — JSONB-GIN shares the same posting-list shape,
1934 // so it joins the same re-attach path.
1935 IndexKind::Gin(map)
1936 | IndexKind::GinTrgm(map)
1937 | IndexKind::GinFulltext(map)
1938 | IndexKind::GinJsonb(map) => map,
1939 IndexKind::BTree(_) | IndexKind::Nsw(_) | IndexKind::Brin { .. } => {
1940 return Err(StorageError::Corrupt(format!(
1941 "register_gin_cold_locators: index {index_name:?} is not GIN"
1942 )));
1943 }
1944 };
1945 let mut count = 0usize;
1946 for (word, locator) in locators {
1947 if let Some(entries) = map.get_mut(&word) {
1948 entries.push(locator);
1949 } else {
1950 map.insert_mut(word, alloc::vec![locator]);
1951 }
1952 count += 1;
1953 }
1954 Ok(count)
1955 }
1956
1957 /// v5.2.3: remove every `Cold` locator currently registered on
1958 /// `index_name` under the given `key`. `Hot` locators for the
1959 /// same key are left in place — useful when a row has just been
1960 /// promoted hot-side and the caller wants the old Cold pointer
1961 /// retired without losing the new hot entry.
1962 ///
1963 /// Returns the number of cold locators removed (0 when the key
1964 /// has only hot entries or the key isn't present at all).
1965 /// Errors when the index doesn't exist or isn't a `BTree`.
1966 pub fn remove_cold_locators_for_key(
1967 &mut self,
1968 index_name: &str,
1969 key: &IndexKey,
1970 ) -> Result<usize, StorageError> {
1971 let idx = self
1972 .indices
1973 .iter_mut()
1974 .find(|i| i.name == index_name)
1975 .ok_or_else(|| {
1976 StorageError::Corrupt(format!(
1977 "remove_cold_locators_for_key: index {index_name:?} not found"
1978 ))
1979 })?;
1980 let map = match &mut idx.kind {
1981 IndexKind::BTree(map) => map,
1982 IndexKind::Nsw(_)
1983 | IndexKind::Brin { .. }
1984 | IndexKind::Gin(_)
1985 | IndexKind::GinTrgm(_)
1986 | IndexKind::GinFulltext(_)
1987 | IndexKind::GinJsonb(_) => {
1988 return Err(StorageError::Corrupt(format!(
1989 "remove_cold_locators_for_key: index {index_name:?} is not BTree; \
1990 cold locators apply only to BTree indices"
1991 )));
1992 }
1993 };
1994 let Some(entries) = map.get(key) else {
1995 return Ok(0);
1996 };
1997 let mut kept: Vec<RowLocator> =
1998 entries.iter().copied().filter(RowLocator::is_hot).collect();
1999 let removed = entries.len() - kept.len();
2000 if removed == 0 {
2001 return Ok(0);
2002 }
2003 kept.shrink_to_fit();
2004 // PersistentBTreeMap has no remove API in v5.2; when every
2005 // locator for `key` was Cold, the key keeps an empty Vec
2006 // entry. `Index::lookup_eq` already treats `Some(&[])` and
2007 // `None` as the same empty slice (via `Vec::as_slice`), so
2008 // callers can't distinguish the two. The space cost is one
2009 // empty Vec per shadowed-then-promoted key — bounded and
2010 // recoverable when the future compaction job lands.
2011 map.insert_mut(key.clone(), kept);
2012 Ok(removed)
2013 }
2014
2015 /// v7.13.0 — append a new column to the schema and back-fill
2016 /// every existing row with `fill_value`. Used by the engine's
2017 /// `ALTER TABLE t ADD COLUMN …` handler (mailrs round-5 G1).
2018 /// Indices on existing columns keep working — column positions
2019 /// don't shift since the new column lands at the end — so no
2020 /// index rebuild is needed.
2021 pub fn add_column(&mut self, col: ColumnSchema, fill_value: Value<'static>) {
2022 self.schema.columns.push(col);
2023 let mut new_rows: PersistentVec<Row<'static>> = PersistentVec::new();
2024 for row in self.rows.iter() {
2025 let mut values = row.values.clone();
2026 values.push(fill_value.clone());
2027 new_rows.push_mut(Row::new(values));
2028 }
2029 self.rows = new_rows;
2030 }
2031
2032 /// v7.15.0 — replace the partial-index predicate source on
2033 /// the index at slot `idx`. Used by `ALTER TABLE … RENAME
2034 /// COLUMN` after the engine rewrites column-identifier
2035 /// references in the predicate source text. Pure metadata
2036 /// edit; index rows are unaffected (they're keyed by
2037 /// column position, not predicate text).
2038 pub fn set_partial_predicate(&mut self, idx: usize, pred: Option<String>) {
2039 debug_assert!(idx < self.indices.len());
2040 self.indices[idx].partial_predicate = pred;
2041 }
2042
2043 /// v7.15.0 — rename the column at `col_pos` to `new_name`.
2044 /// The on-disk row encoding is positional, so no row rewrite
2045 /// is needed; only the schema's column name changes. Indices,
2046 /// UCs, FKs all key off column positions and are unaffected.
2047 /// Source-text references that hold the column name (CHECK
2048 /// predicates, partial-index predicates, runtime DEFAULT
2049 /// expressions, trigger `UPDATE OF` lists) are rewritten by
2050 /// the engine before this helper is called — the storage
2051 /// layer doesn't depend on `spg-sql` and so can't re-parse the
2052 /// predicate sources itself.
2053 pub fn rename_column(&mut self, col_pos: usize, new_name: &str) {
2054 debug_assert!(col_pos < self.schema.columns.len());
2055 self.schema.columns[col_pos].name = new_name.to_string();
2056 }
2057
2058 /// v7.13.3 — drop the column at `col_pos`. Removes the entry
2059 /// from the schema, the value from every row, any index that
2060 /// references the column (pure drop, not shift), and shifts
2061 /// every remaining index/UC/FK column position that pointed
2062 /// past `col_pos` down by one. Used by `ALTER TABLE t DROP
2063 /// COLUMN <c>` (mailrs round-7 S8). FK dependents on this
2064 /// column must already have been removed by the caller (CASCADE
2065 /// path); the helper assumes only same-column index removal is
2066 /// needed.
2067 pub fn drop_column(&mut self, col_pos: usize) {
2068 debug_assert!(col_pos < self.schema.columns.len());
2069 // v7.39 (round 215) — dropping a column shifts every later column's
2070 // position, which would leave a range-exclusion index pointing at the
2071 // wrong column. Drop the indexes rather than risk a silent-wrong
2072 // probe; enforce falls back to the correct O(n) scan until they are
2073 // rebuilt (`ensure_excl_range_index` from the constraint's updated
2074 // column position).
2075 self.excl_indexes.clear();
2076 // Strip the column from the schema.
2077 self.schema.columns.remove(col_pos);
2078 // Rewrite every row to omit the cell at col_pos.
2079 let mut new_rows: PersistentVec<Row> = PersistentVec::new();
2080 for row in self.rows.iter() {
2081 let mut values = row.values.clone();
2082 if col_pos < values.len() {
2083 values.remove(col_pos);
2084 }
2085 new_rows.push_mut(Row::new(values));
2086 }
2087 self.rows = new_rows;
2088 // Drop indices on the column outright; shift the rest.
2089 self.indices.retain(|idx| idx.column_position != col_pos);
2090 for idx in &mut self.indices {
2091 if idx.column_position > col_pos {
2092 idx.column_position -= 1;
2093 }
2094 // Same shift for any included-columns reference.
2095 for inc in &mut idx.included_columns {
2096 if *inc > col_pos {
2097 *inc -= 1;
2098 }
2099 }
2100 }
2101 // Shift uniqueness-constraint column positions (and drop
2102 // entries that lose all columns, though that shouldn't
2103 // happen in practice — caller has already CASCADE-removed
2104 // FKs and there's no general CASCADE for UCs).
2105 let mut surviving_ucs: Vec<UniquenessConstraint> = Vec::new();
2106 for mut uc in core::mem::take(&mut self.schema.uniqueness_constraints) {
2107 uc.columns.retain(|&c| c != col_pos);
2108 if uc.columns.is_empty() {
2109 continue;
2110 }
2111 for c in &mut uc.columns {
2112 if *c > col_pos {
2113 *c -= 1;
2114 }
2115 }
2116 surviving_ucs.push(uc);
2117 }
2118 self.schema.uniqueness_constraints = surviving_ucs;
2119 // Shift FK local_columns (parent-pointing column positions
2120 // are off-table and untouched).
2121 for fk in &mut self.schema.foreign_keys {
2122 for c in &mut fk.local_columns {
2123 if *c > col_pos {
2124 *c -= 1;
2125 }
2126 }
2127 }
2128 // Rebuild remaining indices' payload — the column-position
2129 // shift means existing IndexKey entries are still keyed by
2130 // the same column data but the position numbers changed;
2131 // existing key→locator maps stay valid because they're
2132 // keyed by Value not position. The rebuild is conservative
2133 // — same pattern delete_rows uses post-mutation.
2134 self.rebuild_indices();
2135 }
2136
2137 /// v4.4: delete the rows at the given positions in one pass.
2138 /// `positions` must be unique; ordering doesn't matter. Indices
2139 /// are rebuilt from scratch (cheaper than tracking incremental
2140 /// shifts across both B-tree and NSW). Returns the number of
2141 /// rows removed.
2142 /// v7.17.0 Phase 1.3 — wipe every row. Used by REFRESH
2143 /// MATERIALIZED VIEW; same effect as `delete_rows((0..N).into())`
2144 /// but skips the per-position bookkeeping for the all-removed
2145 /// fast path. Indices are rebuilt (empty).
2146 pub fn truncate(&mut self) {
2147 self.rows = PersistentVec::new();
2148 // v7.37.15 (Phase A.2) — keep headers lock-step.
2149 self.headers = PersistentVec::new();
2150 // v7.37.15 (Phase C.1) — clear rowids lock-step. `next_rowid`
2151 // is NOT reset: ids stay globally monotonic within the
2152 // relation so a post-truncate insert never reuses a pre-
2153 // truncate id that a stale reference might still name.
2154 self.rowids = PersistentVec::new();
2155 self.hot_bytes = 0;
2156 self.rebuild_indices();
2157 }
2158
2159 pub fn delete_rows(&mut self, positions: &[usize]) -> usize {
2160 // v7.37.15 (Epic W slice 1) — capture the RowIds of the targeted
2161 // rows BEFORE the deletion shifts them out. One id per input
2162 // position (parallel to `positions`), `RowId::UNASSIGNED` for an
2163 // out-of-bounds position. Only pay for it when redo capture is
2164 // on. `writer_version` (xmax) is 0: the deleting TxId is not
2165 // threaded to this layer yet.
2166 let redo_rowids: Vec<crate::row_header::RowId> = if self.redo_log.is_some() {
2167 positions
2168 .iter()
2169 .map(|&p| {
2170 self.rowids()
2171 .get(p)
2172 .copied()
2173 .unwrap_or(crate::row_header::RowId::UNASSIGNED)
2174 })
2175 .collect()
2176 } else {
2177 Vec::new()
2178 };
2179 let removed = self.delete_rows_no_index(positions);
2180 if removed > 0 {
2181 self.rebuild_indices();
2182 // v7.34 — capture row-level redo. Record the input positions
2183 // (replay's `delete_rows` dedups + bounds-filters identically);
2184 // skip a no-op delete so the log stays minimal.
2185 self.record_redo(move |table| RowChange::Delete {
2186 table,
2187 positions: positions.to_vec(),
2188 rowids: redo_rowids,
2189 writer_version: 0,
2190 });
2191 }
2192 removed
2193 }
2194
2195 /// v7.37.5 (mailrs crash-recovery Ask 3) — row-only delete for the
2196 /// WAL-replay batch path: removes the rows + decrements `hot_bytes`,
2197 /// **does NOT** call `rebuild_indices()` and does **NOT** capture
2198 /// redo. The caller is responsible for invoking `rebuild_indices_pub`
2199 /// once after a sequence of `*_no_index` mutations on this table.
2200 /// Skipping the per-call rebuild closes the
2201 /// O(records × rows × indices × log rows) replay blow-up
2202 /// (5000 DELETEs × 100k × 13 × ln 100k ≈ minutes → seconds).
2203 /// Returns the number of rows actually removed (dedup + bounds-
2204 /// filtered identically to `delete_rows`).
2205 pub fn delete_rows_no_index(&mut self, positions: &[usize]) -> usize {
2206 if positions.is_empty() {
2207 return 0;
2208 }
2209 // Mark positions; v4.39: PV has no in-place retain, so we rebuild
2210 // a fresh PV by pushing the survivors. Still O(n log₃₂ n); the
2211 // structural-sharing win shows up at `Catalog::clone()`, not here.
2212 let mut to_remove = alloc::vec![false; self.rows.len()];
2213 let mut removed = 0;
2214 for &p in positions {
2215 if p < to_remove.len() && !to_remove[p] {
2216 to_remove[p] = true;
2217 removed += 1;
2218 }
2219 }
2220 if removed == 0 {
2221 return 0;
2222 }
2223 let mut new_rows: PersistentVec<Row> = PersistentVec::new();
2224 let mut new_headers: PersistentVec<crate::row_header::RowHeader> = PersistentVec::new();
2225 // v7.37.15 (Phase C.1) — survivors carry their stable RowId
2226 // across the compaction so a held lock / redo reference keeps
2227 // naming the same row while its physical slot shifts down.
2228 let mut new_rowids: PersistentVec<crate::row_header::RowId> = PersistentVec::new();
2229 let mut removed_bytes: u64 = 0;
2230 // v7.37.16 (autovacuum) — recount dead survivors: this rebuild
2231 // is the compaction hub (vacuum and physical delete both land
2232 // here), so the incremental counter re-bases exactly.
2233 let mut surviving_dead: u64 = 0;
2234 for (i, row) in self.rows.iter().enumerate() {
2235 if to_remove[i] {
2236 removed_bytes =
2237 removed_bytes.saturating_add(row_body_encoded_len(row, &self.schema) as u64);
2238 } else {
2239 new_rows.push_mut(row.clone());
2240 // v7.37.15 (Phase A.2) — keep headers lock-step.
2241 // Phase C will stamp xmax with the deleting tx's
2242 // id INSTEAD of physically dropping the row; Phase
2243 // A.2 keeps physical-delete semantics so
2244 // serialisation + WAL paths stay identical.
2245 if let Some(h) = self.headers.get(i) {
2246 if h.xmax != crate::row_header::XMAX_ALIVE {
2247 surviving_dead += 1;
2248 }
2249 new_headers.push_mut(*h);
2250 } else {
2251 new_headers.push_mut(crate::row_header::RowHeader::frozen());
2252 }
2253 if let Some(rid) = self.rowids.get(i) {
2254 new_rowids.push_mut(*rid);
2255 } else {
2256 // Should not happen once C.1 is wired everywhere;
2257 // allocate a fresh id as a defensive fallback so
2258 // the lock-step invariant survives a legacy path.
2259 let rid = crate::row_header::RowId(self.next_rowid);
2260 self.next_rowid += 1;
2261 new_rowids.push_mut(rid);
2262 }
2263 }
2264 }
2265 self.rows = new_rows;
2266 self.headers = new_headers;
2267 self.rowids = new_rowids;
2268 self.hot_bytes = self.hot_bytes.saturating_sub(removed_bytes);
2269 self.dead_rows = surviving_dead;
2270 debug_assert_eq!(
2271 self.rows.len(),
2272 self.headers.len(),
2273 "headers must stay in lock-step with rows after delete_rows_no_index"
2274 );
2275 removed
2276 }
2277
2278 /// v7.37.5 — public alias for the private `rebuild_indices` helper.
2279 /// Used by `Catalog::apply_redo` to coalesce per-record rebuilds
2280 /// across a batch of `RowChange`s into one rebuild per touched table.
2281 pub fn rebuild_indices_pub(&mut self) {
2282 self.rebuild_indices();
2283 }
2284
2285 /// v7.37.5 (mailrs crash-recovery Ask 3) — replace the table's
2286 /// row vector + `hot_bytes` in one shot, then rebuild every
2287 /// index from the new rows. Used by `Catalog::apply_redo`'s
2288 /// batched run: a contiguous slice of `RowChange`s targeting
2289 /// this table is composed into a final `(PersistentVec<Row>,
2290 /// hot_bytes)` pair via in-memory bookkeeping, then handed to
2291 /// this method ONCE for index regeneration. Replaces N per-
2292 /// record `rebuild_indices` calls with 1 per run.
2293 /// v7.39 (flip crash-replay P0) — like
2294 /// [`Self::set_rows_and_rebuild_indices`] but KEEPS the caller's
2295 /// per-slot RowIds. Redo replay applies one WAL record per
2296 /// statement; reassigning ids between records broke every later
2297 /// record's tombstone targets (they name the ids the crashed
2298 /// process allocated), resurrecting deleted rows. The id
2299 /// allocator advances past every preserved id so post-replay
2300 /// inserts never collide.
2301 pub fn set_rows_and_rebuild_indices_with_rowids(
2302 &mut self,
2303 new_rows: PersistentVec<Row<'static>>,
2304 new_hot_bytes: u64,
2305 rowids: &[crate::row_header::RowId],
2306 headers: &[crate::row_header::RowHeader],
2307 ) {
2308 debug_assert_eq!(new_rows.len(), rowids.len());
2309 debug_assert_eq!(new_rows.len(), headers.len());
2310 let mut new_headers: PersistentVec<crate::row_header::RowHeader> = PersistentVec::new();
2311 let mut new_rowids: PersistentVec<crate::row_header::RowId> = PersistentVec::new();
2312 let mut dead: u64 = 0;
2313 for (rid, h) in rowids.iter().zip(headers) {
2314 // Preserve the caller's header — an earlier replayed WAL
2315 // record's tombstone stamp must survive this record's
2316 // rebuild (per-statement replay re-freezing every header
2317 // resurrected every previously-deleted row).
2318 if h.xmax != crate::row_header::XMAX_ALIVE {
2319 dead += 1;
2320 }
2321 new_headers.push_mut(*h);
2322 let rid = if *rid == crate::row_header::RowId::UNASSIGNED {
2323 let fresh = crate::row_header::RowId(self.next_rowid);
2324 self.next_rowid += 1;
2325 fresh
2326 } else {
2327 if rid.0 >= self.next_rowid {
2328 self.next_rowid = rid.0 + 1;
2329 }
2330 *rid
2331 };
2332 new_rowids.push_mut(rid);
2333 }
2334 self.rows = new_rows;
2335 self.headers = new_headers;
2336 self.rowids = new_rowids;
2337 self.hot_bytes = new_hot_bytes;
2338 self.dead_rows = dead;
2339 debug_assert_eq!(self.rows.len(), self.headers.len());
2340 debug_assert_eq!(self.rows.len(), self.rowids.len());
2341 self.rebuild_indices();
2342 }
2343
2344 pub fn set_rows_and_rebuild_indices(
2345 &mut self,
2346 new_rows: PersistentVec<Row<'static>>,
2347 new_hot_bytes: u64,
2348 ) {
2349 // v7.37.15 (Phase A.2) — synthesise frozen headers for
2350 // the replacement rows. Phase D's catalog snapshot format
2351 // (bumped to V6) will start carrying headers verbatim,
2352 // letting recovery preserve real xmin/xmax instead of
2353 // freezing everything; until then frozen is the safe
2354 // default for replay (all visible to every snapshot).
2355 let mut new_headers: PersistentVec<crate::row_header::RowHeader> = PersistentVec::new();
2356 // v7.37.15 (Phase C.1) — fresh monotonic ids for the
2357 // replacement rows drawn from the relation allocator, so a
2358 // post-replay id never collides with a pre-replay one.
2359 let mut new_rowids: PersistentVec<crate::row_header::RowId> = PersistentVec::new();
2360 for _ in 0..new_rows.len() {
2361 new_headers.push_mut(crate::row_header::RowHeader::frozen());
2362 let rid = crate::row_header::RowId(self.next_rowid);
2363 self.next_rowid += 1;
2364 new_rowids.push_mut(rid);
2365 }
2366 self.rows = new_rows;
2367 self.headers = new_headers;
2368 self.rowids = new_rowids;
2369 self.hot_bytes = new_hot_bytes;
2370 // All-frozen replacement headers → no dead rows by construction.
2371 self.dead_rows = 0;
2372 debug_assert_eq!(
2373 self.rows.len(),
2374 self.headers.len(),
2375 "headers must stay in lock-step with rows after set_rows_and_rebuild_indices"
2376 );
2377 debug_assert_eq!(
2378 self.rows.len(),
2379 self.rowids.len(),
2380 "rowids must stay in lock-step with rows after set_rows_and_rebuild_indices"
2381 );
2382 self.rebuild_indices();
2383 }
2384
2385 /// v7.37.5 (mailrs crash-recovery Ask 3) — row-only insert for the
2386 /// WAL-replay batch path: pushes the row + bumps `hot_bytes`, and
2387 /// **does NOT** update any index (B-tree, GIN, NSW). The caller is
2388 /// responsible for invoking `rebuild_indices_pub` once after a
2389 /// sequence of `*_no_index` mutations on this table.
2390 /// Schema validation (arity + per-column type compatibility) is
2391 /// applied so a malformed redo log surfaces honestly.
2392 pub fn insert_no_index(&mut self, row: Row<'static>) -> Result<(), StorageError> {
2393 if row.len() != self.schema.columns.len() {
2394 return Err(StorageError::ArityMismatch {
2395 expected: self.schema.columns.len(),
2396 actual: row.len(),
2397 });
2398 }
2399 validate_row_against_schema(&row.values, &self.schema)?;
2400 self.hot_bytes = self
2401 .hot_bytes
2402 .saturating_add(row_body_encoded_len(&row, &self.schema) as u64);
2403 self.rows.push_mut(row);
2404 // v7.37.15 (Phase A.2) — keep headers lock-step for the
2405 // WAL replay path. Replay-time headers are frozen because
2406 // pre-V6 envelopes carry no header info; Phase D will
2407 // restore the original xmin/xmax once the V6 catalog
2408 // format ships.
2409 self.headers
2410 .push_mut(crate::row_header::RowHeader::frozen());
2411 // v7.37.15 (Phase C.1) — RowId lock-step for the WAL-replay
2412 // append path.
2413 let rid = self.alloc_rowid();
2414 self.rowids.push_mut(rid);
2415 debug_assert_eq!(
2416 self.rows.len(),
2417 self.headers.len(),
2418 "headers must stay in lock-step with rows after insert_no_index"
2419 );
2420 debug_assert_eq!(
2421 self.rows.len(),
2422 self.rowids.len(),
2423 "rowids must stay in lock-step with rows after insert_no_index"
2424 );
2425 Ok(())
2426 }
2427
2428 /// v7.37.5 (mailrs crash-recovery Ask 3) — row-only update for the
2429 /// WAL-replay batch path: replaces the row at `position` + adjusts
2430 /// `hot_bytes`, and **does NOT** touch any index. Skipping the
2431 /// per-update incremental index work is safe because the trailing
2432 /// `rebuild_indices_pub` regenerates indices from `self.rows` in
2433 /// their final state.
2434 pub fn update_row_no_index(
2435 &mut self,
2436 position: usize,
2437 new_values: Vec<Value<'static>>,
2438 ) -> Result<(), StorageError> {
2439 if position >= self.rows.len() {
2440 return Err(StorageError::Corrupt(alloc::format!(
2441 "update_row_no_index: position {position} out of bounds (rows={})",
2442 self.rows.len()
2443 )));
2444 }
2445 if new_values.len() != self.schema.columns.len() {
2446 return Err(StorageError::ArityMismatch {
2447 expected: self.schema.columns.len(),
2448 actual: new_values.len(),
2449 });
2450 }
2451 validate_row_against_schema(&new_values, &self.schema)?;
2452 let old_row = self
2453 .rows
2454 .get(position)
2455 .expect("position bounds-checked above");
2456 let old_bytes = row_body_encoded_len(old_row, &self.schema) as u64;
2457 let new_row = Row::new(new_values);
2458 let new_bytes = row_body_encoded_len(&new_row, &self.schema) as u64;
2459 self.rows = self
2460 .rows
2461 .set(position, new_row)
2462 .expect("position bounds-checked above");
2463 self.hot_bytes = self
2464 .hot_bytes
2465 .saturating_sub(old_bytes)
2466 .saturating_add(new_bytes);
2467 Ok(())
2468 }
2469
2470 /// v4.4: replace the row at `position` with `new_values` (must
2471 /// match the schema arity + types). v7.20: index maintenance is
2472 /// incremental — only indices whose key value changed are
2473 /// touched (B-tree entry move in place; NSW / BRIN / GIN fall
2474 /// back to a full rebuild when their column changed).
2475 pub fn update_row(
2476 &mut self,
2477 position: usize,
2478 new_values: Vec<Value<'static>>,
2479 ) -> Result<(), StorageError> {
2480 if position >= self.rows.len() {
2481 return Err(StorageError::Corrupt(alloc::format!(
2482 "update_row: position {position} out of bounds (rows={})",
2483 self.rows.len()
2484 )));
2485 }
2486 if new_values.len() != self.schema.columns.len() {
2487 return Err(StorageError::ArityMismatch {
2488 expected: self.schema.columns.len(),
2489 actual: new_values.len(),
2490 });
2491 }
2492 // Reuse the per-cell type-compat validation that `insert`
2493 // applies. The body below mirrors that check intentionally —
2494 // factoring it would be more code than the duplication.
2495 for (i, (val, col)) in new_values.iter().zip(&self.schema.columns).enumerate() {
2496 if val.is_null() {
2497 if !col.nullable {
2498 return Err(StorageError::NullInNotNull {
2499 column: col.name.clone(),
2500 });
2501 }
2502 continue;
2503 }
2504 // v7.39 (read01 round 54) — `data_type()` is None for the
2505 // eval-only variants that carry no DataType (RegClass, Composite).
2506 // They are NOT NULL, so `.expect("non-null")` PANICKED on them —
2507 // materialising a CTE like `WITH w AS (SELECT 't'::regclass)` blew
2508 // up the query with an "internal error". Report a clean type
2509 // mismatch instead; the engine coerces these before they get here
2510 // on every path that knows how.
2511 let Some(actual) = val.data_type() else {
2512 // An eval-only value (RegClass carries oid + name, Composite a
2513 // field tuple) has no DataType in the storage lattice. It is
2514 // NOT NULL, so the old `.expect("non-null")` PANICKED — which
2515 // is how `WITH w AS (SELECT 't'::regclass)` blew up with an
2516 // "internal error". Accept it: the value keeps its dual shape
2517 // and downstream comparisons (RegClass vs BigInt oid) handle it.
2518 continue;
2519 };
2520 let compatible = column_accepts(actual, col.ty);
2521 if !compatible {
2522 return Err(StorageError::TypeMismatch {
2523 column: col.name.clone(),
2524 expected: col.ty,
2525 actual,
2526 position: i,
2527 });
2528 }
2529 }
2530 let old_row = self
2531 .rows
2532 .get(position)
2533 .expect("position bounds-checked above");
2534 let old_bytes = row_body_encoded_len(old_row, &self.schema) as u64;
2535 let new_row = Row::new(new_values);
2536 let new_bytes = row_body_encoded_len(&new_row, &self.schema) as u64;
2537 // v7.20 P4 — incremental index maintenance. `rows.set`
2538 // replaces the row in place, so every OTHER row's Hot
2539 // locator stays valid; only indices whose key value
2540 // actually changed at `position` need touching. The
2541 // common OLTP shape (`UPDATE … SET non_indexed_col = …
2542 // WHERE pk = $1`) touches no index at all — pre-v7.20
2543 // this path paid a full rebuild_indices() (O(rows ×
2544 // indices)) per UPDATE, which dominated the profiled
2545 // write cost on a 5k-row table (~1 ms/stmt).
2546 //
2547 // BTree gets an in-place entry move (drop Hot(position)
2548 // from the old key's locator list, append to the new
2549 // key's). NSW graphs / BRIN summaries / GIN posting
2550 // lists have no cheap single-key move — a changed column
2551 // under one of those falls back to the full rebuild.
2552 enum IdxFix {
2553 BTreeMove {
2554 idx_pos: usize,
2555 old_key: Option<IndexKey>,
2556 new_key: Option<IndexKey>,
2557 },
2558 FullRebuild,
2559 }
2560 let mut fixes: Vec<IdxFix> = Vec::new();
2561 for (idx_pos, idx) in self.indices.iter().enumerate() {
2562 let col = idx.column_position;
2563 let old_v = &old_row.values[col];
2564 let new_v = &new_row.values[col];
2565 if old_v == new_v {
2566 continue;
2567 }
2568 match &idx.kind {
2569 IndexKind::BTree(_) => fixes.push(IdxFix::BTreeMove {
2570 idx_pos,
2571 old_key: IndexKey::from_value(old_v),
2572 new_key: IndexKey::from_value(new_v),
2573 }),
2574 IndexKind::Nsw(_)
2575 | IndexKind::Brin { .. }
2576 | IndexKind::Gin(_)
2577 | IndexKind::GinTrgm(_)
2578 | IndexKind::GinFulltext(_)
2579 | IndexKind::GinJsonb(_) => {
2580 fixes.clear();
2581 fixes.push(IdxFix::FullRebuild);
2582 break;
2583 }
2584 }
2585 }
2586 // v7.39 (round 215) — capture the range-exclusion key move BEFORE the
2587 // in-place `set` consumes `new_row`. A `FullRebuild` (a GIN/NSW/BRIN
2588 // column changed) rebuilds the excl indexes too via `rebuild_indices`,
2589 // so only apply the incremental move on the pure-BTreeMove path.
2590 let excl_has_full = fixes.iter().any(|f| matches!(f, IdxFix::FullRebuild));
2591 let excl_moves: Vec<(usize, Option<(i128, u8)>, Option<(i128, u8)>)> =
2592 if self.excl_indexes.is_empty() || excl_has_full {
2593 Vec::new()
2594 } else {
2595 self.excl_indexes
2596 .iter()
2597 .filter_map(|e| {
2598 let c = e.column_position;
2599 let old_k = old_row.values.get(c).and_then(crate::range_excl_index_key);
2600 let new_k = new_row.values.get(c).and_then(crate::range_excl_index_key);
2601 if old_k == new_k {
2602 None // range bound unchanged — no index touch
2603 } else {
2604 Some((c, old_k, new_k))
2605 }
2606 })
2607 .collect()
2608 };
2609 self.rows = self
2610 .rows
2611 .set(position, new_row)
2612 .expect("position bounds-checked above");
2613 self.hot_bytes = self
2614 .hot_bytes
2615 .saturating_sub(old_bytes)
2616 .saturating_add(new_bytes);
2617 // v7.34 — capture row-level redo (after the row is in place; the
2618 // immutable read of the new values is dropped before record_redo's
2619 // mutable borrow, and gated so capture-off pays nothing).
2620 if self.redo_log.is_some() {
2621 let new_row = self
2622 .rows
2623 .get(position)
2624 .map(|r| r.values.clone())
2625 .unwrap_or_default();
2626 // v7.37.15 (Epic W slice 1) — carry the stable RowId of the
2627 // updated row (`position` is bounds-checked above, so the id
2628 // is present). `writer_version` (xmax of the superseded
2629 // tuple) is 0: the writing TxId is not threaded here yet.
2630 let redo_rowid = self
2631 .rowids()
2632 .get(position)
2633 .copied()
2634 .unwrap_or(crate::row_header::RowId::UNASSIGNED);
2635 self.record_redo(|table| RowChange::Update {
2636 table,
2637 pos: position,
2638 new_row,
2639 rowid: redo_rowid,
2640 writer_version: 0,
2641 });
2642 }
2643 for fix in fixes {
2644 match fix {
2645 IdxFix::FullRebuild => {
2646 self.rebuild_indices();
2647 break;
2648 }
2649 IdxFix::BTreeMove {
2650 idx_pos,
2651 old_key,
2652 new_key,
2653 } => {
2654 let IndexKind::BTree(map) = &mut self.indices[idx_pos].kind else {
2655 unreachable!("IdxFix::BTreeMove built from a BTree index");
2656 };
2657 // NULL keys never enter the B-tree (from_value
2658 // returns None), so a None on either side means
2659 // "no entry on that side".
2660 if let Some(k) = old_key
2661 && let Some(locs) = map.get(&k)
2662 {
2663 let mut locs = locs.clone();
2664 locs.retain(|l| *l != RowLocator::Hot(position));
2665 // No remove_mut on the persistent map: an
2666 // empty locator list is the tombstone —
2667 // lookup_eq returns an empty slice, and the
2668 // next rebuild_indices() drops the key.
2669 map.insert_mut(k, locs);
2670 }
2671 if let Some(k) = new_key {
2672 if let Some(entries) = map.get_mut(&k) {
2673 entries.push(RowLocator::Hot(position));
2674 } else {
2675 map.insert_mut(k, alloc::vec![RowLocator::Hot(position)]);
2676 }
2677 }
2678 }
2679 }
2680 }
2681 // v7.39 (round 215) — apply the range-exclusion key moves captured
2682 // above (skipped when a FullRebuild already re-emitted every excl
2683 // index). Same shape as the BTreeMove: drop Hot(position) from the
2684 // old key, append it to the new key.
2685 for (col, old_k, new_k) in excl_moves {
2686 let Some(ex) = self
2687 .excl_indexes
2688 .iter_mut()
2689 .find(|e| e.column_position == col)
2690 else {
2691 continue;
2692 };
2693 if let Some(k) = old_k
2694 && let Some(locs) = ex.map.get(&k)
2695 {
2696 let mut locs = locs.clone();
2697 locs.retain(|l| *l != RowLocator::Hot(position));
2698 ex.map.insert_mut(k, locs);
2699 }
2700 if let Some(k) = new_k {
2701 if let Some(entries) = ex.map.get_mut(&k) {
2702 entries.push(RowLocator::Hot(position));
2703 } else {
2704 ex.map.insert_mut(k, alloc::vec![RowLocator::Hot(position)]);
2705 }
2706 }
2707 }
2708 Ok(())
2709 }
2710
2711 /// v4.4 helper used by `delete_rows` / `update_row`: discard all
2712 /// index payloads and rebuild from `self.rows`. Cheap enough
2713 /// for typical SPG scale (catalogs in the docker-compose
2714 /// deployment shape are small); the alternative — incremental
2715 /// shift bookkeeping across B-tree + NSW — would be far more
2716 /// invasive than the savings justify.
2717 fn rebuild_indices(&mut self) {
2718 // v5.2.3: capture every `Cold` locator on every BTree index
2719 // before the rebuild, so the from-rows re-emission below
2720 // (which only produces `Hot` locators) doesn't drop cold-
2721 // tier entries on keys unrelated to the row that changed.
2722 // Pre-v5.2.3 this was a `freeze_oldest_to_cold` worry only
2723 // and the freezer did its own capture-then-reregister; v5.2.3
2724 // promotes that pattern into the base helper because UPDATE
2725 // / DELETE now run rebuild_indices on tables with cold rows.
2726 let preserved_cold: Vec<(String, Vec<(IndexKey, RowLocator)>)> = self
2727 .indices
2728 .iter()
2729 .filter_map(|idx| match &idx.kind {
2730 IndexKind::BTree(map) => {
2731 let cold: Vec<(IndexKey, RowLocator)> = map
2732 .iter()
2733 .flat_map(|(k, locs)| {
2734 locs.iter()
2735 .filter(|l| l.is_cold())
2736 .copied()
2737 .map(move |l| (k.clone(), l))
2738 })
2739 .collect();
2740 if cold.is_empty() {
2741 None
2742 } else {
2743 Some((idx.name.clone(), cold))
2744 }
2745 }
2746 // BRIN / NSW carry no key→locator map. GIN handles
2747 // its own cold preservation below in `preserved_gin_cold`.
2748 IndexKind::Nsw(_)
2749 | IndexKind::Brin { .. }
2750 | IndexKind::Gin(_)
2751 | IndexKind::GinTrgm(_)
2752 | IndexKind::GinFulltext(_)
2753 | IndexKind::GinJsonb(_) => None,
2754 })
2755 .collect();
2756
2757 // v7.12.3 — same cold-preservation pattern for GIN's
2758 // `word → Vec<RowLocator>` posting lists. Parallel to the
2759 // BTree pass above (different key type so a separate vec is
2760 // cleaner than a generic merge). v7.15.0: trigram-GIN
2761 // (`gin_trgm_ops`) shares the same posting-list shape, so
2762 // one pass handles both — the `RebuildKind` carries the
2763 // kind tag to drive resurrection.
2764 let preserved_gin_cold: Vec<(String, Vec<(String, RowLocator)>)> = self
2765 .indices
2766 .iter()
2767 .filter_map(|idx| match &idx.kind {
2768 // v7.17.0 Phase 2.2 — fulltext-GIN posting lists
2769 // share the `String → Vec<RowLocator>` shape, so
2770 // cold preservation handles all three GIN flavours
2771 // in one pass.
2772 IndexKind::Gin(map)
2773 | IndexKind::GinTrgm(map)
2774 | IndexKind::GinFulltext(map)
2775 | IndexKind::GinJsonb(map) => {
2776 let cold: Vec<(String, RowLocator)> = map
2777 .iter()
2778 .flat_map(|(w, locs)| {
2779 locs.iter()
2780 .filter(|l| l.is_cold())
2781 .copied()
2782 .map(move |l| (w.clone(), l))
2783 })
2784 .collect();
2785 if cold.is_empty() {
2786 None
2787 } else {
2788 Some((idx.name.clone(), cold))
2789 }
2790 }
2791 IndexKind::BTree(_) | IndexKind::Nsw(_) | IndexKind::Brin { .. } => None,
2792 })
2793 .collect();
2794
2795 // v6.7.1 — descriptor needs to capture index kind so the
2796 // rebuild loop can resurrect BTree / NSW / BRIN / GIN exactly
2797 // as they were. (NSW carries m; BRIN carries the column type
2798 // snapshot; BTree / GIN need no extra payload.)
2799 #[derive(Clone)]
2800 enum RebuildKind {
2801 BTree,
2802 Nsw(usize),
2803 Brin(DataType),
2804 Gin,
2805 GinTrgm,
2806 GinFulltext,
2807 GinJsonb,
2808 }
2809 // v7.39 (round 170) — the descriptor must carry the FULL index
2810 // metadata: the rebuild used to reconstruct via bare
2811 // `Index::new_btree(name, pos)`, silently DROPPING is_unique /
2812 // extra_column_positions / partial_predicate / expression /
2813 // included_columns / nulls_not_distinct — so the first VACUUM
2814 // (or any delete-path rebuild) turned every UNIQUE INDEX into a
2815 // plain one and stopped enforcing it (probe-reproduced:
2816 // duplicate keys inserted silently after VACUUM).
2817 struct RebuildDesc {
2818 name: String,
2819 column_position: usize,
2820 kind: RebuildKind,
2821 is_unique: bool,
2822 extra_column_positions: Vec<usize>,
2823 partial_predicate: Option<String>,
2824 expression: Option<String>,
2825 included_columns: Vec<usize>,
2826 nulls_not_distinct: bool,
2827 // v7.39 (round 537) — carried through a rebuild like the rest.
2828 descending: bool,
2829 nulls_first: Option<bool>,
2830 collation: Option<String>,
2831 }
2832 let descriptors: Vec<RebuildDesc> = self
2833 .indices
2834 .iter()
2835 .map(|idx| {
2836 let kind = match &idx.kind {
2837 IndexKind::Nsw(g) => RebuildKind::Nsw(g.m),
2838 IndexKind::Brin { column_type } => RebuildKind::Brin(*column_type),
2839 IndexKind::BTree(_) => RebuildKind::BTree,
2840 IndexKind::Gin(_) => RebuildKind::Gin,
2841 IndexKind::GinTrgm(_) => RebuildKind::GinTrgm,
2842 IndexKind::GinFulltext(_) => RebuildKind::GinFulltext,
2843 IndexKind::GinJsonb(_) => RebuildKind::GinJsonb,
2844 };
2845 RebuildDesc {
2846 name: idx.name.clone(),
2847 column_position: idx.column_position,
2848 kind,
2849 is_unique: idx.is_unique,
2850 extra_column_positions: idx.extra_column_positions.clone(),
2851 partial_predicate: idx.partial_predicate.clone(),
2852 expression: idx.expression.clone(),
2853 included_columns: idx.included_columns.clone(),
2854 nulls_not_distinct: idx.nulls_not_distinct,
2855 descending: idx.descending,
2856 nulls_first: idx.nulls_first,
2857 collation: idx.collation.clone(),
2858 }
2859 })
2860 .collect();
2861 self.indices.clear();
2862 for desc in descriptors {
2863 let RebuildDesc {
2864 name,
2865 column_position,
2866 kind: rebuild_kind,
2867 is_unique,
2868 extra_column_positions,
2869 partial_predicate,
2870 expression,
2871 included_columns,
2872 nulls_not_distinct,
2873 descending,
2874 nulls_first,
2875 collation,
2876 } = desc;
2877 let pre_len = self.indices.len();
2878 match rebuild_kind {
2879 RebuildKind::Nsw(m) => {
2880 let idx = Index::new_nsw(name, column_position, m);
2881 self.indices.push(idx);
2882 let idx_pos = self.indices.len() - 1;
2883 let row_indices: Vec<usize> = (0..self.rows.len()).collect();
2884 for row_idx in row_indices {
2885 nsw_insert_at(self, idx_pos, row_idx);
2886 }
2887 }
2888 RebuildKind::Brin(column_type) => {
2889 // BRIN has no in-memory rebuild — the summaries
2890 // live in cold segments which freeze emits.
2891 self.indices
2892 .push(Index::new_brin(name, column_position, column_type));
2893 }
2894 RebuildKind::BTree => {
2895 // v7.39 (round 170) — bulk build: collect + sort +
2896 // group + from_sorted. The per-row insert_mut paid a
2897 // path-copy allocation per row per index (~15ms per
2898 // index on a 50k-row VACUUM, the dominant cost).
2899 let mut idx = Index::new_btree(name, column_position);
2900 let mut pairs: Vec<(IndexKey, usize)> = Vec::with_capacity(self.rows.len());
2901 for (i, row) in self.rows.iter().enumerate() {
2902 if let Some(key) = IndexKey::from_value(&row.values[column_position]) {
2903 pairs.push((key, i));
2904 }
2905 }
2906 pairs.sort_by(|a, b| a.0.cmp(&b.0));
2907 let mut grouped: Vec<(IndexKey, Vec<RowLocator>)> = Vec::new();
2908 for (key, i) in pairs {
2909 match grouped.last_mut() {
2910 Some((k, locs)) if *k == key => locs.push(RowLocator::Hot(i)),
2911 _ => grouped.push((key, alloc::vec![RowLocator::Hot(i)])),
2912 }
2913 }
2914 idx.kind = IndexKind::BTree(
2915 crate::persistent_btree::PersistentBTreeMap::from_sorted(grouped),
2916 );
2917 self.indices.push(idx);
2918 }
2919 RebuildKind::Gin => {
2920 let mut idx = Index::new_gin(name, column_position);
2921 if let IndexKind::Gin(map) = &mut idx.kind {
2922 for (i, row) in self.rows.iter().enumerate() {
2923 if let Value::TsVector(lexemes) = &row.values[column_position] {
2924 for lex in lexemes {
2925 if let Some(entries) = map.get_mut(&lex.word) {
2926 entries.push(RowLocator::Hot(i));
2927 } else {
2928 map.insert_mut(
2929 lex.word.clone(),
2930 alloc::vec![RowLocator::Hot(i)],
2931 );
2932 }
2933 }
2934 }
2935 }
2936 }
2937 self.indices.push(idx);
2938 }
2939 RebuildKind::GinTrgm => {
2940 let mut idx = Index::new_gin_trgm(name, column_position);
2941 if let IndexKind::GinTrgm(map) = &mut idx.kind {
2942 for (i, row) in self.rows.iter().enumerate() {
2943 if let Value::Text(s) = &row.values[column_position] {
2944 for tri in trgm::extract_trigrams(s) {
2945 // r1019 — address the String-keyed map with the borrowed
2946 // trigram; allocate one only for a key the map has never
2947 // seen, which after the first rows is rare.
2948 let key = trgm::trigram_str(&tri);
2949 if let Some(entries) = map.get_mut_by(key) {
2950 entries.push(RowLocator::Hot(i));
2951 } else {
2952 map.insert_mut(
2953 alloc::string::ToString::to_string(key),
2954 alloc::vec![RowLocator::Hot(i)],
2955 );
2956 }
2957 }
2958 }
2959 }
2960 }
2961 self.indices.push(idx);
2962 }
2963 RebuildKind::GinFulltext => {
2964 // v7.17.0 Phase 2.2 — re-derive the lexeme
2965 // posting list from each TEXT/VARCHAR cell.
2966 // Mirrors the GinTrgm rebuild shape but
2967 // tokenises via `fts_simple::simple_lex`
2968 // (same rule as `to_tsvector('simple')`).
2969 let mut idx = Index::new_gin_fulltext(name, column_position);
2970 if let IndexKind::GinFulltext(map) = &mut idx.kind {
2971 for (i, row) in self.rows.iter().enumerate() {
2972 if let Value::Text(s) = &row.values[column_position] {
2973 for lex in fts_simple::simple_lex(s) {
2974 if let Some(entries) = map.get_mut(&lex) {
2975 entries.push(RowLocator::Hot(i));
2976 } else {
2977 map.insert_mut(lex, alloc::vec![RowLocator::Hot(i)]);
2978 }
2979 }
2980 }
2981 }
2982 }
2983 self.indices.push(idx);
2984 }
2985 RebuildKind::GinJsonb => {
2986 // v7.37.8 — re-derive the JSONB posting list
2987 // from each `Value::Json` cell.
2988 let mut idx = Index::new_gin_jsonb(name, column_position);
2989 if let IndexKind::GinJsonb(map) = &mut idx.kind {
2990 for (i, row) in self.rows.iter().enumerate() {
2991 if let Value::Json(s) = &row.values[column_position] {
2992 for tok in jsonb_gin::extract_tokens(s) {
2993 if let Some(entries) = map.get_mut(&tok) {
2994 entries.push(RowLocator::Hot(i));
2995 } else {
2996 map.insert_mut(tok, alloc::vec![RowLocator::Hot(i)]);
2997 }
2998 }
2999 }
3000 }
3001 }
3002 self.indices.push(idx);
3003 }
3004 }
3005 // v7.39 (round 170) — restore the captured metadata onto
3006 // whatever this arm pushed (see RebuildDesc above).
3007 if let Some(idx) = self.indices.get_mut(pre_len) {
3008 idx.is_unique = is_unique;
3009 idx.extra_column_positions = extra_column_positions;
3010 idx.partial_predicate = partial_predicate;
3011 idx.expression = expression;
3012 idx.included_columns = included_columns;
3013 idx.nulls_not_distinct = nulls_not_distinct;
3014 idx.descending = descending;
3015 idx.nulls_first = nulls_first;
3016 idx.collation = collation;
3017 }
3018 }
3019
3020 // Re-attach preserved cold locators after the from-rows
3021 // rebuild. `register_cold_locators` handles the per-key
3022 // entries-vec append; no key collisions arise because the
3023 // rebuild loop above produced only Hot locators.
3024 for (idx_name, locators) in preserved_cold {
3025 // Errors here would only fire if the index disappeared
3026 // between snapshot and rebuild, which can't happen
3027 // because the rebuild restores the same descriptor set.
3028 let _ = self.register_cold_locators(&idx_name, locators);
3029 }
3030 // v7.12.3 — same for GIN posting-list cold locators.
3031 for (idx_name, locators) in preserved_gin_cold {
3032 let _ = self.register_gin_cold_locators(&idx_name, locators);
3033 }
3034 // v7.39 (round 215) — the range-exclusion indexes address rows by the
3035 // same physical slot, so a compaction that shifted slots invalidates
3036 // their Hot locators too. Re-emit them from the (post-compaction) rows.
3037 if !self.excl_indexes.is_empty() {
3038 self.rebuild_excl_indexes();
3039 }
3040 }
3041
3042 fn add_nsw_index_inner(
3043 &mut self,
3044 name: String,
3045 column_name: &str,
3046 m: usize,
3047 restore: Option<NswGraph>,
3048 ) -> Result<(), StorageError> {
3049 if self.indices.iter().any(|i| i.name == name) {
3050 return Err(StorageError::DuplicateIndex { name });
3051 }
3052 let column_position = self.schema.column_position(column_name).ok_or_else(|| {
3053 StorageError::ColumnNotFound {
3054 column: column_name.into(),
3055 }
3056 })?;
3057 if !matches!(
3058 self.schema.columns[column_position].ty,
3059 DataType::Vector { .. }
3060 ) {
3061 return Err(StorageError::TypeMismatch {
3062 column: column_name.into(),
3063 expected: DataType::Vector {
3064 dim: 0,
3065 encoding: VecEncoding::F32,
3066 },
3067 actual: self.schema.columns[column_position].ty,
3068 position: column_position,
3069 });
3070 }
3071 if let Some(graph) = restore {
3072 self.indices.push(Index {
3073 name,
3074 column_position,
3075 kind: IndexKind::Nsw(graph),
3076 included_columns: Vec::new(),
3077 partial_predicate: None,
3078 expression: None,
3079 is_unique: false,
3080 nulls_not_distinct: false,
3081 descending: false,
3082 nulls_first: None,
3083 collation: None,
3084 extra_column_positions: Vec::new(),
3085 });
3086 return Ok(());
3087 }
3088 let idx = Index::new_nsw(name, column_position, m);
3089 self.indices.push(idx);
3090 let idx_pos = self.indices.len() - 1;
3091 // Bulk-build by walking the existing rows in order — each insert
3092 // sees the partial graph and links into it.
3093 let row_indices: Vec<usize> = (0..self.rows.len()).collect();
3094 for row_idx in row_indices {
3095 nsw_insert_at(self, idx_pos, row_idx);
3096 }
3097 Ok(())
3098 }
3099}
3100
3101/// v7.37.5 (mailrs crash-recovery Ask 3) — per-cell schema-compat
3102/// check shared by `insert_no_index` and `update_row_no_index`. The
3103/// logic mirrors the inline body in `insert` / `update_row` (NULL
3104/// handling, the cross-type compatibility map: TEXT ↔ VARCHAR/CHAR/
3105/// JSON/JSONB, TIMESTAMP ↔ TIMESTAMPTZ, BIT ↔ VARBIT, INET ↔ CIDR,
3106/// NUMERIC scale match).
3107/// v7.39 (round 642/643) — does a value of type `actual` belong in a
3108/// column declared `declared`?
3109///
3110/// This existed in THREE copies — insert, update and the standalone
3111/// row validator — and they had drifted apart in three independent
3112/// places: only insert accepted the `name` pairs, only insert and
3113/// update accepted a bit-to-bit pair with differing typmods, and only
3114/// update accepted a NEGATIVE declared numeric scale. Each omission was
3115/// a hole waiting for a value to reach that path; none was a deliberate
3116/// tightening, so the union below is the rule and all three now ask it.
3117///
3118/// Measured before converging: every shape the three disagreed about
3119/// answers identically to PG18 today, so this fixes nothing observable.
3120/// What it fixes is the next type — adding `xid` in round 640 meant
3121/// remembering to patch three places, and forgetting one would have
3122/// half-wired it.
3123///
3124/// The rule itself: a pair is compatible when the value's storage shape
3125/// is what the column stores. Length and precision contracts are NOT
3126/// checked here — they belong to coercion, which runs first.
3127/// `#[inline]` is not decoration. Extracting this matrix out of its
3128/// three call sites — a change with no semantic content at all — cost
3129/// `SELECT count(*) FROM d WHERE g BETWEEN 10 AND 20` **23x**, 5.8 ms
3130/// to 133 ms over 500 000 rows, reproducibly and outside the panel.
3131/// None of the three callers is on a scan path; taking the matrix out
3132/// of them was enough to move whatever else in this module the row loop
3133/// depends on being inlined. Round 641 learned the same thing about
3134/// `eval::binop::compare`. A refactor that reads as pure structure is
3135/// still a codegen change.
3136#[inline]
3137fn column_accepts(actual: DataType, declared: DataType) -> bool {
3138 if actual == declared {
3139 return true;
3140 }
3141 if matches!(
3142 (actual, declared),
3143 // A NAME column stores a Value::Text: the type identity is the
3144 // schema's and a value can never be one, so both directions.
3145 (
3146 DataType::Text,
3147 DataType::Varchar(_)
3148 | DataType::Char(_)
3149 | DataType::Name
3150 | DataType::Json
3151 | DataType::Jsonb
3152 ) | (DataType::Name, DataType::Text)
3153 // An XID column stores the Value::BigInt a transaction id
3154 // has always been; xid8 has no value of its own at all.
3155 | (DataType::BigInt, DataType::Xid | DataType::Xid8)
3156 | (DataType::Xid | DataType::Xid8, DataType::BigInt)
3157 // v7.39 (round 667) — an OID column likewise stores a plain
3158 // integer. INT is listed as well as BIGINT because a bare
3159 // literal arrives as one: PG takes `INSERT INTO t(o) VALUES
3160 // (42)` into an oid column, and measured, it does NOT take the
3161 // same integer into an xid column ("column is of type xid but
3162 // expression is of type integer"). SPG has been laxer than PG
3163 // on that xid direction since before this round — that is the
3164 // limitation `DataType::Xid8` documents, not something added
3165 // here.
3166 | (
3167 DataType::BigInt | DataType::Int | DataType::SmallInt,
3168 DataType::Oid,
3169 )
3170 | (DataType::Oid, DataType::BigInt | DataType::Int)
3171 // v7.39 (round 694) — `oid[]` rides in a BigIntArray cell, so
3172 // it accepts one either way, exactly as the scalar above does.
3173 | (DataType::BigIntArray | DataType::IntArray, DataType::OidArray)
3174 | (DataType::OidArray, DataType::BigIntArray)
3175 | (DataType::Json | DataType::Jsonb, DataType::Text)
3176 | (DataType::Json, DataType::Jsonb)
3177 | (DataType::Jsonb, DataType::Json)
3178 | (DataType::Timestamp, DataType::Timestamptz)
3179 | (DataType::Timestamptz, DataType::Timestamp)
3180 // BIT / VARBIT share the BitString storage shape; INET /
3181 // CIDR likewise. Same-family pairs with different typmods
3182 // are compatible HERE — the length contract is coercion's.
3183 | (DataType::Bit(_), DataType::BitVarying(_))
3184 | (DataType::BitVarying(_), DataType::Bit(_))
3185 | (DataType::Bit(_), DataType::Bit(_))
3186 | (DataType::BitVarying(_), DataType::BitVarying(_))
3187 | (DataType::Inet, DataType::Cidr)
3188 | (DataType::Cidr, DataType::Inet)
3189 ) {
3190 return true;
3191 }
3192 // NUMERIC carries its own scale in the value while the column
3193 // declares the expected one. An unconstrained `numeric` (the
3194 // precision-0/scale-0 sentinel) takes any scale; a declared
3195 // `numeric(p,s)` needs the rescaled value; and a NEGATIVE declared
3196 // scale stores at display scale 0, having been rounded to a
3197 // multiple of 10^|s|.
3198 matches!(
3199 (actual, declared),
3200 (
3201 DataType::Numeric { scale: a, .. },
3202 DataType::Numeric {
3203 precision: bp,
3204 scale: b,
3205 },
3206 ) if a == b || (bp == 0 && b == 0) || (b < 0 && a == 0)
3207 )
3208}
3209
3210fn validate_row_against_schema(
3211 values: &[Value<'static>],
3212 schema: &TableSchema,
3213) -> Result<(), StorageError> {
3214 for (i, (val, col)) in values.iter().zip(&schema.columns).enumerate() {
3215 if val.is_null() {
3216 if !col.nullable {
3217 return Err(StorageError::NullInNotNull {
3218 column: col.name.clone(),
3219 });
3220 }
3221 continue;
3222 }
3223 // v7.39 (read01 round 54) — see above: no panic on an untyped value.
3224 let Some(actual) = val.data_type() else {
3225 // See above: an eval-only untyped value is accepted, not a panic.
3226 continue;
3227 };
3228 let compatible = column_accepts(actual, col.ty);
3229 if !compatible {
3230 return Err(StorageError::TypeMismatch {
3231 column: col.name.clone(),
3232 expected: col.ty,
3233 actual,
3234 position: i,
3235 });
3236 }
3237 }
3238 Ok(())
3239}
3240
3241/// v6.0.4 — re-encode a single cell to the target `VecEncoding`.
3242/// Used by `Table::rebuild_nsw_index` when ALTER INDEX REBUILD
3243/// includes the optional `WITH (encoding = …)` clause. Round-trip
3244/// goes through f32: `current → Vec<f32> → target`, leaving NULL
3245/// cells untouched. Returns `Unsupported` on a non-vector cell —
3246/// the caller should have rejected the schema before reaching this.
3247fn recode_vector_cell(
3248 cell: Value<'static>,
3249 target: VecEncoding,
3250) -> Result<Value<'static>, StorageError> {
3251 if matches!(cell, Value::Null) {
3252 return Ok(cell);
3253 }
3254 // Step 1 — extract the f32 representation of the source cell.
3255 let as_f32: Vec<f32> = match &cell {
3256 Value::Vector(v) => v.to_vec(),
3257 Value::Sq8Vector(q) => quantize::dequantize(q),
3258 Value::HalfVector(h) => h.to_f32_vec(),
3259 other => {
3260 return Err(StorageError::Unsupported(format!(
3261 "ALTER INDEX REBUILD: cannot recode non-vector cell {:?}",
3262 other.data_type()
3263 )));
3264 }
3265 };
3266 // Step 2 — encode into the target shape. `F32` is the identity
3267 // path (saves one alloc round-trip when the source is already
3268 // F32 — but `Value::Vector(as_f32)` is the right answer
3269 // regardless).
3270 Ok(match target {
3271 VecEncoding::F32 => Value::Vector(Cow::Owned(as_f32)),
3272 VecEncoding::Sq8 => Value::Sq8Vector(quantize::quantize(&as_f32)),
3273 VecEncoding::F16 => Value::HalfVector(halfvec::HalfVector::from_f32_slice(&as_f32)),
3274 })
3275}
3276
3277/// v7.39 (round 562) — a cursor over `Table`'s row headers that holds
3278/// the trie leaf it last descended to.
3279///
3280/// See `Table::header_runs` for why. Ask about ascending positions and
3281/// the descent happens once per 32; ask about scattered ones and it
3282/// happens as often as `position_visible` would have done it.
3283#[derive(Debug)]
3284pub struct HeaderRuns<'a> {
3285 table: &'a Table,
3286 /// `(start, run)` — `run[i - start]` is the header for position `i`.
3287 run: Option<(usize, &'a [crate::row_header::RowHeader])>,
3288}
3289
3290impl HeaderRuns<'_> {
3291 /// Is the row at this position visible to the snapshot?
3292 ///
3293 /// Answers exactly as `Table::position_visible` does — same
3294 /// `SKIP LOCKED` handling, same snapshot rules — and the pins in
3295 /// `e2e_index_only_scan_round560` hold both to it.
3296 pub fn visible(&mut self, idx: usize, snapshot: &crate::snapshot::Snapshot) -> bool {
3297 if let Some((start, run)) = self.run
3298 && idx >= start
3299 && idx - start < run.len()
3300 {
3301 return self.table.header_visible(idx, &run[idx - start], snapshot);
3302 }
3303 let Some((start, run)) = self.table.headers.run_containing(idx) else {
3304 return false;
3305 };
3306 self.run = Some((start, run));
3307 self.table.header_visible(idx, &run[idx - start], snapshot)
3308 }
3309}