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
37mod acl;
38pub mod aggregate;
39pub(crate) mod amcheck;
40mod baregroup;
41pub mod brin;
42mod bytebudget;
43mod cancel;
44mod clock;
45mod collate;
46mod collate_derive;
47mod collation_catalog;
48mod constfold;
49mod constraints;
50mod conversions;
51pub mod copy;
52mod cursor;
53mod ddl;
54pub mod describe;
55mod distinct;
56mod dml;
57mod dump;
58mod envelope;
59pub mod eval;
60mod execute;
61mod explain;
62mod expr_analysis;
63mod expr_index;
64pub(crate) mod extsort;
65pub mod fts;
66mod fts_de;
67mod fts_es;
68mod fts_fr;
69mod fts_stop;
70mod guc_catalog;
71mod immutable_fn;
72mod index_access;
73mod join;
74mod join_using;
75mod joinfold;
76pub mod json;
77pub mod largeobject;
78mod limit_expr;
79pub mod locks;
80mod maintenance;
81pub mod memoize;
82mod notify;
83mod numeric;
84mod opclass;
85mod orderby;
86mod partition;
87pub(crate) mod partition_walks;
88pub mod plan_cache;
89mod plpgsql;
90pub mod publications;
91mod qualorder;
92pub mod query_stats;
93mod readonly;
94pub mod reorder;
95mod rls;
96mod rules;
97pub mod scalarsq_streaming;
98mod select;
99pub mod selectivity;
100mod sequence;
101mod session;
102mod show;
103mod spg_admin;
104pub mod statistics;
105pub mod subquery;
106pub mod subscriptions;
107mod substitute;
108mod system_catalog;
109mod table_access;
110pub mod tempstore;
111pub mod testkit;
112mod transaction;
113pub(crate) use transaction::{TxStmtClass, classify_stmt_for_tx};
114pub mod triggers;
115pub mod users;
116mod window;
117
118pub use crate::users::{Role, ScramSecrets, UserError, UserStore};
119pub use cancel::{CancelToken, MonotonicNowFn};
120pub use execute::{RowCells, StreamItem};
121
122use bytebudget::*;
123pub(crate) use clock::{rewrite_clock_calls, value_to_literal};
124use constraints::*;
125pub use constraints::{UNIQ_FOLD_CHOSEN, UNIQ_PROBE_CALLS, UNIQ_PROBE_LOCATORS};
126use conversions::*;
127pub use conversions::{
128 format_bigint_2d_text_pub, format_bit_string, format_circle, format_hstore_text, format_inet,
129 format_int_2d_text_pub, format_line, format_lseg, format_macaddr, format_macaddr8,
130 format_multirange, format_path, format_pg_box, format_pg_lsn, format_point, format_polygon,
131 format_range_text, format_text_2d_text_pub,
132};
133pub(crate) use ddl::{
134 canonicalize_set_value, enforce_enum_label, eval_runtime_default_free,
135 resolve_column_default_free,
136};
137pub(crate) use envelope::{EnvelopeParse, build_envelope, split_envelope};
138use expr_analysis::*;
139use index_access::*;
140pub use join::{ANTI_JOIN_FAST_PATH_FIRED, ANTI_JOIN_FAST_PATH_TRIED};
141pub(crate) use orderby::{
142 OrderKey, apply_offset_and_limit, apply_offset_and_limit_tagged, build_order_keys,
143 canonical_value_repr, cmp_multi_key, expand_group_by_all, order_by_value_cmp,
144 order_by_value_cmp_in, render_histogram_bounds, resolve_order_by_position, sort_by_keys,
145 sort_values_for_histogram, topk_trim, value_cmp, value_to_f64,
146};
147pub use select::{DISTINCT_DUP_DROPPED, PROJ_DIRECT_FIRE, PROJ_ROW_BUILT, SCAN_PATH_ENTERED};
148pub(crate) use select::{build_projection, infer_column_types, value_to_order_key};
149pub use sequence::MUTATING_CALL_NEEDLES;
150pub(crate) use show::render_create_table;
151pub use subquery::{
152 BATCHED_SCALAR_FALL_THROUGH_COUNT, BATCHED_SCALAR_KEYED_FIRE_COUNT,
153 BATCHED_SCALAR_KEYED_PROBE_COUNT, EXISTS_BATCH_FALL_THROUGH_COUNT, EXISTS_BATCH_FIRE_COUNT,
154 EXISTS_PULLUP_BAIL_INNER_FROM, EXISTS_PULLUP_BAIL_INNER_SHAPE,
155 EXISTS_PULLUP_BAIL_MULTICOL_DISABLED, EXISTS_PULLUP_BAIL_NO_CORR, EXISTS_PULLUP_BAIL_NO_WHERE,
156 EXISTS_PULLUP_BAIL_RESIDUAL_NOT_INNER, EXISTS_PULLUP_BAIL_UNIQUE_KEY_MISSING,
157 EXISTS_PULLUP_CANDIDATE_COUNT, EXISTS_PULLUP_FIRE_COUNT, EXISTS_PULLUP_MULTICOL_DISABLE,
158 PULLUP_LIMIT1_FIRE_COUNT, SCALARSQ_PK_PROBE_FIRED, ScalarPkProbeFastPath,
159 expr_tree_has_subquery,
160};
161pub(crate) use subquery::{build_in_list_set, collect_scalar_subqueries, expr_has_subquery};
162pub use substitute::substitute_placeholders;
163use substitute::*;
164use system_catalog::*;
165use window::*;
166
167use alloc::collections::{BTreeMap, BTreeSet};
168use alloc::string::String;
169use alloc::vec::Vec;
170use core::fmt;
171
172// v7.16.0 — re-export the parsed-statement AST so downstream
173// crates (spg-embedded → spg-sqlx) don't need a direct dep on
174// spg-sql for the prepare/bind handle.
175pub use spg_sql::ast::{SelectStatement, Statement as ParsedStatement};
176// v7.37.15 Phase B — re-export the visibility primitives for engine
177// callers so the per-row MVCC types live behind one stable name
178// (`spg_engine::Snapshot` / `spg_engine::RowHeader`) instead of every
179// caller threading through `spg_storage::snapshot::*` directly.
180pub use spg_storage::RowChange;
181pub use spg_storage::row_header::{RowHeader, XMAX_ALIVE, XMIN_FROZEN};
182pub use spg_storage::snapshot::{
183 AllCommitted, InProgressSet, Snapshot, XactStatus, XactStatusOracle,
184};
185
186/// v7.37.15 (Phase C.2) — the engine is its own visibility oracle.
187/// Scans hold `&Engine` while reading, so a scan site can pass `self`
188/// as the [`XactStatusOracle`] alongside its [`Snapshot`] when the
189/// visibility gate migrates from `visible` to `visible_with_status`
190/// (next Phase C step). Delegates to [`Engine::xact_status`].
191impl XactStatusOracle for Engine {
192 fn status(&self, version: u64) -> XactStatus {
193 self.xact_status(version)
194 }
195}
196// v7.37.14 (A2.5-stub) — re-export the silent-FOR-UPDATE telemetry
197// helper through the engine surface so downstream wrappers
198// (spg-embedded / spg-embedded-tokio / spgctl) and their tests
199// don't need a direct `spg-sql` dep.
200use spg_sql::parser::ParseError;
201pub use spg_sql::silent_for_update_count;
202use spg_storage::{Catalog, ColumnSchema, Row, StorageError};
203
204use crate::eval::EvalError;
205
206/// Result of executing one statement.
207#[derive(Debug, Clone, PartialEq)]
208#[non_exhaustive]
209pub enum QueryResult {
210 /// DDL or DML succeeded.
211 ///
212 /// `affected` is the row count for `INSERT` and 0 elsewhere.
213 /// `modified_catalog` tells the server whether this statement
214 /// caused the *committed* catalog to change — it's the signal to
215 /// snapshot/audit. False for `BEGIN`/`ROLLBACK`, false for writeful
216 /// statements executed inside a transaction (those only touch the
217 /// shadow), and true for `COMMIT` and for writes outside a TX.
218 CommandOk {
219 affected: usize,
220 modified_catalog: bool,
221 },
222 /// `SELECT` returned a (possibly empty) row set.
223 Rows {
224 columns: Vec<ColumnSchema>,
225 rows: Vec<Row<'static>>,
226 },
227}
228
229/// All errors the engine can return.
230///
231/// Marked `#[non_exhaustive]` from v7.5.0 onward: external `match`
232/// must include a `_` arm so new variants in subsequent v7.x releases
233/// are not breaking changes.
234#[derive(Debug, Clone, PartialEq)]
235#[non_exhaustive]
236pub enum EngineError {
237 Parse(ParseError),
238 Storage(StorageError),
239 Eval(EvalError),
240 /// Front-end accepted a construct that the v0.x executor doesn't support.
241 Unsupported(String),
242 /// `BEGIN` while another transaction is already open.
243 TransactionAlreadyOpen,
244 /// `COMMIT` / `ROLLBACK` with no active transaction.
245 NoActiveTransaction,
246 /// v7.38 (read01 P3.26) — a statement other than COMMIT / ROLLBACK /
247 /// ROLLBACK TO SAVEPOINT was issued after an earlier statement in the
248 /// same transaction failed. PG aborts the whole transaction on the
249 /// first error and rejects everything until it is ended (SQLSTATE
250 /// 25P02); this mirrors that so partial work can't slip through.
251 InFailedTransaction,
252 /// v7.39 (round 299, E3 Phase 2) — a row lock is held by another
253 /// transaction and the policy is `Wait`.
254 ///
255 /// Its own variant, not an `Unsupported` string: the SERVER has to
256 /// recognise it to retry, and it cannot block inside the engine
257 /// write lock — doing so would stop the whole server, including the
258 /// transaction whose commit would release the lock.
259 LockWouldBlock,
260 /// v7.39 (round 299) — granting the wait would close a wait-for
261 /// cycle. PG's 40P01.
262 LockDeadlock,
263 /// v7.38 (read01 P4.02) — a scalar / row subquery used as an
264 /// expression returned more than one row. PG raises this as
265 /// SQLSTATE 21000 (CARDINALITY_VIOLATION) with a fixed message.
266 CardinalityViolation,
267 /// v7.37.17 (Phase E3) — a REPEATABLE READ / SERIALIZABLE commit
268 /// found a write-write conflict with a concurrently-committed
269 /// transaction (a row this tx wrote was deleted/updated by another
270 /// committed writer, or a unique key this tx inserted was taken).
271 /// PG raises SQLSTATE 40001; the client retries the transaction.
272 /// The failing COMMIT rolls the transaction back, like PG.
273 SerializationFailure(String),
274 /// v4.0 sentinel: `execute_readonly` got a statement that
275 /// mutates engine state (INSERT / CREATE / BEGIN / COMMIT / …).
276 /// The caller should retake the write lock and dispatch through
277 /// `execute(&mut self)` instead.
278 WriteRequired,
279 /// v4.2: a SELECT would have returned more rows than the
280 /// configured `max_query_rows` cap. Carries the cap.
281 RowLimitExceeded(usize),
282 /// v7.30.3 (mailrs round-26): a SELECT's join/filter
283 /// materialisation would have held more (approximate) heap
284 /// bytes than the configured `max_query_bytes` cap. The row
285 /// cap above counts rows; this counts bytes, because one row
286 /// can be a multi-MB mail body — 1000 fat rows pressure the
287 /// host long before any row ceiling trips. Carries the cap.
288 QueryBytesExceeded(usize),
289 /// v4.5: cooperative cancellation — the host (server's
290 /// per-query watchdog) set the cancel flag while a long-running
291 /// SELECT / UPDATE / DELETE was scanning rows. The partial work
292 /// is discarded; the caller should surface this as a timeout
293 /// to the client.
294 Cancelled,
295 /// v7.39 (round 318, V51) — MySQL `KILL <id>` naming an id no live
296 /// connection carries. MariaDB 11: `ERROR 1094 (HY000) Unknown thread
297 /// id: N`.
298 UnknownThreadId(u32),
299 /// v7.39 (round 318, V51) — MySQL `KILL <own id>`. The connection
300 /// really is killed; MariaDB reports it to the victim as
301 /// `ERROR 1927 (70100) Connection was killed` and closes.
302 ConnectionKilled,
303 /// v7.38 Epic P (panic isolation): a panic unwound out of
304 /// statement execution and was caught at the engine's
305 /// `execute_*` boundary (see `execute_in_with_cancel`). The
306 /// in-flight transaction's shadow was discarded (rollback) and
307 /// the engine left consistent; the caller sees this ordinary
308 /// error instead of a crashed process / poisoned lock. NOTE:
309 /// under the release `panic = "abort"` profile the process
310 /// aborts before any unwind, so this variant only ever surfaces
311 /// in dev/test (`panic = "unwind"`) — and in production once a
312 /// later slice flips the release profile to unwind.
313 Internal(String),
314}
315
316impl fmt::Display for EngineError {
317 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
318 match self {
319 Self::Parse(e) => write!(f, "parse: {e}"),
320 Self::Storage(e) => write!(f, "storage: {e}"),
321 Self::Eval(e) => write!(f, "eval: {e}"),
322 Self::Unsupported(s) => write!(f, "unsupported: {s}"),
323 Self::TransactionAlreadyOpen => f.write_str("a transaction is already open"),
324 Self::NoActiveTransaction => f.write_str("no active transaction"),
325 Self::LockWouldBlock => f.write_str("row is locked by another transaction"),
326 Self::LockDeadlock => f.write_str("deadlock detected"),
327 Self::InFailedTransaction => f.write_str(
328 "current transaction is aborted, commands ignored until end of transaction block",
329 ),
330 Self::CardinalityViolation => {
331 f.write_str("more than one row returned by a subquery used as an expression")
332 }
333 Self::SerializationFailure(detail) => {
334 // v7.39 (round 552) — PG has TWO wordings under 40001 and
335 // they mean different things: "concurrent update" for a
336 // write-write conflict, "read/write dependencies among
337 // transactions" for the antidependency a SERIALIZABLE
338 // transaction hits. A detail that already carries PG's
339 // own sentence is passed through rather than nested
340 // inside the other one.
341 if detail.starts_with("could not serialize access") {
342 f.write_str(detail)
343 } else {
344 write!(
345 f,
346 "could not serialize access due to concurrent update: {detail}"
347 )
348 }
349 }
350 Self::WriteRequired => {
351 f.write_str("statement requires a write lock (use execute, not execute_readonly)")
352 }
353 Self::RowLimitExceeded(n) => {
354 write!(f, "query exceeded max_query_rows={n}")
355 }
356 Self::QueryBytesExceeded(n) => {
357 write!(
358 f,
359 "query materialisation exceeded max_query_bytes={n} (set SPG_MAX_QUERY_BYTES to raise, 0 to disable)"
360 )
361 }
362 Self::Cancelled => f.write_str("query cancelled (timeout or client request)"),
363 Self::UnknownThreadId(id) => write!(f, "Unknown thread id: {id}"),
364 Self::ConnectionKilled => f.write_str("Connection was killed"),
365 Self::Internal(s) => write!(f, "internal error: {s}"),
366 }
367 }
368}
369
370impl From<ParseError> for EngineError {
371 fn from(e: ParseError) -> Self {
372 Self::Parse(e)
373 }
374}
375impl From<StorageError> for EngineError {
376 fn from(e: StorageError) -> Self {
377 Self::Storage(e)
378 }
379}
380impl From<EvalError> for EngineError {
381 fn from(e: EvalError) -> Self {
382 Self::Eval(e)
383 }
384}
385
386/// The execution engine. Holds the catalog and (later) other server-scope
387/// state. `Engine::new()` is intentionally cheap so callers can construct one
388/// per database, per test.
389/// Function pointer that returns "now" as microseconds since Unix
390/// epoch. The engine is `no_std`, so it can't reach for `std::time`
391/// itself — callers (`spg-server`, the sqllogictest runner) inject a
392/// concrete implementation. `None` means `NOW()` / `CURRENT_*` raise
393/// `Unsupported`.
394pub type ClockFn = fn() -> i64;
395
396/// Backing store for `SPG_TEST_FIXED_CLOCK_MICROS` (test-mode GUC).
397/// Only ever written when that GUC is set; production engines never
398/// touch it.
399static FIXED_CLOCK_MICROS: core::sync::atomic::AtomicI64 = core::sync::atomic::AtomicI64::new(0);
400
401fn fixed_clock_from_env() -> i64 {
402 FIXED_CLOCK_MICROS.load(core::sync::atomic::Ordering::Relaxed)
403}
404
405/// v7.39 (pg_stat knife A) — host-provided live connection count for
406/// `pg_stat_database.numbackends`.
407pub type BackendCountFn = fn() -> u32;
408
409/// v7.39 (read01 pgstatfuncs.c) — host-provided identity of the CALLING
410/// connection for `pg_backend_pid()` / the pg_stat_activity self-join.
411/// The host reads a connection-thread-local set at session start; the
412/// no_std engine just calls through. `None` (embedded) → pid 1.
413pub type BackendPidFn = fn() -> u32;
414
415/// v7.39 (round 476) — the WAL's current byte position, as a PG LSN.
416///
417/// `pg_current_wal_lsn()` answered the literal `0/0` forever, so every
418/// monitor watching WAL progress or replication lag saw an instance that
419/// had never written anything. SPG's WAL is a file and its length IS an
420/// LSN in every sense a monitor uses one: monotonic, byte-denominated, and
421/// comparable — `pg_wal_lsn_diff` over two samples gives real bytes.
422///
423/// `None` (embedded, or a server started without a WAL) keeps `0/0`, which
424/// is the honest answer there: nothing is being written.
425pub type WalLsnFn = fn() -> u64;
426
427/// v7.39 (round 318, V51) — host-provided connection control. `terminate`
428/// false = cancel the target's running statement (PG `pg_cancel_backend`,
429/// MySQL `KILL QUERY`); true = also close the connection (PG
430/// `pg_terminate_backend`, MySQL `KILL CONNECTION`). Returns whether a
431/// connection with that id exists — the engine has no registry of its own,
432/// so the answer has to come from the host that accepted the sockets.
433/// `None` (embedded, no connections) ⇒ nothing to signal.
434pub type BackendSignalFn = fn(pid: u32, terminate: bool) -> bool;
435
436pub use tempstore::{SpillStats, TempRun, TempRunFactory, TempStoreError};
437
438/// v7.39 (tz epic) — host-injected IANA timezone lookups (the no_std
439/// engine can't read the system zoneinfo directory; spg-tzif is the
440/// std-side implementation). All instants are MICROSECONDS.
441/// UTC offset (µs east) of a zone at a UTC instant; None = unknown zone.
442/// v7.39 (round 534) — the compiled-in default PG18 reports for a
443/// configuration parameter, for the wire's own SHOW shortcut.
444///
445/// The pgwire layer answers `SHOW <name>` from a small canned list
446/// before the statement ever reaches the engine, so it needs the same
447/// inventory the engine reads or the two disagree — which they did:
448/// `SHOW fsync` over the wire returned an empty row.
449#[must_use]
450pub fn pg_guc_boot_value(name: &str) -> Option<&'static str> {
451 crate::guc_catalog::guc_boot_value(name)
452}
453
454pub type TzOffsetFn = fn(&str, i64) -> Option<i64>;
455/// Local wall-clock µs -> UTC µs with PG's DST disambiguation.
456pub type TzLocalizeFn = fn(&str, i64) -> Option<i64>;
457/// Canonical zone spelling ("asia/tokyo" -> "Asia/Tokyo").
458pub type TzCanonFn = fn(&str) -> Option<alloc::string::String>;
459/// Zone designation ("JST", "EDT") at a UTC instant.
460pub type TzAbbrevFn = fn(&str, i64) -> Option<alloc::string::String>;
461/// v7.39 (round 502) — every zone the host knows at a UTC instant, as
462/// `(name, abbrev, utc_offset_secs, is_dst)`.
463///
464/// Backs `pg_timezone_names`. SPG resolved named zones correctly — round
465/// 502 measured DST boundaries byte-identical to PG18 — but could not
466/// LIST them, so a client populating a timezone picker got "relation
467/// pg_timezone_names does not exist". The data was there, only
468/// unlistable. No hook, or a host with no tzdata, yields an empty view
469/// rather than an error: that is what such a host honestly has.
470pub type TzAllFn =
471 fn(i64) -> alloc::vec::Vec<(alloc::string::String, alloc::string::String, i64, bool)>;
472
473/// v7.39 (tz epic) — per-statement snapshot of the session TimeZone,
474/// consumed per-VALUE by the timestamptz renderers (a DST zone's
475/// offset depends on the instant being rendered).
476#[derive(Debug, Clone)]
477pub enum SessionTz {
478 Utc,
479 /// Fixed offset, µs east.
480 Fixed(i64),
481 /// IANA zone + the host lookups.
482 Named(alloc::string::String, TzOffsetFn, TzAbbrevFn),
483}
484
485impl SessionTz {
486 #[must_use]
487 pub fn is_utc(&self) -> bool {
488 matches!(self, Self::Utc) || matches!(self, Self::Fixed(0))
489 }
490
491 /// Offset (µs east) at a UTC instant.
492 #[must_use]
493 pub fn offset_at(&self, utc_micros: i64) -> i64 {
494 match self {
495 Self::Utc => 0,
496 Self::Fixed(off) => *off,
497 Self::Named(zone, f, _) => f(zone, utc_micros).unwrap_or(0),
498 }
499 }
500
501 /// Designation for the non-ISO DateStyle suffix: a named zone's
502 /// abbreviation at the instant; None for UTC/fixed (callers spell
503 /// "UTC" / "+09" themselves).
504 #[must_use]
505 pub fn abbrev_at(&self, utc_micros: i64) -> Option<alloc::string::String> {
506 match self {
507 Self::Named(zone, _, f) => f(zone, utc_micros),
508 _ => None,
509 }
510 }
511}
512
513/// Function pointer that produces 16 cryptographically random bytes.
514/// Like `ClockFn`, the engine is `no_std` and can't reach for /dev/urandom
515/// itself — host (`spg-server`) injects an OS-backed source. `None`
516/// means SQL-driven `CREATE USER` falls back to a deterministic salt
517/// derived from the username (acceptable in tests; the server always
518/// installs a real RNG so production paths never see this).
519pub type SaltFn = fn() -> [u8; 16];
520
521/// v4.5 cooperative cancellation token. A long-running SELECT /
522/// UPDATE / DELETE checks `is_cancelled` at row-loop checkpoints
523/// and bails with `EngineError::Cancelled`. The host
524/// (`spg-server`) creates an `AtomicBool` per query, spawns a
525/// watchdog thread that sets it after `SPG_QUERY_TIMEOUT_MS`,
526/// and passes it via `execute_with_cancel` / `execute_readonly_with_cancel`.
527///
528/// `CancelToken::none()` is a no-op — used by the legacy `execute`
529/// and `execute_readonly` entry points so existing callers don't
530/// change.
531/// v4.41.1 opaque transaction handle. Returned by `Engine::alloc_tx_id`,
532/// threaded through `Engine::execute_in` so dispatch can identify which
533/// in-flight TX a statement belongs to. `IMPLICIT_TX` is the reserved
534/// slot every legacy caller — engine self-tests, spg-cli, spg-embedded,
535/// startup replay — implicitly uses through the unchanged
536/// `Engine::execute(sql)` API. v4.41.1 keeps at most one active slot at
537/// runtime (dispatch holds `engine.write()` across the wrap, same as
538/// v4.34); the map shape is here to let v4.42 turn on N in-flight
539/// implicit TXs without reshuffling the engine internals.
540#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
541pub struct TxId(pub u64);
542
543/// Reserved slot used by `Engine::execute(sql)` — the legacy single-
544/// global-shadow path. New `alloc_tx_id` handles start at 1.
545pub const IMPLICIT_TX: TxId = TxId(0);
546
547/// v6.7.3 — default segment-size threshold used by `COMPACT COLD
548/// SEGMENTS` when no explicit target is supplied. Segments whose
549/// `OwnedSegment::bytes().len()` is **strictly** less than this
550/// value are eligible to merge. spg-server reads
551/// `SPG_COMPACTION_TARGET_SEGMENT_BYTES` to override.
552pub const COMPACTION_TARGET_DEFAULT_BYTES: u64 = 4 * 1024 * 1024;
553
554/// Per-slot transaction state. Held inside `tx_catalogs[tx_id]` for the
555/// lifetime of a BEGIN..COMMIT (or BEGIN..ROLLBACK) window. Drops when
556/// the TX commits (its `catalog` is moved over `Engine.catalog`) or
557/// rolls back (slot removed, catalog discarded).
558#[derive(Debug, Default, Clone)]
559struct TxState {
560 /// The TX's shadow copy of the catalog. Started as a clone of
561 /// `Engine.catalog` at BEGIN time; writes flow into it; COMMIT
562 /// installs it over `Engine.catalog`. `Catalog::clone()` is O(1)
563 /// since v4.40 (`PersistentVec` rows + `PersistentBTreeMap` indices).
564 catalog: Catalog,
565 /// v7.37 (round 828) — the TX's shadow copy of the user store,
566 /// following exactly the catalog's model one field up: created
567 /// lazily by the first role DDL inside the TX (an ordinary TX
568 /// never pays the clone), written through for the rest of the TX,
569 /// installed over `Engine.users` at COMMIT, discarded on ROLLBACK.
570 /// PG treats roles as ordinary catalog rows — `BEGIN; CREATE ROLE
571 /// r; ROLLBACK` leaves no role — and SPG used to refuse the
572 /// statement instead, which no drop-in client expects.
573 ///
574 /// Other sessions and the auth path keep reading the committed
575 /// store, so an uncommitted role can neither log in nor be seen
576 /// elsewhere — the isolation PG gives via its catalog MVCC.
577 users: Option<crate::users::UserStore>,
578 /// Per-TX savepoint stack. Each entry pairs the savepoint name with
579 /// a clone of `catalog` (and of the role shadow, which subtransactions
580 /// roll back too) at the moment `SAVEPOINT <name>` fired.
581 /// `ROLLBACK TO <name>` restores from the entry and pops everything
582 /// after it; `RELEASE <name>` discards the entry and everything
583 /// after; COMMIT/ROLLBACK clears the whole stack.
584 savepoints: Vec<(String, Catalog, Option<crate::users::UserStore>)>,
585 /// v7.37.15 (Phase E) — cached MVCC snapshot for REPEATABLE
586 /// READ / SERIALIZABLE. Captured at `exec_begin` time when the
587 /// session's `current_isolation_level` is RR/SER; read paths
588 /// inside the TX use this snapshot rather than calling
589 /// `Engine::current_snapshot` per statement, so a row that
590 /// becomes visible mid-tx (because another writer committed)
591 /// is NOT exposed to this tx — preserving RR's invariant.
592 ///
593 /// `None` for READ COMMITTED (default): each statement gets a
594 /// fresh snapshot via `current_snapshot()`.
595 cached_snapshot: Option<spg_storage::snapshot::Snapshot>,
596 /// v7.37.17 (Phase E2 — RC rebase) — tables this tx has run DML
597 /// against. The per-statement rebase extracts/replays write-sets
598 /// only for these (see `maybe_rc_rebase`).
599 touched_tables: alloc::collections::BTreeSet<String>,
600 /// v7.39 (round 552) — tables this tx has READ. PG's SIREAD locks,
601 /// at table granularity: the coarse end of the same idea, and what
602 /// PG itself falls back to when its per-tuple lock memory runs out.
603 ///
604 /// A SERIALIZABLE tx aborts at COMMIT if any table it read was
605 /// written by a transaction that committed after its snapshot —
606 /// the read/write antidependency SI cannot see. Coarse granularity
607 /// means SPG aborts some transactions PG would let through; it
608 /// never lets through one PG would abort.
609 read_tables: alloc::collections::BTreeSet<String>,
610 /// v7.39 (round 552) — was THIS transaction opened SERIALIZABLE?
611 ///
612 /// `Engine::current_isolation_level` is one field for the whole
613 /// engine, not part of the per-session bag, so with two connections
614 /// open one transaction's COMMIT resets it under the other's feet —
615 /// the shared-engine leak rounds 279 and 283 chased through session
616 /// state and advisory locks. The level a transaction runs at has to
617 /// live on the transaction.
618 serializable: bool,
619 /// The engine's commit sequence when this tx began.
620 begin_commit_seq: u64,
621 /// v7.39 (round 494) — has anything asked for this shadow catalog
622 /// MUTABLY since BEGIN?
623 ///
624 /// COMMIT installs the shadow over the committed catalog, so a
625 /// transaction that changed nothing must install nothing — otherwise
626 /// it reverts whatever other sessions committed while it was open.
627 /// `touched_tables` cannot answer this: it records DML targets for the
628 /// rebase, and a `SELECT lo_write(…)` classifies read-only while
629 /// mutating (the large-object pins caught exactly that).
630 ///
631 /// Set in `active_catalog_mut`, the single place a `&mut Catalog` is
632 /// handed out. A caller that takes the mutable handle without writing
633 /// merely keeps the old install behaviour, so the flag errs toward
634 /// installing.
635 shadow_dirty: bool,
636 /// v7.39 (round 298) — this transaction is in the aborted state.
637 ///
638 /// Per SLOT. It used to be one flag on the shared `Engine`, guarded
639 /// by `in_transaction()` — the GLOBAL "is any transaction open"
640 /// test. So an autocommit statement that failed while a DIFFERENT
641 /// connection happened to hold a transaction set the flag, and
642 /// every other connection was then refused with 25P02. Round 283
643 /// fixed seven sites of this exact shape in the server; this one is
644 /// in the engine and was missed.
645 aborted: bool,
646 /// v7.39 (round 288) — `SET CONSTRAINTS … {DEFERRED|IMMEDIATE}`
647 /// override for this transaction. `None` = each constraint uses
648 /// its own declared timing; `Some(true)` = every DEFERRABLE one is
649 /// deferred; `Some(false)` = every one is immediate.
650 constraints_deferred: Option<bool>,
651 /// v7.39 (round 308) — per-constraint overrides from the NAMED form
652 /// of `SET CONSTRAINTS`. Consulted before `constraints_deferred`, so
653 /// `ALL DEFERRED` followed by `fk_a IMMEDIATE` leaves fk_a immediate
654 /// and everything else deferred. An `ALL` form clears this map,
655 /// which is what makes a later blanket setting win — PG resets the
656 /// whole set the same way.
657 constraints_deferred_by_name: BTreeMap<String, bool>,
658 /// v7.37.17 — the tx executed a statement whose effect on the
659 /// shadow catalog can't be expressed as a versioned row write-set
660 /// (DDL, COPY, anything unclassified). The rebase would lose it,
661 /// so the tx degrades to its frozen BEGIN-time view (SI) for the
662 /// rest of its life — the pre-E2 behaviour, honestly kept.
663 rebase_poisoned: bool,
664 /// v7.37.17 — statements successfully run inside this tx. The
665 /// first statement sees the BEGIN-time clone unchanged (it IS the
666 /// latest base at that point); rebasing starts from the second.
667 stmts_run: u32,
668 /// v7.39 (round 196) — the engine `commit_epoch` this tx last
669 /// rebased against (BEGIN seeds it). When the epoch hasn't moved,
670 /// no other path committed to the base catalog, so the
671 /// per-statement RC rebase — whose write-set extraction is a full
672 /// scan of every touched table — is skipped entirely. The r196
673 /// wire panel traced tx_batch's 2.8× LOSS to exactly that scan
674 /// running before EVERY in-tx statement (~200 µs/stmt on a
675 /// 20k-row table, 58× the statement itself). Over-incrementing
676 /// the epoch is safe (an extra rebase is only slower, never
677 /// wrong); missing an increment would be a correctness bug, so
678 /// the epoch bumps on every completed non-tx statement.
679 rebased_at_epoch: u64,
680 /// v7.37.17 (Phase E4 fix) — (old RowId → new RowId) pairs recorded
681 /// by every in-place UPDATE this tx ran, keyed by table (RowIds are
682 /// per-relation). An UPDATE's write-set is tombstone(old) +
683 /// insert(new); when a rebase skips a CONFLICTING tombstone (the
684 /// row was updated/deleted by a concurrently-committed tx), the
685 /// paired insert must be dropped too — otherwise the row
686 /// DUPLICATES (caught by the E4 isolation matrix).
687 update_pairs: alloc::collections::BTreeMap<
688 String,
689 Vec<(
690 spg_storage::row_header::RowId,
691 spg_storage::row_header::RowId,
692 )>,
693 >,
694}
695
696/// v7.11.0 — frozen read-only view of the engine's committed state.
697/// Constructed via [`Engine::clone_snapshot`]. Holds clones of the
698/// catalog, statistics, clock function, and row-cap config — the
699/// four fields the `execute_readonly` path actually reads. Cheap to
700/// `Clone` (each clone shares the underlying `PersistentVec` row
701/// storage; only the trie root pointers copy). Send + Sync so a
702/// snapshot can be moved across `tokio::task::spawn_blocking`
703/// boundaries without coordination.
704///
705/// The contract: a snapshot reflects the engine's state at the
706/// moment `clone_snapshot()` returned. Subsequent writes to the
707/// engine are NOT visible. Callers who need fresher data take a
708/// new snapshot.
709#[derive(Debug, Clone)]
710pub struct CatalogSnapshot {
711 catalog: Catalog,
712 statistics: statistics::Statistics,
713 clock: Option<ClockFn>,
714 max_query_rows: Option<usize>,
715}
716
717/// CoW-1 (v7.34) — frozen view of the *persisted* committed engine
718/// state. Carries every field the `snapshot()` envelope serializes;
719/// v7.39 (round 279) — the per-CONNECTION state, parked while another
720/// connection holds the engine.
721///
722/// The server runs ONE shared `Engine` behind a `RwLock`
723/// (`ServerState.engine`, built once at startup), so everything the
724/// engine called "session state" was in fact process-wide and leaked
725/// between clients: two connections saw each other's prepared
726/// statements, and one client's `SET sql_mode` re-dialected another's
727/// string literals. PG scopes all of this per session.
728///
729/// Rather than thread a session handle through every call site, the
730/// engine keeps the ACTIVE session's state in its own fields — so the
731/// ~40 existing `self.session_params` / `self.backslash_escapes` uses
732/// are untouched — and swaps the whole bag when the caller announces a
733/// different session. Embedded hosts never announce one and stay on
734/// session 0 forever, exactly as before.
735/// v7.38.18 (C12) — one row of MySQL's diagnostics area.
736///
737/// The three columns `SHOW WARNINGS` returns, and the numbers are
738/// MySQL's own: `1265` for a truncated value, `1366` for an integer
739/// column given something that is not one. Measured on 9.7.2 rather
740/// than looked up, because an errno a client switches on has to be
741/// the errno it would have seen.
742#[derive(Debug, Clone, PartialEq, Eq)]
743pub struct MysqlWarning {
744 /// `Warning`, `Error` or `Note` — the `Level` column.
745 pub level: &'static str,
746 /// MySQL's error code.
747 pub code: u16,
748 /// The message, worded as MySQL words it.
749 pub message: String,
750}
751
752#[derive(Debug, Default)]
753pub(crate) struct SessionBag {
754 pub(crate) session_params: BTreeMap<String, String>,
755 pub(crate) backslash_escapes: bool,
756 /// v7.39 (round 470) — is the MySQL session in a strict `sql_mode`?
757 ///
758 /// MariaDB's default includes `STRICT_TRANS_TABLES`, so this starts
759 /// true; `SET sql_mode=''` (or any list without a STRICT_ flag) turns
760 /// it off and a value that would otherwise raise is bent to fit
761 /// instead — the same conversion `INSERT IGNORE` uses.
762 pub(crate) mysql_strict: bool,
763 /// v7.38.18 (C12) — the MySQL diagnostics area: what the last
764 /// warning-generating statement bent, ready for `SHOW WARNINGS`
765 /// and `@@warning_count`.
766 ///
767 /// The value-bending half has been byte-for-byte MySQL's since
768 /// v7.39 round 470 — `INSERT INTO t (i, s) VALUES ('abc', 'toolong')`
769 /// stores `0` and `'too'` exactly as MySQL 9.7.2 does. What was
770 /// missing is that MySQL TELLS you: `Warning 1265 Data truncated
771 /// for column 's' at row 1`. An application that checks
772 /// `@@warning_count` after an insert had no way to learn that its
773 /// data had been changed.
774 ///
775 /// Cleared by the next statement that can produce warnings, which
776 /// is MySQL's rule; `SHOW WARNINGS` and `SELECT @@warning_count`
777 /// read it without clearing.
778 pub(crate) mysql_warnings: Vec<MysqlWarning>,
779 pub(crate) prepared_statements: BTreeMap<String, PreparedSqlStatement>,
780 /// v7.39 (round 499) — the value `nextval` last returned IN THIS
781 /// SESSION, per sequence, and which sequence that was.
782 ///
783 /// PG defines `currval` and `lastval` as session-local: they answer
784 /// the number THIS session was given, and error with 55000 ("not yet
785 /// defined in this session") when it has not called `nextval`. They
786 /// are deliberately not the sequence's current value — another
787 /// session may have advanced it since, and reading that would hand
788 /// back a number this session never owned, which is what a caller
789 /// then uses as a foreign key.
790 ///
791 /// Measured before this (`iso_session` T1/T2): `currval` answered in
792 /// a connection that had never called `nextval`, and `lastval`
793 /// answered across connections, because the tracking lived on the
794 /// shared engine rather than in the bag.
795 pub(crate) seq_currvals: BTreeMap<String, i64>,
796 pub(crate) last_sequence_used: Option<String>,
797 /// v7.39 (round 553) — the isolation level THIS connection is
798 /// running at.
799 ///
800 /// It lived on the shared engine, so it leaked both ways between
801 /// connections. Measured over pgwire against PG18: connection B
802 /// opened a plain BEGIN and `SHOW transaction_isolation` answered
803 /// `serializable` — A's level; and A, still inside its SERIALIZABLE
804 /// block, then read `read committed`, because B's COMMIT reset the
805 /// field under it. PG answers `read committed` and `serializable`
806 /// throughout. So a transaction that asked for SERIALIZABLE ran at
807 /// READ COMMITTED and one that asked for nothing ran at
808 /// SERIALIZABLE, purely because another connection was busy.
809 ///
810 /// Round 552 saw the same field give way and worked around it by
811 /// putting the level on the TRANSACTION; this puts the session's
812 /// own copy where the rest of its state already lives — the place
813 /// r306's comment says every piece of per-connection state belongs
814 /// so it never gets a process-wide version to regress from.
815 pub(crate) isolation_level: spg_sql::ast::IsolationLevel,
816 /// v7.39 (round 306) — open large-object descriptors. Per session
817 /// from the start, deliberately: r277/r279/r283 each landed a piece
818 /// of per-connection state on the process-wide engine first and had
819 /// to be unpicked afterwards, so this one never gets a process-wide
820 /// version to regress from. PG additionally scopes descriptors to
821 /// the transaction, so the table is emptied at COMMIT / ROLLBACK.
822 pub(crate) lo_descriptors: BTreeMap<i32, LargeObjectDescriptor>,
823 /// Next descriptor number to hand out. PG starts at 0 and counts up
824 /// within a transaction, restarting once the transaction ends.
825 pub(crate) lo_next_fd: i32,
826 /// v7.39 (round 321, V54) — open server-side cursors. They lived on
827 /// the shared engine until now, i.e. in ONE namespace for every
828 /// connection: two clients could not both `DECLARE c`, a `FETCH`
829 /// could read another client's rows, and `CLOSE ALL` closed
830 /// everybody's.
831 pub(crate) cursors: BTreeMap<String, cursor::OpenCursor>,
832 /// v7.39 (round 347, M2) — MySQL's `LAST_INSERT_ID()`. Per SESSION
833 /// from the start (r277/r279/r283 each paid for landing per-connection
834 /// state on the shared engine first): one connection's insert must not
835 /// be readable as another's. MariaDB, measured: a fresh session reads
836 /// 0; an insert that generates an AUTO_INCREMENT value sets it to the
837 /// FIRST one generated; a statement that generates none — an explicit
838 /// id, an UPDATE, a DELETE, a plain table — leaves it alone.
839 pub(crate) last_insert_id: i64,
840 /// v7.39 (round 426) — MySQL's `ROW_COUNT()`. Per SESSION like
841 /// `last_insert_id`. MariaDB, measured: a DML statement leaves the
842 /// number of rows it CHANGED (an UPDATE that matched but changed
843 /// nothing leaves 0); a SELECT leaves -1; DDL leaves 0. A FRESH
844 /// session reads 0 (measured), not -1.
845 pub(crate) row_count: i64,
846 /// v7.39 (round 430) — MySQL USER variables (`SET @x = 5`). Per
847 /// SESSION like `last_insert_id` / `row_count`; its own namespace,
848 /// separate from the `@@` session parameters. Reading an unset one
849 /// answers NULL, as MariaDB does.
850 pub(crate) user_vars: BTreeMap<String, spg_storage::Value<'static>>,
851 /// v7.39 (round 436) — the logical names of this session's TEMPORARY
852 /// tables. Each is stored in the catalog under a per-session prefix; this
853 /// set is what says "resolve `t` to my temp one" and what `end_session`
854 /// walks to drop them.
855 pub(crate) temp_tables: alloc::collections::BTreeSet<String>,
856 /// v7.39 (round 469) — the session's TEMPORARY sequences and views, by
857 /// logical name. Separate sets because dropping one at session end has
858 /// to name the catalog map it lives in.
859 pub(crate) temp_sequences: alloc::collections::BTreeSet<String>,
860 pub(crate) temp_views: alloc::collections::BTreeSet<String>,
861}
862
863/// v7.39 (round 306) — one open large-object descriptor.
864#[derive(Debug, Clone, Copy)]
865pub(crate) struct LargeObjectDescriptor {
866 pub(crate) oid: u32,
867 /// Byte offset the next read / write starts at.
868 pub(crate) pos: u64,
869 /// Whether the descriptor was opened with `INV_WRITE`. Reads need no
870 /// permission at all in PG — a write-only descriptor reads fine —
871 /// so only this half is worth remembering.
872 pub(crate) writable: bool,
873}
874
875/// v7.39 (round 277) — one SQL-level prepared statement.
876#[derive(Debug, Clone)]
877pub(crate) struct PreparedSqlStatement {
878 /// The body with its `$N` placeholders still in place.
879 pub(crate) body: spg_sql::ast::Statement,
880 /// Declared parameter type names, in order; empty when PG would
881 /// have inferred them.
882 pub(crate) param_types: alloc::vec::Vec<String>,
883 /// The whole `PREPARE …` text, which `pg_prepared_statements`
884 /// reports verbatim.
885 pub(crate) source: String,
886}
887
888/// `Clone` is O(1) on the catalog (Arc bump) and cheap typed-clones
889/// on the trailers. Decouples "capture state" from "serialize bytes"
890/// so the background-checkpoint worker can hold the snapshot and
891/// produce bytes off the engine write lock.
892#[derive(Debug, Clone)]
893pub struct EngineSnapshot {
894 catalog: Catalog,
895 users: UserStore,
896 publications: publications::Publications,
897 subscriptions: subscriptions::Subscriptions,
898 statistics: statistics::Statistics,
899}
900
901impl EngineSnapshot {
902 /// Same envelope rules as `Engine::snapshot()`: bare catalog when
903 /// every trailer is empty, full envelope otherwise.
904 pub fn serialize(&self) -> Vec<u8> {
905 if self.users.is_empty()
906 && self.publications.is_empty()
907 && self.subscriptions.is_empty()
908 && self.statistics.is_empty()
909 {
910 self.catalog.serialize()
911 } else {
912 build_envelope(
913 &self.catalog.serialize(),
914 &users::serialize_users(&self.users),
915 &self.publications.serialize(),
916 &self.subscriptions.serialize(),
917 &self.statistics.serialize(),
918 )
919 }
920 }
921}
922
923/// v7.39 (parallel-agg P0) — host-injected parallel executor. The
924/// engine is `no_std` and cannot spawn threads; like `ClockFn` /
925/// `RandomFn`, the std-side host (spg-server / embedded-tokio)
926/// injects an implementation at startup. `None` (the default, and
927/// the only option in pure-`no_std` embeddings) keeps every code
928/// path single-threaded and byte-identical to pre-P0 behaviour.
929///
930/// The callback returns `Box<dyn Any + Send>` so one trait serves
931/// any shard-result type; call sites downcast what they produced.
932pub trait ParallelRunner: Send + Sync {
933 /// Run `f(0) .. f(n-1)`, possibly concurrently; return the
934 /// results in shard order. Every call completes before return.
935 fn run_shards(
936 &self,
937 n: usize,
938 f: &(dyn Fn(usize) -> alloc::boxed::Box<dyn core::any::Any + Send> + Sync),
939 ) -> alloc::vec::Vec<alloc::boxed::Box<dyn core::any::Any + Send>>;
940}
941
942/// Engine slot for the injected runner — a newtype so the `Engine`
943/// derive(Debug) keeps working over the non-Debug trait object.
944#[derive(Clone, Default)]
945pub struct ParallelRunnerSlot(pub(crate) Option<alloc::sync::Arc<dyn ParallelRunner>>);
946
947impl core::fmt::Debug for ParallelRunnerSlot {
948 fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
949 f.write_str(if self.0.is_some() {
950 "ParallelRunner(<injected>)"
951 } else {
952 "ParallelRunner(none)"
953 })
954 }
955}
956
957/// v7.39 (parallel-agg P0) — below this many input rows a query
958/// never parallelises: thread spin-up beats the win on small scans.
959pub(crate) const PARALLEL_MIN_ROWS: usize = 100_000;
960
961/// v7.39 — diagnostic counter: how many aggregate scans took the
962/// sharded path (read by benches to ground-truth activation).
963/// v7.39 (round 740) — matview delta ground-truth counters (the r735
964/// lesson: a green content pin cannot distinguish "delta applied" from
965/// "silently fell back to full"; these can).
966pub static MATVIEW_FANOUT_BUFFERED: core::sync::atomic::AtomicU64 =
967 core::sync::atomic::AtomicU64::new(0);
968pub static MATVIEW_DELTA_APPLIED: core::sync::atomic::AtomicU64 =
969 core::sync::atomic::AtomicU64::new(0);
970pub static MATVIEW_DELTA_BAILED: core::sync::atomic::AtomicU64 =
971 core::sync::atomic::AtomicU64::new(0);
972pub static PARALLEL_AGG_FIRED: core::sync::atomic::AtomicU64 =
973 core::sync::atomic::AtomicU64::new(0);
974
975// The engine carries several independent session/capture flags (dialect,
976// FK-checks, meta-view materialisation, redo capture); they're orthogonal
977// switches, not a state enum begging to be modelled.
978#[allow(clippy::struct_excessive_bools)]
979#[derive(Debug, Default)]
980pub struct Engine {
981 /// v7.39 (parallel-agg P0) — see [`ParallelRunner`].
982 pub(crate) parallel_runner: ParallelRunnerSlot,
983 /// Committed catalog — what survives `Engine::snapshot()` and what
984 /// outside-TX `SELECT`s read.
985 catalog: Catalog,
986 /// Active TX slots, keyed by `TxId`. Empty when no TX is in flight.
987 /// v4.41.1 runtime invariant: at most one entry (single-writer
988 /// model unchanged). v4.42 will let dispatch hold multiple entries
989 /// concurrently for group commit + engine MVCC.
990 tx_catalogs: BTreeMap<TxId, TxState>,
991 /// v7.39 (round 552) — the COMMIT SEQUENCE at which each table was
992 /// last written. Commit order, not begin order: a transaction that
993 /// began first can commit last, so the writer version allocated at
994 /// BEGIN cannot answer "did this change after I read it".
995 table_last_commit: BTreeMap<String, u64>,
996 /// Monotonic, bumped once per successful COMMIT.
997 commit_seq: u64,
998 /// Which slot the next exec_* call should mutate. Set by
999 /// `execute_in(sql, tx_id)` at the entry point; legacy `execute(sql)`
1000 /// sets it to `IMPLICIT_TX`. None when no TX is in flight (read /
1001 /// write goes straight against `catalog`).
1002 current_tx: Option<TxId>,
1003 /// Monotonic counter for `alloc_tx_id`. Starts at 1 — slot 0 is
1004 /// reserved for `IMPLICIT_TX`.
1005 next_tx_id: u64,
1006 /// v7.37.15 (Phase C) — versions allocated by in-flight
1007 /// writers. Snapshot construction folds this into the
1008 /// `Snapshot.in_progress` set so concurrent readers (or readers
1009 /// inside an older snapshot's REPEATABLE READ) don't see
1010 /// uncommitted writes.
1011 ///
1012 /// SPG's single-writer invariant means at most one writer
1013 /// version sits here at any moment (the one currently
1014 /// executing inside the engine write lock). The set survives
1015 /// engine clones because tx-commit removes versions before the
1016 /// `Engine::snapshot_data` returns, so a snapshot taken after
1017 /// commit observes an empty set.
1018 ///
1019 /// `BTreeSet` (not Vec) so iteration is sorted — Snapshot
1020 /// constructor expects the input sorted for binary-search
1021 /// `contains` correctness.
1022 active_writer_versions: BTreeSet<u64>,
1023 /// v7.37.15 (Phase C.2) — writer versions that ABORTED (rolled
1024 /// back). The engine-side visibility oracle ([`Self::xact_status`])
1025 /// consults this to report `Aborted` for a version that left the
1026 /// in-flight set via rollback rather than commit — the third state
1027 /// the abort-aware [`spg_storage::snapshot::Snapshot::visible_with_status`]
1028 /// needs once Phase C.3's in-place writes leave aborted stamps in
1029 /// place. Pruned below `oldest_active` by vacuum (Phase D); until
1030 /// then it grows only with rolled-back transactions (a never-die
1031 /// follow-up, not a commit-path leak).
1032 aborted_versions: BTreeSet<u64>,
1033 /// v7.37.15 (Phase C.4) — row-level lock table keyed on stable
1034 /// `(RelId, RowId)`. The in-place write path (C.3) acquires a
1035 /// tuple lock before stamping xmax; `exec_commit` / `exec_rollback`
1036 /// release the whole transaction's locks at end. No writer acquires
1037 /// yet at this commit — the field + delegating methods are the
1038 /// plumbing the write path consumes next. Rides on `Engine` like
1039 /// `active_writer_versions`; the sharded lock-free manager is C.5.
1040 locks: crate::locks::LockTable,
1041 /// v7.37.15 (Phase C.3) — kill switch for the in-place MVCC write
1042 /// path. `false` (default) = legacy physical semantics (DELETE
1043 /// physically removes the row, UPDATE replaces in place). `true` =
1044 /// the C.3 write path (DELETE tombstones via `mark_row_deleted`,
1045 /// UPDATE tombstones the old version + appends the new one, both
1046 /// keeping dead versions physically present for the now-uniformly-
1047 /// gated readers until vacuum reclaims them). `no_std` engine can't
1048 /// read env; the host (spg-server / spg-embedded) reads
1049 /// `SPG_MVCC_INPLACE` and calls [`Self::set_mvcc_inplace`]. Off
1050 /// until the write path + PG18 differential tests are proven.
1051 mvcc_inplace: bool,
1052 /// v7.37.16 — threshold-triggered synchronous vacuum at DML statement
1053 /// exit (autovacuum-lite; see .claude/state/autovacuum-design.md).
1054 /// Default ON; hosts may disable via `SPG_AUTOVACUUM=0`.
1055 autovacuum: bool,
1056 /// v7.39 (round 173) — whether the statement-exit trigger runs the
1057 /// vacuum **inline**. Default ON (embedded: single-threaded host,
1058 /// the statement path is the only place work can happen). A host
1059 /// with a background autovacuum worker (spg-server) flips this off
1060 /// and drives [`Self::autovacuum_tick`] from its own thread instead
1061 /// — PG's shape, where autovacuum never runs inside a client
1062 /// statement. Only meaningful while `autovacuum` itself is on.
1063 autovacuum_inline: bool,
1064 /// v7.37.15 (Phase C) — TxId → writer version registry. When
1065 /// `exec_begin` opens an explicit transaction it allocates a
1066 /// fresh writer version (via [`Self::begin_writer_version`])
1067 /// and stashes the mapping here so the matching `exec_commit`
1068 /// / `exec_rollback` can call
1069 /// [`Self::commit_writer_version`] on the right entry. Empty
1070 /// when no explicit transactions are open.
1071 /// v7.39 (round 295, E3 Phase 1b) — rows a `SKIP LOCKED` pass found
1072 /// held by another transaction. Set by the locking pre-pass and
1073 /// consulted by the base scan, which is READ-only here: the locks
1074 /// themselves were taken under `&mut self` in the pre-pass, so the
1075 /// `&self` scan never mutates the lock table. (A `RefCell` there
1076 /// would cost `Engine: Sync`, which the server's `RwLock<Engine>`
1077 /// needs — see the RFC's §5.6.)
1078 pub(crate) lock_skip_rows: Option<(String, alloc::collections::BTreeSet<usize>)>,
1079 tx_writer_versions: BTreeMap<TxId, u64>,
1080 /// v7.37.15 (Epic W slice 2) — the current statement's autocommit
1081 /// writer version, memoized. In autocommit
1082 /// [`Self::writer_version_for_current_stmt`] mints a fresh version
1083 /// via `next_version()` (a `fetch_add`), so calling it a second
1084 /// time — e.g. when the redo drain post-stamps `RowChange`s —
1085 /// would allocate a *different* number than the writes actually
1086 /// used. Memoizing the first allocation for the duration of one
1087 /// statement makes the drain stamp read back the exact version the
1088 /// rows were written with, without advancing the counter twice.
1089 /// `None` outside a statement / before the first fetch; saved and
1090 /// reset per `execute_in_with_cancel` so it never leaks across
1091 /// statements. Explicit transactions bypass this (their version is
1092 /// the deterministic `tx_writer_versions` entry).
1093 stmt_writer_version: Option<u64>,
1094 /// v7.22 (round-13 T3) — session string-literal dialect. `false`
1095 /// (default) = PG semantics (backslash literal, `''` escape);
1096 /// `true` = MySQL semantics (`\'` etc.). Flipped by the
1097 /// deterministic session signals each dump emits: `SET sql_mode`
1098 /// (only MySQL clients/dumps send it) turns it on,
1099 /// `SET standard_conforming_strings = on` (every pg_dump
1100 /// preamble) turns it off. The plan cache is cleared on every
1101 /// flip — the same SQL text lexes differently per dialect.
1102 backslash_escapes: bool,
1103 /// v7.39 (round 470) — see [`SessionBag::mysql_strict`].
1104 mysql_strict: bool,
1105 /// v7.38.18 (C12) — see [`SessionBag::mysql_warnings`]. Lives on
1106 /// the Engine like every other live-session field and swaps with
1107 /// the bag, because the server runs ONE Engine for every
1108 /// connection and per-connection state that does not swap leaks
1109 /// across them.
1110 mysql_warnings: Vec<MysqlWarning>,
1111
1112 /// v7.38.18 (C12) — the diagnostics area of the statement that is
1113 /// running RIGHT NOW, which is not the one a read returns. MySQL 9
1114 /// answers `SELECT @@warning_count` with the PREVIOUS statement's
1115 /// count and only then replaces the visible set with its own (a
1116 /// second read in a row therefore answers 0). Producers fill this;
1117 /// [`Engine::dispatch_stmt_inner`] publishes it into
1118 /// `mysql_warnings` when the statement ends. Never live across
1119 /// statements, so it is not part of [`SessionBag`].
1120 mysql_stmt_warnings: Vec<MysqlWarning>,
1121 /// v7.39 (round 306) — the live session's open large-object
1122 /// descriptors, swapped in and out with the rest of its bag.
1123 pub(crate) lo_descriptors: BTreeMap<i32, LargeObjectDescriptor>,
1124 pub(crate) lo_next_fd: i32,
1125 /// v7.37.17 — name of the sequence most recently advanced by
1126 /// nextval() in this Engine (session). Backs PG's lastval().
1127 /// None until the first nextval; PG errors in that state.
1128 last_sequence_used: Option<String>,
1129 /// v7.39 (round 499) — per-session `currval` values; see
1130 /// [`SessionBag::seq_currvals`].
1131 seq_currvals: alloc::collections::BTreeMap<String, i64>,
1132 /// v7.39 (round 277) — SQL-level prepared statements, session
1133 /// scoped exactly as in PG. Keyed by name; each entry keeps the
1134 /// parsed body (placeholders intact), the declared parameter type
1135 /// names and the statement text `pg_prepared_statements` reports.
1136 prepared_statements: alloc::collections::BTreeMap<String, PreparedSqlStatement>,
1137 /// v7.39 (round 279) — which connection's state is currently
1138 /// installed in the fields above. 0 is the embedded / default
1139 /// session.
1140 current_session: u32,
1141 /// Parked state for every OTHER connection.
1142 sessions: BTreeMap<u32, SessionBag>,
1143 /// v7.39 (round 279) — advisory locks, held ACROSS sessions and so
1144 /// deliberately NOT part of the swapped bag: the whole purpose of
1145 /// an advisory lock is to be visible to the other connection.
1146 /// key → (owning session, re-entrant depth). PG allows the same
1147 /// session to take a lock it already holds.
1148 advisory_locks: BTreeMap<i64, (u32, u32)>,
1149 /// Optional wall clock used to satisfy `NOW()` / `CURRENT_TIMESTAMP`
1150 /// / `CURRENT_DATE`. Set by the host environment.
1151 clock: Option<ClockFn>,
1152 /// v4.1 cryptographic RNG for per-user password salt. Set by the
1153 /// host. `None` means SQL-driven `CREATE USER` uses a
1154 /// deterministic fallback — see `SaltFn`.
1155 salt_fn: Option<SaltFn>,
1156 /// v4.2 per-query row cap. `None` = unlimited. When set, a
1157 /// SELECT that materialises more than `n` rows returns
1158 /// `EngineError::RowLimitExceeded`. Enforced before the result
1159 /// is shaped into wire frames so a runaway scan can't blow the
1160 /// server's heap.
1161 max_query_rows: Option<usize>,
1162 /// v7.30.3 (mailrs round-26) per-query byte cap on join/filter
1163 /// materialisation. `None` = unlimited. Approximate net
1164 /// accounting (Value heap payloads + per-cell enum overhead)
1165 /// charged at every point the join pipeline clones rows;
1166 /// crossing the cap raises `EngineError::QueryBytesExceeded`
1167 /// instead of pressuring the host into reclaim livelock. The
1168 /// host wires this to `SPG_MAX_QUERY_BYTES` (embed defaults it
1169 /// ON; the server keeps its allocator-precise budget as the
1170 /// outer layer).
1171 pub(crate) max_query_bytes: Option<usize>,
1172 /// v7.39 (round 786, T35 Phase A) — host factory for spill runs.
1173 /// `None` (the default, and every embedded caller that has not opted
1174 /// in) keeps today's behaviour exactly: a sort that outgrows
1175 /// `max_query_bytes` still refuses rather than spilling.
1176 pub(crate) temp_run_factory: Option<crate::TempRunFactory>,
1177 /// v4.1 RBAC user table. Empty means "no RBAC configured yet" —
1178 /// the server decides what that means at the auth boundary
1179 /// (open mode vs legacy single-password mode). User CRUD goes
1180 /// through `create_user`/`drop_user`/`verify_user`; persistence
1181 /// rides the snapshot envelope alongside the catalog.
1182 pub(crate) users: UserStore,
1183 /// v6.1.2 logical-replication publication catalog. Empty until
1184 /// `CREATE PUBLICATION` runs. Persistence rides the v3 envelope
1185 /// trailer (see `build_envelope`).
1186 publications: publications::Publications,
1187 /// v6.1.4 logical-replication subscription catalog. Empty until
1188 /// `CREATE SUBSCRIPTION` runs. Persistence rides the v4 envelope
1189 /// trailer.
1190 subscriptions: subscriptions::Subscriptions,
1191 /// v6.2.0 — per-column statistics for the cost-based optimizer.
1192 /// Populated by `ANALYZE`; queried via `spg_statistic` virtual
1193 /// table. Persistence rides the v5 envelope trailer.
1194 statistics: statistics::Statistics,
1195 /// v6.3.0 — engine-level plan cache. Caches the post-`prepare()`
1196 /// `Statement` keyed on SQL text. In-memory only — does NOT ride
1197 /// the snapshot envelope (rebuilt on demand after restart).
1198 plan_cache: plan_cache::PlanCache,
1199 /// v6.5.1 — per-distinct-SQL execution stats. In-memory only,
1200 /// surfaced via `spg_stat_query` virtual table. Updated by the
1201 /// `execute_*` paths after a successful execute.
1202 query_stats: query_stats::QueryStats,
1203 /// v6.5.2 — connection-state provider callback. spg-server
1204 /// registers a function at startup that snapshots its
1205 /// per-pgwire-connection registry into `ActivityRow`s; engine
1206 /// reads through it on every `SELECT * FROM spg_stat_activity`.
1207 /// `None` ⇒ no-data (returns empty rows; matches the no_std
1208 /// embedded callers that don't run pgwire).
1209 activity_provider: Option<ActivityProvider>,
1210 /// v6.5.3 — audit-chain provider + verifier. Same pattern as
1211 /// activity_provider: spg-server registers both at startup;
1212 /// engine reads through on `SELECT * FROM spg_audit_chain` and
1213 /// `SELECT * FROM spg_audit_verify`. `None` ⇒ no-data.
1214 audit_chain_provider: Option<AuditChainProvider>,
1215 audit_verifier: Option<AuditVerifier>,
1216 /// v6.5.6 — slow-query log threshold in microseconds. When set,
1217 /// every successful execute whose elapsed exceeds the threshold
1218 /// gets fed to the registered slow-query log callback (so
1219 /// spg-server can emit a structured log line). Default `None`
1220 /// = no slow-query logging.
1221 slow_query_threshold_us: Option<u64>,
1222 slow_query_logger: Option<SlowQueryLogger>,
1223 /// v7.12.1 — session parameters set via `SET <name> = <value>`.
1224 /// Only `default_text_search_config` is consumed by the engine
1225 /// today (the FTS function dispatcher reads it when
1226 /// `to_tsvector(text)` is called without an explicit config).
1227 /// All other names are accepted + recorded so PG-dump output
1228 /// loads, but have no behavioural effect.
1229 pub(crate) session_params: BTreeMap<String, String>,
1230 /// v7.39 (round 218) — open server-side cursors (DECLARE … CURSOR),
1231 /// keyed by name. Materialized at DECLARE (INSENSITIVE semantics —
1232 /// PG's only actual behaviour too); FETCH / MOVE walk the stored rows.
1233 /// Lifecycle: created only inside a transaction; COMMIT closes
1234 /// non-HOLD cursors and marks WITH HOLD ones held; ROLLBACK closes
1235 /// everything not already held by an earlier commit. Never serialized.
1236 /// Session-scoped in PG; SPG stores them engine-wide (the same
1237 /// process-level session-state architecture wall as `session_params`).
1238 pub(crate) cursors: BTreeMap<String, cursor::OpenCursor>,
1239 /// v7.39 (round 347, M2) — the current session's LAST_INSERT_ID().
1240 /// Swapped with [`SessionBag`] like every other per-connection slot.
1241 /// An atomic because `LAST_INSERT_ID(expr)` SETS it while evaluation
1242 /// holds only `&Engine` — and `Engine` must stay `Sync`, which a
1243 /// `Cell` would have taken away (spg-embedded-tokio shares one across
1244 /// tasks; clippy caught it there before the tests did).
1245 pub(crate) last_insert_id: core::sync::atomic::AtomicI64,
1246 /// v7.39 (round 426) — the current session's ROW_COUNT(). Swapped
1247 /// with [`SessionBag`] like every other per-connection slot. A plain
1248 /// i64: unlike LAST_INSERT_ID it is only ever WRITTEN from the
1249 /// statement driver, which holds `&mut Engine`.
1250 pub(crate) row_count: i64,
1251 /// v7.39 (round 430) — this session's MySQL USER variables.
1252 /// Swapped with [`SessionBag`] like every other per-connection slot.
1253 pub(crate) user_vars: BTreeMap<String, spg_storage::Value<'static>>,
1254 /// v7.39 (round 436) — the logical names of this session's TEMPORARY
1255 /// tables. Swapped with [`SessionBag`]; see `session_temp_name`.
1256 pub(crate) temp_tables: BTreeSet<String>,
1257 pub(crate) temp_sequences: BTreeSet<String>,
1258 pub(crate) temp_views: BTreeSet<String>,
1259 /// v7.39 (round 222) — channels this session LISTENs on. Engine-wide
1260 /// (the same process-level session-state architecture wall as
1261 /// `session_params`). Never serialized.
1262 pub(crate) listen_channels: BTreeSet<String>,
1263 /// v7.39 (round 222) — NOTIFYs raised inside the current transaction,
1264 /// held until COMMIT (PG: transactional delivery, deduplicated within
1265 /// the tx); dropped at ROLLBACK.
1266 pub(crate) tx_pending_notifies: Vec<(String, String)>,
1267 /// v7.39 (round 222) — committed notifications on LISTENed channels,
1268 /// awaiting a drain by the wire layer ('A' NotificationResponse) or an
1269 /// embedded caller ([`Engine::take_notifications`]).
1270 pub(crate) delivered_notifies: Vec<(String, String)>,
1271 /// v7.39 (read01 round 46) — NOTICEs raised by the statement now
1272 /// executing. PG emits a NoticeResponse whenever an `IF EXISTS` /
1273 /// `IF NOT EXISTS` clause makes it skip work ("table \"t\" does not
1274 /// exist, skipping"). The engine appends the PG-worded text here;
1275 /// the caller drains it with [`Engine::take_notices`] after each
1276 /// statement (pgwire turns each into an 'N' message, embedded
1277 /// callers can ignore or surface them). Cleared at the start of
1278 /// every statement so a notice never leaks into the next one.
1279 pending_notices: Vec<Notice>,
1280 /// v7.38 (read01 P3.12) — cumulative row-write counters feeding
1281 /// `pg_stat_database` (database-wide `tup_inserted` / `tup_updated` /
1282 /// `tup_deleted`). Bumped by the affected-row count of each successful
1283 /// INSERT / UPDATE / DELETE statement. Per-Engine (so tests stay
1284 /// isolated); on the server's shared engine they read as the
1285 /// since-start database totals PG reports.
1286 /// v7.39 (pg_stat knife A) — committed / rolled-back transaction
1287 /// counters for pg_stat_database. Atomics so the read-only
1288 /// autocommit path (&self) can count its implicit commit, matching
1289 /// PG (every successful statement outside a tx block is one
1290 /// xact_commit — SELECTs included).
1291 /// v7.37 (round 884) — what sorts have spilled in this process, for
1292 /// `pg_stat_database` and for EXPLAIN ANALYZE's `Sort Method`.
1293 pub(crate) spill_stats: crate::tempstore::SpillStats,
1294 pub(crate) xact_commit: core::sync::atomic::AtomicU64,
1295 pub(crate) xact_rollback: core::sync::atomic::AtomicU64,
1296 /// v7.39 (pg_stat knife A) — host-injected live backend count for
1297 /// pg_stat_database.numbackends (ClockFn-style fn slot; the server
1298 /// wires its connection registry, embedded stays None -> 1).
1299 pub(crate) backend_count_fn: Option<BackendCountFn>,
1300 pub(crate) backend_pid_fn: Option<BackendPidFn>,
1301 /// v7.39 (round 476) — see [`WalLsnFn`].
1302 pub(crate) wal_lsn_fn: Option<WalLsnFn>,
1303 /// v7.39 (round 318, V51) — host connection-control hook. See
1304 /// [`BackendSignalFn`].
1305 pub(crate) backend_signal_fn: Option<BackendSignalFn>,
1306 /// v7.39 (tz epic) — injected IANA timezone lookups; None on a
1307 /// host without zoneinfo (named zones then fail to SET, honestly).
1308 pub(crate) tz_offset_fn: Option<TzOffsetFn>,
1309 pub(crate) tz_localize_fn: Option<TzLocalizeFn>,
1310 pub(crate) tz_canon_fn: Option<TzCanonFn>,
1311 pub(crate) tz_abbrev_fn: Option<TzAbbrevFn>,
1312 /// v7.39 (round 502) — see [`TzAllFn`].
1313 pub(crate) tz_all_fn: Option<TzAllFn>,
1314 pub(crate) stat_tup_inserted: u64,
1315 pub(crate) stat_tup_updated: u64,
1316 pub(crate) stat_tup_deleted: u64,
1317 /// v7.39 (round 192) — per-table DML counters for
1318 /// pg_stat_user_tables (n_tup_ins / n_tup_upd / n_tup_del).
1319 /// Engine-side and NON-transactional, like PG's stats collector:
1320 /// a rolled-back INSERT still counts, and a tx's counts don't
1321 /// ride the shadow catalog (the RC rebase rebuilt shadow tables
1322 /// from the committed base, silently dropping any counter bumped
1323 /// on the shadow — the r192 probe's tx-wrapped inserts read 0).
1324 /// Keyed by table name; DROP TABLE clears, RENAME re-keys.
1325 pub(crate) table_write_stats: alloc::collections::BTreeMap<String, (u64, u64, u64)>,
1326 /// v7.39 (round 196) — bumped after every completed statement that
1327 /// ran OUTSIDE a transaction block (any autocommit statement, plus
1328 /// COMMIT itself via the post-statement check). An open tx whose
1329 /// `rebased_at_epoch` equals this value knows the committed base
1330 /// hasn't moved and skips the per-statement RC rebase (whose
1331 /// write-set extraction full-scans every touched table).
1332 /// Over-approximation is deliberate: read-only statements bump it
1333 /// too, which only costs an extra (correct) rebase.
1334 pub(crate) commit_epoch: u64,
1335 /// v7.38 (read01 P3.19) — `SET LOCAL` undo log for the current
1336 /// transaction. Each entry is `(param_name, prior_value)` captured
1337 /// just before a `SET LOCAL` overwrote it (`None` = the param had no
1338 /// session value, so restoring means removing it). Replayed in
1339 /// reverse at COMMIT / ROLLBACK to revert transaction-local settings;
1340 /// `savepoint_guc_marks` records the stack depth at each open
1341 /// savepoint so `ROLLBACK TO` can unwind just the later ones.
1342 pub(crate) local_guc_saves: Vec<(String, Option<String>)>,
1343 /// v7.39 (GUC knife 3) — parsed DateStyle / IntervalStyle /
1344 /// extra_float_digits, kept in lockstep with `session_params` so
1345 /// renderers don't re-parse GUC text per cell.
1346 pub(crate) render_style: crate::eval::RenderStyle,
1347 pub(crate) savepoint_guc_marks: Vec<(String, usize)>,
1348 /// v7.12.7 — depth counter for trigger-emitted embedded SQL.
1349 /// Each time the engine executes a `DeferredEmbeddedStmt` it
1350 /// increments this; the recursive `execute_stmt_with_cancel`
1351 /// inside that path checks against [`MAX_TRIGGER_RECURSION`]
1352 /// to bound runaway cascades (trigger A's UPDATE on table B
1353 /// fires trigger B which UPDATEs table A which fires trigger
1354 /// A again…). Reset to 0 once the original DML returns.
1355 trigger_recursion_depth: u32,
1356 /// v7.39 (round 140) — set while a DELETE / UPDATE is being re-run by the
1357 /// DO ALSO rule wrapper so the wrapper's inner call does not re-enter the
1358 /// rule-rewrite path (which would recurse forever). INSERT captures its
1359 /// post-image rows directly and needs no such guard.
1360 rule_rewrite_active: bool,
1361 /// v7.14.0 — when `SET FOREIGN_KEY_CHECKS=0` is in effect
1362 /// (mysqldump preamble), the FK existence + arity check at
1363 /// CREATE TABLE time is deferred. FKs referencing a
1364 /// not-yet-existing parent land in `pending_foreign_keys`
1365 /// keyed by child table; `SET FOREIGN_KEY_CHECKS=1` drains
1366 /// the queue and resolves each FK against the now-complete
1367 /// catalog. Empty by default; the queue is drained on every
1368 /// `RESET ALL` too.
1369 foreign_key_checks: bool,
1370 /// v7.16.2 — true on the temp Engine an outer
1371 /// `exec_select_with_meta_views` builds, telling that
1372 /// temp engine "stop short-circuiting into the meta-view
1373 /// path — your catalog already has the materialised
1374 /// tables; just run the regular SELECT." Without this we'd
1375 /// infinite-loop since the meta-view name (e.g.
1376 /// `__spg_info_columns`) still triggers
1377 /// `select_references_meta_view`.
1378 meta_views_materialised: bool,
1379 pending_foreign_keys: Vec<(alloc::string::String, spg_sql::ast::ForeignKeyConstraint)>,
1380 /// v7.38 元机制 D — frozen snapshot of `SPG_TEST_*` env vars. Read
1381 /// once at construction (`with_env_cfg`) and queried on hot paths
1382 /// via `engine.env_cfg().<field>`. Production builds keep this at
1383 /// `EnvConfig::default()`, so the optimiser can const-fold every
1384 /// `if env_cfg.<field>` gate. See `testkit::env_config` + the
1385 /// `xtests/sigil/test-mode-gucs.md` index.
1386 env_cfg: testkit::EnvConfig,
1387 /// v7.38 P0 元机制 A — per-engine `injection_points` attach
1388 /// table. Only exists when the crate is built with the
1389 /// `injection-points` feature; release builds carry no field.
1390 /// Pushed onto the thread-local stack by
1391 /// `enter_injection_scope()` so the `injection_point!()` macro
1392 /// can find it from anywhere in the executor without rewiring
1393 /// every signature. See
1394 /// `crates/spg-engine/src/testkit/injection.rs`.
1395 #[cfg(feature = "injection-points")]
1396 injection_store: alloc::sync::Arc<crate::testkit::injection::InjectionStore>,
1397 /// v7.34 (crash-recovery P0 #2) — row-level redo capture. When the
1398 /// embedding layer turns this on (persistence enabled), each mutating
1399 /// `execute` records the physical [`RowChange`]s it applied; the
1400 /// engine drains them into `last_redo` on success, and the embedded
1401 /// layer reads them via [`Engine::take_redo`] to write the WAL in
1402 /// place of the SQL text. Off (default) = zero capture overhead.
1403 redo_capture: bool,
1404 /// Redo captured by the most recent successful mutating `execute`,
1405 /// awaiting drain by the embedding layer. Cleared on each capture.
1406 last_redo: Vec<RowChange>,
1407 /// v7.39 (round 735, S14/B3) — per-table change sequence, bumped on
1408 /// every write entry (INSERT / UPDATE / DELETE / TRUNCATE / COPY /
1409 /// table-shape DDL). In-memory only: after a restart the map is
1410 /// empty, every watermark comparison misses, and the next REFRESH
1411 /// is a full one — stale-view-safe by construction. A rolled-back
1412 /// transaction's bump stays too, which can only cause an EXTRA full
1413 /// refresh, never a wrong no-op.
1414 table_change_seq: alloc::collections::BTreeMap<String, u64>,
1415 /// v7.39 (round 735, S14/B3) — per-materialized-view refresh
1416 /// watermark: the (table, change-seq) pairs its last full refresh
1417 /// saw. When every dependency's seq is unchanged, REFRESH is an
1418 /// O(1) no-op — an incremental-maintenance first step PG does not
1419 /// have (its REFRESH always recomputes).
1420 matview_refresh_watermark: alloc::collections::BTreeMap<String, Vec<(String, u64)>>,
1421 /// v7.39 (round 736, S14/B3 knife 2) — delta-maintainable
1422 /// materialized views: mv name -> its single base table. Registered
1423 /// at CREATE MATERIALIZED VIEW / full REFRESH when the body is a
1424 /// single-stored-table pure projection (no aggregates / joins /
1425 /// CTEs / subqueries / DISTINCT / ORDER / LIMIT / windows / SRFs).
1426 matview_maintainable: alloc::collections::BTreeMap<String, String>,
1427 /// Buffered base-table row changes per maintainable view, fanned
1428 /// out from the statement redo drain. Capped (see
1429 /// `MATVIEW_DELTA_CEILING`); an overflowed view falls back to a
1430 /// full refresh — never-die, never-stale.
1431 matview_delta_buf: alloc::collections::BTreeMap<String, Vec<RowChange>>,
1432 matview_delta_overflow: alloc::collections::BTreeSet<String>,
1433 /// v7.39 (round 738, S14/B3 knife 3) — per-view row map: expected
1434 /// PHYSICAL length of the view's backing table, plus base-row
1435 /// RowId -> view row position. Built only by the maintainable full
1436 /// refresh's internal scan (the SQL path cannot see rowids), and
1437 /// consulted by the delete/tombstone delta arms. In-memory: restart
1438 /// or any length mismatch (a vacuum moved rows) -> full refresh.
1439 matview_row_map:
1440 alloc::collections::BTreeMap<String, (usize, alloc::collections::BTreeMap<u64, usize>)>,
1441 /// v7.38 轴 4 — currently-selected SQL isolation level. Set by
1442 /// `SET TRANSACTION ISOLATION LEVEL …`; read by
1443 /// `SHOW transaction_isolation`. v7.37.8 implements the
1444 /// SQL surface; actual semantic differentiation (REPEATABLE READ
1445 /// snapshot / SERIALIZABLE SSI) lands in a separate train.
1446 pub(crate) current_isolation_level: spg_sql::ast::IsolationLevel,
1447}
1448
1449/// v7.12.7 — hard cap on nested trigger-emitted embedded SQL
1450/// fires. 16 deep is well past anything a normal trigger graph
1451/// uses while still preventing infinite-loop wedging.
1452const MAX_TRIGGER_RECURSION: u32 = 16;
1453
1454/// v6.5.6 — callback signature for slow-query log emission. Called
1455/// with `(sql, elapsed_us)` once per successful execute that crosses
1456/// the threshold.
1457pub type SlowQueryLogger = fn(&str, u64);
1458
1459/// v6.5.2 — one row of `spg_stat_activity`. Engine-public so
1460/// spg-server can construct rows without re-exporting internal
1461/// dispatch types.
1462#[derive(Debug, Clone)]
1463pub struct ActivityRow {
1464 pub pid: u32,
1465 pub user: String,
1466 /// v7.39 (round 319, V52) — the peer's IP, empty when the connection
1467 /// has no TCP peer (PG reports NULL there).
1468 pub client_addr: String,
1469 /// v7.39 (round 319, V52) — the peer's port. PG reports **-1**, not
1470 /// NULL, for a connection with no TCP port; measured on PG 18.4.
1471 pub client_port: i32,
1472 /// v7.39 (round 319, V52) — the database this connection named. Empty
1473 /// when it named none; both `pg_stat_activity.datname` and
1474 /// `SHOW PROCESSLIST.db` report that as NULL.
1475 pub database: String,
1476 pub started_at_us: i64,
1477 pub current_sql: String,
1478 /// v7.37.14 (B6.3) — PG-style wait-event categorisation
1479 /// ("Lock", "LWLock", "IPC", "IO", "Timeout", "Client",
1480 /// "BufferPin", "Extension", ""). Empty string means idle.
1481 /// Pair with `wait_event` to identify "what specifically is
1482 /// the backend waiting on" the same way PG does.
1483 pub wait_event_type: String,
1484 pub wait_event: String,
1485 pub elapsed_us: i64,
1486 pub in_transaction: bool,
1487 /// v7.17 Phase 2.4 — startup-param `application_name` (or the
1488 /// last value the client sent via `SET application_name = '...'`).
1489 /// Empty when the client never declared one.
1490 pub application_name: String,
1491 /// v7.39 (round 474) — PG's `backend_type`: `client backend` for a
1492 /// connection, or the worker's own name for a background process.
1493 ///
1494 /// pg_stat_activity used to hardcode `client backend`, so SPG's own
1495 /// background workers — the ones that hold the engine write lock and
1496 /// are exactly what an operator is looking for when a statement
1497 /// stalls — did not appear at all. PG18 lists eight of them beside
1498 /// the single client backend on an idle server.
1499 pub backend_type: String,
1500}
1501
1502impl ActivityRow {
1503 /// The `backend_type` PG gives a background process: no database, no
1504 /// user, no query, and a state PG reports as NULL.
1505 #[must_use]
1506 pub fn background(pid: u32, backend_type: &str) -> Self {
1507 Self {
1508 pid,
1509 user: String::new(),
1510 client_addr: String::new(),
1511 client_port: -1,
1512 database: String::new(),
1513 started_at_us: 0,
1514 current_sql: String::new(),
1515 wait_event_type: String::new(),
1516 wait_event: String::new(),
1517 elapsed_us: 0,
1518 in_transaction: false,
1519 application_name: String::new(),
1520 backend_type: backend_type.into(),
1521 }
1522 }
1523}
1524
1525/// v6.5.2 — provider callback type. Fresh snapshot returned each
1526/// call; engine doesn't cache the slice.
1527pub type ActivityProvider = fn() -> Vec<ActivityRow>;
1528
1529/// v7.39 (round 318, V41) — how loud a diagnostic the statement raised is.
1530/// PG distinguishes them on the wire (`S`/`V` fields of NoticeResponse) and
1531/// clients act on it: psql prints `WARNING:` in a different colour, and
1532/// several drivers surface warnings to the application while dropping
1533/// notices. Emitting everything as NOTICE loses that.
1534#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1535pub enum NoticeSeverity {
1536 Notice,
1537 Warning,
1538 /// v7.39 (round 757, F31-B3) — `RAISE INFO`. PG sends INFO to the
1539 /// client ALWAYS, regardless of `client_min_messages`.
1540 Info,
1541}
1542
1543impl NoticeSeverity {
1544 /// The non-localized severity string PG puts in the `V` field.
1545 #[must_use]
1546 pub const fn as_pg_str(self) -> &'static str {
1547 match self {
1548 Self::Notice => "NOTICE",
1549 Self::Warning => "WARNING",
1550 Self::Info => "INFO",
1551 }
1552 }
1553}
1554
1555/// v7.39 (round 318, V41) — one diagnostic the statement raised, in PG's
1556/// exact wording minus the severity banner (the wire layer adds that).
1557#[derive(Debug, Clone)]
1558pub struct Notice {
1559 pub severity: NoticeSeverity,
1560 pub message: String,
1561}
1562
1563/// v6.5.3 — one row of `spg_audit_chain`. Engine-public so
1564/// spg-server can construct rows directly from `AuditEntry`.
1565#[derive(Debug, Clone)]
1566pub struct AuditRow {
1567 pub seq: i64,
1568 pub ts_ms: i64,
1569 pub prev_hash_hex: String,
1570 pub entry_hash_hex: String,
1571 pub sql: String,
1572}
1573
1574/// v6.5.3 — chain-table provider + verifier. spg-server registers
1575/// fn pointers that snapshot / verify the audit log. `verify`
1576/// returns `(verified_count, broken_at_seq)` — `broken_at_seq` is
1577/// `-1` on a clean chain.
1578pub type AuditChainProvider = fn() -> Vec<AuditRow>;
1579pub type AuditVerifier = fn() -> (i64, i64);
1580
1581impl Engine {
1582 pub fn new() -> Self {
1583 Self {
1584 catalog: Catalog::new(),
1585 parallel_runner: ParallelRunnerSlot::default(),
1586 tx_catalogs: BTreeMap::new(),
1587 table_last_commit: BTreeMap::new(),
1588 commit_seq: 0,
1589 current_tx: None,
1590 backslash_escapes: false,
1591 mysql_strict: true,
1592 mysql_warnings: Vec::new(),
1593 mysql_stmt_warnings: Vec::new(),
1594 lo_descriptors: BTreeMap::new(),
1595 lo_next_fd: 0,
1596 prepared_statements: alloc::collections::BTreeMap::new(),
1597 current_session: 0,
1598 sessions: BTreeMap::new(),
1599 advisory_locks: BTreeMap::new(),
1600 last_sequence_used: None,
1601 seq_currvals: alloc::collections::BTreeMap::new(),
1602 next_tx_id: 1,
1603 active_writer_versions: BTreeSet::new(),
1604 aborted_versions: BTreeSet::new(),
1605 locks: crate::locks::LockTable::new(),
1606 mvcc_inplace: !cfg!(feature = "mvcc-inplace-off"),
1607 autovacuum: true,
1608 autovacuum_inline: true,
1609 lock_skip_rows: None,
1610 tx_writer_versions: BTreeMap::new(),
1611 stmt_writer_version: None,
1612 clock: None,
1613 salt_fn: None,
1614 max_query_rows: None,
1615 max_query_bytes: None,
1616 temp_run_factory: None,
1617 users: UserStore::new(),
1618 publications: publications::Publications::new(),
1619 subscriptions: subscriptions::Subscriptions::new(),
1620 statistics: statistics::Statistics::new(),
1621 plan_cache: plan_cache::PlanCache::new(),
1622 query_stats: query_stats::QueryStats::new(),
1623 activity_provider: None,
1624 audit_chain_provider: None,
1625 audit_verifier: None,
1626 slow_query_threshold_us: None,
1627 slow_query_logger: None,
1628 session_params: BTreeMap::new(),
1629 cursors: BTreeMap::new(),
1630 last_insert_id: core::sync::atomic::AtomicI64::new(0),
1631 row_count: 0,
1632 user_vars: BTreeMap::new(),
1633 temp_tables: BTreeSet::new(),
1634 temp_sequences: BTreeSet::new(),
1635 temp_views: BTreeSet::new(),
1636 listen_channels: BTreeSet::new(),
1637 tx_pending_notifies: Vec::new(),
1638 delivered_notifies: Vec::new(),
1639 pending_notices: Vec::new(),
1640 spill_stats: crate::tempstore::SpillStats::default(),
1641 xact_commit: core::sync::atomic::AtomicU64::new(0),
1642 xact_rollback: core::sync::atomic::AtomicU64::new(0),
1643 backend_count_fn: None,
1644 backend_pid_fn: None,
1645 wal_lsn_fn: None,
1646 backend_signal_fn: None,
1647 tz_offset_fn: None,
1648 tz_localize_fn: None,
1649 tz_canon_fn: None,
1650 tz_abbrev_fn: None,
1651 tz_all_fn: None,
1652 stat_tup_inserted: 0,
1653 table_write_stats: alloc::collections::BTreeMap::new(),
1654 commit_epoch: 0,
1655 stat_tup_updated: 0,
1656 stat_tup_deleted: 0,
1657 local_guc_saves: Vec::new(),
1658 render_style: crate::eval::RenderStyle::default(),
1659 savepoint_guc_marks: Vec::new(),
1660 trigger_recursion_depth: 0,
1661 rule_rewrite_active: false,
1662 foreign_key_checks: true,
1663 meta_views_materialised: false,
1664 pending_foreign_keys: Vec::new(),
1665 env_cfg: testkit::EnvConfig::default(),
1666 #[cfg(feature = "injection-points")]
1667 injection_store: alloc::sync::Arc::new(
1668 crate::testkit::injection::InjectionStore::default(),
1669 ),
1670 redo_capture: false,
1671 current_isolation_level: spg_sql::ast::IsolationLevel::ReadCommitted,
1672 last_redo: Vec::new(),
1673 table_change_seq: alloc::collections::BTreeMap::new(),
1674 matview_refresh_watermark: alloc::collections::BTreeMap::new(),
1675 matview_maintainable: alloc::collections::BTreeMap::new(),
1676 matview_delta_buf: alloc::collections::BTreeMap::new(),
1677 matview_delta_overflow: alloc::collections::BTreeSet::new(),
1678 matview_row_map: alloc::collections::BTreeMap::new(),
1679 }
1680 }
1681
1682 /// v7.11.0 — clone the engine's committed catalog + read-time
1683 /// state into a frozen `CatalogSnapshot`. Cheap (`Catalog` is
1684 /// backed by `PersistentVec`; cloning is O(log n) per table).
1685 /// Subsequent writes to this engine are invisible to the
1686 /// snapshot; the snapshot is self-contained and can be moved
1687 /// to another thread for concurrent `execute_readonly_on_snapshot`
1688 /// calls. The basis for [`AsyncReadHandle`] in spg-embedded-tokio
1689 /// and any other read-fanout pattern.
1690 #[must_use]
1691 pub fn clone_snapshot(&self) -> CatalogSnapshot {
1692 CatalogSnapshot {
1693 catalog: self.active_catalog().clone(),
1694 statistics: self.statistics.clone(),
1695 clock: self.clock,
1696 max_query_rows: self.max_query_rows,
1697 }
1698 }
1699
1700 /// v7.39 (round 513) — does this role exist? `'x'::regrole` needs it,
1701 /// and roles live on the engine rather than the catalog.
1702 #[must_use]
1703 pub fn role_exists(&self, name: &str) -> bool {
1704 // v7.39 (round 696) — the SESSION's own identity, same class as the
1705 // `postgres` case below and missed by it. `current_user` reported
1706 // the connected name while this predicate denied it, so `SET ROLE
1707 // <me>` refused the role the session was already running as.
1708 if name == self.session_user() {
1709 return true;
1710 }
1711 // The engine's default identity exists even before any CREATE USER.
1712 //
1713 // v7.39 (round 652) — and so does `postgres`. `synth_pg_roles`
1714 // has always inserted it as the bootstrap superuser when no user
1715 // by that name was created, so this predicate and the catalogue
1716 // it is supposed to reflect disagreed: `pg_roles` listed
1717 // `postgres` while `'postgres'::regrole` said it did not exist.
1718 // Every pg_dump names it (`OWNER TO postgres`), so the ALTER
1719 // TABLE OWNER check added this round would have refused the one
1720 // role that appears in essentially every dump.
1721 self.effective_users().contains(name)
1722 || name.eq_ignore_ascii_case("admin")
1723 || name.eq_ignore_ascii_case("postgres")
1724 // 7.38.1 S5.2 — PG's predefined pg_* roles exist without
1725 // anyone creating them, and pg_dump's ACL section GRANTs
1726 // to `pg_database_owner` on every dump of `public`.
1727 || matches!(
1728 name.to_ascii_lowercase().as_str(),
1729 "pg_database_owner"
1730 | "pg_read_all_data"
1731 | "pg_write_all_data"
1732 | "pg_monitor"
1733 | "pg_read_all_settings"
1734 | "pg_read_all_stats"
1735 | "pg_stat_scan_tables"
1736 | "pg_signal_backend"
1737 | "pg_checkpoint"
1738 | "pg_maintain"
1739 | "pg_use_reserved_connections"
1740 | "pg_create_subscription"
1741 )
1742 }
1743
1744 /// v7.39 (round 520) — the role an oid names, as `pg_get_userbyid`
1745 /// reports it. The numbering is `synth_pg_roles`': base 10, one per
1746 /// user in catalog order.
1747 #[must_use]
1748 pub fn role_name_for_oid(&self, oid: i64) -> Option<String> {
1749 // Oid 10 is the bootstrap superuser, which `synth_pg_roles` always
1750 // publishes as `postgres`. Following the catalogue rather than the
1751 // session is the point: a join on `relowner = pg_roles.oid` and
1752 // `pg_get_userbyid(relowner)` have to name the same role.
1753 if oid == 10 {
1754 return Some(alloc::string::String::from("postgres"));
1755 }
1756 let idx = usize::try_from(oid - 11).ok()?;
1757 self.users
1758 .iter()
1759 .nth(idx)
1760 .map(|(n, _)| alloc::string::String::from(n))
1761 }
1762
1763 /// v7.37.15 (Phase B / C / E) — current per-row visibility
1764 /// snapshot for in-engine scans. Captures the live writer-
1765 /// version cursor + active-writer set; readers built from this
1766 /// Snapshot see committed state through the moment of capture
1767 /// and DO NOT observe uncommitted writes still inside
1768 /// `active_writer_versions`.
1769 ///
1770 /// Phase E: if there's an explicit transaction in flight under
1771 /// REPEATABLE READ or SERIALIZABLE isolation, returns the
1772 /// snapshot the tx cached at BEGIN time — every statement in
1773 /// the tx sees the same coherent prior-committed view. READ
1774 /// COMMITTED (the default) returns a fresh snapshot per call,
1775 /// matching PG's per-statement visibility semantics.
1776 ///
1777 /// `oldest_active = version` when no writer is in flight (== no
1778 /// dead row could still be observed); else == min of active
1779 /// versions (vacuum-floor).
1780 #[must_use]
1781 pub fn current_snapshot(&self) -> spg_storage::snapshot::Snapshot {
1782 // v7.39 (round 297, E3 Phase 1b) — carry the SKIP LOCKED
1783 // exclusions on the snapshot. Every row source threads a
1784 // snapshot through `is_row_visible`, so this is the one place
1785 // that cannot be routed around; adding the filter per scan site
1786 // missed the live path three times.
1787 let locked_out = self.lock_skip_rows.as_ref().and_then(|(t, set)| {
1788 self.active_catalog()
1789 .get(t)
1790 .map(|tbl| (tbl.rel_id(), set.clone()))
1791 });
1792 let mut snap = self.current_snapshot_inner();
1793 snap.locked_out = locked_out;
1794 snap
1795 }
1796
1797 fn current_snapshot_inner(&self) -> spg_storage::snapshot::Snapshot {
1798 // Phase E — if we're inside a RR/SER tx, return its
1799 // cached snapshot so the whole tx sees one frozen view.
1800 if let Some(tx_id) = self.current_tx
1801 && let Some(state) = self.tx_catalogs.get(&tx_id)
1802 && let Some(s) = state.cached_snapshot.as_ref()
1803 {
1804 return s.clone();
1805 }
1806 // v7.37.15 (Phase C.3, step 1) — carry the current tx's writer
1807 // version as the snapshot's `tx_id` so the visibility gate's
1808 // self-write branch (`visible` step 1) recognises rows this
1809 // transaction stamped (`xmin == v`, with `v` in
1810 // `active_writer_versions`). Without this the tx's own
1811 // uncommitted rows fall to the in-progress step and become
1812 // invisible to itself on the gated read paths. Autocommit reads
1813 // (no `tx_writer_versions` entry) keep `tx_id = 0`.
1814 let reader_tx_id = self
1815 .current_tx
1816 .and_then(|t| self.tx_writer_versions.get(&t).copied())
1817 .unwrap_or(0);
1818 let version = spg_storage::row_header::current_version();
1819 if self.active_writer_versions.is_empty() {
1820 // Hot path: no writer in flight. Snapshot::unbounded()
1821 // would also work, but pinning to the live cursor
1822 // means the snapshot's oldest_active is accurate
1823 // (= version) so subsequent vacuum can advance.
1824 return spg_storage::snapshot::Snapshot::new(
1825 version,
1826 spg_storage::snapshot::InProgressSet::empty(),
1827 version,
1828 reader_tx_id,
1829 );
1830 }
1831 let sorted: alloc::vec::Vec<u64> = self.active_writer_versions.iter().copied().collect();
1832 let oldest = *sorted.first().unwrap_or(&version);
1833 spg_storage::snapshot::Snapshot::new(
1834 version,
1835 spg_storage::snapshot::InProgressSet::from_sorted(sorted),
1836 oldest,
1837 reader_tx_id,
1838 )
1839 }
1840
1841 /// v7.37.15 (Phase C) — allocate the next writer version AND
1842 /// add it to the in-flight set so concurrent snapshots hide
1843 /// the resulting writes until [`Self::commit_writer_version`]
1844 /// removes the entry. Returns the allocated version so the
1845 /// writer can stamp it on `xmin` / `xmax`.
1846 pub fn begin_writer_version(&mut self) -> u64 {
1847 let v = spg_storage::row_header::next_version();
1848 self.active_writer_versions.insert(v);
1849 v
1850 }
1851
1852 /// v7.37.15 (Phase C) — mark a previously-allocated writer
1853 /// version as committed. Subsequent snapshots stop including
1854 /// it in `in_progress`, so the writes the version stamped
1855 /// become visible to new readers.
1856 ///
1857 /// No-op if the version was never allocated; matches PG's
1858 /// idempotent `TransactionIdCommitTree` semantics.
1859 pub fn commit_writer_version(&mut self, v: u64) {
1860 self.active_writer_versions.remove(&v);
1861 }
1862
1863 /// v7.37.15 (Phase C.2) — mark a previously-allocated writer
1864 /// version as ABORTED (rolled back). Removes it from the in-flight
1865 /// set and records it in `aborted_versions` so the visibility
1866 /// oracle ([`Self::xact_status`]) reports `Aborted` rather than
1867 /// silently treating it as committed once it leaves the in-flight
1868 /// set. Phase C.3's in-place write path relies on this: a
1869 /// rolled-back version's xmin/xmax stamps stay physically present
1870 /// until vacuum reclaims them, and readers must NOT see them.
1871 ///
1872 /// Idempotent; a no-op if the version was never allocated.
1873 pub fn abort_writer_version(&mut self, v: u64) {
1874 self.active_writer_versions.remove(&v);
1875 self.aborted_versions.insert(v);
1876 }
1877
1878 /// v7.37.15 (Phase C.2) — the visibility oracle's terminal-status
1879 /// lookup for one version. In-flight if still allocated, Aborted
1880 /// if it rolled back, otherwise Committed (the default for a
1881 /// version that left the in-flight set the normal way, and for
1882 /// every frozen / pruned old version the engine no longer tracks).
1883 ///
1884 /// `aborted_versions` is bounded by pruning below `oldest_active`
1885 /// during vacuum (Phase D): once no live snapshot can still see an
1886 /// aborted version's stamps, its entry is dropped. Until Phase D
1887 /// lands the set only grows with rolled-back transactions — noted
1888 /// as a never-die follow-up, not a steady-state leak on the
1889 /// commit path.
1890 #[must_use]
1891 pub fn xact_status(&self, v: u64) -> spg_storage::snapshot::XactStatus {
1892 use spg_storage::snapshot::XactStatus;
1893 if self.active_writer_versions.contains(&v) {
1894 XactStatus::InProgress
1895 } else if self.aborted_versions.contains(&v) {
1896 XactStatus::Aborted
1897 } else {
1898 XactStatus::Committed
1899 }
1900 }
1901
1902 /// v7.37.15 (Phase C.4) — acquire a tuple lock on a stable
1903 /// `(RelId, RowId)` for writer `version`. The in-place write path
1904 /// (C.3) calls this before stamping xmax; `SELECT ... FOR UPDATE`
1905 /// wires here via the parser's lock-strength clause (C.4). Returns
1906 /// the [`LockOutcome`](crate::locks::LockOutcome) the caller acts on
1907 /// (grant / park / skip / fail / deadlock-abort).
1908 pub fn acquire_row_lock(
1909 &mut self,
1910 rel: spg_storage::row_header::RelId,
1911 row: spg_storage::row_header::RowId,
1912 mode: crate::locks::LockMode,
1913 version: u64,
1914 policy: crate::locks::WaitPolicy,
1915 ) -> crate::locks::LockOutcome {
1916 self.locks.acquire(rel, row, mode, version, policy)
1917 }
1918
1919 /// v7.37.15 (Phase C.4) — release every lock + wait held by
1920 /// `version` at transaction end. Called from `exec_commit` /
1921 /// `exec_rollback` alongside the writer-version bookkeeping.
1922 pub fn release_tx_locks(&mut self, version: u64) {
1923 self.locks.release_all(version);
1924 }
1925
1926 /// 7.38.1 S2.1 — drop the tuple locks an AUTOCOMMIT statement took
1927 /// (its implicit transaction ends with it). A no-op inside an open
1928 /// transaction (those release at COMMIT/ROLLBACK) and when the
1929 /// statement allocated no writer version.
1930 pub(crate) fn release_autocommit_stmt_locks(&mut self) {
1931 let in_tx = self
1932 .current_tx
1933 .is_some_and(|tx| self.tx_writer_versions.contains_key(&tx));
1934 if in_tx {
1935 return;
1936 }
1937 if let Some(v) = self.stmt_writer_version {
1938 self.locks.release_all(v);
1939 }
1940 }
1941
1942 /// v7.37.15 (Phase C.4) — number of rows currently locked, for the
1943 /// `pg_locks` enumeration and tests.
1944 #[must_use]
1945 pub fn locked_row_count(&self) -> usize {
1946 self.locks.locked_row_count()
1947 }
1948
1949 /// v7.37.15 (Phase C.3) — is the in-place MVCC write path enabled?
1950 /// `false` (default) keeps legacy physical DELETE/UPDATE. The C.3
1951 /// writers consult this to choose tombstone-vs-physical.
1952 #[must_use]
1953 pub fn mvcc_inplace(&self) -> bool {
1954 self.mvcc_inplace
1955 }
1956
1957 /// v7.37.15 (Phase C.3) — enable/disable the in-place MVCC write
1958 /// path. Called by the host after reading `SPG_MVCC_INPLACE` (the
1959 /// `no_std` engine can't read the environment itself). Off until
1960 /// the write path is proven against PG18 differential tests.
1961 pub fn set_mvcc_inplace(&mut self, on: bool) {
1962 self.mvcc_inplace = on;
1963 }
1964
1965 /// v7.39 (parallel-agg P0) — inject the host's parallel executor
1966 /// (see [`ParallelRunner`]). Called once at host startup; the
1967 /// engine stays single-threaded without it.
1968 /// v7.39 (pg_stat knife A) — inject the host's live backend count.
1969 pub fn set_backend_count_fn(&mut self, f: BackendCountFn) {
1970 self.backend_count_fn = Some(f);
1971 }
1972
1973 /// v7.39 (read01 pgstatfuncs.c) — inject the host's calling-connection
1974 /// identity for pg_backend_pid().
1975 /// v7.39 (round 476) — register the WAL byte-position provider.
1976 pub fn set_wal_lsn_fn(&mut self, f: WalLsnFn) {
1977 self.wal_lsn_fn = Some(f);
1978 }
1979
1980 pub fn set_backend_pid_fn(&mut self, f: BackendPidFn) {
1981 self.backend_pid_fn = Some(f);
1982 }
1983
1984 /// v7.39 (round 318, V51) — inject the host's connection-control hook,
1985 /// so `pg_cancel_backend` / `pg_terminate_backend` / `KILL` act instead
1986 /// of answering a constant.
1987 pub fn set_backend_signal_fn(&mut self, f: BackendSignalFn) {
1988 self.backend_signal_fn = Some(f);
1989 }
1990
1991 /// v7.39 (round 786, T35 Phase A) — install the host's spill-run
1992 /// factory. Without one the engine cannot spill and a sort that
1993 /// outgrows `max_query_bytes` keeps refusing, which is exactly the
1994 /// behaviour every caller has today.
1995 pub fn set_temp_run_factory(&mut self, f: crate::TempRunFactory) {
1996 self.temp_run_factory = Some(f);
1997 }
1998
1999 /// Whether spilling is available in this process.
2000 #[must_use]
2001 pub fn can_spill(&self) -> bool {
2002 self.temp_run_factory.is_some()
2003 }
2004
2005 /// v7.39 (round 786) — open a fresh spill run, or `None` when no
2006 /// host factory is installed. Phase B's run generation calls this;
2007 /// it lives here so the `None` path stays a single decision point.
2008 pub(crate) fn open_temp_run(
2009 &self,
2010 ) -> Option<Result<alloc::boxed::Box<dyn crate::TempRun>, crate::TempStoreError>> {
2011 self.temp_run_factory.map(|f| f())
2012 }
2013
2014 /// v7.39 (tz epic) — inject the host's IANA timezone lookups
2015 /// (spg-tzif's fn family on std hosts).
2016 pub fn set_tz_fns(
2017 &mut self,
2018 offset: TzOffsetFn,
2019 localize: TzLocalizeFn,
2020 canon: TzCanonFn,
2021 abbrev: TzAbbrevFn,
2022 ) {
2023 self.tz_offset_fn = Some(offset);
2024 self.tz_localize_fn = Some(localize);
2025 self.tz_canon_fn = Some(canon);
2026 self.tz_abbrev_fn = Some(abbrev);
2027 }
2028
2029 /// v7.39 (round 502) — the zone enumerator behind `pg_timezone_names`.
2030 /// Separate from `set_tz_fns` so an embedder that already calls that
2031 /// one keeps compiling.
2032 pub fn set_tz_all_fn(&mut self, all: TzAllFn) {
2033 self.tz_all_fn = Some(all);
2034 }
2035
2036 /// Every zone the host knows at `utc_micros`; empty without a hook.
2037 pub(crate) fn tz_all_at(
2038 &self,
2039 utc_micros: i64,
2040 ) -> alloc::vec::Vec<(alloc::string::String, alloc::string::String, i64, bool)> {
2041 self.tz_all_fn
2042 .map_or_else(alloc::vec::Vec::new, |f| f(utc_micros))
2043 }
2044
2045 pub fn set_parallel_runner(&mut self, runner: alloc::sync::Arc<dyn ParallelRunner>) {
2046 self.parallel_runner = ParallelRunnerSlot(Some(runner));
2047 }
2048
2049 /// v7.37.15 (Phase C) — allocate a fresh version number for
2050 /// the next write. Always strictly monotonic + process-wide
2051 /// shared so concurrent engines on the same process agree on
2052 /// "tx 17 commits before tx 18". Phase C writer paths call
2053 /// this once per INSERT / UPDATE / DELETE statement to obtain
2054 /// the version they'll stamp on the new row's `xmin` (or the
2055 /// existing row's `xmax`).
2056 ///
2057 /// Returns [`XMIN_FROZEN`] when MVCC stamping is intentionally
2058 /// off (legacy `in_memory` flow / WAL replay): the writer
2059 /// then takes the legacy frozen-insert short-circuit path
2060 /// inside `Table::insert_with_xmin`.
2061 #[must_use]
2062 pub fn next_writer_version(&self) -> u64 {
2063 spg_storage::row_header::next_version()
2064 }
2065
2066 /// v7.37.15 (Phase C) — version a writer should stamp on
2067 /// rows produced by the current statement. Inside an explicit
2068 /// transaction the version is the tx's pre-allocated one (so
2069 /// every statement in the tx commits atomically at COMMIT);
2070 /// in autocommit it allocates a fresh version per statement.
2071 ///
2072 /// This is the canonical helper engine writers should call
2073 /// — using it instead of `next_writer_version` ensures
2074 /// explicit-tx semantics where every row produced by the tx
2075 /// shares one xmin and concurrent readers don't see partial
2076 /// state until COMMIT.
2077 ///
2078 /// v7.37.15 (Epic W slice 2) — takes `&mut self` so the autocommit
2079 /// branch can **memoize** its freshly-minted version in
2080 /// `stmt_writer_version`. `next_writer_version()` is a `fetch_add`,
2081 /// so without memoization a second call within one statement (the
2082 /// redo drain post-stamps the captured `RowChange`s) would allocate
2083 /// a *different* version than the writes used. Memoizing makes the
2084 /// value stable for the statement's lifetime; it is reset per
2085 /// `execute_in_with_cancel`, so the counter still advances exactly
2086 /// once per autocommit statement — identical to before.
2087 pub fn writer_version_for_current_stmt(&mut self) -> u64 {
2088 if let Some(tx_id) = self.current_tx
2089 && let Some(&v) = self.tx_writer_versions.get(&tx_id)
2090 {
2091 return v;
2092 }
2093 // Autocommit shape: fresh version, immediately "committed"
2094 // (no entry in active_writer_versions, so subsequent
2095 // readers see the row). Memoized for the statement so the
2096 // redo drain reads back the same version the writes used.
2097 if let Some(v) = self.stmt_writer_version {
2098 return v;
2099 }
2100 let v = self.next_writer_version();
2101 self.stmt_writer_version = Some(v);
2102 v
2103 }
2104
2105 /// Construct an engine restored from a previously-snapshotted catalog
2106 /// (see `snapshot()`).
2107 pub fn restore(catalog: Catalog) -> Self {
2108 Self {
2109 lock_skip_rows: None,
2110 catalog,
2111 parallel_runner: ParallelRunnerSlot::default(),
2112 tx_catalogs: BTreeMap::new(),
2113 table_last_commit: BTreeMap::new(),
2114 commit_seq: 0,
2115 current_tx: None,
2116 backslash_escapes: false,
2117 mysql_strict: true,
2118 mysql_warnings: Vec::new(),
2119 mysql_stmt_warnings: Vec::new(),
2120 lo_descriptors: BTreeMap::new(),
2121 lo_next_fd: 0,
2122 prepared_statements: alloc::collections::BTreeMap::new(),
2123 current_session: 0,
2124 sessions: BTreeMap::new(),
2125 advisory_locks: BTreeMap::new(),
2126 last_sequence_used: None,
2127 seq_currvals: alloc::collections::BTreeMap::new(),
2128 next_tx_id: 1,
2129 active_writer_versions: BTreeSet::new(),
2130 aborted_versions: BTreeSet::new(),
2131 locks: crate::locks::LockTable::new(),
2132 mvcc_inplace: !cfg!(feature = "mvcc-inplace-off"),
2133 autovacuum: true,
2134 autovacuum_inline: true,
2135 tx_writer_versions: BTreeMap::new(),
2136 stmt_writer_version: None,
2137 clock: None,
2138 salt_fn: None,
2139 max_query_rows: None,
2140 max_query_bytes: None,
2141 temp_run_factory: None,
2142 users: UserStore::new(),
2143 publications: publications::Publications::new(),
2144 subscriptions: subscriptions::Subscriptions::new(),
2145 statistics: statistics::Statistics::new(),
2146 plan_cache: plan_cache::PlanCache::new(),
2147 query_stats: query_stats::QueryStats::new(),
2148 activity_provider: None,
2149 audit_chain_provider: None,
2150 audit_verifier: None,
2151 slow_query_threshold_us: None,
2152 slow_query_logger: None,
2153 session_params: BTreeMap::new(),
2154 cursors: BTreeMap::new(),
2155 last_insert_id: core::sync::atomic::AtomicI64::new(0),
2156 row_count: 0,
2157 user_vars: BTreeMap::new(),
2158 temp_tables: BTreeSet::new(),
2159 temp_sequences: BTreeSet::new(),
2160 temp_views: BTreeSet::new(),
2161 listen_channels: BTreeSet::new(),
2162 tx_pending_notifies: Vec::new(),
2163 delivered_notifies: Vec::new(),
2164 pending_notices: Vec::new(),
2165 spill_stats: crate::tempstore::SpillStats::default(),
2166 xact_commit: core::sync::atomic::AtomicU64::new(0),
2167 xact_rollback: core::sync::atomic::AtomicU64::new(0),
2168 backend_count_fn: None,
2169 backend_pid_fn: None,
2170 wal_lsn_fn: None,
2171 backend_signal_fn: None,
2172 tz_offset_fn: None,
2173 tz_localize_fn: None,
2174 tz_canon_fn: None,
2175 tz_abbrev_fn: None,
2176 tz_all_fn: None,
2177 stat_tup_inserted: 0,
2178 table_write_stats: alloc::collections::BTreeMap::new(),
2179 commit_epoch: 0,
2180 stat_tup_updated: 0,
2181 stat_tup_deleted: 0,
2182 local_guc_saves: Vec::new(),
2183 render_style: crate::eval::RenderStyle::default(),
2184 savepoint_guc_marks: Vec::new(),
2185 trigger_recursion_depth: 0,
2186 rule_rewrite_active: false,
2187 foreign_key_checks: true,
2188 meta_views_materialised: false,
2189 pending_foreign_keys: Vec::new(),
2190 env_cfg: testkit::EnvConfig::default(),
2191 #[cfg(feature = "injection-points")]
2192 injection_store: alloc::sync::Arc::new(
2193 crate::testkit::injection::InjectionStore::default(),
2194 ),
2195 redo_capture: false,
2196 current_isolation_level: spg_sql::ast::IsolationLevel::ReadCommitted,
2197 last_redo: Vec::new(),
2198 table_change_seq: alloc::collections::BTreeMap::new(),
2199 matview_refresh_watermark: alloc::collections::BTreeMap::new(),
2200 matview_maintainable: alloc::collections::BTreeMap::new(),
2201 matview_delta_buf: alloc::collections::BTreeMap::new(),
2202 matview_delta_overflow: alloc::collections::BTreeSet::new(),
2203 matview_row_map: alloc::collections::BTreeMap::new(),
2204 }
2205 }
2206
2207 /// Restore an engine + user table from a v4.1 envelope produced
2208 /// by `snapshot_with_users()`. Falls back to plain catalog-only
2209 /// restore if the envelope magic isn't present (so v3.x snapshot
2210 /// files still load). v6.1.2 adds the optional publications
2211 /// trailer (envelope v3); a v1/v2 envelope deserialises to an
2212 /// empty publication table.
2213 pub fn restore_envelope(buf: &[u8]) -> Result<Self, EngineError> {
2214 match split_envelope(buf) {
2215 EnvelopeParse::Pair {
2216 catalog: catalog_bytes,
2217 users: user_bytes,
2218 publications: pub_bytes,
2219 subscriptions: sub_bytes,
2220 statistics: stats_bytes,
2221 } => {
2222 let mut catalog =
2223 Catalog::deserialize(catalog_bytes).map_err(EngineError::Storage)?;
2224 crate::ddl::rebuild_all_excl_indexes(&mut catalog);
2225 // v7.38.16 — and refill the expression indexes, whose keys
2226 // the format cannot carry: what is on disk under one was
2227 // written by a version that stored the wrong values there.
2228 crate::expr_index::rebuild_all(&mut catalog);
2229 let users = users::deserialize_users(user_bytes)
2230 .map_err(|e| EngineError::Unsupported(alloc::format!("users restore: {e}")))?;
2231 let publications = match pub_bytes {
2232 Some(b) => publications::Publications::deserialize(b).map_err(|e| {
2233 EngineError::Unsupported(alloc::format!("publications restore: {e:?}"))
2234 })?,
2235 None => publications::Publications::new(),
2236 };
2237 let subscriptions = match sub_bytes {
2238 Some(b) => subscriptions::Subscriptions::deserialize(b).map_err(|e| {
2239 EngineError::Unsupported(alloc::format!("subscriptions restore: {e:?}"))
2240 })?,
2241 None => subscriptions::Subscriptions::new(),
2242 };
2243 let statistics = match stats_bytes {
2244 Some(b) => statistics::Statistics::deserialize(b).map_err(|e| {
2245 EngineError::Unsupported(alloc::format!("statistics restore: {e:?}"))
2246 })?,
2247 None => statistics::Statistics::new(),
2248 };
2249 Ok(Self {
2250 lock_skip_rows: None,
2251 catalog,
2252 parallel_runner: ParallelRunnerSlot::default(),
2253 tx_catalogs: BTreeMap::new(),
2254 table_last_commit: BTreeMap::new(),
2255 commit_seq: 0,
2256 current_tx: None,
2257 backslash_escapes: false,
2258 mysql_strict: true,
2259 mysql_warnings: Vec::new(),
2260 mysql_stmt_warnings: Vec::new(),
2261 lo_descriptors: BTreeMap::new(),
2262 lo_next_fd: 0,
2263 prepared_statements: alloc::collections::BTreeMap::new(),
2264 current_session: 0,
2265 sessions: BTreeMap::new(),
2266 advisory_locks: BTreeMap::new(),
2267 last_sequence_used: None,
2268 seq_currvals: alloc::collections::BTreeMap::new(),
2269 next_tx_id: 1,
2270 active_writer_versions: BTreeSet::new(),
2271 aborted_versions: BTreeSet::new(),
2272 locks: crate::locks::LockTable::new(),
2273 mvcc_inplace: !cfg!(feature = "mvcc-inplace-off"),
2274 autovacuum: true,
2275 autovacuum_inline: true,
2276 tx_writer_versions: BTreeMap::new(),
2277 stmt_writer_version: None,
2278 clock: None,
2279 salt_fn: None,
2280 max_query_rows: None,
2281 max_query_bytes: None,
2282 temp_run_factory: None,
2283 users,
2284 publications,
2285 subscriptions,
2286 statistics,
2287 plan_cache: plan_cache::PlanCache::new(),
2288 query_stats: query_stats::QueryStats::new(),
2289 activity_provider: None,
2290 audit_chain_provider: None,
2291 audit_verifier: None,
2292 slow_query_threshold_us: None,
2293 slow_query_logger: None,
2294 session_params: BTreeMap::new(),
2295 cursors: BTreeMap::new(),
2296 last_insert_id: core::sync::atomic::AtomicI64::new(0),
2297 row_count: 0,
2298 user_vars: BTreeMap::new(),
2299 temp_tables: BTreeSet::new(),
2300 temp_sequences: BTreeSet::new(),
2301 temp_views: BTreeSet::new(),
2302 listen_channels: BTreeSet::new(),
2303 tx_pending_notifies: Vec::new(),
2304 delivered_notifies: Vec::new(),
2305 pending_notices: Vec::new(),
2306 spill_stats: crate::tempstore::SpillStats::default(),
2307 xact_commit: core::sync::atomic::AtomicU64::new(0),
2308 xact_rollback: core::sync::atomic::AtomicU64::new(0),
2309 backend_count_fn: None,
2310 backend_pid_fn: None,
2311 wal_lsn_fn: None,
2312 backend_signal_fn: None,
2313 tz_offset_fn: None,
2314 tz_localize_fn: None,
2315 tz_canon_fn: None,
2316 tz_abbrev_fn: None,
2317 tz_all_fn: None,
2318 stat_tup_inserted: 0,
2319 table_write_stats: alloc::collections::BTreeMap::new(),
2320 commit_epoch: 0,
2321 stat_tup_updated: 0,
2322 stat_tup_deleted: 0,
2323 local_guc_saves: Vec::new(),
2324 render_style: crate::eval::RenderStyle::default(),
2325 savepoint_guc_marks: Vec::new(),
2326 trigger_recursion_depth: 0,
2327 rule_rewrite_active: false,
2328 foreign_key_checks: true,
2329 meta_views_materialised: false,
2330 pending_foreign_keys: Vec::new(),
2331 env_cfg: testkit::EnvConfig::default(),
2332 #[cfg(feature = "injection-points")]
2333 injection_store: alloc::sync::Arc::new(
2334 crate::testkit::injection::InjectionStore::default(),
2335 ),
2336 redo_capture: false,
2337 current_isolation_level: spg_sql::ast::IsolationLevel::ReadCommitted,
2338 last_redo: Vec::new(),
2339 table_change_seq: alloc::collections::BTreeMap::new(),
2340 matview_refresh_watermark: alloc::collections::BTreeMap::new(),
2341 matview_maintainable: alloc::collections::BTreeMap::new(),
2342 matview_delta_buf: alloc::collections::BTreeMap::new(),
2343 matview_delta_overflow: alloc::collections::BTreeSet::new(),
2344 matview_row_map: alloc::collections::BTreeMap::new(),
2345 })
2346 }
2347 EnvelopeParse::CrcMismatch { expected, computed } => {
2348 Err(EngineError::Storage(StorageError::Corrupt(alloc::format!(
2349 "snapshot envelope CRC32 mismatch (expected={expected:#010x}, computed={computed:#010x})"
2350 ))))
2351 }
2352 EnvelopeParse::Bare => {
2353 let mut catalog = Catalog::deserialize(buf).map_err(EngineError::Storage)?;
2354 crate::ddl::rebuild_all_excl_indexes(&mut catalog);
2355 // v7.38.16 — and refill the expression indexes, whose keys
2356 // the format cannot carry: what is on disk under one was
2357 // written by a version that stored the wrong values there.
2358 crate::expr_index::rebuild_all(&mut catalog);
2359 Ok(Self::restore(catalog))
2360 }
2361 }
2362 }
2363
2364 pub const fn users(&self) -> &UserStore {
2365 &self.users
2366 }
2367
2368 /// Builder: attach a wall clock so `NOW()` / `CURRENT_TIMESTAMP` /
2369 /// `CURRENT_DATE` evaluate to a real value instead of erroring out.
2370 /// v7.39 (round 279) — announce which connection is about to run.
2371 /// The server calls this before every statement; embedded hosts
2372 /// never do and stay on session 0.
2373 ///
2374 /// Swapping parks the outgoing connection's state and installs the
2375 /// incoming one's, creating it on first sight. The plan cache is
2376 /// cleared because the string-literal dialect is part of what
2377 /// swaps and the same SQL text lexes differently under it.
2378 pub fn set_current_session(&mut self, id: u32) {
2379 if id == self.current_session {
2380 return;
2381 }
2382 let outgoing = SessionBag {
2383 session_params: core::mem::take(&mut self.session_params),
2384 backslash_escapes: self.backslash_escapes,
2385 mysql_strict: self.mysql_strict,
2386 mysql_warnings: core::mem::take(&mut self.mysql_warnings),
2387 prepared_statements: core::mem::take(&mut self.prepared_statements),
2388 lo_descriptors: core::mem::take(&mut self.lo_descriptors),
2389 lo_next_fd: self.lo_next_fd,
2390 cursors: core::mem::take(&mut self.cursors),
2391 last_insert_id: self
2392 .last_insert_id
2393 .load(core::sync::atomic::Ordering::Relaxed),
2394 row_count: self.row_count,
2395 user_vars: core::mem::take(&mut self.user_vars),
2396 temp_tables: core::mem::take(&mut self.temp_tables),
2397 temp_sequences: core::mem::take(&mut self.temp_sequences),
2398 temp_views: core::mem::take(&mut self.temp_views),
2399 seq_currvals: core::mem::take(&mut self.seq_currvals),
2400 last_sequence_used: self.last_sequence_used.take(),
2401 isolation_level: self.current_isolation_level,
2402 };
2403 self.sessions.insert(self.current_session, outgoing);
2404 let incoming = self.sessions.remove(&id).unwrap_or_default();
2405 self.session_params = incoming.session_params;
2406 self.backslash_escapes = incoming.backslash_escapes;
2407 self.mysql_strict = incoming.mysql_strict;
2408 self.mysql_warnings = incoming.mysql_warnings;
2409 self.prepared_statements = incoming.prepared_statements;
2410 self.lo_descriptors = incoming.lo_descriptors;
2411 self.lo_next_fd = incoming.lo_next_fd;
2412 self.cursors = incoming.cursors;
2413 self.last_insert_id.store(
2414 incoming.last_insert_id,
2415 core::sync::atomic::Ordering::Relaxed,
2416 );
2417 self.row_count = incoming.row_count;
2418 self.user_vars = incoming.user_vars;
2419 self.temp_tables = incoming.temp_tables;
2420 self.temp_sequences = incoming.temp_sequences;
2421 self.temp_views = incoming.temp_views;
2422 self.seq_currvals = incoming.seq_currvals;
2423 self.last_sequence_used = incoming.last_sequence_used;
2424 self.current_isolation_level = incoming.isolation_level;
2425 self.current_session = id;
2426 // The incoming session's temp namespace must be live before its very
2427 // first statement resolves a name.
2428 self.refresh_temp_prefix();
2429 self.plan_cache.clear();
2430 }
2431
2432 /// v7.39 (round 436) — the catalog-name prefix session `id` stores its
2433 /// TEMPORARY tables under. Mirrors PG's per-session `pg_temp_N` schema;
2434 /// the leading underscores keep it out of any name a client can write.
2435 fn temp_prefix_for(id: u32) -> String {
2436 alloc::format!("__spg_temp_{id}__")
2437 }
2438
2439 /// v7.38.14 — which session's temporary namespace does this catalog
2440 /// name belong to, if any? The inverse of `session_temp_name`, so the
2441 /// system catalog can report `pg_temp_N` for it instead of `public`.
2442 pub(crate) fn temp_session_of(catalog_name: &str) -> Option<u32> {
2443 let rest = catalog_name.strip_prefix("__spg_temp_")?;
2444 let (id, _) = rest.split_once("__")?;
2445 id.parse().ok()
2446 }
2447
2448 /// The catalog name this session's TEMPORARY table `logical` takes.
2449 pub(crate) fn session_temp_name(&self, logical: &str) -> String {
2450 alloc::format!("{}{logical}", Self::temp_prefix_for(self.current_session))
2451 }
2452
2453 /// v7.39 (round 436) — point every catalog this session can reach at its
2454 /// temp namespace, or at none when it owns no temporary tables (so a
2455 /// session that never made one pays a single `Option` check per lookup).
2456 /// Both the committed catalog and any open transaction's shadow are set:
2457 /// a temp table created inside a transaction must resolve there too.
2458 pub(crate) fn refresh_temp_prefix(&mut self) {
2459 let prefix = if self.temp_tables.is_empty()
2460 && self.temp_sequences.is_empty()
2461 && self.temp_views.is_empty()
2462 {
2463 None
2464 } else {
2465 Some(Self::temp_prefix_for(self.current_session))
2466 };
2467 self.catalog.set_temp_prefix(prefix.clone());
2468 for shadow in self.tx_catalogs.values_mut() {
2469 shadow.catalog.set_temp_prefix(prefix.clone());
2470 }
2471 }
2472
2473 /// v7.39 (round 279) — a connection has gone away: drop its parked
2474 /// state and release every advisory lock it still held, which is
2475 /// what PG does at backend exit.
2476 pub fn end_session(&mut self, id: u32) {
2477 // v7.39 (round 436) — a TEMPORARY table dies with its session, in
2478 // both PG and MySQL. Done before the bag is dropped, since the bag
2479 // is what knows which tables the session owns.
2480 let owned: Vec<String> = if id == self.current_session {
2481 self.temp_tables.iter().cloned().collect()
2482 } else {
2483 self.sessions
2484 .get(&id)
2485 .map(|b| b.temp_tables.iter().cloned().collect())
2486 .unwrap_or_default()
2487 };
2488 // v7.39 (round 469) — the same for TEMPORARY sequences and views,
2489 // which PG also drops at backend exit.
2490 let owned_seqs: Vec<String> = if id == self.current_session {
2491 self.temp_sequences.iter().cloned().collect()
2492 } else {
2493 self.sessions
2494 .get(&id)
2495 .map(|b| b.temp_sequences.iter().cloned().collect())
2496 .unwrap_or_default()
2497 };
2498 let owned_views: Vec<String> = if id == self.current_session {
2499 self.temp_views.iter().cloned().collect()
2500 } else {
2501 self.sessions
2502 .get(&id)
2503 .map(|b| b.temp_views.iter().cloned().collect())
2504 .unwrap_or_default()
2505 };
2506 if !owned.is_empty() || !owned_seqs.is_empty() || !owned_views.is_empty() {
2507 let prefix = Self::temp_prefix_for(id);
2508 for logical in owned {
2509 let mangled = alloc::format!("{prefix}{logical}");
2510 self.catalog.drop_table(&mangled);
2511 }
2512 for logical in owned_seqs {
2513 let mangled = alloc::format!("{prefix}{logical}");
2514 self.catalog.drop_sequence(&mangled);
2515 }
2516 for logical in owned_views {
2517 let mangled = alloc::format!("{prefix}{logical}");
2518 self.catalog.drop_view(&mangled);
2519 }
2520 if id == self.current_session {
2521 self.temp_tables.clear();
2522 self.temp_sequences.clear();
2523 self.temp_views.clear();
2524 self.refresh_temp_prefix();
2525 }
2526 }
2527 self.sessions.remove(&id);
2528 self.advisory_locks.retain(|_, (owner, _)| *owner != id);
2529 if id == self.current_session {
2530 self.session_params.clear();
2531 self.prepared_statements.clear();
2532 self.backslash_escapes = false;
2533 self.mysql_strict = true;
2534 self.lo_descriptors.clear();
2535 self.lo_next_fd = 0;
2536 self.cursors.clear();
2537 self.current_session = 0;
2538 }
2539 }
2540
2541 /// v7.39 (round 302, V15) — force the current session's string-literal
2542 /// dialect. A MySQL-protocol connection defaults to MySQL semantics
2543 /// (backslash is an escape: `'\n'` is a newline), which PG's own
2544 /// default (`standard_conforming_strings = on`) does not do. The
2545 /// mysql-wire shim calls this once, right after installing its
2546 /// session, so a client that never sends `SET sql_mode` still gets
2547 /// MySQL string handling; a later `SET sql_mode='NO_BACKSLASH_ESCAPES'`
2548 /// flips it back through the normal SET path. Clearing the plan cache
2549 /// mirrors [`set_current_session`] — the same SQL text lexes
2550 /// differently once the flag moves.
2551 /// Is this session in MySQL dialect right now?
2552 ///
2553 /// v7.38.17 — a test harness cannot take a file's word for which
2554 /// dialect it runs in. `SET sql_mode = 'STRICT_TRANS_TABLES'` puts a
2555 /// session into MySQL semantics mid-file, so six corpus files were
2556 /// entering MySQL through a door no directive named, and a harness
2557 /// that read the directive reported them as PostgreSQL. Ask the
2558 /// session, do not parse the script.
2559 pub fn in_mysql_dialect(&self) -> bool {
2560 self.backslash_escapes
2561 }
2562
2563 pub fn set_backslash_escapes(&mut self, flag: bool) {
2564 if flag != self.backslash_escapes {
2565 self.backslash_escapes = flag;
2566 self.plan_cache.clear();
2567 }
2568 }
2569
2570 /// v7.39 (round 279) — take an advisory lock. Returns false only
2571 /// when ANOTHER session holds it; re-taking one this session
2572 /// already holds bumps a depth counter, as in PG.
2573 pub(crate) fn advisory_try_lock(&mut self, key: i64) -> bool {
2574 let me = self.current_session;
2575 match self.advisory_locks.get_mut(&key) {
2576 Some((owner, depth)) if *owner == me => {
2577 *depth += 1;
2578 true
2579 }
2580 Some(_) => false,
2581 None => {
2582 self.advisory_locks.insert(key, (me, 1));
2583 true
2584 }
2585 }
2586 }
2587
2588 /// Release one level, `false` when this session does not hold it.
2589 ///
2590 /// v7.38.19 — and warn when it does not, which PostgreSQL does and
2591 /// this did not. Measured on PG 18.4:
2592 ///
2593 /// ```text
2594 /// WARNING: you don't own a lock of type ExclusiveLock
2595 /// pg_advisory_unlock
2596 /// f
2597 /// ```
2598 ///
2599 /// The comment here used to record the missing warning as a delta
2600 /// and leave it. Measuring the delta to close it turned up a second
2601 /// one it had never mentioned: the column was called `?column?`
2602 /// rather than `pg_advisory_unlock`, and so was every other
2603 /// statement-resolved call. A recorded delta says what someone
2604 /// noticed, not what is true.
2605 pub(crate) fn advisory_unlock(&mut self, key: i64) -> bool {
2606 let me = self.current_session;
2607 match self.advisory_locks.get_mut(&key) {
2608 Some((owner, depth)) if *owner == me => {
2609 *depth -= 1;
2610 if *depth == 0 {
2611 self.advisory_locks.remove(&key);
2612 }
2613 true
2614 }
2615 _ => {
2616 self.warning(alloc::string::String::from(
2617 "you don't own a lock of type ExclusiveLock",
2618 ));
2619 false
2620 }
2621 }
2622 }
2623
2624 /// Release every advisory lock this session holds.
2625 pub(crate) fn advisory_unlock_all(&mut self) {
2626 let me = self.current_session;
2627 self.advisory_locks.retain(|_, (owner, _)| *owner != me);
2628 }
2629
2630 /// v7.39 (round 417) — the current session's id (for MySQL
2631 /// `IS_USED_LOCK`, which reports the connection that holds a lock).
2632 /// v7.39 (round 430) — read one of this session's MySQL USER
2633 /// variables. `None` when it was never set, which the caller turns
2634 /// into NULL (MariaDB reads an unset user variable as NULL).
2635 pub(crate) fn user_var(&self, name: &str) -> Option<&spg_storage::Value<'static>> {
2636 self.user_vars.get(name)
2637 }
2638
2639 pub(crate) const fn current_session_id(&self) -> u32 {
2640 self.current_session
2641 }
2642
2643 /// v7.39 (round 417) — who holds an advisory-lock key (any session id),
2644 /// or `None` when nobody holds it. Used by MySQL `IS_USED_LOCK` and to
2645 /// separate `RELEASE_LOCK`'s "not held by anyone" (returns NULL) from
2646 /// "held by someone else" (returns 0).
2647 pub(crate) fn advisory_holder(&self, key: i64) -> Option<u32> {
2648 self.advisory_locks.get(&key).map(|(owner, _)| *owner)
2649 }
2650
2651 /// v7.39 (round 417) — MySQL `RELEASE_ALL_LOCKS()` returns the number of
2652 /// locks it released; PG's `pg_advisory_unlock_all()` returns void.
2653 pub(crate) fn advisory_unlock_all_count(&mut self) -> i32 {
2654 let me = self.current_session;
2655 // Total depth held by this session, so re-locked keys count as many.
2656 let mut n: i32 = 0;
2657 for (_, (owner, depth)) in &self.advisory_locks {
2658 if *owner == me {
2659 n = n.saturating_add(*depth as i32);
2660 }
2661 }
2662 self.advisory_locks.retain(|_, (owner, _)| *owner != me);
2663 n
2664 }
2665
2666 #[must_use]
2667 pub const fn with_clock(mut self, clock: ClockFn) -> Self {
2668 self.clock = Some(clock);
2669 self
2670 }
2671
2672 /// Builder: attach an OS-backed RNG for per-user password salts.
2673 /// The host (`spg-server`) typically wires this to `/dev/urandom`.
2674 #[must_use]
2675 pub const fn with_salt_fn(mut self, f: SaltFn) -> Self {
2676 self.salt_fn = Some(f);
2677 self
2678 }
2679
2680 /// v7.38 元机制 D — install a frozen [`testkit::EnvConfig`] snapshot.
2681 ///
2682 /// Hosts (spg-server, spg-embedded, tests) call this once at engine
2683 /// init with either `EnvConfig::from_env()` (production-with-test-vars)
2684 /// or `EnvConfig::builder()....build()` (programmatic). After
2685 /// construction the engine never reads env vars; all test-mode
2686 /// behaviour flows through `self.env_cfg()`.
2687 #[must_use]
2688 pub fn with_env_cfg(mut self, env_cfg: testkit::EnvConfig) -> Self {
2689 // r1051 — a configured seed reaches `random()` too, not only
2690 // the `rng_seed()` accessor: the D-mechanism's "single seed
2691 // source" claim, made true at the SQL level (first pin_v738_
2692 // catch). Production engines carry no seed and never touch
2693 // the PRNG state here.
2694 if let Some(seed) = env_cfg.random_seed {
2695 crate::eval::math::prng_install_seed(seed);
2696 }
2697 // r1058 — `SPG_TEST_FIXED_CLOCK_MICROS` pins the clock for
2698 // hosts that read GUCs from env instead of calling
2699 // `with_clock` (the server). Process-global by necessity: a
2700 // `ClockFn` is a plain fn pointer and cannot capture.
2701 if let Some(micros) = env_cfg.fixed_clock_micros {
2702 FIXED_CLOCK_MICROS.store(micros, core::sync::atomic::Ordering::Relaxed);
2703 self = self.with_clock(fixed_clock_from_env);
2704 }
2705 self.env_cfg = env_cfg;
2706 self
2707 }
2708
2709 /// v7.38 元机制 D — frozen test-mode GUC snapshot. Hot paths gate
2710 /// nondeterministic surfaces on fields of this struct; production
2711 /// default keeps every field at `false / None / Auto` so the
2712 /// optimiser can const-fold the gate.
2713 pub fn env_cfg(&self) -> &testkit::EnvConfig {
2714 &self.env_cfg
2715 }
2716
2717 /// v7.38 元机制 D acceptor — single seed source for every
2718 /// nondeterministic engine subsystem (hash builders, randomised
2719 /// tie-breakers, …). Honour `SPG_TEST_RANDOM_SEED=N` when set;
2720 /// otherwise derive from the engine's wall clock (production) or
2721 /// fall back to a fixed sentinel when the host hasn't installed
2722 /// a clock. Two engines built with the same builder seed return
2723 /// byte-equal output for the same query.
2724 /// See `xtests/sigil/test-mode-gucs.md`.
2725 pub fn rng_seed(&self) -> u64 {
2726 if let Some(seed) = self.env_cfg.random_seed {
2727 return seed;
2728 }
2729 match self.clock {
2730 Some(f) => f() as u64,
2731 // Production engines without a clock installed get a fixed
2732 // non-zero sentinel; same shape as PG's `random()` start
2733 // state under a `setseed(0)`.
2734 None => 0xBAD_5EED_DEAD_BEEF,
2735 }
2736 }
2737
2738 /// v7.38 P0 元机制 A — push this engine's `InjectionStore` onto
2739 /// the thread-local stack so any `injection_point!()` reached
2740 /// during the returned guard's lifetime resolves against this
2741 /// engine. Mirrors PG's per-backend injection table.
2742 ///
2743 /// Returns a no-op guard when the `injection-points` feature is
2744 /// off so call sites don't need `#[cfg]`.
2745 pub fn enter_injection_scope(&self) -> crate::testkit::injection::InjectionGuard {
2746 #[cfg(feature = "injection-points")]
2747 {
2748 crate::testkit::injection::enter_scope(&self.injection_store)
2749 }
2750 #[cfg(not(feature = "injection-points"))]
2751 {
2752 crate::testkit::injection::new_guard()
2753 }
2754 }
2755
2756 /// v7.38 P0 元机制 A — expose the per-engine store so tests can
2757 /// query notice counts / detach actions without parsing SQL
2758 /// output. Only present when the feature is on.
2759 #[cfg(feature = "injection-points")]
2760 pub fn injection_store(&self) -> alloc::sync::Arc<crate::testkit::injection::InjectionStore> {
2761 self.injection_store.clone()
2762 }
2763
2764 /// Builder: cap the number of rows a single SELECT may return.
2765 /// Exceeding the cap raises `EngineError::RowLimitExceeded` —
2766 /// the bound is checked inside the executor so a runaway
2767 /// catalog scan can't allocate millions of rows before the
2768 /// server gets a chance to reject the result.
2769 #[must_use]
2770 pub const fn with_max_query_rows(mut self, n: usize) -> Self {
2771 self.max_query_rows = Some(n);
2772 self
2773 }
2774
2775 /// Builder: cap the approximate heap bytes a single SELECT's
2776 /// join/filter materialisation may hold. Exceeding the cap
2777 /// raises `EngineError::QueryBytesExceeded`. Rows are the wrong
2778 /// unit when one row carries a multi-MB body (mailrs round-26:
2779 /// 1000-row batches of full mail text walked a 15 GiB host into
2780 /// reclaim livelock without ever tripping a row ceiling).
2781 #[must_use]
2782 pub const fn with_max_query_bytes(mut self, n: usize) -> Self {
2783 self.max_query_bytes = Some(n);
2784 self
2785 }
2786
2787 /// The *committed* catalog. Note: during a transaction this returns the
2788 /// pre-TX state — `SELECT` inside a TX goes through `execute()` and reads
2789 /// the shadow. Tests that inspect outside-TX state should use this.
2790 pub const fn catalog(&self) -> &Catalog {
2791 &self.catalog
2792 }
2793
2794 /// Capture a frozen view of the committed engine state. Catalog
2795 /// is O(1) Arc bump; trailers are cheap clones. Decouples "capture"
2796 /// (needs &Engine) from "serialize" (CPU, no engine access) — the
2797 /// seam the background-checkpoint worker rides in CoW-2.
2798 pub fn snapshot_data(&self) -> EngineSnapshot {
2799 EngineSnapshot {
2800 catalog: self.catalog.clone(),
2801 users: self.users.clone(),
2802 publications: self.publications.clone(),
2803 subscriptions: self.subscriptions.clone(),
2804 statistics: self.statistics.clone(),
2805 }
2806 }
2807
2808 /// Serialize the *committed* catalog to bytes. v0.6 was full-snapshot; v0.9
2809 /// adds the rule that an open TX's shadow is never snapshotted — only the
2810 /// post-COMMIT state is persisted. v4.1 wraps the catalog in an envelope
2811 /// when there are users to persist; an empty user table snapshots as the
2812 /// bare catalog format (backwards-compat with v3.x readers). v6.1.2
2813 /// adds publications to the envelope condition: either non-empty
2814 /// users OR non-empty publications now triggers the envelope path.
2815 pub fn snapshot(&self) -> Vec<u8> {
2816 self.snapshot_data().serialize()
2817 }
2818
2819 /// True when at least one TX slot is in flight. v4.41.1 runtime
2820 /// invariant: at most one slot active at a time (dispatch holds
2821 /// `engine.write()` across the entire wrap). v4.42 will let this
2822 /// return true with multiple slots concurrently.
2823 pub fn in_transaction(&self) -> bool {
2824 !self.tx_catalogs.is_empty()
2825 }
2826
2827 /// v7.37 C.5 (A.2) — per-connection in-transaction test. A given
2828 /// connection is "in a transaction" iff its own `tx_id` has an open
2829 /// shadow slot. Unlike [`in_transaction`] (which is true if *any* tx is
2830 /// open), this lets concurrent connections each carry their own explicit
2831 /// transaction without colliding on the global slot. `IMPLICIT_TX` never
2832 /// has a persistent slot (autocommit reads/writes the main catalog), so
2833 /// this is false for the autocommit id.
2834 pub fn is_tx_open(&self, tx_id: TxId) -> bool {
2835 self.tx_catalogs.contains_key(&tx_id)
2836 }
2837
2838 /// v7.37 (round 828) — the user store THIS session should read:
2839 /// its transaction's role shadow when one exists, the committed
2840 /// store otherwise. The auth path and other sessions read
2841 /// `self.users` directly on purpose — an uncommitted role must not
2842 /// be visible to them, let alone able to log in.
2843 pub(crate) fn effective_users(&self) -> &crate::users::UserStore {
2844 if let Some(tx) = self.current_tx
2845 && let Some(state) = self.tx_catalogs.get(&tx)
2846 && let Some(shadow) = &state.users
2847 {
2848 return shadow;
2849 }
2850 &self.users
2851 }
2852
2853 /// v7.37 (round 828) — the store role DDL writes to: the TX's role
2854 /// shadow (created from the committed store on first use) inside a
2855 /// transaction, the committed store in autocommit. Every mutation
2856 /// of roles or memberships goes through here, so `BEGIN; CREATE
2857 /// ROLE r; ROLLBACK` leaves nothing behind — the shadow drops with
2858 /// the TxState — and COMMIT installs the shadow wholesale.
2859 pub(crate) fn role_ddl_users_mut(&mut self) -> &mut crate::users::UserStore {
2860 let tx_slot = self
2861 .current_tx
2862 .filter(|tx| self.tx_catalogs.contains_key(tx));
2863 match tx_slot {
2864 Some(tx) => {
2865 if self
2866 .tx_catalogs
2867 .get(&tx)
2868 .is_some_and(|state| state.users.is_none())
2869 {
2870 let committed = self.users.clone();
2871 if let Some(state) = self.tx_catalogs.get_mut(&tx) {
2872 state.users = Some(committed);
2873 }
2874 }
2875 self.tx_catalogs
2876 .get_mut(&tx)
2877 .and_then(|state| state.users.as_mut())
2878 .expect("role shadow ensured just above for an open tx slot")
2879 }
2880 None => &mut self.users,
2881 }
2882 }
2883
2884 /// v4.41.1 allocate a fresh TX handle. Used by spg-server dispatch
2885 /// to scope each implicit-wrap BEGIN..stmt..COMMIT to its own slot
2886 /// in `tx_catalogs`. v4.42 — the commit-barrier leader allocates
2887 /// one of these per task in its group, runs `BEGIN`+sql+`COMMIT`
2888 /// sequentially under a single `engine.write()` so each task's
2889 /// mutations accumulate into shared state, then either keeps the
2890 /// accumulated state (fsync OK) or restores the pre-image via
2891 /// `replace_catalog` (fsync err).
2892 pub fn alloc_tx_id(&mut self) -> TxId {
2893 let id = TxId(self.next_tx_id);
2894 self.next_tx_id = self.next_tx_id.saturating_add(1);
2895 id
2896 }
2897
2898 /// v4.42 — atomically replace the live catalog. Used by the
2899 /// commit-barrier leader to roll back a group whose batched
2900 /// fsync failed: the leader snapshots `engine.catalog().clone()`
2901 /// (O(1) Arc bump after the v4.39/v4.40 persistent migration)
2902 /// at group start, sequentially applies each task's BEGIN+sql+
2903 /// COMMIT under the same write lock to accumulate mutations
2904 /// into shared state, batches the WAL bytes, fsyncs once, and
2905 /// on failure calls this with the pre-image to undo every
2906 /// task in the group at once.
2907 ///
2908 /// **Does NOT touch `tx_catalogs` / `current_tx`.** Any
2909 /// explicit-TX slot from a concurrent client (created via the
2910 /// legacy `IMPLICIT_TX`-less dispatch path or via the future
2911 /// MVCC-readers v5+ work) has its own snapshot baked into the
2912 /// slot — restoring `self.catalog` to the pre-image leaves
2913 /// those slots untouched, exactly as they were when the leader
2914 /// took the lock. The leader's own implicit-TX slots are all
2915 /// already discarded (`exec_commit` removed them as each
2916 /// task's COMMIT ran) by the time this is reached.
2917 pub fn replace_catalog(&mut self, catalog: Catalog) {
2918 self.catalog = catalog;
2919 }
2920
2921 /// v6.7.0 — public shim around `Catalog::freeze_oldest_to_cold`
2922 /// so tests + the spg-server freezer can drive a freeze without
2923 /// reaching into the private `active_catalog_mut`. v6.7.4
2924 /// parallel freezer will build on this surface.
2925 ///
2926 /// Marks the table's cached `cold_row_count` stale because the
2927 /// freeze added cold locators that ANALYZE hasn't yet refreshed.
2928 pub fn freeze_oldest_to_cold(
2929 &mut self,
2930 table_name: &str,
2931 index_name: &str,
2932 max_rows: usize,
2933 ) -> Result<spg_storage::FreezeReport, EngineError> {
2934 let report = self
2935 .active_catalog_mut()
2936 .freeze_oldest_to_cold(table_name, index_name, max_rows)
2937 .map_err(EngineError::Storage)?;
2938 if let Some(t) = self.active_catalog_mut().get_mut(table_name) {
2939 t.mark_cold_row_count_stale();
2940 }
2941 Ok(report)
2942 }
2943
2944 /// v6.7.5 — public shim used by the spg-server follower's
2945 /// segment-forwarding receiver. Registers a cold-tier segment
2946 /// at a specific id (the master's id, as transmitted on the
2947 /// wire) so the follower's BTree-Cold locators stay byte-
2948 /// identical with the master's. Wraps
2949 /// `Catalog::load_segment_bytes_at` under the standard
2950 /// clone-mutate-replace pattern.
2951 ///
2952 /// Returns `Ok(())` on success **and** on the "slot already
2953 /// occupied" case — a follower mid-reconnect may receive a
2954 /// segment chunk for a segment_id it already has on disk
2955 /// (forwarded last session); the caller should treat that
2956 /// path as a no-op rather than a fatal error.
2957 pub fn receive_cold_segment(
2958 &mut self,
2959 segment_id: u32,
2960 bytes: Vec<u8>,
2961 ) -> Result<(), EngineError> {
2962 let mut new_cat = self.catalog.clone();
2963 match new_cat.load_segment_bytes_at(segment_id, bytes) {
2964 Ok(()) => {
2965 self.replace_catalog(new_cat);
2966 Ok(())
2967 }
2968 Err(StorageError::Corrupt(msg)) if msg.contains("already occupied") => Ok(()),
2969 Err(e) => Err(EngineError::Storage(e)),
2970 }
2971 }
2972
2973 /// v7.39 (round 598) — mutable access to the base catalog, for the
2974 /// recursive-CTE loop.
2975 ///
2976 /// It built a whole `Engine` per iteration to hold the working set:
2977 /// `Engine::restore` initialises 82 fields, and a counting allocator put
2978 /// the loop at 63 allocations and 104 kB PER ITERATION — 1 GB for a
2979 /// 10,000-row recursive CTE, none of it dependent on how much else was
2980 /// in the catalog. One engine, whose CTE table is refilled each round,
2981 /// needs this.
2982 pub(crate) fn base_catalog_mut(&mut self) -> &mut Catalog {
2983 &mut self.catalog
2984 }
2985
2986 /// v7.38.18 (S1) — the collation this database was created with.
2987 pub fn database_collation(&self) -> &str {
2988 self.active_catalog().db_collation()
2989 }
2990
2991 /// Record the collation this database is created with, once.
2992 ///
2993 /// The host calls this before the first table exists, from its
2994 /// environment — `LC_ALL`, then `LC_COLLATE`, then `LANG`, the order
2995 /// `initdb` reads them in. An existing database keeps what it was
2996 /// created with and this answers `Ok(false)`; asking for a DIFFERENT
2997 /// collation on a database that already has tables is an error, the
2998 /// same one PostgreSQL gives, because every index key in it was
2999 /// built under the collation it has.
3000 ///
3001 /// # Errors
3002 /// When the database already has a different collation in force.
3003 pub fn set_database_collation(&mut self, name: &str) -> Result<bool, EngineError> {
3004 // v7.38.18 (G2) — a name PostgreSQL does not have is not a
3005 // collation. Before this, ICU's fallback to root made `zz_ZZ` a
3006 // perfectly acceptable database collation.
3007 if !crate::collate::is_known(name) {
3008 return Err(crate::collate::unknown_collation_error(name));
3009 }
3010 if !crate::collate::is_supported(name) {
3011 return Err(EngineError::Unsupported(alloc::format!(
3012 "collation {name:?} is not one this build can perform; \
3013 a database created under it could not be read back"
3014 )));
3015 }
3016 let changed = self
3017 .catalog
3018 .set_db_collation(name)
3019 .map_err(EngineError::Storage)?;
3020 Ok(changed)
3021 }
3022
3023 /// `CREATE DATABASE … LC_COLLATE 'x'` — the SQL path, as opposed to
3024 /// [`Self::set_database_collation`]'s host-environment one.
3025 ///
3026 /// Returns whether it took effect. A `false` means a table already
3027 /// exists, and the caller warns; see
3028 /// [`Catalog::declare_db_collation`] for why that is not an error.
3029 ///
3030 /// # Errors
3031 /// When the name is not a collation PostgreSQL has, or is one this
3032 /// build cannot perform — the same two refusals as the other path.
3033 pub fn declare_database_collation(&mut self, name: &str) -> Result<bool, EngineError> {
3034 if !crate::collate::is_known(name) {
3035 return Err(crate::collate::unknown_collation_error(name));
3036 }
3037 if !crate::collate::is_supported(name) {
3038 return Err(EngineError::Unsupported(alloc::format!(
3039 "collation {name:?} is not one this build can perform; \
3040 a database created under it could not be read back"
3041 )));
3042 }
3043 Ok(self.catalog.declare_db_collation(name))
3044 }
3045
3046 pub(crate) fn active_catalog(&self) -> &Catalog {
3047 match self.current_tx {
3048 Some(t) => self
3049 .tx_catalogs
3050 .get(&t)
3051 .map_or(&self.catalog, |s| &s.catalog),
3052 None => &self.catalog,
3053 }
3054 }
3055
3056 fn active_catalog_mut(&mut self) -> &mut Catalog {
3057 let tx = self.current_tx;
3058 match tx {
3059 Some(t) => match self.tx_catalogs.get_mut(&t) {
3060 Some(s) => {
3061 // v7.39 (round 494) — see `TxState::shadow_dirty`.
3062 s.shadow_dirty = true;
3063 &mut s.catalog
3064 }
3065 None => &mut self.catalog,
3066 },
3067 None => &mut self.catalog,
3068 }
3069 }
3070
3071 /// v7.34 (crash-recovery P0 #2) — turn row-level redo capture on/off.
3072 /// The embedding layer enables it when persistence is on so each
3073 /// mutating `execute` records the physical [`RowChange`]s it applied
3074 /// (drained via [`Engine::take_redo`]). Off = zero capture overhead.
3075 pub fn set_redo_capture(&mut self, on: bool) {
3076 self.redo_capture = on;
3077 }
3078
3079 /// v7.39 (round 735, S14/B3) — record that `table`'s rows (or shape)
3080 /// changed. Cheap (one BTreeMap bump), called from every write entry;
3081 /// the materialized-view refresh watermark reads it.
3082 /// v7.39 (round 736) — per-view buffered-delta ceiling. Past this,
3083 /// the view's next REFRESH is a full one (the buffer is the
3084 /// optimisation, not the truth).
3085 pub(crate) const MATVIEW_DELTA_CEILING: usize = 65_536;
3086
3087 /// v7.39 (round 736) — fan the drained redo out to every
3088 /// maintainable view whose base table it touches.
3089 pub(crate) fn fan_out_matview_deltas(&mut self, drained: &[RowChange]) {
3090 if self.matview_maintainable.is_empty() {
3091 return;
3092 }
3093 for ch in drained {
3094 let t = ch.table_name().to_ascii_lowercase();
3095 let hit: Vec<String> = self
3096 .matview_maintainable
3097 .iter()
3098 .filter(|(_, base)| **base == t)
3099 .map(|(mv, _)| mv.clone())
3100 .collect();
3101 for mv in hit {
3102 if self.matview_delta_overflow.contains(&mv) {
3103 continue;
3104 }
3105 let buf = self.matview_delta_buf.entry(mv.clone()).or_default();
3106 if buf.len() >= Self::MATVIEW_DELTA_CEILING {
3107 self.matview_delta_overflow.insert(mv.clone());
3108 self.matview_delta_buf.remove(&mv);
3109 } else {
3110 buf.push(ch.clone());
3111 MATVIEW_FANOUT_BUFFERED.fetch_add(1, core::sync::atomic::Ordering::Relaxed);
3112 }
3113 }
3114 }
3115 }
3116
3117 pub(crate) fn bump_table_change(&mut self, table: &str) {
3118 let k = table.to_ascii_lowercase();
3119 *self.table_change_seq.entry(k).or_insert(0) += 1;
3120 }
3121
3122 /// v7.37.8 — read accessor for tests / observability. The
3123 /// embedding layer flips this on once per `open_path` (after
3124 /// replay completes) when `SPG_WAL_ROW_REDO` is enabled (now
3125 /// default in v7.37.8). A consumer that wants to verify the
3126 /// post-upgrade contract ("writes go to V5 ROW_REDO by default")
3127 /// reads this through `Database::engine_redo_capture()` instead
3128 /// of inspecting WAL bytes (which the auto-checkpoint truncates
3129 /// on `Drop`).
3130 pub fn redo_capture_enabled(&self) -> bool {
3131 self.redo_capture
3132 }
3133
3134 /// v7.38 轴 4 — currently-selected SQL isolation level. Default
3135 /// `ReadCommitted` after construction; updated by
3136 /// `SET TRANSACTION ISOLATION LEVEL …`. Read by
3137 /// `SHOW transaction_isolation` and any future MVCC/SSI gate.
3138 pub fn current_isolation_level(&self) -> spg_sql::ast::IsolationLevel {
3139 self.current_isolation_level
3140 }
3141
3142 /// v7.34 — take the redo captured by the most recent successful
3143 /// mutating `execute` (empty when capture is off, the statement was a
3144 /// read, or it changed nothing). The embedding layer writes these to
3145 /// the WAL in place of the SQL text.
3146 pub fn take_redo(&mut self) -> Vec<RowChange> {
3147 core::mem::take(&mut self.last_redo)
3148 }
3149
3150 /// v7.34 (crash-recovery P0 #2) — replay a row-level redo log onto the
3151 /// committed catalog (the row-level WAL recovery primitive: apply the
3152 /// captured physical changes from a checkpoint baseline, in place of
3153 /// re-executing the SQL). Trusts the log — no uniqueness/FK/parse.
3154 pub fn apply_redo(&mut self, changes: &[RowChange]) -> Result<(), EngineError> {
3155 self.catalog
3156 .apply_redo(changes)
3157 .map_err(EngineError::Storage)
3158 }
3159
3160 /// Read-only execute path. Succeeds for `SELECT` / `SHOW TABLES`
3161 /// / `SHOW COLUMNS`; returns `EngineError::WriteRequired` for
3162 /// every other statement, so the caller can fall through to the
3163 /// `&mut self` `execute` path under a write lock. Engine state is
3164 /// not mutated even on the success path (`rewrite_clock_calls`
3165 /// and `resolve_order_by_position` both mutate the locally-owned
3166 /// AST, not `self`).
3167 ///
3168 /// v4.2: cap result-set size. Applied after the executor
3169 /// materialises rows but before they leave the engine — wrapping
3170 /// every Rows-returning exec_* function would scatter the check.
3171 ///
3172 /// v7.31 (memory campaign, bucket A) — the same choke point now
3173 /// also enforces the BYTE budget on the final result set, so
3174 /// single-table and aggregate paths (which don't route through
3175 /// the join materialiser's incremental accounting) still cannot
3176 /// hand the host an unbounded result. Intermediate single-table
3177 /// clones are the 7.31.x follow-up (design doc, bucket A).
3178 fn enforce_row_limit(
3179 &self,
3180 result: Result<QueryResult, EngineError>,
3181 ) -> Result<QueryResult, EngineError> {
3182 if let Ok(QueryResult::Rows { rows, .. }) = &result {
3183 if let Some(cap) = self.max_query_rows
3184 && rows.len() > cap
3185 {
3186 return Err(EngineError::RowLimitExceeded(cap));
3187 }
3188 if let Some(byte_cap) = self.max_query_bytes
3189 && approx_rows_bytes(rows) > byte_cap
3190 {
3191 return Err(EngineError::QueryBytesExceeded(byte_cap));
3192 }
3193 }
3194 result
3195 }
3196}
3197
3198/// v7.31 (memory campaign — ceiling-first / never-die, design v1) —
3199/// per-table slice of the engine's resident-memory accounting.
3200/// `hot_encoded_bytes` is the storage layer's maintained meter (what
3201/// the rows encode to); `approx_resident_bytes` is what they COST in
3202/// RAM (per-cell enum slots + heap payloads via `approx_row_bytes`)
3203/// — the gap between the two is the representation multiplier the
3204/// round-26 report measured at ~11× end-to-end.
3205#[derive(Debug, Clone)]
3206pub struct TableMemoryStats {
3207 pub name: String,
3208 pub hot_rows: u64,
3209 /// Cached cold-row count (refreshed by ANALYZE — see
3210 /// `Table::cold_row_count`'s staleness contract).
3211 pub cold_rows: u64,
3212 pub hot_encoded_bytes: u64,
3213 pub approx_resident_bytes: u64,
3214 pub index_count: u64,
3215 /// v7.31 C2 — sum of `IndexKind::approx_resident_bytes()` over the
3216 /// table's indices: every variant (BTree / NSW / BRIN / GIN family)
3217 /// walks its own structure, so the GIN posting lists and NSW layer
3218 /// adjacency that dominate text/vector tables are counted honestly
3219 /// instead of the old flat-token estimate.
3220 pub approx_index_bytes: u64,
3221}
3222
3223/// v7.31 — whole-engine memory snapshot: the polling form of the
3224/// round-26 ask-4 watermark signal. Hosts compare
3225/// `total_approx_resident_bytes` (+ their own WAL/file accounting)
3226/// against their deployment ceiling and shed/shrink before the
3227/// kernel does it for them.
3228#[derive(Debug, Clone)]
3229pub struct MemoryStats {
3230 pub tables: Vec<TableMemoryStats>,
3231 pub total_hot_encoded_bytes: u64,
3232 pub total_approx_resident_bytes: u64,
3233 pub total_approx_index_bytes: u64,
3234 /// The active per-query materialisation budget (bucket A), so a
3235 /// monitoring host sees ceiling and usage through one call.
3236 pub max_query_bytes: Option<usize>,
3237 /// v7.31 C2 — bucket D: live WAL bytes (active chunk + buffered,
3238 /// uncheckpointed). `None` from the engine itself — it has no WAL;
3239 /// the durable hosts (embed `Database`, server) fill it in from
3240 /// their own WAL accounting. `Some(0)` means "host has a WAL and
3241 /// it is empty"; `None` means "no WAL on this path" (in-memory).
3242 pub wal_bytes: Option<u64>,
3243}
3244
3245/// v6.2.0 — true for engine-managed catalog tables that the bare
3246/// `ANALYZE` (no target) should skip. v6.2.0 has no internal
3247/// tables yet (publications / subscriptions / users / statistics
3248/// all live as engine fields, not catalog tables), so this is a
3249/// reserved future-proofing hook — every existing user table is
3250/// analysed.
3251const fn is_internal_table_name(_name: &str) -> bool {
3252 false
3253}
3254
3255impl Engine {
3256 /// v7.38.18 (C12) — how many warnings the last warning-generating
3257 /// statement produced: `@@warning_count`, and what a wire shim
3258 /// reports in its OK packet.
3259 #[must_use]
3260 pub fn mysql_warning_count(&self) -> usize {
3261 self.mysql_warnings.len()
3262 }
3263
3264 /// The diagnostics area itself, for `SHOW WARNINGS`.
3265 #[must_use]
3266 pub fn mysql_warnings(&self) -> &[MysqlWarning] {
3267 &self.mysql_warnings
3268 }
3269}
3270
3271#[cfg(test)]
3272mod tests;