Skip to main content

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