spg_engine/lib.rs
1//! SPG execution engine — v0.3 wires the SQL front-end to the in-memory
2//! storage layer. Implements `CREATE TABLE`, single-row `INSERT VALUES`, and
3//! `SELECT * FROM <table>` (no WHERE yet — that lands in v0.4 alongside
4//! expression evaluation against rows).
5#![no_std]
6
7extern crate alloc;
8
9// v7.37.9 T3 — `bump_counter!(C)` / `bump_counter!(C, N)` macros for the
10// Step VM + aggregate hot-path diagnostic counters. Gated on the
11// `perf-counters` feature so release builds pay zero cost; the
12// `xtests/dogfood_replay/spg-counter-dump` binary turns the feature on
13// to attribute Class A / B / C cascade cost.
14#[cfg(not(feature = "perf-counters"))]
15#[macro_export]
16macro_rules! bump_counter {
17 ($c:path) => {{
18 let _ = &$c;
19 }};
20 ($c:path, $n:expr) => {{
21 let _ = &$c;
22 let _ = &$n;
23 }};
24}
25
26#[cfg(feature = "perf-counters")]
27#[macro_export]
28macro_rules! bump_counter {
29 ($c:path) => {{
30 $c.fetch_add(1, core::sync::atomic::Ordering::Relaxed);
31 }};
32 ($c:path, $n:expr) => {{
33 $c.fetch_add($n, core::sync::atomic::Ordering::Relaxed);
34 }};
35}
36
37pub mod aggregate;
38mod bytebudget;
39mod cancel;
40mod clock;
41mod constraints;
42mod conversions;
43pub mod copy;
44mod ddl;
45pub mod describe;
46mod dml;
47mod envelope;
48pub mod eval;
49mod execute;
50mod explain;
51mod expr_analysis;
52pub mod fts;
53mod index_access;
54mod join;
55mod joinfold;
56pub mod json;
57mod maintenance;
58pub mod memoize;
59mod numeric;
60mod orderby;
61mod partition;
62pub mod plan_cache;
63mod plpgsql;
64pub mod publications;
65pub mod query_stats;
66mod readonly;
67pub mod reorder;
68pub mod scalarsq_streaming;
69mod select;
70pub mod selectivity;
71mod sequence;
72mod session;
73mod show;
74mod spg_admin;
75pub mod statistics;
76pub mod subquery;
77pub mod subscriptions;
78mod substitute;
79mod system_catalog;
80mod table_access;
81pub mod testkit;
82mod transaction;
83pub mod triggers;
84pub mod users;
85mod window;
86
87pub use crate::users::{Role, ScramSecrets, UserError, UserStore};
88pub use cancel::{CancelToken, MonotonicNowFn};
89pub use execute::StreamItem;
90
91use bytebudget::*;
92pub(crate) use clock::{rewrite_clock_calls, value_to_literal};
93use constraints::*;
94use conversions::*;
95pub use conversions::{
96 format_bigint_2d_text_pub, format_bit_string, format_circle, format_hstore_text, format_inet,
97 format_int_2d_text_pub, format_line, format_lseg, format_macaddr, format_macaddr8,
98 format_multirange, format_path, format_pg_box, format_point, format_polygon, format_range_text,
99 format_text_2d_text_pub,
100};
101pub(crate) use ddl::{
102 canonicalize_set_value, enforce_enum_label, eval_runtime_default_free,
103 resolve_column_default_free,
104};
105pub(crate) use envelope::{EnvelopeParse, build_envelope, split_envelope};
106use expr_analysis::*;
107use index_access::*;
108pub use join::{ANTI_JOIN_FAST_PATH_FIRED, ANTI_JOIN_FAST_PATH_TRIED};
109pub(crate) use orderby::{
110 apply_offset_and_limit, apply_offset_and_limit_tagged, build_order_keys, canonical_value_repr,
111 expand_group_by_all, order_by_value_cmp, partial_sort_tagged, render_histogram_bounds,
112 resolve_order_by_position, sort_by_keys, sort_values_for_histogram, value_cmp, value_to_f64,
113};
114pub(crate) use select::{build_projection, infer_column_types, value_to_order_key};
115pub(crate) use show::render_create_table;
116pub use subquery::{
117 BATCHED_SCALAR_FALL_THROUGH_COUNT, BATCHED_SCALAR_KEYED_FIRE_COUNT,
118 BATCHED_SCALAR_KEYED_PROBE_COUNT, EXISTS_BATCH_FALL_THROUGH_COUNT, EXISTS_BATCH_FIRE_COUNT,
119 EXISTS_PULLUP_BAIL_INNER_FROM, EXISTS_PULLUP_BAIL_INNER_SHAPE,
120 EXISTS_PULLUP_BAIL_MULTICOL_DISABLED, EXISTS_PULLUP_BAIL_NO_CORR,
121 EXISTS_PULLUP_BAIL_NO_WHERE, EXISTS_PULLUP_BAIL_RESIDUAL_NOT_INNER,
122 EXISTS_PULLUP_BAIL_UNIQUE_KEY_MISSING, EXISTS_PULLUP_CANDIDATE_COUNT,
123 EXISTS_PULLUP_FIRE_COUNT, EXISTS_PULLUP_MULTICOL_DISABLE, PULLUP_LIMIT1_FIRE_COUNT,
124 SCALARSQ_PK_PROBE_FIRED, ScalarPkProbeFastPath, expr_tree_has_subquery,
125};
126pub(crate) use subquery::{build_in_list_set, collect_scalar_subqueries, expr_has_subquery};
127pub use substitute::substitute_placeholders;
128use substitute::*;
129use system_catalog::*;
130use window::*;
131
132use alloc::collections::BTreeMap;
133use alloc::string::String;
134use alloc::vec::Vec;
135use core::fmt;
136
137// v7.16.0 — re-export the parsed-statement AST so downstream
138// crates (spg-embedded → spg-sqlx) don't need a direct dep on
139// spg-sql for the prepare/bind handle.
140pub use spg_sql::ast::{SelectStatement, Statement as ParsedStatement};
141use spg_sql::parser::ParseError;
142use spg_storage::{Catalog, ColumnSchema, Row, RowChange, StorageError};
143
144use crate::eval::EvalError;
145
146/// Result of executing one statement.
147#[derive(Debug, Clone, PartialEq)]
148#[non_exhaustive]
149pub enum QueryResult {
150 /// DDL or DML succeeded.
151 ///
152 /// `affected` is the row count for `INSERT` and 0 elsewhere.
153 /// `modified_catalog` tells the server whether this statement
154 /// caused the *committed* catalog to change — it's the signal to
155 /// snapshot/audit. False for `BEGIN`/`ROLLBACK`, false for writeful
156 /// statements executed inside a transaction (those only touch the
157 /// shadow), and true for `COMMIT` and for writes outside a TX.
158 CommandOk {
159 affected: usize,
160 modified_catalog: bool,
161 },
162 /// `SELECT` returned a (possibly empty) row set.
163 Rows {
164 columns: Vec<ColumnSchema>,
165 rows: Vec<Row<'static>>,
166 },
167}
168
169/// All errors the engine can return.
170///
171/// Marked `#[non_exhaustive]` from v7.5.0 onward: external `match`
172/// must include a `_` arm so new variants in subsequent v7.x releases
173/// are not breaking changes.
174#[derive(Debug, Clone, PartialEq)]
175#[non_exhaustive]
176pub enum EngineError {
177 Parse(ParseError),
178 Storage(StorageError),
179 Eval(EvalError),
180 /// Front-end accepted a construct that the v0.x executor doesn't support.
181 Unsupported(String),
182 /// `BEGIN` while another transaction is already open.
183 TransactionAlreadyOpen,
184 /// `COMMIT` / `ROLLBACK` with no active transaction.
185 NoActiveTransaction,
186 /// v4.0 sentinel: `execute_readonly` got a statement that
187 /// mutates engine state (INSERT / CREATE / BEGIN / COMMIT / …).
188 /// The caller should retake the write lock and dispatch through
189 /// `execute(&mut self)` instead.
190 WriteRequired,
191 /// v4.2: a SELECT would have returned more rows than the
192 /// configured `max_query_rows` cap. Carries the cap.
193 RowLimitExceeded(usize),
194 /// v7.30.3 (mailrs round-26): a SELECT's join/filter
195 /// materialisation would have held more (approximate) heap
196 /// bytes than the configured `max_query_bytes` cap. The row
197 /// cap above counts rows; this counts bytes, because one row
198 /// can be a multi-MB mail body — 1000 fat rows pressure the
199 /// host long before any row ceiling trips. Carries the cap.
200 QueryBytesExceeded(usize),
201 /// v4.5: cooperative cancellation — the host (server's
202 /// per-query watchdog) set the cancel flag while a long-running
203 /// SELECT / UPDATE / DELETE was scanning rows. The partial work
204 /// is discarded; the caller should surface this as a timeout
205 /// to the client.
206 Cancelled,
207}
208
209impl fmt::Display for EngineError {
210 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
211 match self {
212 Self::Parse(e) => write!(f, "parse: {e}"),
213 Self::Storage(e) => write!(f, "storage: {e}"),
214 Self::Eval(e) => write!(f, "eval: {e}"),
215 Self::Unsupported(s) => write!(f, "unsupported: {s}"),
216 Self::TransactionAlreadyOpen => f.write_str("a transaction is already open"),
217 Self::NoActiveTransaction => f.write_str("no active transaction"),
218 Self::WriteRequired => {
219 f.write_str("statement requires a write lock (use execute, not execute_readonly)")
220 }
221 Self::RowLimitExceeded(n) => {
222 write!(f, "query exceeded max_query_rows={n}")
223 }
224 Self::QueryBytesExceeded(n) => {
225 write!(
226 f,
227 "query materialisation exceeded max_query_bytes={n} (set SPG_MAX_QUERY_BYTES to raise, 0 to disable)"
228 )
229 }
230 Self::Cancelled => f.write_str("query cancelled (timeout or client request)"),
231 }
232 }
233}
234
235impl From<ParseError> for EngineError {
236 fn from(e: ParseError) -> Self {
237 Self::Parse(e)
238 }
239}
240impl From<StorageError> for EngineError {
241 fn from(e: StorageError) -> Self {
242 Self::Storage(e)
243 }
244}
245impl From<EvalError> for EngineError {
246 fn from(e: EvalError) -> Self {
247 Self::Eval(e)
248 }
249}
250
251/// The execution engine. Holds the catalog and (later) other server-scope
252/// state. `Engine::new()` is intentionally cheap so callers can construct one
253/// per database, per test.
254/// Function pointer that returns "now" as microseconds since Unix
255/// epoch. The engine is `no_std`, so it can't reach for `std::time`
256/// itself — callers (`spg-server`, the sqllogictest runner) inject a
257/// concrete implementation. `None` means `NOW()` / `CURRENT_*` raise
258/// `Unsupported`.
259pub type ClockFn = fn() -> i64;
260
261/// Function pointer that produces 16 cryptographically random bytes.
262/// Like `ClockFn`, the engine is `no_std` and can't reach for /dev/urandom
263/// itself — host (`spg-server`) injects an OS-backed source. `None`
264/// means SQL-driven `CREATE USER` falls back to a deterministic salt
265/// derived from the username (acceptable in tests; the server always
266/// installs a real RNG so production paths never see this).
267pub type SaltFn = fn() -> [u8; 16];
268
269/// v4.5 cooperative cancellation token. A long-running SELECT /
270/// UPDATE / DELETE checks `is_cancelled` at row-loop checkpoints
271/// and bails with `EngineError::Cancelled`. The host
272/// (`spg-server`) creates an `AtomicBool` per query, spawns a
273/// watchdog thread that sets it after `SPG_QUERY_TIMEOUT_MS`,
274/// and passes it via `execute_with_cancel` / `execute_readonly_with_cancel`.
275///
276/// `CancelToken::none()` is a no-op — used by the legacy `execute`
277/// and `execute_readonly` entry points so existing callers don't
278/// change.
279/// v4.41.1 opaque transaction handle. Returned by `Engine::alloc_tx_id`,
280/// threaded through `Engine::execute_in` so dispatch can identify which
281/// in-flight TX a statement belongs to. `IMPLICIT_TX` is the reserved
282/// slot every legacy caller — engine self-tests, spg-cli, spg-embedded,
283/// startup replay — implicitly uses through the unchanged
284/// `Engine::execute(sql)` API. v4.41.1 keeps at most one active slot at
285/// runtime (dispatch holds `engine.write()` across the wrap, same as
286/// v4.34); the map shape is here to let v4.42 turn on N in-flight
287/// implicit TXs without reshuffling the engine internals.
288#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
289pub struct TxId(pub u64);
290
291/// Reserved slot used by `Engine::execute(sql)` — the legacy single-
292/// global-shadow path. New `alloc_tx_id` handles start at 1.
293pub const IMPLICIT_TX: TxId = TxId(0);
294
295/// v6.7.3 — default segment-size threshold used by `COMPACT COLD
296/// SEGMENTS` when no explicit target is supplied. Segments whose
297/// `OwnedSegment::bytes().len()` is **strictly** less than this
298/// value are eligible to merge. spg-server reads
299/// `SPG_COMPACTION_TARGET_SEGMENT_BYTES` to override.
300pub const COMPACTION_TARGET_DEFAULT_BYTES: u64 = 4 * 1024 * 1024;
301
302/// Per-slot transaction state. Held inside `tx_catalogs[tx_id]` for the
303/// lifetime of a BEGIN..COMMIT (or BEGIN..ROLLBACK) window. Drops when
304/// the TX commits (its `catalog` is moved over `Engine.catalog`) or
305/// rolls back (slot removed, catalog discarded).
306#[derive(Debug, Default, Clone)]
307struct TxState {
308 /// The TX's shadow copy of the catalog. Started as a clone of
309 /// `Engine.catalog` at BEGIN time; writes flow into it; COMMIT
310 /// installs it over `Engine.catalog`. `Catalog::clone()` is O(1)
311 /// since v4.40 (`PersistentVec` rows + `PersistentBTreeMap` indices).
312 catalog: Catalog,
313 /// Per-TX savepoint stack. Each entry pairs the savepoint name with
314 /// a clone of `catalog` at the moment `SAVEPOINT <name>` fired.
315 /// `ROLLBACK TO <name>` restores from the entry and pops everything
316 /// after it; `RELEASE <name>` discards the entry and everything
317 /// after; COMMIT/ROLLBACK clears the whole stack.
318 savepoints: Vec<(String, Catalog)>,
319}
320
321/// v7.11.0 — frozen read-only view of the engine's committed state.
322/// Constructed via [`Engine::clone_snapshot`]. Holds clones of the
323/// catalog, statistics, clock function, and row-cap config — the
324/// four fields the `execute_readonly` path actually reads. Cheap to
325/// `Clone` (each clone shares the underlying `PersistentVec` row
326/// storage; only the trie root pointers copy). Send + Sync so a
327/// snapshot can be moved across `tokio::task::spawn_blocking`
328/// boundaries without coordination.
329///
330/// The contract: a snapshot reflects the engine's state at the
331/// moment `clone_snapshot()` returned. Subsequent writes to the
332/// engine are NOT visible. Callers who need fresher data take a
333/// new snapshot.
334#[derive(Debug, Clone)]
335pub struct CatalogSnapshot {
336 catalog: Catalog,
337 statistics: statistics::Statistics,
338 clock: Option<ClockFn>,
339 max_query_rows: Option<usize>,
340}
341
342/// CoW-1 (v7.34) — frozen view of the *persisted* committed engine
343/// state. Carries every field the `snapshot()` envelope serializes;
344/// `Clone` is O(1) on the catalog (Arc bump) and cheap typed-clones
345/// on the trailers. Decouples "capture state" from "serialize bytes"
346/// so the background-checkpoint worker can hold the snapshot and
347/// produce bytes off the engine write lock.
348#[derive(Debug, Clone)]
349pub struct EngineSnapshot {
350 catalog: Catalog,
351 users: UserStore,
352 publications: publications::Publications,
353 subscriptions: subscriptions::Subscriptions,
354 statistics: statistics::Statistics,
355}
356
357impl EngineSnapshot {
358 /// Same envelope rules as `Engine::snapshot()`: bare catalog when
359 /// every trailer is empty, full envelope otherwise.
360 pub fn serialize(&self) -> Vec<u8> {
361 if self.users.is_empty()
362 && self.publications.is_empty()
363 && self.subscriptions.is_empty()
364 && self.statistics.is_empty()
365 {
366 self.catalog.serialize()
367 } else {
368 build_envelope(
369 &self.catalog.serialize(),
370 &users::serialize_users(&self.users),
371 &self.publications.serialize(),
372 &self.subscriptions.serialize(),
373 &self.statistics.serialize(),
374 )
375 }
376 }
377}
378
379// The engine carries several independent session/capture flags (dialect,
380// FK-checks, meta-view materialisation, redo capture); they're orthogonal
381// switches, not a state enum begging to be modelled.
382#[allow(clippy::struct_excessive_bools)]
383#[derive(Debug, Default)]
384pub struct Engine {
385 /// Committed catalog — what survives `Engine::snapshot()` and what
386 /// outside-TX `SELECT`s read.
387 catalog: Catalog,
388 /// Active TX slots, keyed by `TxId`. Empty when no TX is in flight.
389 /// v4.41.1 runtime invariant: at most one entry (single-writer
390 /// model unchanged). v4.42 will let dispatch hold multiple entries
391 /// concurrently for group commit + engine MVCC.
392 tx_catalogs: BTreeMap<TxId, TxState>,
393 /// Which slot the next exec_* call should mutate. Set by
394 /// `execute_in(sql, tx_id)` at the entry point; legacy `execute(sql)`
395 /// sets it to `IMPLICIT_TX`. None when no TX is in flight (read /
396 /// write goes straight against `catalog`).
397 current_tx: Option<TxId>,
398 /// Monotonic counter for `alloc_tx_id`. Starts at 1 — slot 0 is
399 /// reserved for `IMPLICIT_TX`.
400 next_tx_id: u64,
401 /// v7.22 (round-13 T3) — session string-literal dialect. `false`
402 /// (default) = PG semantics (backslash literal, `''` escape);
403 /// `true` = MySQL semantics (`\'` etc.). Flipped by the
404 /// deterministic session signals each dump emits: `SET sql_mode`
405 /// (only MySQL clients/dumps send it) turns it on,
406 /// `SET standard_conforming_strings = on` (every pg_dump
407 /// preamble) turns it off. The plan cache is cleared on every
408 /// flip — the same SQL text lexes differently per dialect.
409 backslash_escapes: bool,
410 /// Optional wall clock used to satisfy `NOW()` / `CURRENT_TIMESTAMP`
411 /// / `CURRENT_DATE`. Set by the host environment.
412 clock: Option<ClockFn>,
413 /// v4.1 cryptographic RNG for per-user password salt. Set by the
414 /// host. `None` means SQL-driven `CREATE USER` uses a
415 /// deterministic fallback — see `SaltFn`.
416 salt_fn: Option<SaltFn>,
417 /// v4.2 per-query row cap. `None` = unlimited. When set, a
418 /// SELECT that materialises more than `n` rows returns
419 /// `EngineError::RowLimitExceeded`. Enforced before the result
420 /// is shaped into wire frames so a runaway scan can't blow the
421 /// server's heap.
422 max_query_rows: Option<usize>,
423 /// v7.30.3 (mailrs round-26) per-query byte cap on join/filter
424 /// materialisation. `None` = unlimited. Approximate net
425 /// accounting (Value heap payloads + per-cell enum overhead)
426 /// charged at every point the join pipeline clones rows;
427 /// crossing the cap raises `EngineError::QueryBytesExceeded`
428 /// instead of pressuring the host into reclaim livelock. The
429 /// host wires this to `SPG_MAX_QUERY_BYTES` (embed defaults it
430 /// ON; the server keeps its allocator-precise budget as the
431 /// outer layer).
432 pub(crate) max_query_bytes: Option<usize>,
433 /// v4.1 RBAC user table. Empty means "no RBAC configured yet" —
434 /// the server decides what that means at the auth boundary
435 /// (open mode vs legacy single-password mode). User CRUD goes
436 /// through `create_user`/`drop_user`/`verify_user`; persistence
437 /// rides the snapshot envelope alongside the catalog.
438 pub(crate) users: UserStore,
439 /// v6.1.2 logical-replication publication catalog. Empty until
440 /// `CREATE PUBLICATION` runs. Persistence rides the v3 envelope
441 /// trailer (see `build_envelope`).
442 publications: publications::Publications,
443 /// v6.1.4 logical-replication subscription catalog. Empty until
444 /// `CREATE SUBSCRIPTION` runs. Persistence rides the v4 envelope
445 /// trailer.
446 subscriptions: subscriptions::Subscriptions,
447 /// v6.2.0 — per-column statistics for the cost-based optimizer.
448 /// Populated by `ANALYZE`; queried via `spg_statistic` virtual
449 /// table. Persistence rides the v5 envelope trailer.
450 statistics: statistics::Statistics,
451 /// v6.3.0 — engine-level plan cache. Caches the post-`prepare()`
452 /// `Statement` keyed on SQL text. In-memory only — does NOT ride
453 /// the snapshot envelope (rebuilt on demand after restart).
454 plan_cache: plan_cache::PlanCache,
455 /// v6.5.1 — per-distinct-SQL execution stats. In-memory only,
456 /// surfaced via `spg_stat_query` virtual table. Updated by the
457 /// `execute_*` paths after a successful execute.
458 query_stats: query_stats::QueryStats,
459 /// v6.5.2 — connection-state provider callback. spg-server
460 /// registers a function at startup that snapshots its
461 /// per-pgwire-connection registry into `ActivityRow`s; engine
462 /// reads through it on every `SELECT * FROM spg_stat_activity`.
463 /// `None` ⇒ no-data (returns empty rows; matches the no_std
464 /// embedded callers that don't run pgwire).
465 activity_provider: Option<ActivityProvider>,
466 /// v6.5.3 — audit-chain provider + verifier. Same pattern as
467 /// activity_provider: spg-server registers both at startup;
468 /// engine reads through on `SELECT * FROM spg_audit_chain` and
469 /// `SELECT * FROM spg_audit_verify`. `None` ⇒ no-data.
470 audit_chain_provider: Option<AuditChainProvider>,
471 audit_verifier: Option<AuditVerifier>,
472 /// v6.5.6 — slow-query log threshold in microseconds. When set,
473 /// every successful execute whose elapsed exceeds the threshold
474 /// gets fed to the registered slow-query log callback (so
475 /// spg-server can emit a structured log line). Default `None`
476 /// = no slow-query logging.
477 slow_query_threshold_us: Option<u64>,
478 slow_query_logger: Option<SlowQueryLogger>,
479 /// v7.12.1 — session parameters set via `SET <name> = <value>`.
480 /// Only `default_text_search_config` is consumed by the engine
481 /// today (the FTS function dispatcher reads it when
482 /// `to_tsvector(text)` is called without an explicit config).
483 /// All other names are accepted + recorded so PG-dump output
484 /// loads, but have no behavioural effect.
485 pub(crate) session_params: BTreeMap<String, String>,
486 /// v7.12.7 — depth counter for trigger-emitted embedded SQL.
487 /// Each time the engine executes a `DeferredEmbeddedStmt` it
488 /// increments this; the recursive `execute_stmt_with_cancel`
489 /// inside that path checks against [`MAX_TRIGGER_RECURSION`]
490 /// to bound runaway cascades (trigger A's UPDATE on table B
491 /// fires trigger B which UPDATEs table A which fires trigger
492 /// A again…). Reset to 0 once the original DML returns.
493 trigger_recursion_depth: u32,
494 /// v7.14.0 — when `SET FOREIGN_KEY_CHECKS=0` is in effect
495 /// (mysqldump preamble), the FK existence + arity check at
496 /// CREATE TABLE time is deferred. FKs referencing a
497 /// not-yet-existing parent land in `pending_foreign_keys`
498 /// keyed by child table; `SET FOREIGN_KEY_CHECKS=1` drains
499 /// the queue and resolves each FK against the now-complete
500 /// catalog. Empty by default; the queue is drained on every
501 /// `RESET ALL` too.
502 foreign_key_checks: bool,
503 /// v7.16.2 — true on the temp Engine an outer
504 /// `exec_select_with_meta_views` builds, telling that
505 /// temp engine "stop short-circuiting into the meta-view
506 /// path — your catalog already has the materialised
507 /// tables; just run the regular SELECT." Without this we'd
508 /// infinite-loop since the meta-view name (e.g.
509 /// `__spg_info_columns`) still triggers
510 /// `select_references_meta_view`.
511 meta_views_materialised: bool,
512 pending_foreign_keys: Vec<(alloc::string::String, spg_sql::ast::ForeignKeyConstraint)>,
513 /// v7.38 元机制 D — frozen snapshot of `SPG_TEST_*` env vars. Read
514 /// once at construction (`with_env_cfg`) and queried on hot paths
515 /// via `engine.env_cfg().<field>`. Production builds keep this at
516 /// `EnvConfig::default()`, so the optimiser can const-fold every
517 /// `if env_cfg.<field>` gate. See `testkit::env_config` + the
518 /// `xtests/sigil/test-mode-gucs.md` index.
519 env_cfg: testkit::EnvConfig,
520 /// v7.38 P0 元机制 A — per-engine `injection_points` attach
521 /// table. Only exists when the crate is built with the
522 /// `injection-points` feature; release builds carry no field.
523 /// Pushed onto the thread-local stack by
524 /// `enter_injection_scope()` so the `injection_point!()` macro
525 /// can find it from anywhere in the executor without rewiring
526 /// every signature. See
527 /// `crates/spg-engine/src/testkit/injection.rs`.
528 #[cfg(feature = "injection-points")]
529 injection_store: alloc::sync::Arc<crate::testkit::injection::InjectionStore>,
530 /// v7.34 (crash-recovery P0 #2) — row-level redo capture. When the
531 /// embedding layer turns this on (persistence enabled), each mutating
532 /// `execute` records the physical [`RowChange`]s it applied; the
533 /// engine drains them into `last_redo` on success, and the embedded
534 /// layer reads them via [`Engine::take_redo`] to write the WAL in
535 /// place of the SQL text. Off (default) = zero capture overhead.
536 redo_capture: bool,
537 /// Redo captured by the most recent successful mutating `execute`,
538 /// awaiting drain by the embedding layer. Cleared on each capture.
539 last_redo: Vec<RowChange>,
540 /// v7.38 轴 4 — currently-selected SQL isolation level. Set by
541 /// `SET TRANSACTION ISOLATION LEVEL …`; read by
542 /// `SHOW transaction_isolation`. v7.37.8 implements the
543 /// SQL surface; actual semantic differentiation (REPEATABLE READ
544 /// snapshot / SERIALIZABLE SSI) lands in a separate train.
545 pub(crate) current_isolation_level: spg_sql::ast::IsolationLevel,
546}
547
548/// v7.12.7 — hard cap on nested trigger-emitted embedded SQL
549/// fires. 16 deep is well past anything a normal trigger graph
550/// uses while still preventing infinite-loop wedging.
551const MAX_TRIGGER_RECURSION: u32 = 16;
552
553/// v6.5.6 — callback signature for slow-query log emission. Called
554/// with `(sql, elapsed_us)` once per successful execute that crosses
555/// the threshold.
556pub type SlowQueryLogger = fn(&str, u64);
557
558/// v6.5.2 — one row of `spg_stat_activity`. Engine-public so
559/// spg-server can construct rows without re-exporting internal
560/// dispatch types.
561#[derive(Debug, Clone)]
562pub struct ActivityRow {
563 pub pid: u32,
564 pub user: String,
565 pub started_at_us: i64,
566 pub current_sql: String,
567 pub wait_event: String,
568 pub elapsed_us: i64,
569 pub in_transaction: bool,
570 /// v7.17 Phase 2.4 — startup-param `application_name` (or the
571 /// last value the client sent via `SET application_name = '...'`).
572 /// Empty when the client never declared one.
573 pub application_name: String,
574}
575
576/// v6.5.2 — provider callback type. Fresh snapshot returned each
577/// call; engine doesn't cache the slice.
578pub type ActivityProvider = fn() -> Vec<ActivityRow>;
579
580/// v6.5.3 — one row of `spg_audit_chain`. Engine-public so
581/// spg-server can construct rows directly from `AuditEntry`.
582#[derive(Debug, Clone)]
583pub struct AuditRow {
584 pub seq: i64,
585 pub ts_ms: i64,
586 pub prev_hash_hex: String,
587 pub entry_hash_hex: String,
588 pub sql: String,
589}
590
591/// v6.5.3 — chain-table provider + verifier. spg-server registers
592/// fn pointers that snapshot / verify the audit log. `verify`
593/// returns `(verified_count, broken_at_seq)` — `broken_at_seq` is
594/// `-1` on a clean chain.
595pub type AuditChainProvider = fn() -> Vec<AuditRow>;
596pub type AuditVerifier = fn() -> (i64, i64);
597
598impl Engine {
599 pub fn new() -> Self {
600 Self {
601 catalog: Catalog::new(),
602 tx_catalogs: BTreeMap::new(),
603 current_tx: None,
604 backslash_escapes: false,
605 next_tx_id: 1,
606 clock: None,
607 salt_fn: None,
608 max_query_rows: None,
609 max_query_bytes: None,
610 users: UserStore::new(),
611 publications: publications::Publications::new(),
612 subscriptions: subscriptions::Subscriptions::new(),
613 statistics: statistics::Statistics::new(),
614 plan_cache: plan_cache::PlanCache::new(),
615 query_stats: query_stats::QueryStats::new(),
616 activity_provider: None,
617 audit_chain_provider: None,
618 audit_verifier: None,
619 slow_query_threshold_us: None,
620 slow_query_logger: None,
621 session_params: BTreeMap::new(),
622 trigger_recursion_depth: 0,
623 foreign_key_checks: true,
624 meta_views_materialised: false,
625 pending_foreign_keys: Vec::new(),
626 env_cfg: testkit::EnvConfig::default(),
627 #[cfg(feature = "injection-points")]
628 injection_store: alloc::sync::Arc::new(
629 crate::testkit::injection::InjectionStore::default(),
630 ),
631 redo_capture: false,
632 current_isolation_level: spg_sql::ast::IsolationLevel::ReadCommitted,
633 last_redo: Vec::new(),
634 }
635 }
636
637 /// v7.11.0 — clone the engine's committed catalog + read-time
638 /// state into a frozen `CatalogSnapshot`. Cheap (`Catalog` is
639 /// backed by `PersistentVec`; cloning is O(log n) per table).
640 /// Subsequent writes to this engine are invisible to the
641 /// snapshot; the snapshot is self-contained and can be moved
642 /// to another thread for concurrent `execute_readonly_on_snapshot`
643 /// calls. The basis for [`AsyncReadHandle`] in spg-embedded-tokio
644 /// and any other read-fanout pattern.
645 #[must_use]
646 pub fn clone_snapshot(&self) -> CatalogSnapshot {
647 CatalogSnapshot {
648 catalog: self.active_catalog().clone(),
649 statistics: self.statistics.clone(),
650 clock: self.clock,
651 max_query_rows: self.max_query_rows,
652 }
653 }
654
655 /// Construct an engine restored from a previously-snapshotted catalog
656 /// (see `snapshot()`).
657 pub fn restore(catalog: Catalog) -> Self {
658 Self {
659 catalog,
660 tx_catalogs: BTreeMap::new(),
661 current_tx: None,
662 backslash_escapes: false,
663 next_tx_id: 1,
664 clock: None,
665 salt_fn: None,
666 max_query_rows: None,
667 max_query_bytes: None,
668 users: UserStore::new(),
669 publications: publications::Publications::new(),
670 subscriptions: subscriptions::Subscriptions::new(),
671 statistics: statistics::Statistics::new(),
672 plan_cache: plan_cache::PlanCache::new(),
673 query_stats: query_stats::QueryStats::new(),
674 activity_provider: None,
675 audit_chain_provider: None,
676 audit_verifier: None,
677 slow_query_threshold_us: None,
678 slow_query_logger: None,
679 session_params: BTreeMap::new(),
680 trigger_recursion_depth: 0,
681 foreign_key_checks: true,
682 meta_views_materialised: false,
683 pending_foreign_keys: Vec::new(),
684 env_cfg: testkit::EnvConfig::default(),
685 #[cfg(feature = "injection-points")]
686 injection_store: alloc::sync::Arc::new(
687 crate::testkit::injection::InjectionStore::default(),
688 ),
689 redo_capture: false,
690 current_isolation_level: spg_sql::ast::IsolationLevel::ReadCommitted,
691 last_redo: Vec::new(),
692 }
693 }
694
695 /// Restore an engine + user table from a v4.1 envelope produced
696 /// by `snapshot_with_users()`. Falls back to plain catalog-only
697 /// restore if the envelope magic isn't present (so v3.x snapshot
698 /// files still load). v6.1.2 adds the optional publications
699 /// trailer (envelope v3); a v1/v2 envelope deserialises to an
700 /// empty publication table.
701 pub fn restore_envelope(buf: &[u8]) -> Result<Self, EngineError> {
702 match split_envelope(buf) {
703 EnvelopeParse::Pair {
704 catalog: catalog_bytes,
705 users: user_bytes,
706 publications: pub_bytes,
707 subscriptions: sub_bytes,
708 statistics: stats_bytes,
709 } => {
710 let catalog = Catalog::deserialize(catalog_bytes).map_err(EngineError::Storage)?;
711 let users = users::deserialize_users(user_bytes)
712 .map_err(|e| EngineError::Unsupported(alloc::format!("users restore: {e}")))?;
713 let publications = match pub_bytes {
714 Some(b) => publications::Publications::deserialize(b).map_err(|e| {
715 EngineError::Unsupported(alloc::format!("publications restore: {e:?}"))
716 })?,
717 None => publications::Publications::new(),
718 };
719 let subscriptions = match sub_bytes {
720 Some(b) => subscriptions::Subscriptions::deserialize(b).map_err(|e| {
721 EngineError::Unsupported(alloc::format!("subscriptions restore: {e:?}"))
722 })?,
723 None => subscriptions::Subscriptions::new(),
724 };
725 let statistics = match stats_bytes {
726 Some(b) => statistics::Statistics::deserialize(b).map_err(|e| {
727 EngineError::Unsupported(alloc::format!("statistics restore: {e:?}"))
728 })?,
729 None => statistics::Statistics::new(),
730 };
731 Ok(Self {
732 catalog,
733 tx_catalogs: BTreeMap::new(),
734 current_tx: None,
735 backslash_escapes: false,
736 next_tx_id: 1,
737 clock: None,
738 salt_fn: None,
739 max_query_rows: None,
740 max_query_bytes: None,
741 users,
742 publications,
743 subscriptions,
744 statistics,
745 plan_cache: plan_cache::PlanCache::new(),
746 query_stats: query_stats::QueryStats::new(),
747 activity_provider: None,
748 audit_chain_provider: None,
749 audit_verifier: None,
750 slow_query_threshold_us: None,
751 slow_query_logger: None,
752 session_params: BTreeMap::new(),
753 trigger_recursion_depth: 0,
754 foreign_key_checks: true,
755 meta_views_materialised: false,
756 pending_foreign_keys: Vec::new(),
757 env_cfg: testkit::EnvConfig::default(),
758 #[cfg(feature = "injection-points")]
759 injection_store: alloc::sync::Arc::new(
760 crate::testkit::injection::InjectionStore::default(),
761 ),
762 redo_capture: false,
763 current_isolation_level: spg_sql::ast::IsolationLevel::ReadCommitted,
764 last_redo: Vec::new(),
765 })
766 }
767 EnvelopeParse::CrcMismatch { expected, computed } => {
768 Err(EngineError::Storage(StorageError::Corrupt(alloc::format!(
769 "snapshot envelope CRC32 mismatch (expected={expected:#010x}, computed={computed:#010x})"
770 ))))
771 }
772 EnvelopeParse::Bare => {
773 let catalog = Catalog::deserialize(buf).map_err(EngineError::Storage)?;
774 Ok(Self::restore(catalog))
775 }
776 }
777 }
778
779 pub const fn users(&self) -> &UserStore {
780 &self.users
781 }
782
783 /// Builder: attach a wall clock so `NOW()` / `CURRENT_TIMESTAMP` /
784 /// `CURRENT_DATE` evaluate to a real value instead of erroring out.
785 #[must_use]
786 pub const fn with_clock(mut self, clock: ClockFn) -> Self {
787 self.clock = Some(clock);
788 self
789 }
790
791 /// Builder: attach an OS-backed RNG for per-user password salts.
792 /// The host (`spg-server`) typically wires this to `/dev/urandom`.
793 #[must_use]
794 pub const fn with_salt_fn(mut self, f: SaltFn) -> Self {
795 self.salt_fn = Some(f);
796 self
797 }
798
799 /// v7.38 元机制 D — install a frozen [`testkit::EnvConfig`] snapshot.
800 ///
801 /// Hosts (spg-server, spg-embedded, tests) call this once at engine
802 /// init with either `EnvConfig::from_env()` (production-with-test-vars)
803 /// or `EnvConfig::builder()....build()` (programmatic). After
804 /// construction the engine never reads env vars; all test-mode
805 /// behaviour flows through `self.env_cfg()`.
806 #[must_use]
807 pub fn with_env_cfg(mut self, env_cfg: testkit::EnvConfig) -> Self {
808 self.env_cfg = env_cfg;
809 self
810 }
811
812 /// v7.38 元机制 D — frozen test-mode GUC snapshot. Hot paths gate
813 /// nondeterministic surfaces on fields of this struct; production
814 /// default keeps every field at `false / None / Auto` so the
815 /// optimiser can const-fold the gate.
816 pub fn env_cfg(&self) -> &testkit::EnvConfig {
817 &self.env_cfg
818 }
819
820 /// v7.38 元机制 D acceptor — single seed source for every
821 /// nondeterministic engine subsystem (hash builders, randomised
822 /// tie-breakers, …). Honour `SPG_TEST_RANDOM_SEED=N` when set;
823 /// otherwise derive from the engine's wall clock (production) or
824 /// fall back to a fixed sentinel when the host hasn't installed
825 /// a clock. Two engines built with the same builder seed return
826 /// byte-equal output for the same query.
827 /// See `xtests/sigil/test-mode-gucs.md`.
828 pub fn rng_seed(&self) -> u64 {
829 if let Some(seed) = self.env_cfg.random_seed {
830 return seed;
831 }
832 match self.clock {
833 Some(f) => f() as u64,
834 // Production engines without a clock installed get a fixed
835 // non-zero sentinel; same shape as PG's `random()` start
836 // state under a `setseed(0)`.
837 None => 0xBAD_5EED_DEAD_BEEF,
838 }
839 }
840
841 /// v7.38 P0 元机制 A — push this engine's `InjectionStore` onto
842 /// the thread-local stack so any `injection_point!()` reached
843 /// during the returned guard's lifetime resolves against this
844 /// engine. Mirrors PG's per-backend injection table.
845 ///
846 /// Returns a no-op guard when the `injection-points` feature is
847 /// off so call sites don't need `#[cfg]`.
848 #[must_use]
849 pub fn enter_injection_scope(&self) -> crate::testkit::injection::InjectionGuard {
850 #[cfg(feature = "injection-points")]
851 {
852 crate::testkit::injection::enter_scope(&self.injection_store)
853 }
854 #[cfg(not(feature = "injection-points"))]
855 {
856 crate::testkit::injection::new_guard()
857 }
858 }
859
860 /// v7.38 P0 元机制 A — expose the per-engine store so tests can
861 /// query notice counts / detach actions without parsing SQL
862 /// output. Only present when the feature is on.
863 #[cfg(feature = "injection-points")]
864 pub fn injection_store(
865 &self,
866 ) -> alloc::sync::Arc<crate::testkit::injection::InjectionStore> {
867 self.injection_store.clone()
868 }
869
870 /// Builder: cap the number of rows a single SELECT may return.
871 /// Exceeding the cap raises `EngineError::RowLimitExceeded` —
872 /// the bound is checked inside the executor so a runaway
873 /// catalog scan can't allocate millions of rows before the
874 /// server gets a chance to reject the result.
875 #[must_use]
876 pub const fn with_max_query_rows(mut self, n: usize) -> Self {
877 self.max_query_rows = Some(n);
878 self
879 }
880
881 /// Builder: cap the approximate heap bytes a single SELECT's
882 /// join/filter materialisation may hold. Exceeding the cap
883 /// raises `EngineError::QueryBytesExceeded`. Rows are the wrong
884 /// unit when one row carries a multi-MB body (mailrs round-26:
885 /// 1000-row batches of full mail text walked a 15 GiB host into
886 /// reclaim livelock without ever tripping a row ceiling).
887 #[must_use]
888 pub const fn with_max_query_bytes(mut self, n: usize) -> Self {
889 self.max_query_bytes = Some(n);
890 self
891 }
892
893 /// The *committed* catalog. Note: during a transaction this returns the
894 /// pre-TX state — `SELECT` inside a TX goes through `execute()` and reads
895 /// the shadow. Tests that inspect outside-TX state should use this.
896 pub const fn catalog(&self) -> &Catalog {
897 &self.catalog
898 }
899
900 /// Capture a frozen view of the committed engine state. Catalog
901 /// is O(1) Arc bump; trailers are cheap clones. Decouples "capture"
902 /// (needs &Engine) from "serialize" (CPU, no engine access) — the
903 /// seam the background-checkpoint worker rides in CoW-2.
904 pub fn snapshot_data(&self) -> EngineSnapshot {
905 EngineSnapshot {
906 catalog: self.catalog.clone(),
907 users: self.users.clone(),
908 publications: self.publications.clone(),
909 subscriptions: self.subscriptions.clone(),
910 statistics: self.statistics.clone(),
911 }
912 }
913
914 /// Serialize the *committed* catalog to bytes. v0.6 was full-snapshot; v0.9
915 /// adds the rule that an open TX's shadow is never snapshotted — only the
916 /// post-COMMIT state is persisted. v4.1 wraps the catalog in an envelope
917 /// when there are users to persist; an empty user table snapshots as the
918 /// bare catalog format (backwards-compat with v3.x readers). v6.1.2
919 /// adds publications to the envelope condition: either non-empty
920 /// users OR non-empty publications now triggers the envelope path.
921 pub fn snapshot(&self) -> Vec<u8> {
922 self.snapshot_data().serialize()
923 }
924
925 /// True when at least one TX slot is in flight. v4.41.1 runtime
926 /// invariant: at most one slot active at a time (dispatch holds
927 /// `engine.write()` across the entire wrap). v4.42 will let this
928 /// return true with multiple slots concurrently.
929 pub fn in_transaction(&self) -> bool {
930 !self.tx_catalogs.is_empty()
931 }
932
933 /// v4.41.1 allocate a fresh TX handle. Used by spg-server dispatch
934 /// to scope each implicit-wrap BEGIN..stmt..COMMIT to its own slot
935 /// in `tx_catalogs`. v4.42 — the commit-barrier leader allocates
936 /// one of these per task in its group, runs `BEGIN`+sql+`COMMIT`
937 /// sequentially under a single `engine.write()` so each task's
938 /// mutations accumulate into shared state, then either keeps the
939 /// accumulated state (fsync OK) or restores the pre-image via
940 /// `replace_catalog` (fsync err).
941 pub fn alloc_tx_id(&mut self) -> TxId {
942 let id = TxId(self.next_tx_id);
943 self.next_tx_id = self.next_tx_id.saturating_add(1);
944 id
945 }
946
947 /// v4.42 — atomically replace the live catalog. Used by the
948 /// commit-barrier leader to roll back a group whose batched
949 /// fsync failed: the leader snapshots `engine.catalog().clone()`
950 /// (O(1) Arc bump after the v4.39/v4.40 persistent migration)
951 /// at group start, sequentially applies each task's BEGIN+sql+
952 /// COMMIT under the same write lock to accumulate mutations
953 /// into shared state, batches the WAL bytes, fsyncs once, and
954 /// on failure calls this with the pre-image to undo every
955 /// task in the group at once.
956 ///
957 /// **Does NOT touch `tx_catalogs` / `current_tx`.** Any
958 /// explicit-TX slot from a concurrent client (created via the
959 /// legacy `IMPLICIT_TX`-less dispatch path or via the future
960 /// MVCC-readers v5+ work) has its own snapshot baked into the
961 /// slot — restoring `self.catalog` to the pre-image leaves
962 /// those slots untouched, exactly as they were when the leader
963 /// took the lock. The leader's own implicit-TX slots are all
964 /// already discarded (`exec_commit` removed them as each
965 /// task's COMMIT ran) by the time this is reached.
966 pub fn replace_catalog(&mut self, catalog: Catalog) {
967 self.catalog = catalog;
968 }
969
970 /// v6.7.0 — public shim around `Catalog::freeze_oldest_to_cold`
971 /// so tests + the spg-server freezer can drive a freeze without
972 /// reaching into the private `active_catalog_mut`. v6.7.4
973 /// parallel freezer will build on this surface.
974 ///
975 /// Marks the table's cached `cold_row_count` stale because the
976 /// freeze added cold locators that ANALYZE hasn't yet refreshed.
977 pub fn freeze_oldest_to_cold(
978 &mut self,
979 table_name: &str,
980 index_name: &str,
981 max_rows: usize,
982 ) -> Result<spg_storage::FreezeReport, EngineError> {
983 let report = self
984 .active_catalog_mut()
985 .freeze_oldest_to_cold(table_name, index_name, max_rows)
986 .map_err(EngineError::Storage)?;
987 if let Some(t) = self.active_catalog_mut().get_mut(table_name) {
988 t.mark_cold_row_count_stale();
989 }
990 Ok(report)
991 }
992
993 /// v6.7.5 — public shim used by the spg-server follower's
994 /// segment-forwarding receiver. Registers a cold-tier segment
995 /// at a specific id (the master's id, as transmitted on the
996 /// wire) so the follower's BTree-Cold locators stay byte-
997 /// identical with the master's. Wraps
998 /// `Catalog::load_segment_bytes_at` under the standard
999 /// clone-mutate-replace pattern.
1000 ///
1001 /// Returns `Ok(())` on success **and** on the "slot already
1002 /// occupied" case — a follower mid-reconnect may receive a
1003 /// segment chunk for a segment_id it already has on disk
1004 /// (forwarded last session); the caller should treat that
1005 /// path as a no-op rather than a fatal error.
1006 pub fn receive_cold_segment(
1007 &mut self,
1008 segment_id: u32,
1009 bytes: Vec<u8>,
1010 ) -> Result<(), EngineError> {
1011 let mut new_cat = self.catalog.clone();
1012 match new_cat.load_segment_bytes_at(segment_id, bytes) {
1013 Ok(()) => {
1014 self.replace_catalog(new_cat);
1015 Ok(())
1016 }
1017 Err(StorageError::Corrupt(msg)) if msg.contains("already occupied") => Ok(()),
1018 Err(e) => Err(EngineError::Storage(e)),
1019 }
1020 }
1021
1022 pub(crate) fn active_catalog(&self) -> &Catalog {
1023 match self.current_tx {
1024 Some(t) => self
1025 .tx_catalogs
1026 .get(&t)
1027 .map_or(&self.catalog, |s| &s.catalog),
1028 None => &self.catalog,
1029 }
1030 }
1031
1032 fn active_catalog_mut(&mut self) -> &mut Catalog {
1033 let tx = self.current_tx;
1034 match tx {
1035 Some(t) => match self.tx_catalogs.get_mut(&t) {
1036 Some(s) => &mut s.catalog,
1037 None => &mut self.catalog,
1038 },
1039 None => &mut self.catalog,
1040 }
1041 }
1042
1043 /// v7.34 (crash-recovery P0 #2) — turn row-level redo capture on/off.
1044 /// The embedding layer enables it when persistence is on so each
1045 /// mutating `execute` records the physical [`RowChange`]s it applied
1046 /// (drained via [`Engine::take_redo`]). Off = zero capture overhead.
1047 pub fn set_redo_capture(&mut self, on: bool) {
1048 self.redo_capture = on;
1049 }
1050
1051 /// v7.37.8 — read accessor for tests / observability. The
1052 /// embedding layer flips this on once per `open_path` (after
1053 /// replay completes) when `SPG_WAL_ROW_REDO` is enabled (now
1054 /// default in v7.37.8). A consumer that wants to verify the
1055 /// post-upgrade contract ("writes go to V5 ROW_REDO by default")
1056 /// reads this through `Database::engine_redo_capture()` instead
1057 /// of inspecting WAL bytes (which the auto-checkpoint truncates
1058 /// on `Drop`).
1059 pub fn redo_capture_enabled(&self) -> bool {
1060 self.redo_capture
1061 }
1062
1063 /// v7.38 轴 4 — currently-selected SQL isolation level. Default
1064 /// `ReadCommitted` after construction; updated by
1065 /// `SET TRANSACTION ISOLATION LEVEL …`. Read by
1066 /// `SHOW transaction_isolation` and any future MVCC/SSI gate.
1067 pub fn current_isolation_level(&self) -> spg_sql::ast::IsolationLevel {
1068 self.current_isolation_level
1069 }
1070
1071 /// v7.34 — take the redo captured by the most recent successful
1072 /// mutating `execute` (empty when capture is off, the statement was a
1073 /// read, or it changed nothing). The embedding layer writes these to
1074 /// the WAL in place of the SQL text.
1075 pub fn take_redo(&mut self) -> Vec<RowChange> {
1076 core::mem::take(&mut self.last_redo)
1077 }
1078
1079 /// v7.34 (crash-recovery P0 #2) — replay a row-level redo log onto the
1080 /// committed catalog (the row-level WAL recovery primitive: apply the
1081 /// captured physical changes from a checkpoint baseline, in place of
1082 /// re-executing the SQL). Trusts the log — no uniqueness/FK/parse.
1083 pub fn apply_redo(&mut self, changes: &[RowChange]) -> Result<(), EngineError> {
1084 self.catalog
1085 .apply_redo(changes)
1086 .map_err(EngineError::Storage)
1087 }
1088
1089 /// Read-only execute path. Succeeds for `SELECT` / `SHOW TABLES`
1090 /// / `SHOW COLUMNS`; returns `EngineError::WriteRequired` for
1091 /// every other statement, so the caller can fall through to the
1092 /// `&mut self` `execute` path under a write lock. Engine state is
1093 /// not mutated even on the success path (`rewrite_clock_calls`
1094 /// and `resolve_order_by_position` both mutate the locally-owned
1095 /// AST, not `self`).
1096 ///
1097 /// v4.2: cap result-set size. Applied after the executor
1098 /// materialises rows but before they leave the engine — wrapping
1099 /// every Rows-returning exec_* function would scatter the check.
1100 ///
1101 /// v7.31 (memory campaign, bucket A) — the same choke point now
1102 /// also enforces the BYTE budget on the final result set, so
1103 /// single-table and aggregate paths (which don't route through
1104 /// the join materialiser's incremental accounting) still cannot
1105 /// hand the host an unbounded result. Intermediate single-table
1106 /// clones are the 7.31.x follow-up (design doc, bucket A).
1107 fn enforce_row_limit(
1108 &self,
1109 result: Result<QueryResult, EngineError>,
1110 ) -> Result<QueryResult, EngineError> {
1111 if let Ok(QueryResult::Rows { rows, .. }) = &result {
1112 if let Some(cap) = self.max_query_rows
1113 && rows.len() > cap
1114 {
1115 return Err(EngineError::RowLimitExceeded(cap));
1116 }
1117 if let Some(byte_cap) = self.max_query_bytes
1118 && approx_rows_bytes(rows) > byte_cap
1119 {
1120 return Err(EngineError::QueryBytesExceeded(byte_cap));
1121 }
1122 }
1123 result
1124 }
1125}
1126
1127/// v7.31 (memory campaign — ceiling-first / never-die, design v1) —
1128/// per-table slice of the engine's resident-memory accounting.
1129/// `hot_encoded_bytes` is the storage layer's maintained meter (what
1130/// the rows encode to); `approx_resident_bytes` is what they COST in
1131/// RAM (per-cell enum slots + heap payloads via `approx_row_bytes`)
1132/// — the gap between the two is the representation multiplier the
1133/// round-26 report measured at ~11× end-to-end.
1134#[derive(Debug, Clone)]
1135pub struct TableMemoryStats {
1136 pub name: String,
1137 pub hot_rows: u64,
1138 /// Cached cold-row count (refreshed by ANALYZE — see
1139 /// `Table::cold_row_count`'s staleness contract).
1140 pub cold_rows: u64,
1141 pub hot_encoded_bytes: u64,
1142 pub approx_resident_bytes: u64,
1143 pub index_count: u64,
1144 /// v7.31 C2 — sum of `IndexKind::approx_resident_bytes()` over the
1145 /// table's indices: every variant (BTree / NSW / BRIN / GIN family)
1146 /// walks its own structure, so the GIN posting lists and NSW layer
1147 /// adjacency that dominate text/vector tables are counted honestly
1148 /// instead of the old flat-token estimate.
1149 pub approx_index_bytes: u64,
1150}
1151
1152/// v7.31 — whole-engine memory snapshot: the polling form of the
1153/// round-26 ask-4 watermark signal. Hosts compare
1154/// `total_approx_resident_bytes` (+ their own WAL/file accounting)
1155/// against their deployment ceiling and shed/shrink before the
1156/// kernel does it for them.
1157#[derive(Debug, Clone)]
1158pub struct MemoryStats {
1159 pub tables: Vec<TableMemoryStats>,
1160 pub total_hot_encoded_bytes: u64,
1161 pub total_approx_resident_bytes: u64,
1162 pub total_approx_index_bytes: u64,
1163 /// The active per-query materialisation budget (bucket A), so a
1164 /// monitoring host sees ceiling and usage through one call.
1165 pub max_query_bytes: Option<usize>,
1166 /// v7.31 C2 — bucket D: live WAL bytes (active chunk + buffered,
1167 /// uncheckpointed). `None` from the engine itself — it has no WAL;
1168 /// the durable hosts (embed `Database`, server) fill it in from
1169 /// their own WAL accounting. `Some(0)` means "host has a WAL and
1170 /// it is empty"; `None` means "no WAL on this path" (in-memory).
1171 pub wal_bytes: Option<u64>,
1172}
1173
1174/// v6.2.0 — true for engine-managed catalog tables that the bare
1175/// `ANALYZE` (no target) should skip. v6.2.0 has no internal
1176/// tables yet (publications / subscriptions / users / statistics
1177/// all live as engine fields, not catalog tables), so this is a
1178/// reserved future-proofing hook — every existing user table is
1179/// analysed.
1180const fn is_internal_table_name(_name: &str) -> bool {
1181 false
1182}
1183
1184#[cfg(test)]
1185mod tests;