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