Skip to main content

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);