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 /// v7.37.16 — turn the slow-query log off, the state PG expresses
327 /// as `log_min_duration_statement = -1`. Clears the floor and the
328 /// callback together, so an engine re-registered in the same
329 /// process cannot inherit a threshold from an earlier boot.
330 #[must_use]
331 pub const fn without_slow_query_log(mut self) -> Self {
332 self.slow_query_threshold_us = None;
333 self.slow_query_logger = None;
334 self
335 }
336
337 /// v6.5.6 — operator knob for plan cache cap. spg-server reads
338 /// `SPG_PLAN_CACHE_MAX` env at startup; uses this to override
339 /// the compile-time default of 256.
340 pub fn set_plan_cache_max(&mut self, n: usize) {
341 self.plan_cache.set_max_entries(n);
342 }
343
344 /// v6.5.2 — materialise `spg_stat_activity` rows. Pulls a fresh
345 /// snapshot from the registered `ActivityProvider`. Returns an
346 /// empty result set when no provider is registered (the no_std
347 /// embedded path with no pgwire layer).
348 pub(crate) fn exec_spg_stat_activity(&self) -> QueryResult {
349 // v7.37.14 (B6.3) — column order matches PG's
350 // pg_stat_activity for `wait_event_type` immediately before
351 // `wait_event` so client-side projection by ordinal stays
352 // robust even before adopters update to named projection.
353 let columns = alloc::vec![
354 ColumnSchema::new("pid", DataType::Int, false),
355 ColumnSchema::new("user", DataType::Text, false),
356 ColumnSchema::new("started_at_us", DataType::BigInt, false),
357 ColumnSchema::new("current_sql", DataType::Text, false),
358 ColumnSchema::new("wait_event_type", DataType::Text, false),
359 ColumnSchema::new("wait_event", DataType::Text, false),
360 ColumnSchema::new("elapsed_us", DataType::BigInt, false),
361 ColumnSchema::new("in_transaction", DataType::Bool, false),
362 ColumnSchema::new("application_name", DataType::Text, false),
363 ];
364 let rows: Vec<Row<'static>> = self
365 .activity_provider
366 .map(|f| f())
367 .unwrap_or_default()
368 .into_iter()
369 .map(|r| {
370 Row::new(alloc::vec![
371 Value::Int(i32::try_from(r.pid).unwrap_or(i32::MAX)),
372 Value::text(r.user),
373 Value::BigInt(r.started_at_us),
374 Value::text(r.current_sql),
375 Value::text(r.wait_event_type),
376 Value::text(r.wait_event),
377 Value::BigInt(r.elapsed_us),
378 Value::Bool(r.in_transaction),
379 Value::text(r.application_name),
380 ])
381 })
382 .collect();
383 QueryResult::Rows { columns, rows }
384 }
385
386 /// v7.38 (read01 P3.10) — the canonical `pg_stat_activity` view with
387 /// PG's column names, so monitoring tools (which query the standard
388 /// name + columns) work. SPG's `spg_stat_activity` carries the same
389 /// data under SPG-native names; here it is re-projected to PG's 22
390 /// columns, with the fields SPG doesn't track surfaced as NULL and
391 /// `state` derived from the in-transaction / running-query flags.
392 pub(crate) fn exec_pg_stat_activity(&self) -> QueryResult {
393 let columns = alloc::vec![
394 ColumnSchema::new("datid", DataType::BigInt, true),
395 ColumnSchema::new("datname", DataType::Text, true),
396 ColumnSchema::new("pid", DataType::Int, false),
397 ColumnSchema::new("leader_pid", DataType::Int, true),
398 ColumnSchema::new("usesysid", DataType::BigInt, true),
399 ColumnSchema::new("usename", DataType::Text, true),
400 ColumnSchema::new("application_name", DataType::Text, false),
401 ColumnSchema::new("client_addr", DataType::Text, true),
402 ColumnSchema::new("client_hostname", DataType::Text, true),
403 ColumnSchema::new("client_port", DataType::Int, true),
404 ColumnSchema::new("backend_start", DataType::Timestamptz, true),
405 ColumnSchema::new("xact_start", DataType::Timestamptz, true),
406 ColumnSchema::new("query_start", DataType::Timestamptz, true),
407 ColumnSchema::new("state_change", DataType::Timestamptz, true),
408 ColumnSchema::new("wait_event_type", DataType::Text, true),
409 ColumnSchema::new("wait_event", DataType::Text, true),
410 ColumnSchema::new("state", DataType::Text, true),
411 ColumnSchema::new("backend_xid", DataType::BigInt, true),
412 ColumnSchema::new("backend_xmin", DataType::BigInt, true),
413 ColumnSchema::new("query_id", DataType::BigInt, true),
414 ColumnSchema::new("query", DataType::Text, false),
415 ColumnSchema::new("backend_type", DataType::Text, false),
416 ];
417 let rows: Vec<Row<'static>> = self
418 .activity_provider
419 .map(|f| f())
420 .unwrap_or_default()
421 .into_iter()
422 .map(|r| {
423 // PG `state`: a running query is 'active'; otherwise
424 // 'idle in transaction' inside a txn, else 'idle'.
425 // v7.39 (round 474) — PG reports NULL state for a background
426 // process; only a client backend is idle or active.
427 let state = if r.backend_type != "client backend" {
428 ""
429 } else if !r.current_sql.is_empty() {
430 "active"
431 } else if r.in_transaction {
432 "idle in transaction"
433 } else {
434 "idle"
435 };
436 let started = Value::Timestamp(r.started_at_us);
437 Row::new(alloc::vec![
438 Value::Null, // datid
439 // v7.39 (round 319, V52) — each row's OWN database.
440 // This used to read the ASKING session's GUC and stamp
441 // it on every row, so one connection's database was
442 // reported as everybody's.
443 if r.database.is_empty() {
444 Value::Null
445 } else {
446 Value::text(r.database)
447 }, // datname
448 Value::Int(i32::try_from(r.pid).unwrap_or(i32::MAX)),
449 Value::Null, // leader_pid
450 Value::Null, // usesysid
451 Value::text(r.user), // usename
452 Value::text(r.application_name),
453 // v7.39 (round 319, V52) — the real peer. PG leaves
454 // client_hostname NULL unless log_hostname is on, which
455 // SPG has no equivalent of, so it stays NULL; the port
456 // is -1 for a connection with no TCP peer, as in PG.
457 if r.client_addr.is_empty() {
458 Value::Null
459 } else {
460 Value::text(r.client_addr)
461 }, // client_addr
462 Value::Null, // client_hostname
463 Value::Int(r.client_port), // client_port
464 started.clone(), // backend_start
465 if r.in_transaction {
466 started.clone()
467 } else {
468 Value::Null
469 }, // xact_start
470 if r.current_sql.is_empty() {
471 Value::Null
472 } else {
473 started
474 }, // query_start
475 Value::Null, // state_change
476 Value::text(r.wait_event_type),
477 Value::text(r.wait_event),
478 if state.is_empty() {
479 Value::Null
480 } else {
481 Value::text(alloc::string::String::from(state))
482 },
483 Value::Null, // backend_xid
484 Value::Null, // backend_xmin
485 Value::Null, // query_id
486 Value::text(r.current_sql), // query
487 // v7.39 (round 474) — the row's own backend_type, so a
488 // background worker reports as itself rather than as a
489 // client connection.
490 Value::text(r.backend_type.clone()),
491 ])
492 })
493 .collect();
494 QueryResult::Rows { columns, rows }
495 }
496
497 /// v7.37.15 (Phase F) — MVCC diagnostic view. Single-row
498 /// snapshot of the engine's per-row visibility state so
499 /// `spgctl` / monitoring can observe vacuum lag + in-flight
500 /// transaction count without reaching into engine internals.
501 ///
502 /// Columns:
503 /// - `current_version` — the live monotonic writer-version
504 /// cursor (next allocated version comes after this).
505 /// - `active_writer_count` — number of writer versions in
506 /// flight (= concurrent transactions). 0 means quiescent.
507 /// - `oldest_active_version` — floor of the active set;
508 /// vacuum can reclaim any row whose `xmax < this`.
509 pub(crate) fn exec_spg_stat_mvcc(&self) -> QueryResult {
510 let columns = alloc::vec![
511 ColumnSchema::new("current_version", DataType::BigInt, false),
512 ColumnSchema::new("active_writer_count", DataType::Int, false),
513 ColumnSchema::new("oldest_active_version", DataType::BigInt, false),
514 ];
515 let cv = spg_storage::row_header::current_version() as i64;
516 let active = self.active_writer_versions.len() as i32;
517 let oldest = self
518 .active_writer_versions
519 .iter()
520 .next()
521 .copied()
522 .unwrap_or(cv as u64) as i64;
523 let rows = alloc::vec![Row::new(alloc::vec![
524 Value::BigInt(cv),
525 Value::Int(active),
526 Value::BigInt(oldest),
527 ])];
528 QueryResult::Rows { columns, rows }
529 }
530
531 /// v7.37.16 (16.11 [PG+]) — materialise `spg_partition_health`
532 /// rows: one row per partition (Range / List / Hash / Default /
533 /// Parent), plus a "row_count" / "bound" diag column so dashboard
534 /// queries can size a partitioned table at a glance without
535 /// joining catalog tables. PG provides `pg_partitioned_table` +
536 /// `pg_inherits` + per-child `pg_class.reltuples`; SPG bundles
537 /// them into one easy view because dogfood / sentori dashboards
538 /// kept reaching for it.
539 ///
540 /// Columns:
541 /// parent_name TEXT NOT NULL -- parent table name, or
542 /// the partition name itself
543 /// when role == 'Parent'
544 /// partition_name TEXT NOT NULL -- the partition (or
545 /// parent) name
546 /// role TEXT NOT NULL -- 'Parent' | 'Range'
547 /// | 'List' | 'Hash'
548 /// | 'Default'
549 /// row_count BIGINT NOT NULL -- live row count
550 /// bound_desc TEXT NOT NULL -- human-readable bound for
551 /// diagnostics ('' for
552 /// Parent + DEFAULT)
553 pub(crate) fn exec_spg_partition_health(&self) -> QueryResult {
554 use spg_storage::PartitionRole;
555 let columns = alloc::vec![
556 ColumnSchema::new("parent_name", DataType::Text, false),
557 ColumnSchema::new("partition_name", DataType::Text, false),
558 ColumnSchema::new("role", DataType::Text, false),
559 ColumnSchema::new("row_count", DataType::BigInt, false),
560 ColumnSchema::new("bound_desc", DataType::Text, false),
561 ];
562 let mut rows: Vec<Row<'static>> = Vec::new();
563 for name in self.catalog.table_names() {
564 let Some(t) = self.catalog.get(&name) else {
565 continue;
566 };
567 let role = match &t.schema().partition_role {
568 None => continue,
569 Some(r) => r,
570 };
571 let row_count = t.rows().len() as i64;
572 let (parent, role_str, bound) = match role {
573 // v7.39 (round 645) — an inheritance child reports every
574 // parent it names; the diagnostic view lists the first,
575 // which is the only one single inheritance ever has.
576 PartitionRole::Inherits { parent_names } => (
577 parent_names.first().cloned().unwrap_or_default(),
578 alloc::string::String::from("Inherits"),
579 alloc::format!("INHERITS ({})", parent_names.join(", ")),
580 ),
581 PartitionRole::Parent { kind, .. } => {
582 let kind_str = match kind {
583 spg_storage::PartitionKind::Range => "RANGE",
584 spg_storage::PartitionKind::List => "LIST",
585 spg_storage::PartitionKind::Hash => "HASH",
586 };
587 (
588 name.clone(),
589 alloc::string::String::from("Parent"),
590 alloc::format!("PARTITION BY {kind_str}"),
591 )
592 }
593 PartitionRole::Range {
594 parent_name,
595 lower,
596 upper,
597 } => (
598 parent_name.clone(),
599 alloc::string::String::from("Range"),
600 alloc::format!(
601 "FROM ({}) TO ({})",
602 partition_bound_diag(lower),
603 partition_bound_diag(upper)
604 ),
605 ),
606 PartitionRole::List {
607 parent_name,
608 values,
609 } => {
610 let mut diag = alloc::string::String::from("IN (");
611 for (i, v) in values.iter().enumerate() {
612 if i > 0 {
613 diag.push_str(", ");
614 }
615 diag.push_str(&partition_bound_diag(v));
616 }
617 diag.push(')');
618 (
619 parent_name.clone(),
620 alloc::string::String::from("List"),
621 diag,
622 )
623 }
624 PartitionRole::Hash {
625 parent_name,
626 modulus,
627 remainder,
628 } => (
629 parent_name.clone(),
630 alloc::string::String::from("Hash"),
631 alloc::format!("WITH (MODULUS {modulus}, REMAINDER {remainder})"),
632 ),
633 PartitionRole::Default { parent_name } => (
634 parent_name.clone(),
635 alloc::string::String::from("Default"),
636 alloc::string::String::new(),
637 ),
638 };
639 rows.push(Row::new(alloc::vec![
640 Value::Text(alloc::borrow::Cow::Owned(parent)),
641 Value::Text(alloc::borrow::Cow::Owned(name)),
642 Value::Text(alloc::borrow::Cow::Owned(role_str)),
643 Value::BigInt(row_count),
644 Value::Text(alloc::borrow::Cow::Owned(bound)),
645 ]));
646 }
647 QueryResult::Rows { columns, rows }
648 }
649
650 /// v7.37.22 (22.1) — materialise `pg_stat_statements` rows with
651 /// PG's exact column shape. The data source is the same
652 /// `query_stats` registry that backs `spg_stat_query`, but the
653 /// surface is PG-compatible so dashboards/queries written
654 /// against `SELECT … FROM pg_stat_statements ORDER BY
655 /// total_exec_time DESC LIMIT 10` keep working.
656 ///
657 /// SPG ↔ PG mapping:
658 /// query ← stats.sql
659 /// calls ← stats.exec_count
660 /// total_exec_time ← stats.total_us / 1000 (ms)
661 /// min_exec_time ← 0 (no per-call min tracked yet)
662 /// max_exec_time ← stats.max_us / 1000
663 /// mean_exec_time ← derived
664 /// stddev_exec_time ← 0
665 /// rows ← 0 (per-row count tracking lands later)
666 /// userid ← 10 (PG's "postgres" superuser oid)
667 /// dbid ← 16384 (SPG single-db OID)
668 /// queryid ← hash of sql
669 /// plans ← stats.exec_count (one plan per call)
670 /// shared_blks_* ← 0 (no shared-buffer accounting)
671 /// local_blks_* ← 0
672 /// temp_blks_* ← 0
673 /// *_blk_*_time ← 0
674 /// wal_records / wal_fpi / wal_bytes ← 0 (per-stmt accounting)
675 /// jit_* ← 0 (no JIT)
676 /// stats_since / minmax_stats_since ← stats.last_seen_us
677 ///
678 /// 38 columns total to cover PG 18's pg_stat_statements view.
679 pub(crate) fn exec_pg_stat_statements(&self) -> QueryResult {
680 let columns = alloc::vec![
681 ColumnSchema::new("userid", DataType::BigInt, false),
682 ColumnSchema::new("dbid", DataType::BigInt, false),
683 ColumnSchema::new("toplevel", DataType::Bool, false),
684 ColumnSchema::new("queryid", DataType::BigInt, false),
685 ColumnSchema::new("query", DataType::Text, false),
686 ColumnSchema::new("plans", DataType::BigInt, false),
687 ColumnSchema::new("total_plan_time", DataType::Float, false),
688 ColumnSchema::new("min_plan_time", DataType::Float, false),
689 ColumnSchema::new("max_plan_time", DataType::Float, false),
690 ColumnSchema::new("mean_plan_time", DataType::Float, false),
691 ColumnSchema::new("stddev_plan_time", DataType::Float, false),
692 ColumnSchema::new("calls", DataType::BigInt, false),
693 ColumnSchema::new("total_exec_time", DataType::Float, false),
694 ColumnSchema::new("min_exec_time", DataType::Float, false),
695 ColumnSchema::new("max_exec_time", DataType::Float, false),
696 ColumnSchema::new("mean_exec_time", DataType::Float, false),
697 ColumnSchema::new("stddev_exec_time", DataType::Float, false),
698 ColumnSchema::new("rows", DataType::BigInt, false),
699 ColumnSchema::new("shared_blks_hit", DataType::BigInt, false),
700 ColumnSchema::new("shared_blks_read", DataType::BigInt, false),
701 ColumnSchema::new("shared_blks_dirtied", DataType::BigInt, false),
702 ColumnSchema::new("shared_blks_written", DataType::BigInt, false),
703 ColumnSchema::new("local_blks_hit", DataType::BigInt, false),
704 ColumnSchema::new("local_blks_read", DataType::BigInt, false),
705 ColumnSchema::new("local_blks_dirtied", DataType::BigInt, false),
706 ColumnSchema::new("local_blks_written", DataType::BigInt, false),
707 ColumnSchema::new("temp_blks_read", DataType::BigInt, false),
708 ColumnSchema::new("temp_blks_written", DataType::BigInt, false),
709 ColumnSchema::new("blk_read_time", DataType::Float, false),
710 ColumnSchema::new("blk_write_time", DataType::Float, false),
711 ColumnSchema::new("wal_records", DataType::BigInt, false),
712 ColumnSchema::new("wal_fpi", DataType::BigInt, false),
713 ColumnSchema::new("wal_bytes", DataType::BigInt, false),
714 ColumnSchema::new("jit_functions", DataType::BigInt, false),
715 ColumnSchema::new("jit_generation_time", DataType::Float, false),
716 ColumnSchema::new("jit_inlining_count", DataType::BigInt, false),
717 ColumnSchema::new("jit_inlining_time", DataType::Float, false),
718 ColumnSchema::new("jit_emission_count", DataType::BigInt, false),
719 ];
720 let rows: Vec<Row<'static>> = self
721 .query_stats
722 .snapshot()
723 .into_iter()
724 .map(|(sql, s)| {
725 let calls = i64::try_from(s.exec_count).unwrap_or(i64::MAX);
726 let total_ms = (s.total_us as f64) / 1000.0;
727 let max_ms = (s.max_us as f64) / 1000.0;
728 let mean_ms = if s.exec_count == 0 {
729 0.0
730 } else {
731 (s.total_us as f64) / 1000.0 / (s.exec_count as f64)
732 };
733 // queryid: PG uses a 64-bit hash of the normalised
734 // query text. SPG hashes the raw sql with FNV-1a-64
735 // (matches what pg_compatible_hash uses for HASH
736 // partitions). Stable across runs as long as the
737 // sql text is byte-identical.
738 let queryid = crate::partition::pg_compatible_hash(&spg_storage::Value::Text(
739 alloc::borrow::Cow::Borrowed(&sql),
740 )) as i64;
741 Row::new(alloc::vec![
742 Value::BigInt(10), // userid (PG superuser)
743 Value::BigInt(16384), // dbid
744 Value::Bool(true), // toplevel
745 Value::BigInt(queryid),
746 Value::Text(alloc::borrow::Cow::Owned(sql)),
747 Value::BigInt(calls), // plans
748 Value::Float(0.0), // total_plan_time
749 Value::Float(0.0), // min_plan_time
750 Value::Float(0.0), // max_plan_time
751 Value::Float(0.0), // mean_plan_time
752 Value::Float(0.0), // stddev_plan_time
753 Value::BigInt(calls), // calls
754 Value::Float(total_ms),
755 Value::Float(0.0), // min_exec_time
756 Value::Float(max_ms),
757 Value::Float(mean_ms),
758 Value::Float(0.0), // stddev_exec_time
759 // v7.37.22 (22.9) — total rows produced /
760 // affected, mapped from query_stats.total_rows.
761 Value::BigInt(i64::try_from(s.total_rows).unwrap_or(i64::MAX)),
762 // 8 shared_blks_*, 4 local_blks_*, 2 temp_blks_*
763 Value::BigInt(0),
764 Value::BigInt(0),
765 Value::BigInt(0),
766 Value::BigInt(0),
767 Value::BigInt(0),
768 Value::BigInt(0),
769 Value::BigInt(0),
770 Value::BigInt(0),
771 Value::BigInt(0),
772 Value::BigInt(0),
773 Value::Float(0.0), // blk_read_time
774 Value::Float(0.0), // blk_write_time
775 Value::BigInt(0), // wal_records
776 Value::BigInt(0), // wal_fpi
777 Value::BigInt(0), // wal_bytes
778 Value::BigInt(0), // jit_functions
779 Value::Float(0.0), // jit_generation_time
780 Value::BigInt(0), // jit_inlining_count
781 Value::Float(0.0), // jit_inlining_time
782 Value::BigInt(0), // jit_emission_count
783 ])
784 })
785 .collect();
786 QueryResult::Rows { columns, rows }
787 }
788
789 /// v7.37.22 (22.2) — materialise `pg_statio_user_tables` rows.
790 /// PG exposes per-relation I/O counters that monitoring tools
791 /// (pgwatch / pganalyze / Datadog) query routinely. SPG's
792 /// storage model is hot-tier rows + cold-tier segments, both
793 /// of which the engine tracks at finer granularity than PG's
794 /// shared-buffer hit/read split. v7.37.22 (22.2) ships the
795 /// SQL shape with the columns PG dashboards expect; the
796 /// `heap_blks_*` / `idx_blks_*` numbers map to SPG's
797 /// hot/cold accounting where the mapping is unambiguous and
798 /// stay 0 otherwise.
799 ///
800 /// Columns (PG-exact order):
801 /// relid OID NOT NULL -- monotonic per table
802 /// schemaname TEXT NOT NULL -- always 'public'
803 /// relname TEXT NOT NULL -- table name
804 /// heap_blks_read BIGINT NOT NULL -- cold-tier reads (stub: 0)
805 /// heap_blks_hit BIGINT NOT NULL -- hot-tier reads (live row count)
806 /// idx_blks_read BIGINT NOT NULL -- cold-tier index reads (0)
807 /// idx_blks_hit BIGINT NOT NULL -- hot-tier index hits (sum of NSW + BTree probe counters, future)
808 /// toast_blks_read BIGINT NOT NULL -- 0 (SPG has no TOAST)
809 /// toast_blks_hit BIGINT NOT NULL -- 0
810 /// tidx_blks_read BIGINT NOT NULL -- 0
811 /// tidx_blks_hit BIGINT NOT NULL -- 0
812 pub(crate) fn exec_pg_statio_user_tables(&self) -> QueryResult {
813 let columns = alloc::vec![
814 ColumnSchema::new("relid", DataType::BigInt, false),
815 ColumnSchema::new("schemaname", DataType::Text, false),
816 ColumnSchema::new("relname", DataType::Text, false),
817 ColumnSchema::new("heap_blks_read", DataType::BigInt, false),
818 ColumnSchema::new("heap_blks_hit", DataType::BigInt, false),
819 ColumnSchema::new("idx_blks_read", DataType::BigInt, false),
820 ColumnSchema::new("idx_blks_hit", DataType::BigInt, false),
821 ColumnSchema::new("toast_blks_read", DataType::BigInt, false),
822 ColumnSchema::new("toast_blks_hit", DataType::BigInt, false),
823 ColumnSchema::new("tidx_blks_read", DataType::BigInt, false),
824 ColumnSchema::new("tidx_blks_hit", DataType::BigInt, false),
825 ];
826 let mut rows: Vec<Row<'static>> = Vec::new();
827 let mut relid: i64 = 16384; // PG starts user-relation OIDs above 16384
828 for name in self.catalog.table_names() {
829 if is_internal_table_name(&name) {
830 continue;
831 }
832 let Some(t) = self.catalog.get(&name) else {
833 continue;
834 };
835 let live_rows = t.rows().len() as i64;
836 rows.push(Row::new(alloc::vec![
837 Value::BigInt(relid),
838 Value::text::<String>("public".into()),
839 Value::Text(alloc::borrow::Cow::Owned(name)),
840 Value::BigInt(0),
841 Value::BigInt(live_rows),
842 Value::BigInt(0),
843 Value::BigInt(0),
844 Value::BigInt(0),
845 Value::BigInt(0),
846 Value::BigInt(0),
847 Value::BigInt(0),
848 ]));
849 relid += 1;
850 }
851 QueryResult::Rows { columns, rows }
852 }
853
854 /// v7.37.14 (B6.5) — materialise `pg_locks` rows. PG exposes a
855 /// detailed lock table (locktype / database / relation /
856 /// virtualtransaction / pid / mode / granted / fastpath /
857 /// waitstart). SPG's single-writer + Arc-snapshot model means
858 /// the v7.37.14 row set is structurally empty most of the time
859 /// — there are no per-tuple locks to enumerate, and the global
860 /// engine RwLock is either held or not (no chain to walk).
861 /// v7.37.15 (per-row tuple lock implementation) populates rows
862 /// from the live LockTable; the SQL surface ships now so
863 /// adopters can already write monitoring queries / dashboards
864 /// against the stable column set.
865 pub(crate) fn exec_pg_locks(&self) -> QueryResult {
866 let columns = alloc::vec![
867 ColumnSchema::new("locktype", DataType::Text, false),
868 ColumnSchema::new("database", DataType::Text, false),
869 ColumnSchema::new("relation", DataType::Text, false),
870 ColumnSchema::new("virtualtransaction", DataType::Text, false),
871 ColumnSchema::new("pid", DataType::Int, false),
872 ColumnSchema::new("mode", DataType::Text, false),
873 ColumnSchema::new("granted", DataType::Bool, false),
874 ColumnSchema::new("fastpath", DataType::Bool, false),
875 ColumnSchema::new("waitstart_us", DataType::BigInt, false),
876 ];
877 // Empty row set until v7.37.15. Documented as the stable
878 // SQL surface — the row content fills in once tuple locks
879 // exist (B2.5 in AUDIT-3-categories).
880 let rows: Vec<Row<'static>> = Vec::new();
881 QueryResult::Rows { columns, rows }
882 }
883
884 /// v6.5.4 — materialise `spg_table_ddl` rows. One row per user
885 /// table with `(table_name, ddl)`. Reconstructed from catalog
886 /// state on demand.
887 pub(crate) fn exec_spg_table_ddl(&self) -> QueryResult {
888 let columns = alloc::vec![
889 ColumnSchema::new("table_name", DataType::Text, false),
890 ColumnSchema::new("ddl", DataType::Text, false),
891 ];
892 let rows: Vec<Row<'static>> = self
893 .catalog
894 .table_names()
895 .into_iter()
896 .filter(|n| !is_internal_table_name(n))
897 .filter_map(|name| {
898 let table = self.catalog.get(&name)?;
899 let ddl = render_create_table(&name, &table.schema().columns);
900 Some(Row::new(alloc::vec![Value::text(name), Value::text(ddl),]))
901 })
902 .collect();
903 QueryResult::Rows { columns, rows }
904 }
905
906 /// v6.5.4 — materialise `spg_role_ddl` rows. One row per user
907 /// with `(role_name, ddl)`. Password is redacted (matches the
908 /// `Statement::CreateUser` Display which prints `'<redacted>'`).
909 pub(crate) fn exec_spg_role_ddl(&self) -> QueryResult {
910 let columns = alloc::vec![
911 ColumnSchema::new("role_name", DataType::Text, false),
912 ColumnSchema::new("ddl", DataType::Text, false),
913 ];
914 let rows: Vec<Row<'static>> = self
915 .users
916 .iter()
917 .map(|(name, rec)| {
918 let ddl = alloc::format!(
919 "CREATE USER {name} WITH PASSWORD '<redacted>' ROLE '{}'",
920 rec.role.as_str(),
921 );
922 Row::new(alloc::vec![
923 Value::text(String::from(name)),
924 Value::text(ddl)
925 ])
926 })
927 .collect();
928 QueryResult::Rows { columns, rows }
929 }
930
931 /// v6.5.4 — materialise `spg_database_ddl`: single row whose
932 /// `ddl` column concatenates every user table's CREATE +
933 /// every role's CREATE in deterministic catalog order. Suitable
934 /// for piping back through `Engine::execute` to recreate a
935 /// schema-equivalent database.
936 pub(crate) fn exec_spg_database_ddl(&self) -> QueryResult {
937 let columns = alloc::vec![ColumnSchema::new("ddl", DataType::Text, false)];
938 let mut out = String::new();
939 for (name, rec) in self.effective_users().iter() {
940 out.push_str(&alloc::format!(
941 "CREATE USER {name} WITH PASSWORD '<redacted>' ROLE '{}';\n",
942 rec.role.as_str(),
943 ));
944 }
945 for name in self.catalog.table_names() {
946 if is_internal_table_name(&name) {
947 continue;
948 }
949 if let Some(table) = self.catalog.get(&name) {
950 out.push_str(&render_create_table(&name, &table.schema().columns));
951 out.push_str(";\n");
952 }
953 }
954 QueryResult::Rows {
955 columns,
956 rows: alloc::vec![Row::new(alloc::vec![Value::text(out)])],
957 }
958 }
959
960 /// v6.5.3 — materialise `spg_audit_chain` rows. Pulls a fresh
961 /// snapshot from the registered provider; empty when no
962 /// provider is set.
963 pub(crate) fn exec_spg_audit_chain(&self) -> QueryResult {
964 let columns = alloc::vec![
965 ColumnSchema::new("seq", DataType::BigInt, false),
966 ColumnSchema::new("ts_ms", DataType::BigInt, false),
967 ColumnSchema::new("prev_hash", DataType::Text, false),
968 ColumnSchema::new("entry_hash", DataType::Text, false),
969 ColumnSchema::new("sql", DataType::Text, false),
970 ];
971 let rows: Vec<Row<'static>> = self
972 .audit_chain_provider
973 .map(|f| f())
974 .unwrap_or_default()
975 .into_iter()
976 .map(|r| {
977 Row::new(alloc::vec![
978 Value::BigInt(r.seq),
979 Value::BigInt(r.ts_ms),
980 Value::text(r.prev_hash_hex),
981 Value::text(r.entry_hash_hex),
982 Value::text(r.sql),
983 ])
984 })
985 .collect();
986 QueryResult::Rows { columns, rows }
987 }
988
989 /// v6.5.3 — materialise `spg_audit_verify` single-row result.
990 /// `(verified_count, broken_at_seq)` — broken_at_seq is `-1`
991 /// on a clean chain. Returns one row with both values 0 when
992 /// no verifier is registered (no-data fallback for embedded
993 /// callers).
994 pub(crate) fn exec_spg_audit_verify(&self) -> QueryResult {
995 let columns = alloc::vec![
996 ColumnSchema::new("verified_count", DataType::BigInt, false),
997 ColumnSchema::new("broken_at_seq", DataType::BigInt, false),
998 ];
999 let (verified, broken) = self.audit_verifier.map(|f| f()).unwrap_or((0, -1));
1000 let row = Row::new(alloc::vec![Value::BigInt(verified), Value::BigInt(broken),]);
1001 QueryResult::Rows {
1002 columns,
1003 rows: alloc::vec![row],
1004 }
1005 }
1006
1007 /// v6.5.1 — read-only accessor for tests + v6.5.6 ops resets.
1008 pub fn query_stats(&self) -> &query_stats::QueryStats {
1009 &self.query_stats
1010 }
1011
1012 /// v6.5.1 — mutable accessor (clear, etc).
1013 pub fn query_stats_mut(&mut self) -> &mut query_stats::QueryStats {
1014 &mut self.query_stats
1015 }
1016
1017 /// v6.2.0 — read access to the per-column statistics table.
1018 /// Used by the planner (v6.2.2 selectivity functions read this),
1019 /// by `SELECT * FROM spg_statistic`, and by e2e tests.
1020 pub const fn statistics(&self) -> &statistics::Statistics {
1021 &self.statistics
1022 }
1023
1024 /// v6.2.1 — return tables whose modified-row count crossed the
1025 /// auto-analyze threshold since the last ANALYZE on that table.
1026 /// The threshold is `0.1 × max(row_count, MIN_ROWS_FOR_AUTO_
1027 /// ANALYZE)` — combines PG-style fractional + absolute lower
1028 /// bound so a fresh / tiny table doesn't get hammered on every
1029 /// INSERT.
1030 ///
1031 /// Designed to be cheap: walks every user table's
1032 /// `Catalog::table_names()` + reads `statistics::modified_
1033 /// since_last_analyze()` (BTreeMap lookup). The background
1034 /// worker calls this under `engine.read()` then drops the lock
1035 /// before re-acquiring `engine.write()` for the actual ANALYZE.
1036 pub fn tables_needing_analyze(&self) -> Vec<String> {
1037 // v7.38 (read01 P5.29) — PG's autovacuum analyze threshold:
1038 // autovacuum_analyze_threshold (50) + autovacuum_analyze_scale_factor
1039 // (0.1) × reltuples. The prior formula (0.1 × max(rows, 100)) dropped
1040 // the 50-row base, so it re-analyzed small and mid-size tables far
1041 // more eagerly than PG.
1042 const ANALYZE_THRESHOLD_BASE: u64 = 50;
1043 let mut out = Vec::new();
1044 for name in self.catalog.table_names() {
1045 if is_internal_table_name(&name) {
1046 continue;
1047 }
1048 let Some(table) = self.catalog.get(&name) else {
1049 continue;
1050 };
1051 let row_count = table.rows().len() as u64;
1052 let modified = self.statistics.modified_since_last_analyze(&name);
1053 // `(n + 9) / 10` is `ceil(n / 10)` for non-negative `n`, computed
1054 // in integer arithmetic so spg-engine stays no_std (no libm).
1055 let threshold = ANALYZE_THRESHOLD_BASE.saturating_add(row_count.saturating_add(9) / 10);
1056 if modified >= threshold {
1057 out.push(name);
1058 }
1059 }
1060 out
1061 }
1062
1063 /// v7.37.22 (22.3) — autoanalyze pass.
1064 ///
1065 /// PG runs autovacuum + autoanalyze on a background timer.
1066 /// SPG's spg-embedded / spg-server hosts call this from their
1067 /// maintenance loop on a configurable cadence (default 60s,
1068 /// matching PG's `autovacuum_naptime`). Each call:
1069 ///
1070 /// 1. Walks `tables_needing_analyze()` (same threshold as the
1071 /// existing introspection API).
1072 /// 2. Runs `ANALYZE <table>` on each candidate.
1073 /// 3. Returns the names that were analyzed so the host can
1074 /// log / emit metrics.
1075 ///
1076 /// Internally identical to `ANALYZE name1; ANALYZE name2; …`
1077 /// but bundled so the plan-cache invalidation runs once at the
1078 /// end (cheaper than invalidating per-table). The host can call
1079 /// this under the engine write-lock without splicing extra
1080 /// SQL through the parser.
1081 ///
1082 /// Returns the (possibly empty) list of tables analyzed.
1083 pub fn autoanalyze_pass(&mut self) -> Result<Vec<String>, EngineError> {
1084 let candidates = self.tables_needing_analyze();
1085 if candidates.is_empty() {
1086 return Ok(Vec::new());
1087 }
1088 for name in &candidates {
1089 // `exec_analyze` for a single table also bumps
1090 // version + evicts that table's plans. Doing it
1091 // per-table here matches `ANALYZE a; ANALYZE b;`
1092 // semantics — a host that wants the bundled
1093 // optimisation can call `exec_analyze(None)` for the
1094 // bare ANALYZE-all path instead.
1095 self.exec_analyze(Some(name))?;
1096 }
1097 Ok(candidates)
1098 }
1099
1100 /// v7.37.15 (Phase D) — dead-tuple vacuum pass. The engine-level
1101 /// companion to [`Self::autoanalyze_pass`]: physically reclaims
1102 /// committed-tombstoned rows so a gate-on
1103 /// (`SPG_MVCC_INPLACE`) in-place table's storage stays bounded.
1104 ///
1105 /// Under gate-on a DELETE stamps `xmax` and keeps the row
1106 /// physically present (an UPDATE tombstones the old version and
1107 /// appends the new one); those dead rows accumulate until vacuum
1108 /// removes them. SPG's `xmin` / `xmax` are u64 with no wraparound,
1109 /// so this is pure dead-tuple reclamation — no anti-wraparound
1110 /// freeze is ever needed.
1111 ///
1112 /// # Safety predicate
1113 /// A tombstoned row (`xmax != XMAX_ALIVE`) is reclaimed iff its
1114 /// delete-commit version is **strictly below** `oldest_active` —
1115 /// the floor of every version any live reader could still resolve
1116 /// as visible (see [`Self::vacuum_oldest_active`]). `xmax <
1117 /// oldest_active` means every current and future snapshot already
1118 /// observes the delete, so no reader can still see the row. When in
1119 /// doubt the row is left in place — never reclaim a row that could
1120 /// still be visible.
1121 ///
1122 /// # Gate-off (default) is a provable no-op
1123 /// Under the default gate-off path DELETE removes rows *physically*,
1124 /// so no header ever carries a non-`XMAX_ALIVE` `xmax` and there is
1125 /// nothing to reclaim. The explicit guard below returns an empty
1126 /// report without walking any table, so gate-off behaviour is
1127 /// byte-for-byte unchanged.
1128 ///
1129 /// # RowId stability
1130 /// Reclaiming compacts `rows` / `headers` / `rowids` lock-step
1131 /// (via `Table::delete_rows_no_index`): every surviving row keeps
1132 /// its stable, never-reused `RowId`, so held row-locks and
1133 /// tombstone-redo references stay attached to the same row while its
1134 /// physical slot shifts down. Indices are rebuilt against the
1135 /// compacted rows.
1136 ///
1137 /// # Not a daemon (follow-up)
1138 /// This ships the callable primitive only. Wiring a background
1139 /// thread that calls it on a cadence (PG's `autovacuum_naptime`) is
1140 /// a separate concern — a host schedules it under the engine write
1141 /// lock, mirroring how it drives [`Self::autoanalyze_pass`]. Noted
1142 /// as a follow-up, not built in this slice.
1143 ///
1144 /// v7.37.16 — enable/disable the threshold-triggered autovacuum
1145 /// (default ON). Hosts wire `SPG_AUTOVACUUM=0|false|off` to this.
1146 pub fn set_autovacuum(&mut self, on: bool) {
1147 self.autovacuum = on;
1148 }
1149
1150 /// v7.39 (round 173) — turn the statement-exit **inline** vacuum
1151 /// off. A host that flips this off MUST drive
1152 /// [`Self::autovacuum_tick`] from a background worker, or dead
1153 /// rows accumulate without bound (spg-server couples the two: the
1154 /// flag only flips when the worker actually spawns).
1155 pub fn set_autovacuum_inline(&mut self, on: bool) {
1156 self.autovacuum_inline = on;
1157 }
1158
1159 /// v7.39 (round 173) — one background-worker autovacuum pass: walk
1160 /// every table, vacuum those over the PG-inspired threshold
1161 /// (`dead >= 1000 && dead*4 >= live` — same rule as the inline
1162 /// trigger). Returns how many tables were vacuumed. No-op while an
1163 /// explicit transaction is open (its tombstones aren't committed;
1164 /// the next tick picks the backlog up), when autovacuum is off, or
1165 /// when the in-place gate is off (no tombstones exist).
1166 pub fn autovacuum_tick(&mut self) -> usize {
1167 if !self.autovacuum || !self.mvcc_inplace || self.in_transaction() {
1168 return 0;
1169 }
1170 let candidates: Vec<String> = self
1171 .active_catalog()
1172 .table_names()
1173 .into_iter()
1174 .filter(|name| {
1175 self.active_catalog().get(name).is_some_and(|t| {
1176 let dead = t.dead_rows();
1177 let live = (t.row_count() as u64).saturating_sub(dead);
1178 dead >= 1000 && dead * 4 >= live
1179 })
1180 })
1181 .collect();
1182 if candidates.is_empty() {
1183 return 0;
1184 }
1185 let oldest_active = self.vacuum_oldest_active();
1186 let now_us = self.clock.map(|f| f());
1187 let mut vacuumed = 0;
1188 for name in candidates {
1189 if let Some(t) = self.active_catalog_mut().get_mut(&name) {
1190 let _report = t.vacuum(oldest_active, false);
1191 if let Some(us) = now_us {
1192 t.stamp_autovacuum(us);
1193 }
1194 crate::bump_counter!(AUTOVACUUM_FIRE_COUNT);
1195 vacuumed += 1;
1196 }
1197 }
1198 vacuumed
1199 }
1200
1201 /// v7.37.16 (autovacuum-lite, see .claude/state/autovacuum-design.md)
1202 /// — threshold check + single-table synchronous vacuum, called at the
1203 /// exit of an in-place DML statement. Fires only when: autovacuum is
1204 /// on, the in-place gate is on (gate-off produces no tombstones — the
1205 /// counter stays 0 and this returns immediately), NO explicit tx is
1206 /// open (an open tx's tombstones aren't committed and its rollback
1207 /// needs the xmax intact; the backlog is picked up by the next
1208 /// autocommit DML on the table), and the table's dead-row meter
1209 /// crosses the PG-inspired threshold: `dead >= 1000 && dead*4 >=
1210 /// live` (absolute floor keeps small tables from thrashing). The
1211 /// vacuum floor (`vacuum_oldest_active`) is conservative, so rows a
1212 /// held snapshot can still see survive and re-trigger later.
1213 /// v7.39 (round 169) — explicit per-table vacuum, the manual twin of
1214 /// `maybe_autovacuum` without the thresholds (a customer's VACUUM
1215 /// means "reclaim now"). Gate-off / unknown table are no-ops.
1216 pub(crate) fn vacuum_one_table(&mut self, table_name: &str) {
1217 if !self.mvcc_inplace {
1218 return;
1219 }
1220 let oldest_active = self.vacuum_oldest_active();
1221 let now_us = self.clock.map(|f| f());
1222 if let Some(t) = self.active_catalog_mut().get_mut(table_name) {
1223 let _report = t.vacuum(oldest_active, false);
1224 if let Some(us) = now_us {
1225 t.stamp_autovacuum(us);
1226 }
1227 }
1228 }
1229
1230 pub(crate) fn maybe_autovacuum(&mut self, table_name: &str) {
1231 // NB: `in_transaction()` (an EXPLICIT tx has a shadow catalog),
1232 // not `current_tx.is_some()` — the execute path pins an
1233 // IMPLICIT_TX marker for every autocommit statement too.
1234 // v7.39 (round 173) — `autovacuum_inline` off means a
1235 // background worker owns vacuum scheduling (autovacuum_tick);
1236 // the statement path only keeps the dead-row meters current.
1237 if !self.autovacuum
1238 || !self.autovacuum_inline
1239 || !self.mvcc_inplace
1240 || self.in_transaction()
1241 {
1242 return;
1243 }
1244 let Some(t) = self.active_catalog().get(table_name) else {
1245 return;
1246 };
1247 let dead = t.dead_rows();
1248 let live = (t.row_count() as u64).saturating_sub(dead);
1249 if dead < 1000 || dead * 4 < live {
1250 return;
1251 }
1252 let oldest_active = self.vacuum_oldest_active();
1253 // v7.39 (pg_stat knife C) — stamp last_autovacuum (host clock;
1254 // None on clockless embedded engines leaves the column NULL).
1255 let now_us = self.clock.map(|f| f());
1256 if let Some(t) = self.active_catalog_mut().get_mut(table_name) {
1257 let _report = t.vacuum(oldest_active, false);
1258 if let Some(us) = now_us {
1259 t.stamp_autovacuum(us);
1260 }
1261 crate::bump_counter!(AUTOVACUUM_FIRE_COUNT);
1262 }
1263 }
1264
1265 /// `dry_run = true` counts the reclaimable rows without mutating.
1266 pub fn vacuum_pass(&mut self, dry_run: bool) -> spg_storage::vacuum::VacuumReport {
1267 // Gate-off: physical delete leaves no tombstones. Provable
1268 // no-op — do not even walk the tables so the default path is
1269 // byte-for-byte unchanged.
1270 if !self.mvcc_inplace {
1271 return spg_storage::vacuum::VacuumReport::default();
1272 }
1273 let oldest_active = self.vacuum_oldest_active();
1274 self.catalog.vacuum_all(oldest_active, dry_run)
1275 }
1276
1277 /// v7.37.15 (Phase D) — the conservative vacuum floor: the smallest
1278 /// version any live reader could still resolve as visible. A
1279 /// tombstone with `xmax < this` is dead to every reader — current
1280 /// and future — so it is safe to reclaim.
1281 ///
1282 /// Computed as the **minimum** of:
1283 /// * `current_version()` — a *fresh* reader's floor: any new
1284 /// snapshot is taken at (or after) the live cursor and sees
1285 /// every delete stamped at or below it, so nothing below the
1286 /// cursor can be resurrected by a future reader;
1287 /// * `min(active_writer_versions)` — an in-flight writer reads at
1288 /// its own version and can still see rows deleted *after* it;
1289 /// * `min(cached RR/SER reader snapshot versions)` — a held
1290 /// REPEATABLE READ / SERIALIZABLE snapshot froze its view at
1291 /// capture and can still see rows deleted after that point.
1292 ///
1293 /// Taking the minimum is deliberately conservative: any live reader
1294 /// drags the floor down, leaving a row that *might* still be visible
1295 /// in place. Under SPG's single-global-`current_tx` serialized model
1296 /// there is at most one in-flight writer, so in the common quiescent
1297 /// case this collapses to `current_version()`.
1298 #[must_use]
1299 pub fn vacuum_oldest_active(&self) -> u64 {
1300 let mut floor = spg_storage::row_header::current_version();
1301 // In-flight writers: `active_writer_versions` is a BTreeSet, so
1302 // `.iter().next()` is its minimum.
1303 if let Some(&min_writer) = self.active_writer_versions.iter().next() {
1304 floor = floor.min(min_writer);
1305 }
1306 // Held REPEATABLE READ / SERIALIZABLE reader snapshots.
1307 for st in self.tx_catalogs.values() {
1308 if let Some(s) = st.cached_snapshot.as_ref() {
1309 floor = floor.min(s.version);
1310 }
1311 }
1312 floor
1313 }
1314}
1315
1316/// v7.37.16 — autovacuum trigger counter (counter-first observability;
1317/// same read-only diagnostic model as the Step-VM counters).
1318pub static AUTOVACUUM_FIRE_COUNT: core::sync::atomic::AtomicU64 =
1319 core::sync::atomic::AtomicU64::new(0);