spg_engine/spg_admin.rs
1//! `spg_*` introspection views and admin/stats API. Lifted out of
2//! `lib.rs` (v7.32 engine modularisation). The `exec_spg_*` methods
3//! materialise the `spg_statistic` / `spg_stat_*` / `spg_*_ddl` /
4//! `spg_audit_*` meta-views dispatched from the meta-view SELECT path;
5//! the public `memory_stats` / `set_plan_cache_max` / `query_stats` /
6//! `tables_needing_analyze` methods form the embedded admin surface.
7
8use alloc::collections::BTreeMap;
9use alloc::string::String;
10use alloc::vec::Vec;
11
12use spg_storage::{ColumnSchema, DataType, Row, Value};
13
14use crate::{
15 ActivityProvider, AuditChainProvider, AuditVerifier, Engine, EngineError, MemoryStats,
16 QueryResult, SlowQueryLogger, TableMemoryStats, approx_row_bytes, is_internal_table_name,
17 render_create_table, render_histogram_bounds,
18};
19use crate::{query_stats, statistics};
20
21/// v7.37.16 (16.11) — render a PartitionBound for the
22/// `spg_partition_health.bound_desc` column. Mirrors
23/// `crate::partition::bound_to_diag` but lives here so this
24/// crate's `spg_admin` module doesn't need to depend on the
25/// engine-private partition helpers.
26fn partition_bound_diag(b: &spg_storage::PartitionBound) -> String {
27 use spg_storage::PartitionBound;
28 match b {
29 PartitionBound::MinValue => "MINVALUE".into(),
30 PartitionBound::MaxValue => "MAXVALUE".into(),
31 PartitionBound::TimestampTz(m) => alloc::format!("'{m}'::timestamptz"),
32 PartitionBound::BigInt(n) => alloc::format!("{n}::bigint"),
33 PartitionBound::Int(n) => alloc::format!("{n}::integer"),
34 PartitionBound::SmallInt(n) => alloc::format!("{n}::smallint"),
35 PartitionBound::Date(d) => alloc::format!("{d}::date"),
36 PartitionBound::Text(s) => alloc::format!("'{}'", s.replace('\'', "''")),
37 }
38}
39
40impl Engine {
41 /// v6.2.0 — materialise `spg_statistic` rows. One row per
42 /// `(table, column)` pair tracked in `Statistics`, with
43 /// `histogram_bounds` rendered as a `[v0, v1, ...]` string —
44 /// the same canonical form vector literals use for round-trip.
45 pub(crate) fn exec_spg_statistic(&self) -> QueryResult {
46 let columns = alloc::vec![
47 ColumnSchema::new("table_name", DataType::Text, false),
48 ColumnSchema::new("column_name", DataType::Text, false),
49 ColumnSchema::new("null_frac", DataType::Float, false),
50 ColumnSchema::new("n_distinct", DataType::BigInt, false),
51 ColumnSchema::new("histogram_bounds", DataType::Text, false),
52 // v6.7.0 — appended column (v6.2.0 stability contract
53 // allows APPEND to spg_statistic, not reorder/rename).
54 // Reports the cached per-table cold-row count; same
55 // value across every column row of the same table.
56 ColumnSchema::new("cold_row_count", DataType::BigInt, false),
57 ];
58 let rows: Vec<Row<'static>> = self
59 .statistics
60 .iter()
61 .map(|((t, c), s)| {
62 let cold = self
63 .catalog
64 .get(t)
65 .map_or(0, |table| table.cold_row_count());
66 Row::new(alloc::vec![
67 Value::text(t.clone()),
68 Value::text(c.clone()),
69 Value::Float(f64::from(s.null_frac)),
70 Value::BigInt(i64::try_from(s.n_distinct).unwrap_or(i64::MAX)),
71 Value::text(render_histogram_bounds(&s.histogram_bounds)),
72 Value::BigInt(i64::try_from(cold).unwrap_or(i64::MAX)),
73 ])
74 })
75 .collect();
76 QueryResult::Rows { columns, rows }
77 }
78
79 /// v6.5.0 — materialise `spg_stat_replication` rows. One row
80 /// per subscription with `(name, conn_str, publications,
81 /// last_received_pos, enabled)`. Surface mirrors
82 /// `SHOW SUBSCRIPTIONS` but follows the virtual-table dispatch
83 /// shape so it composes with SELECT clauses (WHERE, projection
84 /// onto specific columns, etc).
85 pub(crate) fn exec_spg_stat_replication(&self) -> QueryResult {
86 let columns = alloc::vec![
87 ColumnSchema::new("name", DataType::Text, false),
88 ColumnSchema::new("conn_str", DataType::Text, false),
89 ColumnSchema::new("publications", DataType::Text, false),
90 ColumnSchema::new("last_received_pos", DataType::BigInt, false),
91 ColumnSchema::new("enabled", DataType::Bool, false),
92 ];
93 let rows: Vec<Row<'static>> = self
94 .subscriptions
95 .iter()
96 .map(|(name, sub)| {
97 Row::new(alloc::vec![
98 Value::text(name.clone()),
99 Value::text(sub.conn_str.clone()),
100 Value::text(sub.publications.join(",")),
101 Value::BigInt(i64::try_from(sub.last_received_pos).unwrap_or(i64::MAX)),
102 Value::Bool(sub.enabled),
103 ])
104 })
105 .collect();
106 QueryResult::Rows { columns, rows }
107 }
108
109 /// v6.5.0 — materialise `spg_stat_segment` rows. One row per
110 /// cold-tier segment with `(segment_id, num_rows, num_pages,
111 /// total_bytes)`.
112 ///
113 /// v6.7.0 — appended `table_name` column resolves the v6.5.0
114 /// carve-out. Walks every user table's BTree indices to find
115 /// which table's Cold locators point at each segment. Empty
116 /// string for orphan segments (loaded via SPG_PRELOAD_COLD_SEGMENT
117 /// before any index registered a locator). The walk is
118 /// O(tables × indices × keys); cached per call, not across
119 /// calls — re-walked on every `SELECT * FROM spg_stat_segment`.
120 /// v7.31 (memory campaign) — walk the committed catalog and
121 /// build the per-bucket memory snapshot. O(rows + index
122 /// entries): operator/monitoring surface, not a query path.
123 pub fn memory_stats(&self) -> MemoryStats {
124 let mut tables: Vec<TableMemoryStats> = Vec::new();
125 let (mut total_enc, mut total_res, mut total_idx) = (0u64, 0u64, 0u64);
126 for tname in self.catalog.table_names() {
127 if is_internal_table_name(&tname) {
128 continue;
129 }
130 let Some(t) = self.catalog.get(&tname) else {
131 continue;
132 };
133 let resident: u64 = t.rows().iter().map(|r| approx_row_bytes(r) as u64).sum();
134 // v7.31 C2 — each index variant accounts for its own
135 // resident bytes by walking its real structure (NSW layer
136 // adjacency, GIN posting lists), replacing the old inline
137 // parametric estimate that mis-sized NSW and flat-tokened
138 // every GIN family index.
139 let mut idx_bytes: u64 = 0;
140 for idx in t.indices() {
141 idx_bytes += idx.kind.approx_resident_bytes();
142 }
143 total_enc += t.hot_bytes();
144 total_res += resident;
145 total_idx += idx_bytes;
146 tables.push(TableMemoryStats {
147 name: tname.clone(),
148 hot_rows: t.rows().len() as u64,
149 cold_rows: t.cold_row_count(),
150 hot_encoded_bytes: t.hot_bytes(),
151 approx_resident_bytes: resident,
152 index_count: t.indices().len() as u64,
153 approx_index_bytes: idx_bytes,
154 });
155 }
156 MemoryStats {
157 tables,
158 total_hot_encoded_bytes: total_enc,
159 total_approx_resident_bytes: total_res,
160 total_approx_index_bytes: total_idx,
161 max_query_bytes: self.max_query_bytes,
162 // Bucket D belongs to the durable host (embed / server),
163 // not the engine — filled in there (C2).
164 wal_bytes: None,
165 }
166 }
167
168 /// v7.31 — `SELECT * FROM spg_memory_stats`: one row per user
169 /// table (same numbers as `Engine::memory_stats()`), so the
170 /// server path gets the meter through plain SQL.
171 pub(crate) fn exec_spg_memory_stats(&self) -> QueryResult {
172 let columns = alloc::vec![
173 ColumnSchema::new("table_name", DataType::Text, false),
174 ColumnSchema::new("hot_rows", DataType::BigInt, false),
175 ColumnSchema::new("cold_rows", DataType::BigInt, false),
176 ColumnSchema::new("hot_encoded_bytes", DataType::BigInt, false),
177 ColumnSchema::new("approx_resident_bytes", DataType::BigInt, false),
178 ColumnSchema::new("index_count", DataType::BigInt, false),
179 ColumnSchema::new("approx_index_bytes", DataType::BigInt, false),
180 ];
181 #[allow(clippy::cast_possible_wrap)]
182 let rows: Vec<Row<'static>> = self
183 .memory_stats()
184 .tables
185 .into_iter()
186 .map(|t| {
187 Row::new(alloc::vec![
188 Value::text(t.name),
189 Value::BigInt(t.hot_rows as i64),
190 Value::BigInt(t.cold_rows as i64),
191 Value::BigInt(t.hot_encoded_bytes as i64),
192 Value::BigInt(t.approx_resident_bytes as i64),
193 Value::BigInt(t.index_count as i64),
194 Value::BigInt(t.approx_index_bytes as i64),
195 ])
196 })
197 .collect();
198 QueryResult::Rows { columns, rows }
199 }
200
201 pub(crate) fn exec_spg_stat_segment(&self) -> QueryResult {
202 let columns = alloc::vec![
203 ColumnSchema::new("segment_id", DataType::BigInt, false),
204 ColumnSchema::new("table_name", DataType::Text, false),
205 ColumnSchema::new("num_rows", DataType::BigInt, false),
206 ColumnSchema::new("num_pages", DataType::BigInt, false),
207 ColumnSchema::new("total_bytes", DataType::BigInt, false),
208 ];
209 // v6.7.0 — build a segment_id → table_name map by walking
210 // every user table's BTree indices once. O(tables × indices
211 // × keys) for the v6.5.0 carve-out resolution; acceptable
212 // because spg_stat_segment is operator-facing (not on a
213 // hot-loop path).
214 let mut segment_owners: alloc::collections::BTreeMap<u32, String> = BTreeMap::new();
215 for tname in self.catalog.table_names() {
216 if is_internal_table_name(&tname) {
217 continue;
218 }
219 let Some(t) = self.catalog.get(&tname) else {
220 continue;
221 };
222 for idx in t.indices() {
223 if let spg_storage::IndexKind::BTree(map) = &idx.kind {
224 for (_, locs) in map.iter() {
225 for loc in locs {
226 if let spg_storage::RowLocator::Cold { segment_id, .. } = loc {
227 segment_owners
228 .entry(*segment_id)
229 .or_insert_with(|| tname.clone());
230 }
231 }
232 }
233 }
234 }
235 }
236 let rows: Vec<Row<'static>> = self
237 .catalog
238 .cold_segment_ids_global()
239 .iter()
240 .filter_map(|&id| {
241 let seg = self.catalog.cold_segment(id)?;
242 let meta = seg.meta();
243 let owner = segment_owners.get(&id).cloned().unwrap_or_default();
244 Some(Row::new(alloc::vec![
245 Value::BigInt(i64::from(id)),
246 Value::text(owner),
247 Value::BigInt(i64::try_from(meta.num_rows).unwrap_or(i64::MAX)),
248 Value::BigInt(i64::from(meta.num_pages)),
249 Value::BigInt(i64::try_from(meta.total_bytes).unwrap_or(i64::MAX)),
250 ]))
251 })
252 .collect();
253 QueryResult::Rows { columns, rows }
254 }
255
256 /// v6.5.1 — materialise `spg_stat_query` rows. One row per
257 /// distinct SQL text recorded since the engine booted, capped
258 /// at `QUERY_STATS_MAX` (1024). Columns:
259 /// sql, exec_count, total_us, mean_us, max_us, last_seen_us
260 /// mean_us = total_us / exec_count (saturating).
261 pub(crate) fn exec_spg_stat_query(&self) -> QueryResult {
262 let columns = alloc::vec![
263 ColumnSchema::new("sql", DataType::Text, false),
264 ColumnSchema::new("exec_count", DataType::BigInt, false),
265 ColumnSchema::new("total_us", DataType::BigInt, false),
266 ColumnSchema::new("mean_us", DataType::BigInt, false),
267 ColumnSchema::new("max_us", DataType::BigInt, false),
268 ColumnSchema::new("last_seen_us", DataType::BigInt, false),
269 ];
270 let rows: Vec<Row<'static>> = self
271 .query_stats
272 .snapshot()
273 .into_iter()
274 .map(|(sql, s)| {
275 let mean = if s.exec_count == 0 {
276 0
277 } else {
278 s.total_us / s.exec_count
279 };
280 Row::new(alloc::vec![
281 Value::text(sql),
282 Value::BigInt(i64::try_from(s.exec_count).unwrap_or(i64::MAX)),
283 Value::BigInt(i64::try_from(s.total_us).unwrap_or(i64::MAX)),
284 Value::BigInt(i64::try_from(mean).unwrap_or(i64::MAX)),
285 Value::BigInt(i64::try_from(s.max_us).unwrap_or(i64::MAX)),
286 Value::BigInt(i64::try_from(s.last_seen_us).unwrap_or(i64::MAX)),
287 ])
288 })
289 .collect();
290 QueryResult::Rows { columns, rows }
291 }
292
293 /// v6.5.2 — register a connection-state provider. spg-server
294 /// calls this at startup with a function that snapshots its
295 /// per-pgwire-connection registry. Engine reads through the
296 /// callback on `SELECT * FROM spg_stat_activity`.
297 #[must_use]
298 pub const fn with_activity_provider(mut self, f: ActivityProvider) -> Self {
299 self.activity_provider = Some(f);
300 self
301 }
302
303 /// v6.5.3 — register audit chain provider + verifier.
304 #[must_use]
305 pub const fn with_audit_providers(
306 mut self,
307 chain: AuditChainProvider,
308 verify: AuditVerifier,
309 ) -> Self {
310 self.audit_chain_provider = Some(chain);
311 self.audit_verifier = Some(verify);
312 self
313 }
314
315 /// v6.5.6 — register a slow-query log callback. `threshold_us`
316 /// is the floor (in microseconds); only executes above the floor
317 /// fire the callback. spg-server wires this from
318 /// `SPG_SLOW_QUERY_THRESHOLD_MS` (default 100 ms).
319 #[must_use]
320 pub const fn with_slow_query_log(mut self, threshold_us: u64, logger: SlowQueryLogger) -> Self {
321 self.slow_query_threshold_us = Some(threshold_us);
322 self.slow_query_logger = Some(logger);
323 self
324 }
325
326 /// v6.5.6 — operator knob for plan cache cap. spg-server reads
327 /// `SPG_PLAN_CACHE_MAX` env at startup; uses this to override
328 /// the compile-time default of 256.
329 pub fn set_plan_cache_max(&mut self, n: usize) {
330 self.plan_cache.set_max_entries(n);
331 }
332
333 /// v6.5.2 — materialise `spg_stat_activity` rows. Pulls a fresh
334 /// snapshot from the registered `ActivityProvider`. Returns an
335 /// empty result set when no provider is registered (the no_std
336 /// embedded path with no pgwire layer).
337 pub(crate) fn exec_spg_stat_activity(&self) -> QueryResult {
338 // v7.37.14 (B6.3) — column order matches PG's
339 // pg_stat_activity for `wait_event_type` immediately before
340 // `wait_event` so client-side projection by ordinal stays
341 // robust even before adopters update to named projection.
342 let columns = alloc::vec![
343 ColumnSchema::new("pid", DataType::Int, false),
344 ColumnSchema::new("user", DataType::Text, false),
345 ColumnSchema::new("started_at_us", DataType::BigInt, false),
346 ColumnSchema::new("current_sql", DataType::Text, false),
347 ColumnSchema::new("wait_event_type", DataType::Text, false),
348 ColumnSchema::new("wait_event", DataType::Text, false),
349 ColumnSchema::new("elapsed_us", DataType::BigInt, false),
350 ColumnSchema::new("in_transaction", DataType::Bool, false),
351 ColumnSchema::new("application_name", DataType::Text, false),
352 ];
353 let rows: Vec<Row<'static>> = self
354 .activity_provider
355 .map(|f| f())
356 .unwrap_or_default()
357 .into_iter()
358 .map(|r| {
359 Row::new(alloc::vec![
360 Value::Int(i32::try_from(r.pid).unwrap_or(i32::MAX)),
361 Value::text(r.user),
362 Value::BigInt(r.started_at_us),
363 Value::text(r.current_sql),
364 Value::text(r.wait_event_type),
365 Value::text(r.wait_event),
366 Value::BigInt(r.elapsed_us),
367 Value::Bool(r.in_transaction),
368 Value::text(r.application_name),
369 ])
370 })
371 .collect();
372 QueryResult::Rows { columns, rows }
373 }
374
375 /// v7.38 (read01 P3.10) — the canonical `pg_stat_activity` view with
376 /// PG's column names, so monitoring tools (which query the standard
377 /// name + columns) work. SPG's `spg_stat_activity` carries the same
378 /// data under SPG-native names; here it is re-projected to PG's 22
379 /// columns, with the fields SPG doesn't track surfaced as NULL and
380 /// `state` derived from the in-transaction / running-query flags.
381 pub(crate) fn exec_pg_stat_activity(&self) -> QueryResult {
382 let columns = alloc::vec![
383 ColumnSchema::new("datid", DataType::BigInt, true),
384 ColumnSchema::new("datname", DataType::Text, true),
385 ColumnSchema::new("pid", DataType::Int, false),
386 ColumnSchema::new("leader_pid", DataType::Int, true),
387 ColumnSchema::new("usesysid", DataType::BigInt, true),
388 ColumnSchema::new("usename", DataType::Text, true),
389 ColumnSchema::new("application_name", DataType::Text, false),
390 ColumnSchema::new("client_addr", DataType::Text, true),
391 ColumnSchema::new("client_hostname", DataType::Text, true),
392 ColumnSchema::new("client_port", DataType::Int, true),
393 ColumnSchema::new("backend_start", DataType::Timestamptz, true),
394 ColumnSchema::new("xact_start", DataType::Timestamptz, true),
395 ColumnSchema::new("query_start", DataType::Timestamptz, true),
396 ColumnSchema::new("state_change", DataType::Timestamptz, true),
397 ColumnSchema::new("wait_event_type", DataType::Text, true),
398 ColumnSchema::new("wait_event", DataType::Text, true),
399 ColumnSchema::new("state", DataType::Text, true),
400 ColumnSchema::new("backend_xid", DataType::BigInt, true),
401 ColumnSchema::new("backend_xmin", DataType::BigInt, true),
402 ColumnSchema::new("query_id", DataType::BigInt, true),
403 ColumnSchema::new("query", DataType::Text, false),
404 ColumnSchema::new("backend_type", DataType::Text, false),
405 ];
406 let rows: Vec<Row<'static>> = self
407 .activity_provider
408 .map(|f| f())
409 .unwrap_or_default()
410 .into_iter()
411 .map(|r| {
412 // PG `state`: a running query is 'active'; otherwise
413 // 'idle in transaction' inside a txn, else 'idle'.
414 // v7.39 (round 474) — PG reports NULL state for a background
415 // process; only a client backend is idle or active.
416 let state = if r.backend_type != "client backend" {
417 ""
418 } else if !r.current_sql.is_empty() {
419 "active"
420 } else if r.in_transaction {
421 "idle in transaction"
422 } else {
423 "idle"
424 };
425 let started = Value::Timestamp(r.started_at_us);
426 Row::new(alloc::vec![
427 Value::Null, // datid
428 // v7.39 (round 319, V52) — each row's OWN database.
429 // This used to read the ASKING session's GUC and stamp
430 // it on every row, so one connection's database was
431 // reported as everybody's.
432 if r.database.is_empty() {
433 Value::Null
434 } else {
435 Value::text(r.database)
436 }, // datname
437 Value::Int(i32::try_from(r.pid).unwrap_or(i32::MAX)),
438 Value::Null, // leader_pid
439 Value::Null, // usesysid
440 Value::text(r.user), // usename
441 Value::text(r.application_name),
442 // v7.39 (round 319, V52) — the real peer. PG leaves
443 // client_hostname NULL unless log_hostname is on, which
444 // SPG has no equivalent of, so it stays NULL; the port
445 // is -1 for a connection with no TCP peer, as in PG.
446 if r.client_addr.is_empty() {
447 Value::Null
448 } else {
449 Value::text(r.client_addr)
450 }, // client_addr
451 Value::Null, // client_hostname
452 Value::Int(r.client_port), // client_port
453 started.clone(), // backend_start
454 if r.in_transaction {
455 started.clone()
456 } else {
457 Value::Null
458 }, // xact_start
459 if r.current_sql.is_empty() {
460 Value::Null
461 } else {
462 started
463 }, // query_start
464 Value::Null, // state_change
465 Value::text(r.wait_event_type),
466 Value::text(r.wait_event),
467 if state.is_empty() {
468 Value::Null
469 } else {
470 Value::text(alloc::string::String::from(state))
471 },
472 Value::Null, // backend_xid
473 Value::Null, // backend_xmin
474 Value::Null, // query_id
475 Value::text(r.current_sql), // query
476 // v7.39 (round 474) — the row's own backend_type, so a
477 // background worker reports as itself rather than as a
478 // client connection.
479 Value::text(r.backend_type.clone()),
480 ])
481 })
482 .collect();
483 QueryResult::Rows { columns, rows }
484 }
485
486 /// v7.37.15 (Phase F) — MVCC diagnostic view. Single-row
487 /// snapshot of the engine's per-row visibility state so
488 /// `spgctl` / monitoring can observe vacuum lag + in-flight
489 /// transaction count without reaching into engine internals.
490 ///
491 /// Columns:
492 /// - `current_version` — the live monotonic writer-version
493 /// cursor (next allocated version comes after this).
494 /// - `active_writer_count` — number of writer versions in
495 /// flight (= concurrent transactions). 0 means quiescent.
496 /// - `oldest_active_version` — floor of the active set;
497 /// vacuum can reclaim any row whose `xmax < this`.
498 pub(crate) fn exec_spg_stat_mvcc(&self) -> QueryResult {
499 let columns = alloc::vec![
500 ColumnSchema::new("current_version", DataType::BigInt, false),
501 ColumnSchema::new("active_writer_count", DataType::Int, false),
502 ColumnSchema::new("oldest_active_version", DataType::BigInt, false),
503 ];
504 let cv = spg_storage::row_header::current_version() as i64;
505 let active = self.active_writer_versions.len() as i32;
506 let oldest = self
507 .active_writer_versions
508 .iter()
509 .next()
510 .copied()
511 .unwrap_or(cv as u64) as i64;
512 let rows = alloc::vec![Row::new(alloc::vec![
513 Value::BigInt(cv),
514 Value::Int(active),
515 Value::BigInt(oldest),
516 ])];
517 QueryResult::Rows { columns, rows }
518 }
519
520 /// v7.37.16 (16.11 [PG+]) — materialise `spg_partition_health`
521 /// rows: one row per partition (Range / List / Hash / Default /
522 /// Parent), plus a "row_count" / "bound" diag column so dashboard
523 /// queries can size a partitioned table at a glance without
524 /// joining catalog tables. PG provides `pg_partitioned_table` +
525 /// `pg_inherits` + per-child `pg_class.reltuples`; SPG bundles
526 /// them into one easy view because dogfood / sentori dashboards
527 /// kept reaching for it.
528 ///
529 /// Columns:
530 /// parent_name TEXT NOT NULL -- parent table name, or
531 /// the partition name itself
532 /// when role == 'Parent'
533 /// partition_name TEXT NOT NULL -- the partition (or
534 /// parent) name
535 /// role TEXT NOT NULL -- 'Parent' | 'Range'
536 /// | 'List' | 'Hash'
537 /// | 'Default'
538 /// row_count BIGINT NOT NULL -- live row count
539 /// bound_desc TEXT NOT NULL -- human-readable bound for
540 /// diagnostics ('' for
541 /// Parent + DEFAULT)
542 pub(crate) fn exec_spg_partition_health(&self) -> QueryResult {
543 use spg_storage::PartitionRole;
544 let columns = alloc::vec![
545 ColumnSchema::new("parent_name", DataType::Text, false),
546 ColumnSchema::new("partition_name", DataType::Text, false),
547 ColumnSchema::new("role", DataType::Text, false),
548 ColumnSchema::new("row_count", DataType::BigInt, false),
549 ColumnSchema::new("bound_desc", DataType::Text, false),
550 ];
551 let mut rows: Vec<Row<'static>> = Vec::new();
552 for name in self.catalog.table_names() {
553 let Some(t) = self.catalog.get(&name) else {
554 continue;
555 };
556 let role = match &t.schema().partition_role {
557 None => continue,
558 Some(r) => r,
559 };
560 let row_count = t.rows().len() as i64;
561 let (parent, role_str, bound) = match role {
562 // v7.39 (round 645) — an inheritance child reports every
563 // parent it names; the diagnostic view lists the first,
564 // which is the only one single inheritance ever has.
565 PartitionRole::Inherits { parent_names } => (
566 parent_names.first().cloned().unwrap_or_default(),
567 alloc::string::String::from("Inherits"),
568 alloc::format!("INHERITS ({})", parent_names.join(", ")),
569 ),
570 PartitionRole::Parent { kind, .. } => {
571 let kind_str = match kind {
572 spg_storage::PartitionKind::Range => "RANGE",
573 spg_storage::PartitionKind::List => "LIST",
574 spg_storage::PartitionKind::Hash => "HASH",
575 };
576 (
577 name.clone(),
578 alloc::string::String::from("Parent"),
579 alloc::format!("PARTITION BY {kind_str}"),
580 )
581 }
582 PartitionRole::Range {
583 parent_name,
584 lower,
585 upper,
586 } => (
587 parent_name.clone(),
588 alloc::string::String::from("Range"),
589 alloc::format!(
590 "FROM ({}) TO ({})",
591 partition_bound_diag(lower),
592 partition_bound_diag(upper)
593 ),
594 ),
595 PartitionRole::List {
596 parent_name,
597 values,
598 } => {
599 let mut diag = alloc::string::String::from("IN (");
600 for (i, v) in values.iter().enumerate() {
601 if i > 0 {
602 diag.push_str(", ");
603 }
604 diag.push_str(&partition_bound_diag(v));
605 }
606 diag.push(')');
607 (
608 parent_name.clone(),
609 alloc::string::String::from("List"),
610 diag,
611 )
612 }
613 PartitionRole::Hash {
614 parent_name,
615 modulus,
616 remainder,
617 } => (
618 parent_name.clone(),
619 alloc::string::String::from("Hash"),
620 alloc::format!("WITH (MODULUS {modulus}, REMAINDER {remainder})"),
621 ),
622 PartitionRole::Default { parent_name } => (
623 parent_name.clone(),
624 alloc::string::String::from("Default"),
625 alloc::string::String::new(),
626 ),
627 };
628 rows.push(Row::new(alloc::vec![
629 Value::Text(alloc::borrow::Cow::Owned(parent)),
630 Value::Text(alloc::borrow::Cow::Owned(name)),
631 Value::Text(alloc::borrow::Cow::Owned(role_str)),
632 Value::BigInt(row_count),
633 Value::Text(alloc::borrow::Cow::Owned(bound)),
634 ]));
635 }
636 QueryResult::Rows { columns, rows }
637 }
638
639 /// v7.37.22 (22.1) — materialise `pg_stat_statements` rows with
640 /// PG's exact column shape. The data source is the same
641 /// `query_stats` registry that backs `spg_stat_query`, but the
642 /// surface is PG-compatible so dashboards/queries written
643 /// against `SELECT … FROM pg_stat_statements ORDER BY
644 /// total_exec_time DESC LIMIT 10` keep working.
645 ///
646 /// SPG ↔ PG mapping:
647 /// query ← stats.sql
648 /// calls ← stats.exec_count
649 /// total_exec_time ← stats.total_us / 1000 (ms)
650 /// min_exec_time ← 0 (no per-call min tracked yet)
651 /// max_exec_time ← stats.max_us / 1000
652 /// mean_exec_time ← derived
653 /// stddev_exec_time ← 0
654 /// rows ← 0 (per-row count tracking lands later)
655 /// userid ← 10 (PG's "postgres" superuser oid)
656 /// dbid ← 16384 (SPG single-db OID)
657 /// queryid ← hash of sql
658 /// plans ← stats.exec_count (one plan per call)
659 /// shared_blks_* ← 0 (no shared-buffer accounting)
660 /// local_blks_* ← 0
661 /// temp_blks_* ← 0
662 /// *_blk_*_time ← 0
663 /// wal_records / wal_fpi / wal_bytes ← 0 (per-stmt accounting)
664 /// jit_* ← 0 (no JIT)
665 /// stats_since / minmax_stats_since ← stats.last_seen_us
666 ///
667 /// 38 columns total to cover PG 18's pg_stat_statements view.
668 pub(crate) fn exec_pg_stat_statements(&self) -> QueryResult {
669 let columns = alloc::vec![
670 ColumnSchema::new("userid", DataType::BigInt, false),
671 ColumnSchema::new("dbid", DataType::BigInt, false),
672 ColumnSchema::new("toplevel", DataType::Bool, false),
673 ColumnSchema::new("queryid", DataType::BigInt, false),
674 ColumnSchema::new("query", DataType::Text, false),
675 ColumnSchema::new("plans", DataType::BigInt, false),
676 ColumnSchema::new("total_plan_time", DataType::Float, false),
677 ColumnSchema::new("min_plan_time", DataType::Float, false),
678 ColumnSchema::new("max_plan_time", DataType::Float, false),
679 ColumnSchema::new("mean_plan_time", DataType::Float, false),
680 ColumnSchema::new("stddev_plan_time", DataType::Float, false),
681 ColumnSchema::new("calls", DataType::BigInt, false),
682 ColumnSchema::new("total_exec_time", DataType::Float, false),
683 ColumnSchema::new("min_exec_time", DataType::Float, false),
684 ColumnSchema::new("max_exec_time", DataType::Float, false),
685 ColumnSchema::new("mean_exec_time", DataType::Float, false),
686 ColumnSchema::new("stddev_exec_time", DataType::Float, false),
687 ColumnSchema::new("rows", DataType::BigInt, false),
688 ColumnSchema::new("shared_blks_hit", DataType::BigInt, false),
689 ColumnSchema::new("shared_blks_read", DataType::BigInt, false),
690 ColumnSchema::new("shared_blks_dirtied", DataType::BigInt, false),
691 ColumnSchema::new("shared_blks_written", DataType::BigInt, false),
692 ColumnSchema::new("local_blks_hit", DataType::BigInt, false),
693 ColumnSchema::new("local_blks_read", DataType::BigInt, false),
694 ColumnSchema::new("local_blks_dirtied", DataType::BigInt, false),
695 ColumnSchema::new("local_blks_written", DataType::BigInt, false),
696 ColumnSchema::new("temp_blks_read", DataType::BigInt, false),
697 ColumnSchema::new("temp_blks_written", DataType::BigInt, false),
698 ColumnSchema::new("blk_read_time", DataType::Float, false),
699 ColumnSchema::new("blk_write_time", DataType::Float, false),
700 ColumnSchema::new("wal_records", DataType::BigInt, false),
701 ColumnSchema::new("wal_fpi", DataType::BigInt, false),
702 ColumnSchema::new("wal_bytes", DataType::BigInt, false),
703 ColumnSchema::new("jit_functions", DataType::BigInt, false),
704 ColumnSchema::new("jit_generation_time", DataType::Float, false),
705 ColumnSchema::new("jit_inlining_count", DataType::BigInt, false),
706 ColumnSchema::new("jit_inlining_time", DataType::Float, false),
707 ColumnSchema::new("jit_emission_count", DataType::BigInt, false),
708 ];
709 let rows: Vec<Row<'static>> = self
710 .query_stats
711 .snapshot()
712 .into_iter()
713 .map(|(sql, s)| {
714 let calls = i64::try_from(s.exec_count).unwrap_or(i64::MAX);
715 let total_ms = (s.total_us as f64) / 1000.0;
716 let max_ms = (s.max_us as f64) / 1000.0;
717 let mean_ms = if s.exec_count == 0 {
718 0.0
719 } else {
720 (s.total_us as f64) / 1000.0 / (s.exec_count as f64)
721 };
722 // queryid: PG uses a 64-bit hash of the normalised
723 // query text. SPG hashes the raw sql with FNV-1a-64
724 // (matches what pg_compatible_hash uses for HASH
725 // partitions). Stable across runs as long as the
726 // sql text is byte-identical.
727 let queryid = crate::partition::pg_compatible_hash(&spg_storage::Value::Text(
728 alloc::borrow::Cow::Borrowed(&sql),
729 )) as i64;
730 Row::new(alloc::vec![
731 Value::BigInt(10), // userid (PG superuser)
732 Value::BigInt(16384), // dbid
733 Value::Bool(true), // toplevel
734 Value::BigInt(queryid),
735 Value::Text(alloc::borrow::Cow::Owned(sql)),
736 Value::BigInt(calls), // plans
737 Value::Float(0.0), // total_plan_time
738 Value::Float(0.0), // min_plan_time
739 Value::Float(0.0), // max_plan_time
740 Value::Float(0.0), // mean_plan_time
741 Value::Float(0.0), // stddev_plan_time
742 Value::BigInt(calls), // calls
743 Value::Float(total_ms),
744 Value::Float(0.0), // min_exec_time
745 Value::Float(max_ms),
746 Value::Float(mean_ms),
747 Value::Float(0.0), // stddev_exec_time
748 // v7.37.22 (22.9) — total rows produced /
749 // affected, mapped from query_stats.total_rows.
750 Value::BigInt(i64::try_from(s.total_rows).unwrap_or(i64::MAX)),
751 // 8 shared_blks_*, 4 local_blks_*, 2 temp_blks_*
752 Value::BigInt(0),
753 Value::BigInt(0),
754 Value::BigInt(0),
755 Value::BigInt(0),
756 Value::BigInt(0),
757 Value::BigInt(0),
758 Value::BigInt(0),
759 Value::BigInt(0),
760 Value::BigInt(0),
761 Value::BigInt(0),
762 Value::Float(0.0), // blk_read_time
763 Value::Float(0.0), // blk_write_time
764 Value::BigInt(0), // wal_records
765 Value::BigInt(0), // wal_fpi
766 Value::BigInt(0), // wal_bytes
767 Value::BigInt(0), // jit_functions
768 Value::Float(0.0), // jit_generation_time
769 Value::BigInt(0), // jit_inlining_count
770 Value::Float(0.0), // jit_inlining_time
771 Value::BigInt(0), // jit_emission_count
772 ])
773 })
774 .collect();
775 QueryResult::Rows { columns, rows }
776 }
777
778 /// v7.37.22 (22.2) — materialise `pg_statio_user_tables` rows.
779 /// PG exposes per-relation I/O counters that monitoring tools
780 /// (pgwatch / pganalyze / Datadog) query routinely. SPG's
781 /// storage model is hot-tier rows + cold-tier segments, both
782 /// of which the engine tracks at finer granularity than PG's
783 /// shared-buffer hit/read split. v7.37.22 (22.2) ships the
784 /// SQL shape with the columns PG dashboards expect; the
785 /// `heap_blks_*` / `idx_blks_*` numbers map to SPG's
786 /// hot/cold accounting where the mapping is unambiguous and
787 /// stay 0 otherwise.
788 ///
789 /// Columns (PG-exact order):
790 /// relid OID NOT NULL -- monotonic per table
791 /// schemaname TEXT NOT NULL -- always 'public'
792 /// relname TEXT NOT NULL -- table name
793 /// heap_blks_read BIGINT NOT NULL -- cold-tier reads (stub: 0)
794 /// heap_blks_hit BIGINT NOT NULL -- hot-tier reads (live row count)
795 /// idx_blks_read BIGINT NOT NULL -- cold-tier index reads (0)
796 /// idx_blks_hit BIGINT NOT NULL -- hot-tier index hits (sum of NSW + BTree probe counters, future)
797 /// toast_blks_read BIGINT NOT NULL -- 0 (SPG has no TOAST)
798 /// toast_blks_hit BIGINT NOT NULL -- 0
799 /// tidx_blks_read BIGINT NOT NULL -- 0
800 /// tidx_blks_hit BIGINT NOT NULL -- 0
801 pub(crate) fn exec_pg_statio_user_tables(&self) -> QueryResult {
802 let columns = alloc::vec![
803 ColumnSchema::new("relid", DataType::BigInt, false),
804 ColumnSchema::new("schemaname", DataType::Text, false),
805 ColumnSchema::new("relname", DataType::Text, false),
806 ColumnSchema::new("heap_blks_read", DataType::BigInt, false),
807 ColumnSchema::new("heap_blks_hit", DataType::BigInt, false),
808 ColumnSchema::new("idx_blks_read", DataType::BigInt, false),
809 ColumnSchema::new("idx_blks_hit", DataType::BigInt, false),
810 ColumnSchema::new("toast_blks_read", DataType::BigInt, false),
811 ColumnSchema::new("toast_blks_hit", DataType::BigInt, false),
812 ColumnSchema::new("tidx_blks_read", DataType::BigInt, false),
813 ColumnSchema::new("tidx_blks_hit", DataType::BigInt, false),
814 ];
815 let mut rows: Vec<Row<'static>> = Vec::new();
816 let mut relid: i64 = 16384; // PG starts user-relation OIDs above 16384
817 for name in self.catalog.table_names() {
818 if is_internal_table_name(&name) {
819 continue;
820 }
821 let Some(t) = self.catalog.get(&name) else {
822 continue;
823 };
824 let live_rows = t.rows().len() as i64;
825 rows.push(Row::new(alloc::vec![
826 Value::BigInt(relid),
827 Value::text::<String>("public".into()),
828 Value::Text(alloc::borrow::Cow::Owned(name)),
829 Value::BigInt(0),
830 Value::BigInt(live_rows),
831 Value::BigInt(0),
832 Value::BigInt(0),
833 Value::BigInt(0),
834 Value::BigInt(0),
835 Value::BigInt(0),
836 Value::BigInt(0),
837 ]));
838 relid += 1;
839 }
840 QueryResult::Rows { columns, rows }
841 }
842
843 /// v7.37.14 (B6.5) — materialise `pg_locks` rows. PG exposes a
844 /// detailed lock table (locktype / database / relation /
845 /// virtualtransaction / pid / mode / granted / fastpath /
846 /// waitstart). SPG's single-writer + Arc-snapshot model means
847 /// the v7.37.14 row set is structurally empty most of the time
848 /// — there are no per-tuple locks to enumerate, and the global
849 /// engine RwLock is either held or not (no chain to walk).
850 /// v7.37.15 (per-row tuple lock implementation) populates rows
851 /// from the live LockTable; the SQL surface ships now so
852 /// adopters can already write monitoring queries / dashboards
853 /// against the stable column set.
854 pub(crate) fn exec_pg_locks(&self) -> QueryResult {
855 let columns = alloc::vec![
856 ColumnSchema::new("locktype", DataType::Text, false),
857 ColumnSchema::new("database", DataType::Text, false),
858 ColumnSchema::new("relation", DataType::Text, false),
859 ColumnSchema::new("virtualtransaction", DataType::Text, false),
860 ColumnSchema::new("pid", DataType::Int, false),
861 ColumnSchema::new("mode", DataType::Text, false),
862 ColumnSchema::new("granted", DataType::Bool, false),
863 ColumnSchema::new("fastpath", DataType::Bool, false),
864 ColumnSchema::new("waitstart_us", DataType::BigInt, false),
865 ];
866 // Empty row set until v7.37.15. Documented as the stable
867 // SQL surface — the row content fills in once tuple locks
868 // exist (B2.5 in AUDIT-3-categories).
869 let rows: Vec<Row<'static>> = Vec::new();
870 QueryResult::Rows { columns, rows }
871 }
872
873 /// v6.5.4 — materialise `spg_table_ddl` rows. One row per user
874 /// table with `(table_name, ddl)`. Reconstructed from catalog
875 /// state on demand.
876 pub(crate) fn exec_spg_table_ddl(&self) -> QueryResult {
877 let columns = alloc::vec![
878 ColumnSchema::new("table_name", DataType::Text, false),
879 ColumnSchema::new("ddl", DataType::Text, false),
880 ];
881 let rows: Vec<Row<'static>> = self
882 .catalog
883 .table_names()
884 .into_iter()
885 .filter(|n| !is_internal_table_name(n))
886 .filter_map(|name| {
887 let table = self.catalog.get(&name)?;
888 let ddl = render_create_table(&name, &table.schema().columns);
889 Some(Row::new(alloc::vec![Value::text(name), Value::text(ddl),]))
890 })
891 .collect();
892 QueryResult::Rows { columns, rows }
893 }
894
895 /// v6.5.4 — materialise `spg_role_ddl` rows. One row per user
896 /// with `(role_name, ddl)`. Password is redacted (matches the
897 /// `Statement::CreateUser` Display which prints `'<redacted>'`).
898 pub(crate) fn exec_spg_role_ddl(&self) -> QueryResult {
899 let columns = alloc::vec![
900 ColumnSchema::new("role_name", DataType::Text, false),
901 ColumnSchema::new("ddl", DataType::Text, false),
902 ];
903 let rows: Vec<Row<'static>> = self
904 .users
905 .iter()
906 .map(|(name, rec)| {
907 let ddl = alloc::format!(
908 "CREATE USER {name} WITH PASSWORD '<redacted>' ROLE '{}'",
909 rec.role.as_str(),
910 );
911 Row::new(alloc::vec![
912 Value::text(String::from(name)),
913 Value::text(ddl)
914 ])
915 })
916 .collect();
917 QueryResult::Rows { columns, rows }
918 }
919
920 /// v6.5.4 — materialise `spg_database_ddl`: single row whose
921 /// `ddl` column concatenates every user table's CREATE +
922 /// every role's CREATE in deterministic catalog order. Suitable
923 /// for piping back through `Engine::execute` to recreate a
924 /// schema-equivalent database.
925 pub(crate) fn exec_spg_database_ddl(&self) -> QueryResult {
926 let columns = alloc::vec![ColumnSchema::new("ddl", DataType::Text, false)];
927 let mut out = String::new();
928 for (name, rec) in self.effective_users().iter() {
929 out.push_str(&alloc::format!(
930 "CREATE USER {name} WITH PASSWORD '<redacted>' ROLE '{}';\n",
931 rec.role.as_str(),
932 ));
933 }
934 for name in self.catalog.table_names() {
935 if is_internal_table_name(&name) {
936 continue;
937 }
938 if let Some(table) = self.catalog.get(&name) {
939 out.push_str(&render_create_table(&name, &table.schema().columns));
940 out.push_str(";\n");
941 }
942 }
943 QueryResult::Rows {
944 columns,
945 rows: alloc::vec![Row::new(alloc::vec![Value::text(out)])],
946 }
947 }
948
949 /// v6.5.3 — materialise `spg_audit_chain` rows. Pulls a fresh
950 /// snapshot from the registered provider; empty when no
951 /// provider is set.
952 pub(crate) fn exec_spg_audit_chain(&self) -> QueryResult {
953 let columns = alloc::vec![
954 ColumnSchema::new("seq", DataType::BigInt, false),
955 ColumnSchema::new("ts_ms", DataType::BigInt, false),
956 ColumnSchema::new("prev_hash", DataType::Text, false),
957 ColumnSchema::new("entry_hash", DataType::Text, false),
958 ColumnSchema::new("sql", DataType::Text, false),
959 ];
960 let rows: Vec<Row<'static>> = self
961 .audit_chain_provider
962 .map(|f| f())
963 .unwrap_or_default()
964 .into_iter()
965 .map(|r| {
966 Row::new(alloc::vec![
967 Value::BigInt(r.seq),
968 Value::BigInt(r.ts_ms),
969 Value::text(r.prev_hash_hex),
970 Value::text(r.entry_hash_hex),
971 Value::text(r.sql),
972 ])
973 })
974 .collect();
975 QueryResult::Rows { columns, rows }
976 }
977
978 /// v6.5.3 — materialise `spg_audit_verify` single-row result.
979 /// `(verified_count, broken_at_seq)` — broken_at_seq is `-1`
980 /// on a clean chain. Returns one row with both values 0 when
981 /// no verifier is registered (no-data fallback for embedded
982 /// callers).
983 pub(crate) fn exec_spg_audit_verify(&self) -> QueryResult {
984 let columns = alloc::vec![
985 ColumnSchema::new("verified_count", DataType::BigInt, false),
986 ColumnSchema::new("broken_at_seq", DataType::BigInt, false),
987 ];
988 let (verified, broken) = self.audit_verifier.map(|f| f()).unwrap_or((0, -1));
989 let row = Row::new(alloc::vec![Value::BigInt(verified), Value::BigInt(broken),]);
990 QueryResult::Rows {
991 columns,
992 rows: alloc::vec![row],
993 }
994 }
995
996 /// v6.5.1 — read-only accessor for tests + v6.5.6 ops resets.
997 pub fn query_stats(&self) -> &query_stats::QueryStats {
998 &self.query_stats
999 }
1000
1001 /// v6.5.1 — mutable accessor (clear, etc).
1002 pub fn query_stats_mut(&mut self) -> &mut query_stats::QueryStats {
1003 &mut self.query_stats
1004 }
1005
1006 /// v6.2.0 — read access to the per-column statistics table.
1007 /// Used by the planner (v6.2.2 selectivity functions read this),
1008 /// by `SELECT * FROM spg_statistic`, and by e2e tests.
1009 pub const fn statistics(&self) -> &statistics::Statistics {
1010 &self.statistics
1011 }
1012
1013 /// v6.2.1 — return tables whose modified-row count crossed the
1014 /// auto-analyze threshold since the last ANALYZE on that table.
1015 /// The threshold is `0.1 × max(row_count, MIN_ROWS_FOR_AUTO_
1016 /// ANALYZE)` — combines PG-style fractional + absolute lower
1017 /// bound so a fresh / tiny table doesn't get hammered on every
1018 /// INSERT.
1019 ///
1020 /// Designed to be cheap: walks every user table's
1021 /// `Catalog::table_names()` + reads `statistics::modified_
1022 /// since_last_analyze()` (BTreeMap lookup). The background
1023 /// worker calls this under `engine.read()` then drops the lock
1024 /// before re-acquiring `engine.write()` for the actual ANALYZE.
1025 pub fn tables_needing_analyze(&self) -> Vec<String> {
1026 // v7.38 (read01 P5.29) — PG's autovacuum analyze threshold:
1027 // autovacuum_analyze_threshold (50) + autovacuum_analyze_scale_factor
1028 // (0.1) × reltuples. The prior formula (0.1 × max(rows, 100)) dropped
1029 // the 50-row base, so it re-analyzed small and mid-size tables far
1030 // more eagerly than PG.
1031 const ANALYZE_THRESHOLD_BASE: u64 = 50;
1032 let mut out = Vec::new();
1033 for name in self.catalog.table_names() {
1034 if is_internal_table_name(&name) {
1035 continue;
1036 }
1037 let Some(table) = self.catalog.get(&name) else {
1038 continue;
1039 };
1040 let row_count = table.rows().len() as u64;
1041 let modified = self.statistics.modified_since_last_analyze(&name);
1042 // `(n + 9) / 10` is `ceil(n / 10)` for non-negative `n`, computed
1043 // in integer arithmetic so spg-engine stays no_std (no libm).
1044 let threshold = ANALYZE_THRESHOLD_BASE.saturating_add(row_count.saturating_add(9) / 10);
1045 if modified >= threshold {
1046 out.push(name);
1047 }
1048 }
1049 out
1050 }
1051
1052 /// v7.37.22 (22.3) — autoanalyze pass.
1053 ///
1054 /// PG runs autovacuum + autoanalyze on a background timer.
1055 /// SPG's spg-embedded / spg-server hosts call this from their
1056 /// maintenance loop on a configurable cadence (default 60s,
1057 /// matching PG's `autovacuum_naptime`). Each call:
1058 ///
1059 /// 1. Walks `tables_needing_analyze()` (same threshold as the
1060 /// existing introspection API).
1061 /// 2. Runs `ANALYZE <table>` on each candidate.
1062 /// 3. Returns the names that were analyzed so the host can
1063 /// log / emit metrics.
1064 ///
1065 /// Internally identical to `ANALYZE name1; ANALYZE name2; …`
1066 /// but bundled so the plan-cache invalidation runs once at the
1067 /// end (cheaper than invalidating per-table). The host can call
1068 /// this under the engine write-lock without splicing extra
1069 /// SQL through the parser.
1070 ///
1071 /// Returns the (possibly empty) list of tables analyzed.
1072 pub fn autoanalyze_pass(&mut self) -> Result<Vec<String>, EngineError> {
1073 let candidates = self.tables_needing_analyze();
1074 if candidates.is_empty() {
1075 return Ok(Vec::new());
1076 }
1077 for name in &candidates {
1078 // `exec_analyze` for a single table also bumps
1079 // version + evicts that table's plans. Doing it
1080 // per-table here matches `ANALYZE a; ANALYZE b;`
1081 // semantics — a host that wants the bundled
1082 // optimisation can call `exec_analyze(None)` for the
1083 // bare ANALYZE-all path instead.
1084 self.exec_analyze(Some(name))?;
1085 }
1086 Ok(candidates)
1087 }
1088
1089 /// v7.37.15 (Phase D) — dead-tuple vacuum pass. The engine-level
1090 /// companion to [`Self::autoanalyze_pass`]: physically reclaims
1091 /// committed-tombstoned rows so a gate-on
1092 /// (`SPG_MVCC_INPLACE`) in-place table's storage stays bounded.
1093 ///
1094 /// Under gate-on a DELETE stamps `xmax` and keeps the row
1095 /// physically present (an UPDATE tombstones the old version and
1096 /// appends the new one); those dead rows accumulate until vacuum
1097 /// removes them. SPG's `xmin` / `xmax` are u64 with no wraparound,
1098 /// so this is pure dead-tuple reclamation — no anti-wraparound
1099 /// freeze is ever needed.
1100 ///
1101 /// # Safety predicate
1102 /// A tombstoned row (`xmax != XMAX_ALIVE`) is reclaimed iff its
1103 /// delete-commit version is **strictly below** `oldest_active` —
1104 /// the floor of every version any live reader could still resolve
1105 /// as visible (see [`Self::vacuum_oldest_active`]). `xmax <
1106 /// oldest_active` means every current and future snapshot already
1107 /// observes the delete, so no reader can still see the row. When in
1108 /// doubt the row is left in place — never reclaim a row that could
1109 /// still be visible.
1110 ///
1111 /// # Gate-off (default) is a provable no-op
1112 /// Under the default gate-off path DELETE removes rows *physically*,
1113 /// so no header ever carries a non-`XMAX_ALIVE` `xmax` and there is
1114 /// nothing to reclaim. The explicit guard below returns an empty
1115 /// report without walking any table, so gate-off behaviour is
1116 /// byte-for-byte unchanged.
1117 ///
1118 /// # RowId stability
1119 /// Reclaiming compacts `rows` / `headers` / `rowids` lock-step
1120 /// (via `Table::delete_rows_no_index`): every surviving row keeps
1121 /// its stable, never-reused `RowId`, so held row-locks and
1122 /// tombstone-redo references stay attached to the same row while its
1123 /// physical slot shifts down. Indices are rebuilt against the
1124 /// compacted rows.
1125 ///
1126 /// # Not a daemon (follow-up)
1127 /// This ships the callable primitive only. Wiring a background
1128 /// thread that calls it on a cadence (PG's `autovacuum_naptime`) is
1129 /// a separate concern — a host schedules it under the engine write
1130 /// lock, mirroring how it drives [`Self::autoanalyze_pass`]. Noted
1131 /// as a follow-up, not built in this slice.
1132 ///
1133 /// v7.37.16 — enable/disable the threshold-triggered autovacuum
1134 /// (default ON). Hosts wire `SPG_AUTOVACUUM=0|false|off` to this.
1135 pub fn set_autovacuum(&mut self, on: bool) {
1136 self.autovacuum = on;
1137 }
1138
1139 /// v7.39 (round 173) — turn the statement-exit **inline** vacuum
1140 /// off. A host that flips this off MUST drive
1141 /// [`Self::autovacuum_tick`] from a background worker, or dead
1142 /// rows accumulate without bound (spg-server couples the two: the
1143 /// flag only flips when the worker actually spawns).
1144 pub fn set_autovacuum_inline(&mut self, on: bool) {
1145 self.autovacuum_inline = on;
1146 }
1147
1148 /// v7.39 (round 173) — one background-worker autovacuum pass: walk
1149 /// every table, vacuum those over the PG-inspired threshold
1150 /// (`dead >= 1000 && dead*4 >= live` — same rule as the inline
1151 /// trigger). Returns how many tables were vacuumed. No-op while an
1152 /// explicit transaction is open (its tombstones aren't committed;
1153 /// the next tick picks the backlog up), when autovacuum is off, or
1154 /// when the in-place gate is off (no tombstones exist).
1155 pub fn autovacuum_tick(&mut self) -> usize {
1156 if !self.autovacuum || !self.mvcc_inplace || self.in_transaction() {
1157 return 0;
1158 }
1159 let candidates: Vec<String> = self
1160 .active_catalog()
1161 .table_names()
1162 .into_iter()
1163 .filter(|name| {
1164 self.active_catalog().get(name).is_some_and(|t| {
1165 let dead = t.dead_rows();
1166 let live = (t.row_count() as u64).saturating_sub(dead);
1167 dead >= 1000 && dead * 4 >= live
1168 })
1169 })
1170 .collect();
1171 if candidates.is_empty() {
1172 return 0;
1173 }
1174 let oldest_active = self.vacuum_oldest_active();
1175 let now_us = self.clock.map(|f| f());
1176 let mut vacuumed = 0;
1177 for name in candidates {
1178 if let Some(t) = self.active_catalog_mut().get_mut(&name) {
1179 let _report = t.vacuum(oldest_active, false);
1180 if let Some(us) = now_us {
1181 t.stamp_autovacuum(us);
1182 }
1183 crate::bump_counter!(AUTOVACUUM_FIRE_COUNT);
1184 vacuumed += 1;
1185 }
1186 }
1187 vacuumed
1188 }
1189
1190 /// v7.37.16 (autovacuum-lite, see .claude/state/autovacuum-design.md)
1191 /// — threshold check + single-table synchronous vacuum, called at the
1192 /// exit of an in-place DML statement. Fires only when: autovacuum is
1193 /// on, the in-place gate is on (gate-off produces no tombstones — the
1194 /// counter stays 0 and this returns immediately), NO explicit tx is
1195 /// open (an open tx's tombstones aren't committed and its rollback
1196 /// needs the xmax intact; the backlog is picked up by the next
1197 /// autocommit DML on the table), and the table's dead-row meter
1198 /// crosses the PG-inspired threshold: `dead >= 1000 && dead*4 >=
1199 /// live` (absolute floor keeps small tables from thrashing). The
1200 /// vacuum floor (`vacuum_oldest_active`) is conservative, so rows a
1201 /// held snapshot can still see survive and re-trigger later.
1202 /// v7.39 (round 169) — explicit per-table vacuum, the manual twin of
1203 /// `maybe_autovacuum` without the thresholds (a customer's VACUUM
1204 /// means "reclaim now"). Gate-off / unknown table are no-ops.
1205 pub(crate) fn vacuum_one_table(&mut self, table_name: &str) {
1206 if !self.mvcc_inplace {
1207 return;
1208 }
1209 let oldest_active = self.vacuum_oldest_active();
1210 let now_us = self.clock.map(|f| f());
1211 if let Some(t) = self.active_catalog_mut().get_mut(table_name) {
1212 let _report = t.vacuum(oldest_active, false);
1213 if let Some(us) = now_us {
1214 t.stamp_autovacuum(us);
1215 }
1216 }
1217 }
1218
1219 pub(crate) fn maybe_autovacuum(&mut self, table_name: &str) {
1220 // NB: `in_transaction()` (an EXPLICIT tx has a shadow catalog),
1221 // not `current_tx.is_some()` — the execute path pins an
1222 // IMPLICIT_TX marker for every autocommit statement too.
1223 // v7.39 (round 173) — `autovacuum_inline` off means a
1224 // background worker owns vacuum scheduling (autovacuum_tick);
1225 // the statement path only keeps the dead-row meters current.
1226 if !self.autovacuum
1227 || !self.autovacuum_inline
1228 || !self.mvcc_inplace
1229 || self.in_transaction()
1230 {
1231 return;
1232 }
1233 let Some(t) = self.active_catalog().get(table_name) else {
1234 return;
1235 };
1236 let dead = t.dead_rows();
1237 let live = (t.row_count() as u64).saturating_sub(dead);
1238 if dead < 1000 || dead * 4 < live {
1239 return;
1240 }
1241 let oldest_active = self.vacuum_oldest_active();
1242 // v7.39 (pg_stat knife C) — stamp last_autovacuum (host clock;
1243 // None on clockless embedded engines leaves the column NULL).
1244 let now_us = self.clock.map(|f| f());
1245 if let Some(t) = self.active_catalog_mut().get_mut(table_name) {
1246 let _report = t.vacuum(oldest_active, false);
1247 if let Some(us) = now_us {
1248 t.stamp_autovacuum(us);
1249 }
1250 crate::bump_counter!(AUTOVACUUM_FIRE_COUNT);
1251 }
1252 }
1253
1254 /// `dry_run = true` counts the reclaimable rows without mutating.
1255 pub fn vacuum_pass(&mut self, dry_run: bool) -> spg_storage::vacuum::VacuumReport {
1256 // Gate-off: physical delete leaves no tombstones. Provable
1257 // no-op — do not even walk the tables so the default path is
1258 // byte-for-byte unchanged.
1259 if !self.mvcc_inplace {
1260 return spg_storage::vacuum::VacuumReport::default();
1261 }
1262 let oldest_active = self.vacuum_oldest_active();
1263 self.catalog.vacuum_all(oldest_active, dry_run)
1264 }
1265
1266 /// v7.37.15 (Phase D) — the conservative vacuum floor: the smallest
1267 /// version any live reader could still resolve as visible. A
1268 /// tombstone with `xmax < this` is dead to every reader — current
1269 /// and future — so it is safe to reclaim.
1270 ///
1271 /// Computed as the **minimum** of:
1272 /// * `current_version()` — a *fresh* reader's floor: any new
1273 /// snapshot is taken at (or after) the live cursor and sees
1274 /// every delete stamped at or below it, so nothing below the
1275 /// cursor can be resurrected by a future reader;
1276 /// * `min(active_writer_versions)` — an in-flight writer reads at
1277 /// its own version and can still see rows deleted *after* it;
1278 /// * `min(cached RR/SER reader snapshot versions)` — a held
1279 /// REPEATABLE READ / SERIALIZABLE snapshot froze its view at
1280 /// capture and can still see rows deleted after that point.
1281 ///
1282 /// Taking the minimum is deliberately conservative: any live reader
1283 /// drags the floor down, leaving a row that *might* still be visible
1284 /// in place. Under SPG's single-global-`current_tx` serialized model
1285 /// there is at most one in-flight writer, so in the common quiescent
1286 /// case this collapses to `current_version()`.
1287 #[must_use]
1288 pub fn vacuum_oldest_active(&self) -> u64 {
1289 let mut floor = spg_storage::row_header::current_version();
1290 // In-flight writers: `active_writer_versions` is a BTreeSet, so
1291 // `.iter().next()` is its minimum.
1292 if let Some(&min_writer) = self.active_writer_versions.iter().next() {
1293 floor = floor.min(min_writer);
1294 }
1295 // Held REPEATABLE READ / SERIALIZABLE reader snapshots.
1296 for st in self.tx_catalogs.values() {
1297 if let Some(s) = st.cached_snapshot.as_ref() {
1298 floor = floor.min(s.version);
1299 }
1300 }
1301 floor
1302 }
1303}
1304
1305/// v7.37.16 — autovacuum trigger counter (counter-first observability;
1306/// same read-only diagnostic model as the Step-VM counters).
1307pub static AUTOVACUUM_FIRE_COUNT: core::sync::atomic::AtomicU64 =
1308 core::sync::atomic::AtomicU64::new(0);