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 let mut entries = map.get(&lex.word).cloned().unwrap_or_default();
1130 entries.push(RowLocator::Hot(new_row_idx));
1131 map.insert_mut(lex.word.clone(), entries);
1132 }
1133 }
1134 }
1135 IndexKind::GinTrgm(map) => {
1136 // v7.15.0 — trigram GIN. Shingle the TEXT cell
1137 // into PG-compatible 3-byte trigrams and extend
1138 // each trigram's posting list.
1139 if let Value::Text(s) = &row.values[idx.column_position] {
1140 for tri in trgm::extract_trigrams(s) {
1141 let mut entries = map.get(&tri).cloned().unwrap_or_default();
1142 entries.push(RowLocator::Hot(new_row_idx));
1143 map.insert_mut(tri, entries);
1144 }
1145 }
1146 }
1147 IndexKind::GinFulltext(map) => {
1148 // v7.17.0 Phase 2.2 — MySQL FULLTEXT-shape
1149 // GIN over a TEXT / VARCHAR cell. Tokenise
1150 // via the storage-local `simple_lex` (same
1151 // rule as `to_tsvector('simple', text)`) and
1152 // extend each lexeme's posting list.
1153 let text_cell = match &row.values[idx.column_position] {
1154 Value::Text(s) => Some(s.as_ref()),
1155 // mysqldump-style mediumtext / longtext
1156 // land as Value::Text on insert; varchar
1157 // cells likewise. Anything else (NULL,
1158 // integer, …) contributes no lexemes.
1159 _ => None,
1160 };
1161 if let Some(s) = text_cell {
1162 for lex in fts_simple::simple_lex(s) {
1163 let mut entries = map.get(&lex).cloned().unwrap_or_default();
1164 entries.push(RowLocator::Hot(new_row_idx));
1165 map.insert_mut(lex, entries);
1166 }
1167 }
1168 }
1169 IndexKind::GinJsonb(map) => {
1170 // v7.37.8(sentori Epic 5 P2)— real JSONB-GIN.
1171 // Extract canonical `(path, leaf)` tokens from
1172 // the cell text and extend each token's posting
1173 // list. NULL or non-Json cell contributes no
1174 // tokens(`labels @> '...'` against a NULL row
1175 // is always false so absence here is correct).
1176 let json_cell = match &row.values[idx.column_position] {
1177 Value::Json(s) => Some(s.as_ref()),
1178 _ => None,
1179 };
1180 if let Some(s) = json_cell {
1181 for tok in jsonb_gin::extract_tokens(s) {
1182 let mut entries = map.get(&tok).cloned().unwrap_or_default();
1183 entries.push(RowLocator::Hot(new_row_idx));
1184 map.insert_mut(tok, entries);
1185 }
1186 }
1187 }
1188 // NSW handled below after the row push (so the new row
1189 // is visible to the kNN-graph connect step). BRIN
1190 // carries no per-row state.
1191 IndexKind::Nsw(_) | IndexKind::Brin { .. } => {}
1192 }
1193 }
1194 // v7.39 (round 215) — maintain the range-exclusion indexes for the
1195 // freshly-inserted row (before the move; `new_row_idx` is the slot it
1196 // will occupy). Mirrors the BTree maintenance above.
1197 if !self.excl_indexes.is_empty() {
1198 self.excl_indexes_on_insert(&row, new_row_idx);
1199 }
1200 // v5.2.1: maintain incremental hot-tier byte counter. Computed
1201 // before the move so we don't need to borrow `row` after push.
1202 self.hot_bytes = self
1203 .hot_bytes
1204 .saturating_add(row_body_encoded_len(&row, &self.schema) as u64);
1205 // v7.34 — capture the row-level redo before the row is moved in.
1206 // v7.37.15 (Epic W slice 1) — carry the stable RowId this insert
1207 // will receive. `alloc_rowid` below hands out `RowId(next_rowid)`
1208 // and bumps the counter unconditionally, so the id read here is
1209 // exactly the one the row ends up with. `writer_version` (xmin)
1210 // is 0: the writing TxId is not threaded to this layer yet (the
1211 // header pushed below is `RowHeader::frozen()`).
1212 let redo_rowid = crate::row_header::RowId(self.next_rowid);
1213 self.record_redo(|table| RowChange::Insert {
1214 table,
1215 row: row.clone(),
1216 rowid: redo_rowid,
1217 writer_version: 0,
1218 });
1219 // v4.39.1: push_mut keeps streaming inserts at Vec::push speed when
1220 // the table is uniquely owned (the spg-embedded path); inside a TX
1221 // wrap where a Catalog snapshot exists, push_mut path-copies the
1222 // tail just like push() and the snapshot stays valid.
1223 self.rows.push_mut(row);
1224 // v7.37.15 (Phase A.2) — keep `headers` lock-step with `rows`.
1225 // Phase A defaults every new insert to RowHeader::frozen() so
1226 // visibility checks against any snapshot return true; Phase C
1227 // upgrades the inserter to stamp the writing tx's xmin.
1228 self.headers
1229 .push_mut(crate::row_header::RowHeader::frozen());
1230 // v7.37.15 (Phase C.1) — allocate + push the stable RowId in
1231 // lock-step with rows/headers. Index locators still address
1232 // by physical slot at this commit; the id is additive
1233 // bookkeeping the lock table / HOT chains / WAL migrate to.
1234 let rid = self.alloc_rowid();
1235 self.rowids.push_mut(rid);
1236 // v7.37.15 (Epic W slice 1) — the id captured for the redo log
1237 // above must be the one actually assigned to the row.
1238 debug_assert_eq!(
1239 rid, redo_rowid,
1240 "redo-captured RowId must match the allocated RowId"
1241 );
1242 debug_assert_eq!(
1243 self.rows.len(),
1244 self.headers.len(),
1245 "headers must stay in lock-step with rows after insert"
1246 );
1247 debug_assert_eq!(
1248 self.rows.len(),
1249 self.rowids.len(),
1250 "rowids must stay in lock-step with rows after insert"
1251 );
1252 // NSW updates after the push so the new row is visible to the
1253 // greedy search used during connect.
1254 let new_row_idx = self.rows.len() - 1;
1255 let nsw_targets: Vec<usize> = self
1256 .indices
1257 .iter()
1258 .enumerate()
1259 .filter_map(|(i, idx)| {
1260 if matches!(idx.kind, IndexKind::Nsw(_)) {
1261 Some(i)
1262 } else {
1263 None
1264 }
1265 })
1266 .collect();
1267 for idx_pos in nsw_targets {
1268 nsw_insert_at(self, idx_pos, new_row_idx);
1269 }
1270 Ok(())
1271 }
1272
1273 /// Build a new B-tree index over the named column. Rebuilds from
1274 /// existing rows. Errors if `column_name` doesn't exist or the index
1275 /// name is taken.
1276 pub fn add_index(&mut self, name: String, column_name: &str) -> Result<(), StorageError> {
1277 if self.indices.iter().any(|i| i.name == name) {
1278 return Err(StorageError::DuplicateIndex { name });
1279 }
1280 let column_position = self.schema.column_position(column_name).ok_or_else(|| {
1281 StorageError::ColumnNotFound {
1282 column: column_name.into(),
1283 }
1284 })?;
1285 let mut idx = Index::new_btree(name, column_position);
1286 if let IndexKind::BTree(map) = &mut idx.kind {
1287 for (i, row) in self.rows.iter().enumerate() {
1288 if let Some(key) = IndexKey::from_value(&row.values[column_position]) {
1289 let mut entries = map.get(&key).cloned().unwrap_or_default();
1290 entries.push(RowLocator::Hot(i));
1291 map.insert_mut(key, entries);
1292 }
1293 }
1294 }
1295 self.indices.push(idx);
1296 Ok(())
1297 }
1298
1299 /// v7.39 (round 215) — ensure a range-exclusion index exists on
1300 /// `column_position`, building it from the current rows. Idempotent: a
1301 /// second call for the same column is a no-op. Called at CREATE TABLE /
1302 /// ALTER ADD EXCLUDE and on catalog load (rebuild-from-constraints).
1303 /// Tombstoned rows are indexed too (they are filtered by the consumer via
1304 /// `is_deleted()` at query time — the established index pattern).
1305 pub fn ensure_excl_range_index(&mut self, column_position: usize) {
1306 if self
1307 .excl_indexes
1308 .iter()
1309 .any(|e| e.column_position == column_position)
1310 {
1311 return;
1312 }
1313 let mut map: crate::PersistentBTreeMap<(i128, u8), Vec<RowLocator>> =
1314 crate::PersistentBTreeMap::new();
1315 for (i, row) in self.rows.iter().enumerate() {
1316 if let Some(v) = row.values.get(column_position)
1317 && let Some(key) = crate::range_excl_index_key(v)
1318 {
1319 let mut entries = map.get(&key).cloned().unwrap_or_default();
1320 entries.push(RowLocator::Hot(i));
1321 map.insert_mut(key, entries);
1322 }
1323 }
1324 self.excl_indexes.push(crate::ExclRangeIndex {
1325 column_position,
1326 map,
1327 });
1328 }
1329
1330 /// v7.39 (round 215) — the range-exclusion index on `column_position`, if
1331 /// one was built. The EXCLUDE enforcement path probes its
1332 /// [`predecessor`](crate::PersistentBTreeMap::predecessor) + successors to
1333 /// find candidate overlaps in O(log n).
1334 #[must_use]
1335 pub fn excl_range_index(
1336 &self,
1337 column_position: usize,
1338 ) -> Option<&crate::PersistentBTreeMap<(i128, u8), Vec<RowLocator>>> {
1339 self.excl_indexes
1340 .iter()
1341 .find(|e| e.column_position == column_position)
1342 .map(|e| &e.map)
1343 }
1344
1345 /// v7.39 (round 215) — add a freshly-appended row at `row_idx` to every
1346 /// range-exclusion index. Called from `insert` after the row is pushed,
1347 /// mirroring the BTree secondary-index maintenance.
1348 fn excl_indexes_on_insert(&mut self, row: &Row<'static>, row_idx: usize) {
1349 for ex in &mut self.excl_indexes {
1350 if let Some(v) = row.values.get(ex.column_position)
1351 && let Some(key) = crate::range_excl_index_key(v)
1352 {
1353 let mut entries = ex.map.get(&key).cloned().unwrap_or_default();
1354 entries.push(RowLocator::Hot(row_idx));
1355 ex.map.insert_mut(key, entries);
1356 }
1357 }
1358 }
1359
1360 /// v7.39 (round 215) — rebuild every range-exclusion index from the
1361 /// current rows (called from `rebuild_indices`, i.e. after a physical
1362 /// compaction/delete that shifted slots). Preserves which columns are
1363 /// indexed; re-emits all `Hot` locators.
1364 fn rebuild_excl_indexes(&mut self) {
1365 let cols: Vec<usize> = self
1366 .excl_indexes
1367 .iter()
1368 .map(|e| e.column_position)
1369 .collect();
1370 self.excl_indexes.clear();
1371 for c in cols {
1372 self.ensure_excl_range_index(c);
1373 }
1374 }
1375
1376 /// Build a new NSW (HNSW-flavoured) index over the named column.
1377 /// Required for `ORDER BY col <-> literal LIMIT k` to plan as a
1378 /// graph traversal instead of a full scan. Column must be a Vector
1379 /// type. `m` is the maximum number of neighbours per node.
1380 pub fn add_nsw_index(
1381 &mut self,
1382 name: String,
1383 column_name: &str,
1384 m: usize,
1385 ) -> Result<(), StorageError> {
1386 self.add_nsw_index_inner(name, column_name, m, None)
1387 }
1388
1389 /// v6.0.4 — synchronous rebuild of the named NSW index. If
1390 /// `new_encoding` is `Some(target)` and differs from the column's
1391 /// current encoding, every stored cell at the indexed column is
1392 /// re-coded into the target encoding before the new graph
1393 /// builds. Returns `IndexNotFound` if no index by that name exists
1394 /// and `Unsupported` for non-NSW indexes (`BTree` REBUILD is a no-op
1395 /// the engine layer rejects, not a storage-level concept).
1396 ///
1397 /// Holds the caller's `&mut self` for the duration — no
1398 /// concurrency / staging / WAL-replay machinery in v6.0.4. The
1399 /// "live" optimisation lands as v6.0.4.1.
1400 pub fn rebuild_nsw_index(
1401 &mut self,
1402 name: &str,
1403 new_encoding: Option<VecEncoding>,
1404 ) -> Result<(), StorageError> {
1405 let idx_pos = self
1406 .indices
1407 .iter()
1408 .position(|i| i.name == name)
1409 .ok_or_else(|| StorageError::IndexNotFound {
1410 name: String::from(name),
1411 })?;
1412 let col_pos = self.indices[idx_pos].column_position;
1413 let m = match &self.indices[idx_pos].kind {
1414 IndexKind::Nsw(g) => g.m,
1415 IndexKind::BTree(_)
1416 | IndexKind::Brin { .. }
1417 | IndexKind::Gin(_)
1418 | IndexKind::GinTrgm(_)
1419 | IndexKind::GinFulltext(_)
1420 | IndexKind::GinJsonb(_) => {
1421 return Err(StorageError::Unsupported(format!(
1422 "ALTER INDEX REBUILD on non-NSW index {name:?} — only NSW indexes can rebuild"
1423 )));
1424 }
1425 };
1426 let col_name = self.schema.columns[col_pos].name.clone();
1427 // 1. Optional re-encoding pass. Done first so the cells
1428 // match the schema before the graph rebuild walks them.
1429 if let Some(target) = new_encoding {
1430 let current = match self.schema.columns[col_pos].ty {
1431 DataType::Vector { encoding, .. } => encoding,
1432 ref other => {
1433 return Err(StorageError::Unsupported(format!(
1434 "ALTER INDEX REBUILD WITH (encoding=…) on non-vector column type {other:?}"
1435 )));
1436 }
1437 };
1438 if target != current {
1439 let DataType::Vector { dim, .. } = self.schema.columns[col_pos].ty else {
1440 unreachable!("checked above")
1441 };
1442 let n = self.rows.len();
1443 for i in 0..n {
1444 let row = self
1445 .rows
1446 .get_mut(i)
1447 .expect("row index in bounds (we iterated up to len())");
1448 let cell = core::mem::replace(&mut row.values[col_pos], Value::Null);
1449 let recoded = recode_vector_cell(cell, target)?;
1450 row.values[col_pos] = recoded;
1451 }
1452 self.schema.columns[col_pos].ty = DataType::Vector {
1453 dim,
1454 encoding: target,
1455 };
1456 }
1457 }
1458 // 2. Drop the existing index slot + rebuild from row payload.
1459 self.indices.remove(idx_pos);
1460 self.add_nsw_index_inner(String::from(name), &col_name, m, None)?;
1461 Ok(())
1462 }
1463
1464 /// Restore an NSW index from a pre-built graph (used on
1465 /// deserialize). Skips the bulk-build pass since the topology is
1466 /// already known. Returns `DuplicateIndex` or `ColumnNotFound` on
1467 /// schema mismatch as usual.
1468 pub fn restore_nsw_index(
1469 &mut self,
1470 name: String,
1471 column_name: &str,
1472 graph: NswGraph,
1473 ) -> Result<(), StorageError> {
1474 self.add_nsw_index_inner(name, column_name, graph.m, Some(graph))
1475 }
1476
1477 /// Restore a `BTree` index from a pre-built `(IndexKey, Vec<RowLocator>)`
1478 /// map. Used by [`Catalog::deserialize`] when reading a v9 (or later)
1479 /// catalog snapshot — the map travels on disk so cold-tier locators
1480 /// survive a round-trip, instead of being rebuilt from `self.rows`
1481 /// (which would lose every Cold entry). Same error contract as
1482 /// [`Table::add_index`].
1483 pub fn restore_btree_index(
1484 &mut self,
1485 name: String,
1486 column_name: &str,
1487 map: PersistentBTreeMap<IndexKey, Vec<RowLocator>>,
1488 ) -> Result<(), StorageError> {
1489 if self.indices.iter().any(|i| i.name == name) {
1490 return Err(StorageError::DuplicateIndex { name });
1491 }
1492 let column_position = self.schema.column_position(column_name).ok_or_else(|| {
1493 StorageError::ColumnNotFound {
1494 column: column_name.into(),
1495 }
1496 })?;
1497 self.indices.push(Index {
1498 name,
1499 column_position,
1500 kind: IndexKind::BTree(map),
1501 included_columns: Vec::new(),
1502 partial_predicate: None,
1503 expression: None,
1504 is_unique: false,
1505 nulls_not_distinct: false,
1506 descending: false,
1507 nulls_first: None,
1508 collation: None,
1509 extra_column_positions: Vec::new(),
1510 });
1511 Ok(())
1512 }
1513
1514 /// v6.7.1 — public restore counterpart for BRIN indices. Used
1515 /// by `Catalog::deserialize` when a v10 snapshot carries a
1516 /// BRIN index entry. BRIN carries no in-memory data — only the
1517 /// `column_type` snapshot is restored.
1518 pub fn restore_brin_index(
1519 &mut self,
1520 name: String,
1521 column_name: &str,
1522 column_type: DataType,
1523 ) -> Result<(), StorageError> {
1524 if self.indices.iter().any(|i| i.name == name) {
1525 return Err(StorageError::DuplicateIndex { name });
1526 }
1527 let column_position = self.schema.column_position(column_name).ok_or_else(|| {
1528 StorageError::ColumnNotFound {
1529 column: column_name.into(),
1530 }
1531 })?;
1532 self.indices
1533 .push(Index::new_brin(name, column_position, column_type));
1534 Ok(())
1535 }
1536
1537 /// v6.7.1 — public CREATE INDEX counterpart for BRIN. Creates
1538 /// the index entry with a snapshot of the indexed column's
1539 /// current `DataType`.
1540 pub fn add_brin_index(&mut self, name: String, column_name: &str) -> Result<(), StorageError> {
1541 if self.indices.iter().any(|i| i.name == name) {
1542 return Err(StorageError::DuplicateIndex { name });
1543 }
1544 let column_position = self.schema.column_position(column_name).ok_or_else(|| {
1545 StorageError::ColumnNotFound {
1546 column: column_name.into(),
1547 }
1548 })?;
1549 let column_type = self.schema.columns[column_position].ty;
1550 self.indices
1551 .push(Index::new_brin(name, column_position, column_type));
1552 Ok(())
1553 }
1554
1555 /// v7.12.3 — Build a new GIN inverted index over a `tsvector`
1556 /// column. Populates posting lists from existing rows. Errors
1557 /// if the column doesn't exist, isn't `TsVector`, or the index
1558 /// name is taken.
1559 pub fn add_gin_index(&mut self, name: String, column_name: &str) -> Result<(), StorageError> {
1560 if self.indices.iter().any(|i| i.name == name) {
1561 return Err(StorageError::DuplicateIndex { name });
1562 }
1563 let column_position = self.schema.column_position(column_name).ok_or_else(|| {
1564 StorageError::ColumnNotFound {
1565 column: column_name.into(),
1566 }
1567 })?;
1568 if self.schema.columns[column_position].ty != DataType::TsVector {
1569 return Err(StorageError::Corrupt(format!(
1570 "GIN index {name:?} requires a tsvector column; \
1571 {column_name:?} is {:?}",
1572 self.schema.columns[column_position].ty
1573 )));
1574 }
1575 let mut idx = Index::new_gin(name, column_position);
1576 if let IndexKind::Gin(map) = &mut idx.kind {
1577 for (i, row) in self.rows.iter().enumerate() {
1578 if let Value::TsVector(lexemes) = &row.values[column_position] {
1579 for lex in lexemes {
1580 let mut entries = map.get(&lex.word).cloned().unwrap_or_default();
1581 entries.push(RowLocator::Hot(i));
1582 map.insert_mut(lex.word.clone(), entries);
1583 }
1584 }
1585 }
1586 }
1587 self.indices.push(idx);
1588 Ok(())
1589 }
1590
1591 /// v7.12.3 — Restore a GIN index from a deserialised snapshot.
1592 /// Mirrors [`Self::restore_btree_index`] but takes the GIN's
1593 /// `word → Vec<RowLocator>` posting-list map (already populated
1594 /// from the catalog stream) instead of an `IndexKey` map.
1595 pub fn restore_gin_index(
1596 &mut self,
1597 name: String,
1598 column_name: &str,
1599 map: PersistentBTreeMap<String, Vec<RowLocator>>,
1600 ) -> Result<(), StorageError> {
1601 if self.indices.iter().any(|i| i.name == name) {
1602 return Err(StorageError::DuplicateIndex { name });
1603 }
1604 let column_position = self.schema.column_position(column_name).ok_or_else(|| {
1605 StorageError::ColumnNotFound {
1606 column: column_name.into(),
1607 }
1608 })?;
1609 let mut idx = Index::new_gin(name, column_position);
1610 idx.kind = IndexKind::Gin(map);
1611 self.indices.push(idx);
1612 Ok(())
1613 }
1614
1615 /// v7.15.0 — `gin_trgm_ops` GIN over a TEXT column. Walks
1616 /// every row, shingles the cell into PG-compatible trigrams,
1617 /// and builds the posting-list map. NULL / non-TEXT cells
1618 /// contribute nothing (no trigrams).
1619 pub fn add_gin_trgm_index(
1620 &mut self,
1621 name: String,
1622 column_name: &str,
1623 ) -> Result<(), StorageError> {
1624 if self.indices.iter().any(|i| i.name == name) {
1625 return Err(StorageError::DuplicateIndex { name });
1626 }
1627 let column_position = self.schema.column_position(column_name).ok_or_else(|| {
1628 StorageError::ColumnNotFound {
1629 column: column_name.into(),
1630 }
1631 })?;
1632 if !matches!(
1633 self.schema.columns[column_position].ty,
1634 DataType::Text | DataType::Varchar(_)
1635 ) {
1636 return Err(StorageError::Corrupt(format!(
1637 "trigram-GIN index {name:?} requires a TEXT/VARCHAR column; \
1638 {column_name:?} is {:?}",
1639 self.schema.columns[column_position].ty
1640 )));
1641 }
1642 let mut idx = Index::new_gin_trgm(name, column_position);
1643 if let IndexKind::GinTrgm(map) = &mut idx.kind {
1644 for (i, row) in self.rows.iter().enumerate() {
1645 if let Value::Text(s) = &row.values[column_position] {
1646 for tri in trgm::extract_trigrams(s) {
1647 let mut entries = map.get(&tri).cloned().unwrap_or_default();
1648 entries.push(RowLocator::Hot(i));
1649 map.insert_mut(tri, entries);
1650 }
1651 }
1652 }
1653 }
1654 self.indices.push(idx);
1655 Ok(())
1656 }
1657
1658 /// v7.15.0 — restore a trigram-GIN from its catalog snapshot
1659 /// payload. Mirrors [`Self::restore_gin_index`].
1660 pub fn restore_gin_trgm_index(
1661 &mut self,
1662 name: String,
1663 column_name: &str,
1664 map: PersistentBTreeMap<String, Vec<RowLocator>>,
1665 ) -> Result<(), StorageError> {
1666 if self.indices.iter().any(|i| i.name == name) {
1667 return Err(StorageError::DuplicateIndex { name });
1668 }
1669 let column_position = self.schema.column_position(column_name).ok_or_else(|| {
1670 StorageError::ColumnNotFound {
1671 column: column_name.into(),
1672 }
1673 })?;
1674 let mut idx = Index::new_gin_trgm(name, column_position);
1675 idx.kind = IndexKind::GinTrgm(map);
1676 self.indices.push(idx);
1677 Ok(())
1678 }
1679
1680 /// v7.17.0 Phase 2.2 — MySQL `FULLTEXT KEY` GIN over a TEXT
1681 /// column. Walks every row, tokenises the cell into lower-
1682 /// cased word lexemes (`fts_simple::simple_lex` — same rule
1683 /// as `to_tsvector('simple', text)`), and builds the
1684 /// posting-list map. NULL / non-TEXT cells contribute
1685 /// nothing (no lexemes).
1686 pub fn add_gin_fulltext_index(
1687 &mut self,
1688 name: String,
1689 column_name: &str,
1690 ) -> Result<(), StorageError> {
1691 if self.indices.iter().any(|i| i.name == name) {
1692 return Err(StorageError::DuplicateIndex { name });
1693 }
1694 let column_position = self.schema.column_position(column_name).ok_or_else(|| {
1695 StorageError::ColumnNotFound {
1696 column: column_name.into(),
1697 }
1698 })?;
1699 if !matches!(
1700 self.schema.columns[column_position].ty,
1701 DataType::Text | DataType::Varchar(_)
1702 ) {
1703 return Err(StorageError::Corrupt(format!(
1704 "fulltext-GIN index {name:?} requires a TEXT/VARCHAR column; \
1705 {column_name:?} is {:?}",
1706 self.schema.columns[column_position].ty
1707 )));
1708 }
1709 let mut idx = Index::new_gin_fulltext(name, column_position);
1710 if let IndexKind::GinFulltext(map) = &mut idx.kind {
1711 for (i, row) in self.rows.iter().enumerate() {
1712 if let Value::Text(s) = &row.values[column_position] {
1713 for lex in fts_simple::simple_lex(s) {
1714 let mut entries = map.get(&lex).cloned().unwrap_or_default();
1715 entries.push(RowLocator::Hot(i));
1716 map.insert_mut(lex, entries);
1717 }
1718 }
1719 }
1720 }
1721 self.indices.push(idx);
1722 Ok(())
1723 }
1724
1725 /// v7.17.0 Phase 2.2 — restore a fulltext-GIN from its
1726 /// catalog snapshot payload. Mirrors
1727 /// [`Self::restore_gin_trgm_index`].
1728 pub fn restore_gin_fulltext_index(
1729 &mut self,
1730 name: String,
1731 column_name: &str,
1732 map: PersistentBTreeMap<String, Vec<RowLocator>>,
1733 ) -> Result<(), StorageError> {
1734 if self.indices.iter().any(|i| i.name == name) {
1735 return Err(StorageError::DuplicateIndex { name });
1736 }
1737 let column_position = self.schema.column_position(column_name).ok_or_else(|| {
1738 StorageError::ColumnNotFound {
1739 column: column_name.into(),
1740 }
1741 })?;
1742 let mut idx = Index::new_gin_fulltext(name, column_position);
1743 idx.kind = IndexKind::GinFulltext(map);
1744 self.indices.push(idx);
1745 Ok(())
1746 }
1747
1748 /// v7.37.8(sentori Epic 5 P2)— JSONB-GIN over a `Json` /
1749 /// `Jsonb` column. Walks every row, extracts canonical
1750 /// `(path, leaf)` tokens via
1751 /// [`crate::jsonb_gin::extract_tokens`], and builds the
1752 /// posting-list map. NULL or non-Json cells contribute no
1753 /// tokens(`<col> @> <jsonb>` against a NULL row is always
1754 /// false so absence here is correct).
1755 pub fn add_gin_jsonb_index(
1756 &mut self,
1757 name: String,
1758 column_name: &str,
1759 ) -> Result<(), StorageError> {
1760 if self.indices.iter().any(|i| i.name == name) {
1761 return Err(StorageError::DuplicateIndex { name });
1762 }
1763 let column_position = self.schema.column_position(column_name).ok_or_else(|| {
1764 StorageError::ColumnNotFound {
1765 column: column_name.into(),
1766 }
1767 })?;
1768 if !matches!(
1769 self.schema.columns[column_position].ty,
1770 DataType::Json | DataType::Jsonb
1771 ) {
1772 return Err(StorageError::Corrupt(format!(
1773 "JSONB-GIN index {name:?} requires a JSON/JSONB column; \
1774 {column_name:?} is {:?}",
1775 self.schema.columns[column_position].ty
1776 )));
1777 }
1778 let mut idx = Index::new_gin_jsonb(name, column_position);
1779 if let IndexKind::GinJsonb(map) = &mut idx.kind {
1780 for (i, row) in self.rows.iter().enumerate() {
1781 if let Value::Json(s) = &row.values[column_position] {
1782 for tok in jsonb_gin::extract_tokens(s) {
1783 let mut entries = map.get(&tok).cloned().unwrap_or_default();
1784 entries.push(RowLocator::Hot(i));
1785 map.insert_mut(tok, entries);
1786 }
1787 }
1788 }
1789 }
1790 self.indices.push(idx);
1791 Ok(())
1792 }
1793
1794 /// v7.37.8 — restore a JSONB-GIN from its catalog snapshot
1795 /// payload. Mirrors [`Self::restore_gin_fulltext_index`].
1796 pub fn restore_gin_jsonb_index(
1797 &mut self,
1798 name: String,
1799 column_name: &str,
1800 map: PersistentBTreeMap<String, Vec<RowLocator>>,
1801 ) -> Result<(), StorageError> {
1802 if self.indices.iter().any(|i| i.name == name) {
1803 return Err(StorageError::DuplicateIndex { name });
1804 }
1805 let column_position = self.schema.column_position(column_name).ok_or_else(|| {
1806 StorageError::ColumnNotFound {
1807 column: column_name.into(),
1808 }
1809 })?;
1810 let mut idx = Index::new_gin_jsonb(name, column_position);
1811 idx.kind = IndexKind::GinJsonb(map);
1812 self.indices.push(idx);
1813 Ok(())
1814 }
1815
1816 /// v5.1: register cold-tier locators on a `BTree` index. Used
1817 /// after [`Catalog::load_segment_bytes`] to wire every cold-
1818 /// tier row's PK back to its segment so
1819 /// [`Catalog::lookup_by_pk`] can resolve it. Each call
1820 /// appends to the index — keys that already have hot or cold
1821 /// locators keep them. Returns the number of locators
1822 /// registered.
1823 ///
1824 /// Pre-v5.2 (freezer) this is the only path that adds Cold
1825 /// variants to a PB; post-freezer the background freezer
1826 /// thread produces these as a batch under the engine write
1827 /// lock and this API becomes its in-memory primitive.
1828 ///
1829 /// Errors if `index_name` doesn't exist or names an NSW graph
1830 /// (NSW indices don't carry per-key row locators — they're
1831 /// vector-search structures).
1832 pub fn register_cold_locators<I>(
1833 &mut self,
1834 index_name: &str,
1835 locators: I,
1836 ) -> Result<usize, StorageError>
1837 where
1838 I: IntoIterator<Item = (IndexKey, RowLocator)>,
1839 {
1840 let idx = self
1841 .indices
1842 .iter_mut()
1843 .find(|i| i.name == index_name)
1844 .ok_or_else(|| StorageError::Corrupt(format!("index {index_name:?} not found")))?;
1845 let map = match &mut idx.kind {
1846 IndexKind::BTree(map) => map,
1847 IndexKind::Nsw(_)
1848 | IndexKind::Brin { .. }
1849 | IndexKind::Gin(_)
1850 | IndexKind::GinTrgm(_)
1851 | IndexKind::GinFulltext(_)
1852 | IndexKind::GinJsonb(_) => {
1853 return Err(StorageError::Corrupt(format!(
1854 "index {index_name:?} is not BTree; cold locators apply only to BTree indices"
1855 )));
1856 }
1857 };
1858 let mut count = 0usize;
1859 for (key, locator) in locators {
1860 let mut entries = map.get(&key).cloned().unwrap_or_default();
1861 entries.push(locator);
1862 map.insert_mut(key, entries);
1863 count += 1;
1864 }
1865 Ok(count)
1866 }
1867
1868 /// v7.12.3 — GIN-side parallel to [`Self::register_cold_locators`].
1869 /// Re-attaches `word → cold RowLocator` posting-list entries after
1870 /// the from-rows rebuild loop. Errors when the index doesn't
1871 /// exist or isn't a GIN. Both tsvector-GIN and trigram-GIN
1872 /// variants share posting-list shape (`String → Vec<RowLocator>`),
1873 /// so this helper accepts either.
1874 pub fn register_gin_cold_locators<I>(
1875 &mut self,
1876 index_name: &str,
1877 locators: I,
1878 ) -> Result<usize, StorageError>
1879 where
1880 I: IntoIterator<Item = (String, RowLocator)>,
1881 {
1882 let idx = self
1883 .indices
1884 .iter_mut()
1885 .find(|i| i.name == index_name)
1886 .ok_or_else(|| StorageError::Corrupt(format!("index {index_name:?} not found")))?;
1887 let map = match &mut idx.kind {
1888 // v7.17.0 Phase 2.2 — fulltext-GIN posting lists are
1889 // shape-compatible with tsvector / trigram GINs, so
1890 // cold-locator re-attach handles all three.
1891 // v7.37.8 — JSONB-GIN shares the same posting-list shape,
1892 // so it joins the same re-attach path.
1893 IndexKind::Gin(map)
1894 | IndexKind::GinTrgm(map)
1895 | IndexKind::GinFulltext(map)
1896 | IndexKind::GinJsonb(map) => map,
1897 IndexKind::BTree(_) | IndexKind::Nsw(_) | IndexKind::Brin { .. } => {
1898 return Err(StorageError::Corrupt(format!(
1899 "register_gin_cold_locators: index {index_name:?} is not GIN"
1900 )));
1901 }
1902 };
1903 let mut count = 0usize;
1904 for (word, locator) in locators {
1905 let mut entries = map.get(&word).cloned().unwrap_or_default();
1906 entries.push(locator);
1907 map.insert_mut(word, entries);
1908 count += 1;
1909 }
1910 Ok(count)
1911 }
1912
1913 /// v5.2.3: remove every `Cold` locator currently registered on
1914 /// `index_name` under the given `key`. `Hot` locators for the
1915 /// same key are left in place — useful when a row has just been
1916 /// promoted hot-side and the caller wants the old Cold pointer
1917 /// retired without losing the new hot entry.
1918 ///
1919 /// Returns the number of cold locators removed (0 when the key
1920 /// has only hot entries or the key isn't present at all).
1921 /// Errors when the index doesn't exist or isn't a `BTree`.
1922 pub fn remove_cold_locators_for_key(
1923 &mut self,
1924 index_name: &str,
1925 key: &IndexKey,
1926 ) -> Result<usize, StorageError> {
1927 let idx = self
1928 .indices
1929 .iter_mut()
1930 .find(|i| i.name == index_name)
1931 .ok_or_else(|| {
1932 StorageError::Corrupt(format!(
1933 "remove_cold_locators_for_key: index {index_name:?} not found"
1934 ))
1935 })?;
1936 let map = match &mut idx.kind {
1937 IndexKind::BTree(map) => map,
1938 IndexKind::Nsw(_)
1939 | IndexKind::Brin { .. }
1940 | IndexKind::Gin(_)
1941 | IndexKind::GinTrgm(_)
1942 | IndexKind::GinFulltext(_)
1943 | IndexKind::GinJsonb(_) => {
1944 return Err(StorageError::Corrupt(format!(
1945 "remove_cold_locators_for_key: index {index_name:?} is not BTree; \
1946 cold locators apply only to BTree indices"
1947 )));
1948 }
1949 };
1950 let Some(entries) = map.get(key) else {
1951 return Ok(0);
1952 };
1953 let mut kept: Vec<RowLocator> =
1954 entries.iter().copied().filter(RowLocator::is_hot).collect();
1955 let removed = entries.len() - kept.len();
1956 if removed == 0 {
1957 return Ok(0);
1958 }
1959 kept.shrink_to_fit();
1960 // PersistentBTreeMap has no remove API in v5.2; when every
1961 // locator for `key` was Cold, the key keeps an empty Vec
1962 // entry. `Index::lookup_eq` already treats `Some(&[])` and
1963 // `None` as the same empty slice (via `Vec::as_slice`), so
1964 // callers can't distinguish the two. The space cost is one
1965 // empty Vec per shadowed-then-promoted key — bounded and
1966 // recoverable when the future compaction job lands.
1967 map.insert_mut(key.clone(), kept);
1968 Ok(removed)
1969 }
1970
1971 /// v7.13.0 — append a new column to the schema and back-fill
1972 /// every existing row with `fill_value`. Used by the engine's
1973 /// `ALTER TABLE t ADD COLUMN …` handler (mailrs round-5 G1).
1974 /// Indices on existing columns keep working — column positions
1975 /// don't shift since the new column lands at the end — so no
1976 /// index rebuild is needed.
1977 pub fn add_column(&mut self, col: ColumnSchema, fill_value: Value<'static>) {
1978 self.schema.columns.push(col);
1979 let mut new_rows: PersistentVec<Row<'static>> = PersistentVec::new();
1980 for row in self.rows.iter() {
1981 let mut values = row.values.clone();
1982 values.push(fill_value.clone());
1983 new_rows.push_mut(Row::new(values));
1984 }
1985 self.rows = new_rows;
1986 }
1987
1988 /// v7.15.0 — replace the partial-index predicate source on
1989 /// the index at slot `idx`. Used by `ALTER TABLE … RENAME
1990 /// COLUMN` after the engine rewrites column-identifier
1991 /// references in the predicate source text. Pure metadata
1992 /// edit; index rows are unaffected (they're keyed by
1993 /// column position, not predicate text).
1994 pub fn set_partial_predicate(&mut self, idx: usize, pred: Option<String>) {
1995 debug_assert!(idx < self.indices.len());
1996 self.indices[idx].partial_predicate = pred;
1997 }
1998
1999 /// v7.15.0 — rename the column at `col_pos` to `new_name`.
2000 /// The on-disk row encoding is positional, so no row rewrite
2001 /// is needed; only the schema's column name changes. Indices,
2002 /// UCs, FKs all key off column positions and are unaffected.
2003 /// Source-text references that hold the column name (CHECK
2004 /// predicates, partial-index predicates, runtime DEFAULT
2005 /// expressions, trigger `UPDATE OF` lists) are rewritten by
2006 /// the engine before this helper is called — the storage
2007 /// layer doesn't depend on `spg-sql` and so can't re-parse the
2008 /// predicate sources itself.
2009 pub fn rename_column(&mut self, col_pos: usize, new_name: &str) {
2010 debug_assert!(col_pos < self.schema.columns.len());
2011 self.schema.columns[col_pos].name = new_name.to_string();
2012 }
2013
2014 /// v7.13.3 — drop the column at `col_pos`. Removes the entry
2015 /// from the schema, the value from every row, any index that
2016 /// references the column (pure drop, not shift), and shifts
2017 /// every remaining index/UC/FK column position that pointed
2018 /// past `col_pos` down by one. Used by `ALTER TABLE t DROP
2019 /// COLUMN <c>` (mailrs round-7 S8). FK dependents on this
2020 /// column must already have been removed by the caller (CASCADE
2021 /// path); the helper assumes only same-column index removal is
2022 /// needed.
2023 pub fn drop_column(&mut self, col_pos: usize) {
2024 debug_assert!(col_pos < self.schema.columns.len());
2025 // v7.39 (round 215) — dropping a column shifts every later column's
2026 // position, which would leave a range-exclusion index pointing at the
2027 // wrong column. Drop the indexes rather than risk a silent-wrong
2028 // probe; enforce falls back to the correct O(n) scan until they are
2029 // rebuilt (`ensure_excl_range_index` from the constraint's updated
2030 // column position).
2031 self.excl_indexes.clear();
2032 // Strip the column from the schema.
2033 self.schema.columns.remove(col_pos);
2034 // Rewrite every row to omit the cell at col_pos.
2035 let mut new_rows: PersistentVec<Row> = PersistentVec::new();
2036 for row in self.rows.iter() {
2037 let mut values = row.values.clone();
2038 if col_pos < values.len() {
2039 values.remove(col_pos);
2040 }
2041 new_rows.push_mut(Row::new(values));
2042 }
2043 self.rows = new_rows;
2044 // Drop indices on the column outright; shift the rest.
2045 self.indices.retain(|idx| idx.column_position != col_pos);
2046 for idx in &mut self.indices {
2047 if idx.column_position > col_pos {
2048 idx.column_position -= 1;
2049 }
2050 // Same shift for any included-columns reference.
2051 for inc in &mut idx.included_columns {
2052 if *inc > col_pos {
2053 *inc -= 1;
2054 }
2055 }
2056 }
2057 // Shift uniqueness-constraint column positions (and drop
2058 // entries that lose all columns, though that shouldn't
2059 // happen in practice — caller has already CASCADE-removed
2060 // FKs and there's no general CASCADE for UCs).
2061 let mut surviving_ucs: Vec<UniquenessConstraint> = Vec::new();
2062 for mut uc in core::mem::take(&mut self.schema.uniqueness_constraints) {
2063 uc.columns.retain(|&c| c != col_pos);
2064 if uc.columns.is_empty() {
2065 continue;
2066 }
2067 for c in &mut uc.columns {
2068 if *c > col_pos {
2069 *c -= 1;
2070 }
2071 }
2072 surviving_ucs.push(uc);
2073 }
2074 self.schema.uniqueness_constraints = surviving_ucs;
2075 // Shift FK local_columns (parent-pointing column positions
2076 // are off-table and untouched).
2077 for fk in &mut self.schema.foreign_keys {
2078 for c in &mut fk.local_columns {
2079 if *c > col_pos {
2080 *c -= 1;
2081 }
2082 }
2083 }
2084 // Rebuild remaining indices' payload — the column-position
2085 // shift means existing IndexKey entries are still keyed by
2086 // the same column data but the position numbers changed;
2087 // existing key→locator maps stay valid because they're
2088 // keyed by Value not position. The rebuild is conservative
2089 // — same pattern delete_rows uses post-mutation.
2090 self.rebuild_indices();
2091 }
2092
2093 /// v4.4: delete the rows at the given positions in one pass.
2094 /// `positions` must be unique; ordering doesn't matter. Indices
2095 /// are rebuilt from scratch (cheaper than tracking incremental
2096 /// shifts across both B-tree and NSW). Returns the number of
2097 /// rows removed.
2098 /// v7.17.0 Phase 1.3 — wipe every row. Used by REFRESH
2099 /// MATERIALIZED VIEW; same effect as `delete_rows((0..N).into())`
2100 /// but skips the per-position bookkeeping for the all-removed
2101 /// fast path. Indices are rebuilt (empty).
2102 pub fn truncate(&mut self) {
2103 self.rows = PersistentVec::new();
2104 // v7.37.15 (Phase A.2) — keep headers lock-step.
2105 self.headers = PersistentVec::new();
2106 // v7.37.15 (Phase C.1) — clear rowids lock-step. `next_rowid`
2107 // is NOT reset: ids stay globally monotonic within the
2108 // relation so a post-truncate insert never reuses a pre-
2109 // truncate id that a stale reference might still name.
2110 self.rowids = PersistentVec::new();
2111 self.hot_bytes = 0;
2112 self.rebuild_indices();
2113 }
2114
2115 pub fn delete_rows(&mut self, positions: &[usize]) -> usize {
2116 // v7.37.15 (Epic W slice 1) — capture the RowIds of the targeted
2117 // rows BEFORE the deletion shifts them out. One id per input
2118 // position (parallel to `positions`), `RowId::UNASSIGNED` for an
2119 // out-of-bounds position. Only pay for it when redo capture is
2120 // on. `writer_version` (xmax) is 0: the deleting TxId is not
2121 // threaded to this layer yet.
2122 let redo_rowids: Vec<crate::row_header::RowId> = if self.redo_log.is_some() {
2123 positions
2124 .iter()
2125 .map(|&p| {
2126 self.rowids()
2127 .get(p)
2128 .copied()
2129 .unwrap_or(crate::row_header::RowId::UNASSIGNED)
2130 })
2131 .collect()
2132 } else {
2133 Vec::new()
2134 };
2135 let removed = self.delete_rows_no_index(positions);
2136 if removed > 0 {
2137 self.rebuild_indices();
2138 // v7.34 — capture row-level redo. Record the input positions
2139 // (replay's `delete_rows` dedups + bounds-filters identically);
2140 // skip a no-op delete so the log stays minimal.
2141 self.record_redo(move |table| RowChange::Delete {
2142 table,
2143 positions: positions.to_vec(),
2144 rowids: redo_rowids,
2145 writer_version: 0,
2146 });
2147 }
2148 removed
2149 }
2150
2151 /// v7.37.5 (mailrs crash-recovery Ask 3) — row-only delete for the
2152 /// WAL-replay batch path: removes the rows + decrements `hot_bytes`,
2153 /// **does NOT** call `rebuild_indices()` and does **NOT** capture
2154 /// redo. The caller is responsible for invoking `rebuild_indices_pub`
2155 /// once after a sequence of `*_no_index` mutations on this table.
2156 /// Skipping the per-call rebuild closes the
2157 /// O(records × rows × indices × log rows) replay blow-up
2158 /// (5000 DELETEs × 100k × 13 × ln 100k ≈ minutes → seconds).
2159 /// Returns the number of rows actually removed (dedup + bounds-
2160 /// filtered identically to `delete_rows`).
2161 pub fn delete_rows_no_index(&mut self, positions: &[usize]) -> usize {
2162 if positions.is_empty() {
2163 return 0;
2164 }
2165 // Mark positions; v4.39: PV has no in-place retain, so we rebuild
2166 // a fresh PV by pushing the survivors. Still O(n log₃₂ n); the
2167 // structural-sharing win shows up at `Catalog::clone()`, not here.
2168 let mut to_remove = alloc::vec![false; self.rows.len()];
2169 let mut removed = 0;
2170 for &p in positions {
2171 if p < to_remove.len() && !to_remove[p] {
2172 to_remove[p] = true;
2173 removed += 1;
2174 }
2175 }
2176 if removed == 0 {
2177 return 0;
2178 }
2179 let mut new_rows: PersistentVec<Row> = PersistentVec::new();
2180 let mut new_headers: PersistentVec<crate::row_header::RowHeader> = PersistentVec::new();
2181 // v7.37.15 (Phase C.1) — survivors carry their stable RowId
2182 // across the compaction so a held lock / redo reference keeps
2183 // naming the same row while its physical slot shifts down.
2184 let mut new_rowids: PersistentVec<crate::row_header::RowId> = PersistentVec::new();
2185 let mut removed_bytes: u64 = 0;
2186 // v7.37.16 (autovacuum) — recount dead survivors: this rebuild
2187 // is the compaction hub (vacuum and physical delete both land
2188 // here), so the incremental counter re-bases exactly.
2189 let mut surviving_dead: u64 = 0;
2190 for (i, row) in self.rows.iter().enumerate() {
2191 if to_remove[i] {
2192 removed_bytes =
2193 removed_bytes.saturating_add(row_body_encoded_len(row, &self.schema) as u64);
2194 } else {
2195 new_rows.push_mut(row.clone());
2196 // v7.37.15 (Phase A.2) — keep headers lock-step.
2197 // Phase C will stamp xmax with the deleting tx's
2198 // id INSTEAD of physically dropping the row; Phase
2199 // A.2 keeps physical-delete semantics so
2200 // serialisation + WAL paths stay identical.
2201 if let Some(h) = self.headers.get(i) {
2202 if h.xmax != crate::row_header::XMAX_ALIVE {
2203 surviving_dead += 1;
2204 }
2205 new_headers.push_mut(*h);
2206 } else {
2207 new_headers.push_mut(crate::row_header::RowHeader::frozen());
2208 }
2209 if let Some(rid) = self.rowids.get(i) {
2210 new_rowids.push_mut(*rid);
2211 } else {
2212 // Should not happen once C.1 is wired everywhere;
2213 // allocate a fresh id as a defensive fallback so
2214 // the lock-step invariant survives a legacy path.
2215 let rid = crate::row_header::RowId(self.next_rowid);
2216 self.next_rowid += 1;
2217 new_rowids.push_mut(rid);
2218 }
2219 }
2220 }
2221 self.rows = new_rows;
2222 self.headers = new_headers;
2223 self.rowids = new_rowids;
2224 self.hot_bytes = self.hot_bytes.saturating_sub(removed_bytes);
2225 self.dead_rows = surviving_dead;
2226 debug_assert_eq!(
2227 self.rows.len(),
2228 self.headers.len(),
2229 "headers must stay in lock-step with rows after delete_rows_no_index"
2230 );
2231 removed
2232 }
2233
2234 /// v7.37.5 — public alias for the private `rebuild_indices` helper.
2235 /// Used by `Catalog::apply_redo` to coalesce per-record rebuilds
2236 /// across a batch of `RowChange`s into one rebuild per touched table.
2237 pub fn rebuild_indices_pub(&mut self) {
2238 self.rebuild_indices();
2239 }
2240
2241 /// v7.37.5 (mailrs crash-recovery Ask 3) — replace the table's
2242 /// row vector + `hot_bytes` in one shot, then rebuild every
2243 /// index from the new rows. Used by `Catalog::apply_redo`'s
2244 /// batched run: a contiguous slice of `RowChange`s targeting
2245 /// this table is composed into a final `(PersistentVec<Row>,
2246 /// hot_bytes)` pair via in-memory bookkeeping, then handed to
2247 /// this method ONCE for index regeneration. Replaces N per-
2248 /// record `rebuild_indices` calls with 1 per run.
2249 /// v7.39 (flip crash-replay P0) — like
2250 /// [`Self::set_rows_and_rebuild_indices`] but KEEPS the caller's
2251 /// per-slot RowIds. Redo replay applies one WAL record per
2252 /// statement; reassigning ids between records broke every later
2253 /// record's tombstone targets (they name the ids the crashed
2254 /// process allocated), resurrecting deleted rows. The id
2255 /// allocator advances past every preserved id so post-replay
2256 /// inserts never collide.
2257 pub fn set_rows_and_rebuild_indices_with_rowids(
2258 &mut self,
2259 new_rows: PersistentVec<Row<'static>>,
2260 new_hot_bytes: u64,
2261 rowids: &[crate::row_header::RowId],
2262 headers: &[crate::row_header::RowHeader],
2263 ) {
2264 debug_assert_eq!(new_rows.len(), rowids.len());
2265 debug_assert_eq!(new_rows.len(), headers.len());
2266 let mut new_headers: PersistentVec<crate::row_header::RowHeader> = PersistentVec::new();
2267 let mut new_rowids: PersistentVec<crate::row_header::RowId> = PersistentVec::new();
2268 let mut dead: u64 = 0;
2269 for (rid, h) in rowids.iter().zip(headers) {
2270 // Preserve the caller's header — an earlier replayed WAL
2271 // record's tombstone stamp must survive this record's
2272 // rebuild (per-statement replay re-freezing every header
2273 // resurrected every previously-deleted row).
2274 if h.xmax != crate::row_header::XMAX_ALIVE {
2275 dead += 1;
2276 }
2277 new_headers.push_mut(*h);
2278 let rid = if *rid == crate::row_header::RowId::UNASSIGNED {
2279 let fresh = crate::row_header::RowId(self.next_rowid);
2280 self.next_rowid += 1;
2281 fresh
2282 } else {
2283 if rid.0 >= self.next_rowid {
2284 self.next_rowid = rid.0 + 1;
2285 }
2286 *rid
2287 };
2288 new_rowids.push_mut(rid);
2289 }
2290 self.rows = new_rows;
2291 self.headers = new_headers;
2292 self.rowids = new_rowids;
2293 self.hot_bytes = new_hot_bytes;
2294 self.dead_rows = dead;
2295 debug_assert_eq!(self.rows.len(), self.headers.len());
2296 debug_assert_eq!(self.rows.len(), self.rowids.len());
2297 self.rebuild_indices();
2298 }
2299
2300 pub fn set_rows_and_rebuild_indices(
2301 &mut self,
2302 new_rows: PersistentVec<Row<'static>>,
2303 new_hot_bytes: u64,
2304 ) {
2305 // v7.37.15 (Phase A.2) — synthesise frozen headers for
2306 // the replacement rows. Phase D's catalog snapshot format
2307 // (bumped to V6) will start carrying headers verbatim,
2308 // letting recovery preserve real xmin/xmax instead of
2309 // freezing everything; until then frozen is the safe
2310 // default for replay (all visible to every snapshot).
2311 let mut new_headers: PersistentVec<crate::row_header::RowHeader> = PersistentVec::new();
2312 // v7.37.15 (Phase C.1) — fresh monotonic ids for the
2313 // replacement rows drawn from the relation allocator, so a
2314 // post-replay id never collides with a pre-replay one.
2315 let mut new_rowids: PersistentVec<crate::row_header::RowId> = PersistentVec::new();
2316 for _ in 0..new_rows.len() {
2317 new_headers.push_mut(crate::row_header::RowHeader::frozen());
2318 let rid = crate::row_header::RowId(self.next_rowid);
2319 self.next_rowid += 1;
2320 new_rowids.push_mut(rid);
2321 }
2322 self.rows = new_rows;
2323 self.headers = new_headers;
2324 self.rowids = new_rowids;
2325 self.hot_bytes = new_hot_bytes;
2326 // All-frozen replacement headers → no dead rows by construction.
2327 self.dead_rows = 0;
2328 debug_assert_eq!(
2329 self.rows.len(),
2330 self.headers.len(),
2331 "headers must stay in lock-step with rows after set_rows_and_rebuild_indices"
2332 );
2333 debug_assert_eq!(
2334 self.rows.len(),
2335 self.rowids.len(),
2336 "rowids must stay in lock-step with rows after set_rows_and_rebuild_indices"
2337 );
2338 self.rebuild_indices();
2339 }
2340
2341 /// v7.37.5 (mailrs crash-recovery Ask 3) — row-only insert for the
2342 /// WAL-replay batch path: pushes the row + bumps `hot_bytes`, and
2343 /// **does NOT** update any index (B-tree, GIN, NSW). The caller is
2344 /// responsible for invoking `rebuild_indices_pub` once after a
2345 /// sequence of `*_no_index` mutations on this table.
2346 /// Schema validation (arity + per-column type compatibility) is
2347 /// applied so a malformed redo log surfaces honestly.
2348 pub fn insert_no_index(&mut self, row: Row<'static>) -> Result<(), StorageError> {
2349 if row.len() != self.schema.columns.len() {
2350 return Err(StorageError::ArityMismatch {
2351 expected: self.schema.columns.len(),
2352 actual: row.len(),
2353 });
2354 }
2355 validate_row_against_schema(&row.values, &self.schema)?;
2356 self.hot_bytes = self
2357 .hot_bytes
2358 .saturating_add(row_body_encoded_len(&row, &self.schema) as u64);
2359 self.rows.push_mut(row);
2360 // v7.37.15 (Phase A.2) — keep headers lock-step for the
2361 // WAL replay path. Replay-time headers are frozen because
2362 // pre-V6 envelopes carry no header info; Phase D will
2363 // restore the original xmin/xmax once the V6 catalog
2364 // format ships.
2365 self.headers
2366 .push_mut(crate::row_header::RowHeader::frozen());
2367 // v7.37.15 (Phase C.1) — RowId lock-step for the WAL-replay
2368 // append path.
2369 let rid = self.alloc_rowid();
2370 self.rowids.push_mut(rid);
2371 debug_assert_eq!(
2372 self.rows.len(),
2373 self.headers.len(),
2374 "headers must stay in lock-step with rows after insert_no_index"
2375 );
2376 debug_assert_eq!(
2377 self.rows.len(),
2378 self.rowids.len(),
2379 "rowids must stay in lock-step with rows after insert_no_index"
2380 );
2381 Ok(())
2382 }
2383
2384 /// v7.37.5 (mailrs crash-recovery Ask 3) — row-only update for the
2385 /// WAL-replay batch path: replaces the row at `position` + adjusts
2386 /// `hot_bytes`, and **does NOT** touch any index. Skipping the
2387 /// per-update incremental index work is safe because the trailing
2388 /// `rebuild_indices_pub` regenerates indices from `self.rows` in
2389 /// their final state.
2390 pub fn update_row_no_index(
2391 &mut self,
2392 position: usize,
2393 new_values: Vec<Value<'static>>,
2394 ) -> Result<(), StorageError> {
2395 if position >= self.rows.len() {
2396 return Err(StorageError::Corrupt(alloc::format!(
2397 "update_row_no_index: position {position} out of bounds (rows={})",
2398 self.rows.len()
2399 )));
2400 }
2401 if new_values.len() != self.schema.columns.len() {
2402 return Err(StorageError::ArityMismatch {
2403 expected: self.schema.columns.len(),
2404 actual: new_values.len(),
2405 });
2406 }
2407 validate_row_against_schema(&new_values, &self.schema)?;
2408 let old_row = self
2409 .rows
2410 .get(position)
2411 .expect("position bounds-checked above");
2412 let old_bytes = row_body_encoded_len(old_row, &self.schema) as u64;
2413 let new_row = Row::new(new_values);
2414 let new_bytes = row_body_encoded_len(&new_row, &self.schema) as u64;
2415 self.rows = self
2416 .rows
2417 .set(position, new_row)
2418 .expect("position bounds-checked above");
2419 self.hot_bytes = self
2420 .hot_bytes
2421 .saturating_sub(old_bytes)
2422 .saturating_add(new_bytes);
2423 Ok(())
2424 }
2425
2426 /// v4.4: replace the row at `position` with `new_values` (must
2427 /// match the schema arity + types). v7.20: index maintenance is
2428 /// incremental — only indices whose key value changed are
2429 /// touched (B-tree entry move in place; NSW / BRIN / GIN fall
2430 /// back to a full rebuild when their column changed).
2431 pub fn update_row(
2432 &mut self,
2433 position: usize,
2434 new_values: Vec<Value<'static>>,
2435 ) -> Result<(), StorageError> {
2436 if position >= self.rows.len() {
2437 return Err(StorageError::Corrupt(alloc::format!(
2438 "update_row: position {position} out of bounds (rows={})",
2439 self.rows.len()
2440 )));
2441 }
2442 if new_values.len() != self.schema.columns.len() {
2443 return Err(StorageError::ArityMismatch {
2444 expected: self.schema.columns.len(),
2445 actual: new_values.len(),
2446 });
2447 }
2448 // Reuse the per-cell type-compat validation that `insert`
2449 // applies. The body below mirrors that check intentionally —
2450 // factoring it would be more code than the duplication.
2451 for (i, (val, col)) in new_values.iter().zip(&self.schema.columns).enumerate() {
2452 if val.is_null() {
2453 if !col.nullable {
2454 return Err(StorageError::NullInNotNull {
2455 column: col.name.clone(),
2456 });
2457 }
2458 continue;
2459 }
2460 // v7.39 (read01 round 54) — `data_type()` is None for the
2461 // eval-only variants that carry no DataType (RegClass, Composite).
2462 // They are NOT NULL, so `.expect("non-null")` PANICKED on them —
2463 // materialising a CTE like `WITH w AS (SELECT 't'::regclass)` blew
2464 // up the query with an "internal error". Report a clean type
2465 // mismatch instead; the engine coerces these before they get here
2466 // on every path that knows how.
2467 let Some(actual) = val.data_type() else {
2468 // An eval-only value (RegClass carries oid + name, Composite a
2469 // field tuple) has no DataType in the storage lattice. It is
2470 // NOT NULL, so the old `.expect("non-null")` PANICKED — which
2471 // is how `WITH w AS (SELECT 't'::regclass)` blew up with an
2472 // "internal error". Accept it: the value keeps its dual shape
2473 // and downstream comparisons (RegClass vs BigInt oid) handle it.
2474 continue;
2475 };
2476 let compatible = column_accepts(actual, col.ty);
2477 if !compatible {
2478 return Err(StorageError::TypeMismatch {
2479 column: col.name.clone(),
2480 expected: col.ty,
2481 actual,
2482 position: i,
2483 });
2484 }
2485 }
2486 let old_row = self
2487 .rows
2488 .get(position)
2489 .expect("position bounds-checked above");
2490 let old_bytes = row_body_encoded_len(old_row, &self.schema) as u64;
2491 let new_row = Row::new(new_values);
2492 let new_bytes = row_body_encoded_len(&new_row, &self.schema) as u64;
2493 // v7.20 P4 — incremental index maintenance. `rows.set`
2494 // replaces the row in place, so every OTHER row's Hot
2495 // locator stays valid; only indices whose key value
2496 // actually changed at `position` need touching. The
2497 // common OLTP shape (`UPDATE … SET non_indexed_col = …
2498 // WHERE pk = $1`) touches no index at all — pre-v7.20
2499 // this path paid a full rebuild_indices() (O(rows ×
2500 // indices)) per UPDATE, which dominated the profiled
2501 // write cost on a 5k-row table (~1 ms/stmt).
2502 //
2503 // BTree gets an in-place entry move (drop Hot(position)
2504 // from the old key's locator list, append to the new
2505 // key's). NSW graphs / BRIN summaries / GIN posting
2506 // lists have no cheap single-key move — a changed column
2507 // under one of those falls back to the full rebuild.
2508 enum IdxFix {
2509 BTreeMove {
2510 idx_pos: usize,
2511 old_key: Option<IndexKey>,
2512 new_key: Option<IndexKey>,
2513 },
2514 FullRebuild,
2515 }
2516 let mut fixes: Vec<IdxFix> = Vec::new();
2517 for (idx_pos, idx) in self.indices.iter().enumerate() {
2518 let col = idx.column_position;
2519 let old_v = &old_row.values[col];
2520 let new_v = &new_row.values[col];
2521 if old_v == new_v {
2522 continue;
2523 }
2524 match &idx.kind {
2525 IndexKind::BTree(_) => fixes.push(IdxFix::BTreeMove {
2526 idx_pos,
2527 old_key: IndexKey::from_value(old_v),
2528 new_key: IndexKey::from_value(new_v),
2529 }),
2530 IndexKind::Nsw(_)
2531 | IndexKind::Brin { .. }
2532 | IndexKind::Gin(_)
2533 | IndexKind::GinTrgm(_)
2534 | IndexKind::GinFulltext(_)
2535 | IndexKind::GinJsonb(_) => {
2536 fixes.clear();
2537 fixes.push(IdxFix::FullRebuild);
2538 break;
2539 }
2540 }
2541 }
2542 // v7.39 (round 215) — capture the range-exclusion key move BEFORE the
2543 // in-place `set` consumes `new_row`. A `FullRebuild` (a GIN/NSW/BRIN
2544 // column changed) rebuilds the excl indexes too via `rebuild_indices`,
2545 // so only apply the incremental move on the pure-BTreeMove path.
2546 let excl_has_full = fixes.iter().any(|f| matches!(f, IdxFix::FullRebuild));
2547 let excl_moves: Vec<(usize, Option<(i128, u8)>, Option<(i128, u8)>)> =
2548 if self.excl_indexes.is_empty() || excl_has_full {
2549 Vec::new()
2550 } else {
2551 self.excl_indexes
2552 .iter()
2553 .filter_map(|e| {
2554 let c = e.column_position;
2555 let old_k = old_row.values.get(c).and_then(crate::range_excl_index_key);
2556 let new_k = new_row.values.get(c).and_then(crate::range_excl_index_key);
2557 if old_k == new_k {
2558 None // range bound unchanged — no index touch
2559 } else {
2560 Some((c, old_k, new_k))
2561 }
2562 })
2563 .collect()
2564 };
2565 self.rows = self
2566 .rows
2567 .set(position, new_row)
2568 .expect("position bounds-checked above");
2569 self.hot_bytes = self
2570 .hot_bytes
2571 .saturating_sub(old_bytes)
2572 .saturating_add(new_bytes);
2573 // v7.34 — capture row-level redo (after the row is in place; the
2574 // immutable read of the new values is dropped before record_redo's
2575 // mutable borrow, and gated so capture-off pays nothing).
2576 if self.redo_log.is_some() {
2577 let new_row = self
2578 .rows
2579 .get(position)
2580 .map(|r| r.values.clone())
2581 .unwrap_or_default();
2582 // v7.37.15 (Epic W slice 1) — carry the stable RowId of the
2583 // updated row (`position` is bounds-checked above, so the id
2584 // is present). `writer_version` (xmax of the superseded
2585 // tuple) is 0: the writing TxId is not threaded here yet.
2586 let redo_rowid = self
2587 .rowids()
2588 .get(position)
2589 .copied()
2590 .unwrap_or(crate::row_header::RowId::UNASSIGNED);
2591 self.record_redo(|table| RowChange::Update {
2592 table,
2593 pos: position,
2594 new_row,
2595 rowid: redo_rowid,
2596 writer_version: 0,
2597 });
2598 }
2599 for fix in fixes {
2600 match fix {
2601 IdxFix::FullRebuild => {
2602 self.rebuild_indices();
2603 break;
2604 }
2605 IdxFix::BTreeMove {
2606 idx_pos,
2607 old_key,
2608 new_key,
2609 } => {
2610 let IndexKind::BTree(map) = &mut self.indices[idx_pos].kind else {
2611 unreachable!("IdxFix::BTreeMove built from a BTree index");
2612 };
2613 // NULL keys never enter the B-tree (from_value
2614 // returns None), so a None on either side means
2615 // "no entry on that side".
2616 if let Some(k) = old_key
2617 && let Some(locs) = map.get(&k)
2618 {
2619 let mut locs = locs.clone();
2620 locs.retain(|l| *l != RowLocator::Hot(position));
2621 // No remove_mut on the persistent map: an
2622 // empty locator list is the tombstone —
2623 // lookup_eq returns an empty slice, and the
2624 // next rebuild_indices() drops the key.
2625 map.insert_mut(k, locs);
2626 }
2627 if let Some(k) = new_key {
2628 let mut entries = map.get(&k).cloned().unwrap_or_default();
2629 entries.push(RowLocator::Hot(position));
2630 map.insert_mut(k, entries);
2631 }
2632 }
2633 }
2634 }
2635 // v7.39 (round 215) — apply the range-exclusion key moves captured
2636 // above (skipped when a FullRebuild already re-emitted every excl
2637 // index). Same shape as the BTreeMove: drop Hot(position) from the
2638 // old key, append it to the new key.
2639 for (col, old_k, new_k) in excl_moves {
2640 let Some(ex) = self
2641 .excl_indexes
2642 .iter_mut()
2643 .find(|e| e.column_position == col)
2644 else {
2645 continue;
2646 };
2647 if let Some(k) = old_k
2648 && let Some(locs) = ex.map.get(&k)
2649 {
2650 let mut locs = locs.clone();
2651 locs.retain(|l| *l != RowLocator::Hot(position));
2652 ex.map.insert_mut(k, locs);
2653 }
2654 if let Some(k) = new_k {
2655 let mut entries = ex.map.get(&k).cloned().unwrap_or_default();
2656 entries.push(RowLocator::Hot(position));
2657 ex.map.insert_mut(k, entries);
2658 }
2659 }
2660 Ok(())
2661 }
2662
2663 /// v4.4 helper used by `delete_rows` / `update_row`: discard all
2664 /// index payloads and rebuild from `self.rows`. Cheap enough
2665 /// for typical SPG scale (catalogs in the docker-compose
2666 /// deployment shape are small); the alternative — incremental
2667 /// shift bookkeeping across B-tree + NSW — would be far more
2668 /// invasive than the savings justify.
2669 fn rebuild_indices(&mut self) {
2670 // v5.2.3: capture every `Cold` locator on every BTree index
2671 // before the rebuild, so the from-rows re-emission below
2672 // (which only produces `Hot` locators) doesn't drop cold-
2673 // tier entries on keys unrelated to the row that changed.
2674 // Pre-v5.2.3 this was a `freeze_oldest_to_cold` worry only
2675 // and the freezer did its own capture-then-reregister; v5.2.3
2676 // promotes that pattern into the base helper because UPDATE
2677 // / DELETE now run rebuild_indices on tables with cold rows.
2678 let preserved_cold: Vec<(String, Vec<(IndexKey, RowLocator)>)> = self
2679 .indices
2680 .iter()
2681 .filter_map(|idx| match &idx.kind {
2682 IndexKind::BTree(map) => {
2683 let cold: Vec<(IndexKey, RowLocator)> = map
2684 .iter()
2685 .flat_map(|(k, locs)| {
2686 locs.iter()
2687 .filter(|l| l.is_cold())
2688 .copied()
2689 .map(move |l| (k.clone(), l))
2690 })
2691 .collect();
2692 if cold.is_empty() {
2693 None
2694 } else {
2695 Some((idx.name.clone(), cold))
2696 }
2697 }
2698 // BRIN / NSW carry no key→locator map. GIN handles
2699 // its own cold preservation below in `preserved_gin_cold`.
2700 IndexKind::Nsw(_)
2701 | IndexKind::Brin { .. }
2702 | IndexKind::Gin(_)
2703 | IndexKind::GinTrgm(_)
2704 | IndexKind::GinFulltext(_)
2705 | IndexKind::GinJsonb(_) => None,
2706 })
2707 .collect();
2708
2709 // v7.12.3 — same cold-preservation pattern for GIN's
2710 // `word → Vec<RowLocator>` posting lists. Parallel to the
2711 // BTree pass above (different key type so a separate vec is
2712 // cleaner than a generic merge). v7.15.0: trigram-GIN
2713 // (`gin_trgm_ops`) shares the same posting-list shape, so
2714 // one pass handles both — the `RebuildKind` carries the
2715 // kind tag to drive resurrection.
2716 let preserved_gin_cold: Vec<(String, Vec<(String, RowLocator)>)> = self
2717 .indices
2718 .iter()
2719 .filter_map(|idx| match &idx.kind {
2720 // v7.17.0 Phase 2.2 — fulltext-GIN posting lists
2721 // share the `String → Vec<RowLocator>` shape, so
2722 // cold preservation handles all three GIN flavours
2723 // in one pass.
2724 IndexKind::Gin(map)
2725 | IndexKind::GinTrgm(map)
2726 | IndexKind::GinFulltext(map)
2727 | IndexKind::GinJsonb(map) => {
2728 let cold: Vec<(String, RowLocator)> = map
2729 .iter()
2730 .flat_map(|(w, locs)| {
2731 locs.iter()
2732 .filter(|l| l.is_cold())
2733 .copied()
2734 .map(move |l| (w.clone(), l))
2735 })
2736 .collect();
2737 if cold.is_empty() {
2738 None
2739 } else {
2740 Some((idx.name.clone(), cold))
2741 }
2742 }
2743 IndexKind::BTree(_) | IndexKind::Nsw(_) | IndexKind::Brin { .. } => None,
2744 })
2745 .collect();
2746
2747 // v6.7.1 — descriptor needs to capture index kind so the
2748 // rebuild loop can resurrect BTree / NSW / BRIN / GIN exactly
2749 // as they were. (NSW carries m; BRIN carries the column type
2750 // snapshot; BTree / GIN need no extra payload.)
2751 #[derive(Clone)]
2752 enum RebuildKind {
2753 BTree,
2754 Nsw(usize),
2755 Brin(DataType),
2756 Gin,
2757 GinTrgm,
2758 GinFulltext,
2759 GinJsonb,
2760 }
2761 // v7.39 (round 170) — the descriptor must carry the FULL index
2762 // metadata: the rebuild used to reconstruct via bare
2763 // `Index::new_btree(name, pos)`, silently DROPPING is_unique /
2764 // extra_column_positions / partial_predicate / expression /
2765 // included_columns / nulls_not_distinct — so the first VACUUM
2766 // (or any delete-path rebuild) turned every UNIQUE INDEX into a
2767 // plain one and stopped enforcing it (probe-reproduced:
2768 // duplicate keys inserted silently after VACUUM).
2769 struct RebuildDesc {
2770 name: String,
2771 column_position: usize,
2772 kind: RebuildKind,
2773 is_unique: bool,
2774 extra_column_positions: Vec<usize>,
2775 partial_predicate: Option<String>,
2776 expression: Option<String>,
2777 included_columns: Vec<usize>,
2778 nulls_not_distinct: bool,
2779 // v7.39 (round 537) — carried through a rebuild like the rest.
2780 descending: bool,
2781 nulls_first: Option<bool>,
2782 collation: Option<String>,
2783 }
2784 let descriptors: Vec<RebuildDesc> = self
2785 .indices
2786 .iter()
2787 .map(|idx| {
2788 let kind = match &idx.kind {
2789 IndexKind::Nsw(g) => RebuildKind::Nsw(g.m),
2790 IndexKind::Brin { column_type } => RebuildKind::Brin(*column_type),
2791 IndexKind::BTree(_) => RebuildKind::BTree,
2792 IndexKind::Gin(_) => RebuildKind::Gin,
2793 IndexKind::GinTrgm(_) => RebuildKind::GinTrgm,
2794 IndexKind::GinFulltext(_) => RebuildKind::GinFulltext,
2795 IndexKind::GinJsonb(_) => RebuildKind::GinJsonb,
2796 };
2797 RebuildDesc {
2798 name: idx.name.clone(),
2799 column_position: idx.column_position,
2800 kind,
2801 is_unique: idx.is_unique,
2802 extra_column_positions: idx.extra_column_positions.clone(),
2803 partial_predicate: idx.partial_predicate.clone(),
2804 expression: idx.expression.clone(),
2805 included_columns: idx.included_columns.clone(),
2806 nulls_not_distinct: idx.nulls_not_distinct,
2807 descending: idx.descending,
2808 nulls_first: idx.nulls_first,
2809 collation: idx.collation.clone(),
2810 }
2811 })
2812 .collect();
2813 self.indices.clear();
2814 for desc in descriptors {
2815 let RebuildDesc {
2816 name,
2817 column_position,
2818 kind: rebuild_kind,
2819 is_unique,
2820 extra_column_positions,
2821 partial_predicate,
2822 expression,
2823 included_columns,
2824 nulls_not_distinct,
2825 descending,
2826 nulls_first,
2827 collation,
2828 } = desc;
2829 let pre_len = self.indices.len();
2830 match rebuild_kind {
2831 RebuildKind::Nsw(m) => {
2832 let idx = Index::new_nsw(name, column_position, m);
2833 self.indices.push(idx);
2834 let idx_pos = self.indices.len() - 1;
2835 let row_indices: Vec<usize> = (0..self.rows.len()).collect();
2836 for row_idx in row_indices {
2837 nsw_insert_at(self, idx_pos, row_idx);
2838 }
2839 }
2840 RebuildKind::Brin(column_type) => {
2841 // BRIN has no in-memory rebuild — the summaries
2842 // live in cold segments which freeze emits.
2843 self.indices
2844 .push(Index::new_brin(name, column_position, column_type));
2845 }
2846 RebuildKind::BTree => {
2847 // v7.39 (round 170) — bulk build: collect + sort +
2848 // group + from_sorted. The per-row insert_mut paid a
2849 // path-copy allocation per row per index (~15ms per
2850 // index on a 50k-row VACUUM, the dominant cost).
2851 let mut idx = Index::new_btree(name, column_position);
2852 let mut pairs: Vec<(IndexKey, usize)> = Vec::with_capacity(self.rows.len());
2853 for (i, row) in self.rows.iter().enumerate() {
2854 if let Some(key) = IndexKey::from_value(&row.values[column_position]) {
2855 pairs.push((key, i));
2856 }
2857 }
2858 pairs.sort_by(|a, b| a.0.cmp(&b.0));
2859 let mut grouped: Vec<(IndexKey, Vec<RowLocator>)> = Vec::new();
2860 for (key, i) in pairs {
2861 match grouped.last_mut() {
2862 Some((k, locs)) if *k == key => locs.push(RowLocator::Hot(i)),
2863 _ => grouped.push((key, alloc::vec![RowLocator::Hot(i)])),
2864 }
2865 }
2866 idx.kind = IndexKind::BTree(
2867 crate::persistent_btree::PersistentBTreeMap::from_sorted(grouped),
2868 );
2869 self.indices.push(idx);
2870 }
2871 RebuildKind::Gin => {
2872 let mut idx = Index::new_gin(name, column_position);
2873 if let IndexKind::Gin(map) = &mut idx.kind {
2874 for (i, row) in self.rows.iter().enumerate() {
2875 if let Value::TsVector(lexemes) = &row.values[column_position] {
2876 for lex in lexemes {
2877 let mut entries =
2878 map.get(&lex.word).cloned().unwrap_or_default();
2879 entries.push(RowLocator::Hot(i));
2880 map.insert_mut(lex.word.clone(), entries);
2881 }
2882 }
2883 }
2884 }
2885 self.indices.push(idx);
2886 }
2887 RebuildKind::GinTrgm => {
2888 let mut idx = Index::new_gin_trgm(name, column_position);
2889 if let IndexKind::GinTrgm(map) = &mut idx.kind {
2890 for (i, row) in self.rows.iter().enumerate() {
2891 if let Value::Text(s) = &row.values[column_position] {
2892 for tri in trgm::extract_trigrams(s) {
2893 let mut entries = map.get(&tri).cloned().unwrap_or_default();
2894 entries.push(RowLocator::Hot(i));
2895 map.insert_mut(tri, entries);
2896 }
2897 }
2898 }
2899 }
2900 self.indices.push(idx);
2901 }
2902 RebuildKind::GinFulltext => {
2903 // v7.17.0 Phase 2.2 — re-derive the lexeme
2904 // posting list from each TEXT/VARCHAR cell.
2905 // Mirrors the GinTrgm rebuild shape but
2906 // tokenises via `fts_simple::simple_lex`
2907 // (same rule as `to_tsvector('simple')`).
2908 let mut idx = Index::new_gin_fulltext(name, column_position);
2909 if let IndexKind::GinFulltext(map) = &mut idx.kind {
2910 for (i, row) in self.rows.iter().enumerate() {
2911 if let Value::Text(s) = &row.values[column_position] {
2912 for lex in fts_simple::simple_lex(s) {
2913 let mut entries = map.get(&lex).cloned().unwrap_or_default();
2914 entries.push(RowLocator::Hot(i));
2915 map.insert_mut(lex, entries);
2916 }
2917 }
2918 }
2919 }
2920 self.indices.push(idx);
2921 }
2922 RebuildKind::GinJsonb => {
2923 // v7.37.8 — re-derive the JSONB posting list
2924 // from each `Value::Json` cell.
2925 let mut idx = Index::new_gin_jsonb(name, column_position);
2926 if let IndexKind::GinJsonb(map) = &mut idx.kind {
2927 for (i, row) in self.rows.iter().enumerate() {
2928 if let Value::Json(s) = &row.values[column_position] {
2929 for tok in jsonb_gin::extract_tokens(s) {
2930 let mut entries = map.get(&tok).cloned().unwrap_or_default();
2931 entries.push(RowLocator::Hot(i));
2932 map.insert_mut(tok, entries);
2933 }
2934 }
2935 }
2936 }
2937 self.indices.push(idx);
2938 }
2939 }
2940 // v7.39 (round 170) — restore the captured metadata onto
2941 // whatever this arm pushed (see RebuildDesc above).
2942 if let Some(idx) = self.indices.get_mut(pre_len) {
2943 idx.is_unique = is_unique;
2944 idx.extra_column_positions = extra_column_positions;
2945 idx.partial_predicate = partial_predicate;
2946 idx.expression = expression;
2947 idx.included_columns = included_columns;
2948 idx.nulls_not_distinct = nulls_not_distinct;
2949 idx.descending = descending;
2950 idx.nulls_first = nulls_first;
2951 idx.collation = collation;
2952 }
2953 }
2954
2955 // Re-attach preserved cold locators after the from-rows
2956 // rebuild. `register_cold_locators` handles the per-key
2957 // entries-vec append; no key collisions arise because the
2958 // rebuild loop above produced only Hot locators.
2959 for (idx_name, locators) in preserved_cold {
2960 // Errors here would only fire if the index disappeared
2961 // between snapshot and rebuild, which can't happen
2962 // because the rebuild restores the same descriptor set.
2963 let _ = self.register_cold_locators(&idx_name, locators);
2964 }
2965 // v7.12.3 — same for GIN posting-list cold locators.
2966 for (idx_name, locators) in preserved_gin_cold {
2967 let _ = self.register_gin_cold_locators(&idx_name, locators);
2968 }
2969 // v7.39 (round 215) — the range-exclusion indexes address rows by the
2970 // same physical slot, so a compaction that shifted slots invalidates
2971 // their Hot locators too. Re-emit them from the (post-compaction) rows.
2972 if !self.excl_indexes.is_empty() {
2973 self.rebuild_excl_indexes();
2974 }
2975 }
2976
2977 fn add_nsw_index_inner(
2978 &mut self,
2979 name: String,
2980 column_name: &str,
2981 m: usize,
2982 restore: Option<NswGraph>,
2983 ) -> Result<(), StorageError> {
2984 if self.indices.iter().any(|i| i.name == name) {
2985 return Err(StorageError::DuplicateIndex { name });
2986 }
2987 let column_position = self.schema.column_position(column_name).ok_or_else(|| {
2988 StorageError::ColumnNotFound {
2989 column: column_name.into(),
2990 }
2991 })?;
2992 if !matches!(
2993 self.schema.columns[column_position].ty,
2994 DataType::Vector { .. }
2995 ) {
2996 return Err(StorageError::TypeMismatch {
2997 column: column_name.into(),
2998 expected: DataType::Vector {
2999 dim: 0,
3000 encoding: VecEncoding::F32,
3001 },
3002 actual: self.schema.columns[column_position].ty,
3003 position: column_position,
3004 });
3005 }
3006 if let Some(graph) = restore {
3007 self.indices.push(Index {
3008 name,
3009 column_position,
3010 kind: IndexKind::Nsw(graph),
3011 included_columns: Vec::new(),
3012 partial_predicate: None,
3013 expression: None,
3014 is_unique: false,
3015 nulls_not_distinct: false,
3016 descending: false,
3017 nulls_first: None,
3018 collation: None,
3019 extra_column_positions: Vec::new(),
3020 });
3021 return Ok(());
3022 }
3023 let idx = Index::new_nsw(name, column_position, m);
3024 self.indices.push(idx);
3025 let idx_pos = self.indices.len() - 1;
3026 // Bulk-build by walking the existing rows in order — each insert
3027 // sees the partial graph and links into it.
3028 let row_indices: Vec<usize> = (0..self.rows.len()).collect();
3029 for row_idx in row_indices {
3030 nsw_insert_at(self, idx_pos, row_idx);
3031 }
3032 Ok(())
3033 }
3034}
3035
3036/// v7.37.5 (mailrs crash-recovery Ask 3) — per-cell schema-compat
3037/// check shared by `insert_no_index` and `update_row_no_index`. The
3038/// logic mirrors the inline body in `insert` / `update_row` (NULL
3039/// handling, the cross-type compatibility map: TEXT ↔ VARCHAR/CHAR/
3040/// JSON/JSONB, TIMESTAMP ↔ TIMESTAMPTZ, BIT ↔ VARBIT, INET ↔ CIDR,
3041/// NUMERIC scale match).
3042/// v7.39 (round 642/643) — does a value of type `actual` belong in a
3043/// column declared `declared`?
3044///
3045/// This existed in THREE copies — insert, update and the standalone
3046/// row validator — and they had drifted apart in three independent
3047/// places: only insert accepted the `name` pairs, only insert and
3048/// update accepted a bit-to-bit pair with differing typmods, and only
3049/// update accepted a NEGATIVE declared numeric scale. Each omission was
3050/// a hole waiting for a value to reach that path; none was a deliberate
3051/// tightening, so the union below is the rule and all three now ask it.
3052///
3053/// Measured before converging: every shape the three disagreed about
3054/// answers identically to PG18 today, so this fixes nothing observable.
3055/// What it fixes is the next type — adding `xid` in round 640 meant
3056/// remembering to patch three places, and forgetting one would have
3057/// half-wired it.
3058///
3059/// The rule itself: a pair is compatible when the value's storage shape
3060/// is what the column stores. Length and precision contracts are NOT
3061/// checked here — they belong to coercion, which runs first.
3062/// `#[inline]` is not decoration. Extracting this matrix out of its
3063/// three call sites — a change with no semantic content at all — cost
3064/// `SELECT count(*) FROM d WHERE g BETWEEN 10 AND 20` **23x**, 5.8 ms
3065/// to 133 ms over 500 000 rows, reproducibly and outside the panel.
3066/// None of the three callers is on a scan path; taking the matrix out
3067/// of them was enough to move whatever else in this module the row loop
3068/// depends on being inlined. Round 641 learned the same thing about
3069/// `eval::binop::compare`. A refactor that reads as pure structure is
3070/// still a codegen change.
3071#[inline]
3072fn column_accepts(actual: DataType, declared: DataType) -> bool {
3073 if actual == declared {
3074 return true;
3075 }
3076 if matches!(
3077 (actual, declared),
3078 // A NAME column stores a Value::Text: the type identity is the
3079 // schema's and a value can never be one, so both directions.
3080 (
3081 DataType::Text,
3082 DataType::Varchar(_)
3083 | DataType::Char(_)
3084 | DataType::Name
3085 | DataType::Json
3086 | DataType::Jsonb
3087 ) | (DataType::Name, DataType::Text)
3088 // An XID column stores the Value::BigInt a transaction id
3089 // has always been; xid8 has no value of its own at all.
3090 | (DataType::BigInt, DataType::Xid | DataType::Xid8)
3091 | (DataType::Xid | DataType::Xid8, DataType::BigInt)
3092 // v7.39 (round 667) — an OID column likewise stores a plain
3093 // integer. INT is listed as well as BIGINT because a bare
3094 // literal arrives as one: PG takes `INSERT INTO t(o) VALUES
3095 // (42)` into an oid column, and measured, it does NOT take the
3096 // same integer into an xid column ("column is of type xid but
3097 // expression is of type integer"). SPG has been laxer than PG
3098 // on that xid direction since before this round — that is the
3099 // limitation `DataType::Xid8` documents, not something added
3100 // here.
3101 | (
3102 DataType::BigInt | DataType::Int | DataType::SmallInt,
3103 DataType::Oid,
3104 )
3105 | (DataType::Oid, DataType::BigInt | DataType::Int)
3106 // v7.39 (round 694) — `oid[]` rides in a BigIntArray cell, so
3107 // it accepts one either way, exactly as the scalar above does.
3108 | (DataType::BigIntArray | DataType::IntArray, DataType::OidArray)
3109 | (DataType::OidArray, DataType::BigIntArray)
3110 | (DataType::Json | DataType::Jsonb, DataType::Text)
3111 | (DataType::Json, DataType::Jsonb)
3112 | (DataType::Jsonb, DataType::Json)
3113 | (DataType::Timestamp, DataType::Timestamptz)
3114 | (DataType::Timestamptz, DataType::Timestamp)
3115 // BIT / VARBIT share the BitString storage shape; INET /
3116 // CIDR likewise. Same-family pairs with different typmods
3117 // are compatible HERE — the length contract is coercion's.
3118 | (DataType::Bit(_), DataType::BitVarying(_))
3119 | (DataType::BitVarying(_), DataType::Bit(_))
3120 | (DataType::Bit(_), DataType::Bit(_))
3121 | (DataType::BitVarying(_), DataType::BitVarying(_))
3122 | (DataType::Inet, DataType::Cidr)
3123 | (DataType::Cidr, DataType::Inet)
3124 ) {
3125 return true;
3126 }
3127 // NUMERIC carries its own scale in the value while the column
3128 // declares the expected one. An unconstrained `numeric` (the
3129 // precision-0/scale-0 sentinel) takes any scale; a declared
3130 // `numeric(p,s)` needs the rescaled value; and a NEGATIVE declared
3131 // scale stores at display scale 0, having been rounded to a
3132 // multiple of 10^|s|.
3133 matches!(
3134 (actual, declared),
3135 (
3136 DataType::Numeric { scale: a, .. },
3137 DataType::Numeric {
3138 precision: bp,
3139 scale: b,
3140 },
3141 ) if a == b || (bp == 0 && b == 0) || (b < 0 && a == 0)
3142 )
3143}
3144
3145fn validate_row_against_schema(
3146 values: &[Value<'static>],
3147 schema: &TableSchema,
3148) -> Result<(), StorageError> {
3149 for (i, (val, col)) in values.iter().zip(&schema.columns).enumerate() {
3150 if val.is_null() {
3151 if !col.nullable {
3152 return Err(StorageError::NullInNotNull {
3153 column: col.name.clone(),
3154 });
3155 }
3156 continue;
3157 }
3158 // v7.39 (read01 round 54) — see above: no panic on an untyped value.
3159 let Some(actual) = val.data_type() else {
3160 // See above: an eval-only untyped value is accepted, not a panic.
3161 continue;
3162 };
3163 let compatible = column_accepts(actual, col.ty);
3164 if !compatible {
3165 return Err(StorageError::TypeMismatch {
3166 column: col.name.clone(),
3167 expected: col.ty,
3168 actual,
3169 position: i,
3170 });
3171 }
3172 }
3173 Ok(())
3174}
3175
3176/// v6.0.4 — re-encode a single cell to the target `VecEncoding`.
3177/// Used by `Table::rebuild_nsw_index` when ALTER INDEX REBUILD
3178/// includes the optional `WITH (encoding = …)` clause. Round-trip
3179/// goes through f32: `current → Vec<f32> → target`, leaving NULL
3180/// cells untouched. Returns `Unsupported` on a non-vector cell —
3181/// the caller should have rejected the schema before reaching this.
3182fn recode_vector_cell(
3183 cell: Value<'static>,
3184 target: VecEncoding,
3185) -> Result<Value<'static>, StorageError> {
3186 if matches!(cell, Value::Null) {
3187 return Ok(cell);
3188 }
3189 // Step 1 — extract the f32 representation of the source cell.
3190 let as_f32: Vec<f32> = match &cell {
3191 Value::Vector(v) => v.to_vec(),
3192 Value::Sq8Vector(q) => quantize::dequantize(q),
3193 Value::HalfVector(h) => h.to_f32_vec(),
3194 other => {
3195 return Err(StorageError::Unsupported(format!(
3196 "ALTER INDEX REBUILD: cannot recode non-vector cell {:?}",
3197 other.data_type()
3198 )));
3199 }
3200 };
3201 // Step 2 — encode into the target shape. `F32` is the identity
3202 // path (saves one alloc round-trip when the source is already
3203 // F32 — but `Value::Vector(as_f32)` is the right answer
3204 // regardless).
3205 Ok(match target {
3206 VecEncoding::F32 => Value::Vector(Cow::Owned(as_f32)),
3207 VecEncoding::Sq8 => Value::Sq8Vector(quantize::quantize(&as_f32)),
3208 VecEncoding::F16 => Value::HalfVector(halfvec::HalfVector::from_f32_slice(&as_f32)),
3209 })
3210}
3211
3212/// v7.39 (round 562) — a cursor over `Table`'s row headers that holds
3213/// the trie leaf it last descended to.
3214///
3215/// See `Table::header_runs` for why. Ask about ascending positions and
3216/// the descent happens once per 32; ask about scattered ones and it
3217/// happens as often as `position_visible` would have done it.
3218#[derive(Debug)]
3219pub struct HeaderRuns<'a> {
3220 table: &'a Table,
3221 /// `(start, run)` — `run[i - start]` is the header for position `i`.
3222 run: Option<(usize, &'a [crate::row_header::RowHeader])>,
3223}
3224
3225impl HeaderRuns<'_> {
3226 /// Is the row at this position visible to the snapshot?
3227 ///
3228 /// Answers exactly as `Table::position_visible` does — same
3229 /// `SKIP LOCKED` handling, same snapshot rules — and the pins in
3230 /// `e2e_index_only_scan_round560` hold both to it.
3231 pub fn visible(&mut self, idx: usize, snapshot: &crate::snapshot::Snapshot) -> bool {
3232 if let Some((start, run)) = self.run
3233 && idx >= start
3234 && idx - start < run.len()
3235 {
3236 return self.table.header_visible(idx, &run[idx - start], snapshot);
3237 }
3238 let Some((start, run)) = self.table.headers.run_containing(idx) else {
3239 return false;
3240 };
3241 self.run = Some((start, run));
3242 self.table.header_visible(idx, &run[idx - start], snapshot)
3243 }
3244}