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