Skip to main content

spg_engine/
execute.rs

1//! Statement execution + prepared-statement dispatch, split out of
2//! `lib.rs` (lib.rs split 17). The public `execute` / `execute_in` /
3//! `execute_with_cancel` entry points, the `prepare` / `prepare_cached`
4//! / `describe_prepared` / `execute_prepared` prepared-statement path,
5//! and the internal pipeline (`execute_inner_with_cancel` →
6//! `execute_stmt_with_cancel`) that pre-resolves clock / sequence /
7//! placeholder rewrites and routes each parsed Statement to its domain
8//! handler (DDL / DML / SELECT / transaction / SHOW / …). Whole
9//! `impl Engine` methods reached via the Engine type, so the public
10//! surface is unchanged; `execute_stmt_with_cancel` is pub(crate) for
11//! the plpgsql + trigger re-entry paths.
12
13use alloc::string::String;
14use alloc::vec::Vec;
15
16use spg_sql::ast::Statement;
17use spg_sql::parser::{self, ParseError};
18use spg_storage::{ColumnSchema, Value};
19
20use crate::describe;
21use crate::{
22    CancelToken, Engine, EngineError, IMPLICIT_TX, QueryResult, TxId, expand_group_by_all,
23    plan_cache, reorder, resolve_order_by_position, rewrite_clock_calls, substitute_placeholders,
24};
25
26/// v7.38 Epic P — turn a caught panic payload into an
27/// [`EngineError::Internal`]. Recovers a human-readable detail from the
28/// common payload shapes (`&str` / `String`, and the injection framework's
29/// typed `InjectedError`) so the wire layer sends a clean message; falls
30/// back to a generic string when the payload type is opaque.
31#[cfg(feature = "std")]
32/// v7.38 (read01 P3.17) — reject a clearly-invalid value for a handful of
33/// well-known typed GUCs (boolean / memory-size / duration), so a typo
34/// like `SET work_mem = 'bogus'` errors like PG instead of silently
35/// storing junk. Conservative: only GUCs whose type is unambiguous are
36/// checked; every other name is accepted so pg_dump preambles and
37/// unknown settings still load.
38fn validate_known_guc(name: &str, value: &str) -> Result<(), EngineError> {
39    let key = name.to_ascii_lowercase();
40    let bad = || {
41        EngineError::Unsupported(alloc::format!(
42            "invalid value for parameter \"{name}\": \"{value}\""
43        ))
44    };
45    let is_bool = matches!(
46        value.trim().to_ascii_lowercase().as_str(),
47        "on" | "off" | "true" | "false" | "yes" | "no" | "1" | "0"
48    );
49    // Split a `<number><unit>` GUC value into its numeric head + unit tail.
50    let split_unit = |s: &str| -> (String, String) {
51        let st = s.trim();
52        let cut = st
53            .find(|c: char| c.is_ascii_alphabetic())
54            .unwrap_or(st.len());
55        (
56            String::from(st[..cut].trim()),
57            st[cut..].trim().to_ascii_lowercase(),
58        )
59    };
60    let (num, unit) = split_unit(value);
61    let is_size =
62        num.parse::<f64>().is_ok() && matches!(unit.as_str(), "" | "b" | "kb" | "mb" | "gb" | "tb");
63    let is_duration = num.parse::<i64>().is_ok()
64        && matches!(unit.as_str(), "" | "us" | "ms" | "s" | "min" | "h" | "d");
65    match key.as_str() {
66        "enable_seqscan"
67        | "enable_indexscan"
68        | "enable_bitmapscan"
69        | "enable_indexonlyscan"
70        | "enable_hashjoin"
71        | "enable_mergejoin"
72        | "enable_nestloop"
73        | "autovacuum"
74        | "fsync"
75        | "full_page_writes" => {
76            if !is_bool {
77                return Err(bad());
78            }
79        }
80        "work_mem"
81        | "maintenance_work_mem"
82        | "shared_buffers"
83        | "temp_buffers"
84        | "effective_cache_size"
85        | "wal_buffers" => {
86            if !is_size {
87                return Err(bad());
88            }
89        }
90        "statement_timeout" | "lock_timeout" | "idle_in_transaction_session_timeout" => {
91            if !is_duration {
92                return Err(bad());
93            }
94        }
95        // v7.39 (round 171) — synchronous_commit is a real, session-level
96        // durability control now (the embedded execute path gates its
97        // WAL-fsync wait on it); validate PG's value domain.
98        "synchronous_commit" => {
99            if !matches!(
100                value.to_ascii_lowercase().as_str(),
101                "on" | "off"
102                    | "local"
103                    | "remote_write"
104                    | "remote_apply"
105                    | "true"
106                    | "false"
107                    | "0"
108                    | "1"
109            ) {
110                return Err(bad());
111            }
112        }
113        // v7.39 (round 204) — enum GUCs reject an out-of-domain value
114        // like PG (`SET client_min_messages = bogus` errors). PG's
115        // message quotes the value with a trailing hint listing the
116        // valid set; we match the leading, stable clause.
117        "client_min_messages" => {
118            if !matches!(
119                value.trim().to_ascii_lowercase().as_str(),
120                "debug5"
121                    | "debug4"
122                    | "debug3"
123                    | "debug2"
124                    | "debug1"
125                    | "log"
126                    | "notice"
127                    | "warning"
128                    | "error"
129                    | "fatal"
130                    | "panic"
131            ) {
132                return Err(bad());
133            }
134        }
135        // v7.39 (GUC knife 3) — the render GUCs reject invalid values
136        // with PG's own texts (canonical-caps parameter names).
137        "datestyle" => {
138            if crate::session::parse_datestyle_parts(value, crate::eval::RenderStyle::default())
139                .is_none()
140            {
141                return Err(EngineError::Unsupported(alloc::format!(
142                    "invalid value for parameter \"DateStyle\": \"{value}\""
143                )));
144            }
145        }
146        "intervalstyle" => {
147            if crate::session::parse_intervalstyle(value).is_none() {
148                return Err(EngineError::Unsupported(alloc::format!(
149                    "invalid value for parameter \"IntervalStyle\": \"{value}\""
150                )));
151            }
152        }
153        "extra_float_digits" => match value.trim().parse::<i64>() {
154            Ok(n) if (-15..=3).contains(&n) => {}
155            Ok(n) => {
156                return Err(EngineError::Unsupported(alloc::format!(
157                    "{n} is outside the valid range for parameter \
158                         \"extra_float_digits\" (-15 .. 3)"
159                )));
160            }
161            Err(_) => return Err(bad()),
162        },
163        _ => {}
164    }
165    Ok(())
166}
167
168fn panic_payload_to_engine_error(payload: &(dyn core::any::Any + Send)) -> EngineError {
169    // The injection framework panics with a typed error; surface its
170    // message so tests get a deterministic, informative string.
171    #[cfg(feature = "injection-points")]
172    if let Some(inj) = payload.downcast_ref::<crate::testkit::injection::InjectedError>() {
173        return EngineError::Internal(alloc::format!("query aborted by internal error: {inj}"));
174    }
175    let detail = payload
176        .downcast_ref::<&'static str>()
177        .map(|s| String::from(*s))
178        .or_else(|| payload.downcast_ref::<String>().cloned());
179    match detail {
180        Some(d) => EngineError::Internal(alloc::format!("query aborted by internal error: {d}")),
181        None => EngineError::Internal(String::from("query aborted by internal error")),
182    }
183}
184
185impl Engine {
186    pub fn execute(&mut self, sql: &str) -> Result<QueryResult, EngineError> {
187        self.execute_in_with_cancel(sql, IMPLICIT_TX, CancelToken::none())
188    }
189
190    /// v7.38 (read01 P3.20) — handle a bare `SELECT set_config(name, value,
191    /// is_local)` by writing the GUC to the same session store `SET` uses
192    /// (honouring `is_local` via the transaction undo log), so set_config,
193    /// SHOW, current_setting, and pg_settings all agree. Returns `None`
194    /// (fall through to the ordinary read-only path) unless the statement is
195    /// exactly that shape — set_config buried in a FROM/WHERE/CTE, or over a
196    /// non-text name, keeps the old value-returning behaviour.
197    fn try_exec_set_config(
198        &mut self,
199        s: &spg_sql::ast::SelectStatement,
200    ) -> Result<Option<QueryResult>, EngineError> {
201        use spg_sql::ast::{Expr, SelectItem};
202        if s.from.is_some() || s.where_.is_some() || !s.ctes.is_empty() || s.items.len() != 1 {
203            return Ok(None);
204        }
205        let SelectItem::Expr { expr, .. } = &s.items[0] else {
206            return Ok(None);
207        };
208        let Expr::FunctionCall { name, args } = expr else {
209            return Ok(None);
210        };
211        if !(name.eq_ignore_ascii_case("set_config")
212            || name.eq_ignore_ascii_case("pg_catalog.set_config"))
213            || !(args.len() == 2 || args.len() == 3)
214        {
215            return Ok(None);
216        }
217        // Evaluate the arguments against an empty row.
218        let empty: Vec<ColumnSchema> = Vec::new();
219        let (name_v, value_v, local_v);
220        {
221            let ctx = self.ev_ctx(&empty, None);
222            let dummy = spg_storage::Row::new(Vec::new());
223            name_v = crate::eval::eval_expr(&args[0], &dummy, &ctx).map_err(EngineError::Eval)?;
224            value_v = crate::eval::eval_expr(&args[1], &dummy, &ctx).map_err(EngineError::Eval)?;
225            local_v = if args.len() == 3 {
226                crate::eval::eval_expr(&args[2], &dummy, &ctx).map_err(EngineError::Eval)?
227            } else {
228                Value::Bool(false)
229            };
230        }
231        let single = |v: Value<'static>| QueryResult::Rows {
232            columns: alloc::vec![ColumnSchema::new(
233                "set_config",
234                spg_storage::DataType::Text,
235                true
236            )],
237            rows: alloc::vec![spg_storage::Row::new(alloc::vec![v])],
238        };
239        let pname = match name_v {
240            Value::Text(s) => s.into_owned(),
241            // set_config(NULL, …) is a no-op returning NULL (PG).
242            Value::Null => return Ok(Some(single(Value::Null))),
243            _ => return Ok(None),
244        };
245        let is_local = matches!(local_v, Value::Bool(true));
246        // A NULL value resets the GUC to its default (PG), returning NULL.
247        let pval = match value_v {
248            Value::Text(s) => s.into_owned(),
249            Value::Null => {
250                self.session_params.remove(&pname.to_ascii_lowercase());
251                self.refresh_render_style();
252                return Ok(Some(single(Value::Null)));
253            }
254            _ => return Ok(None),
255        };
256        validate_known_guc(&pname, &pval)?;
257        if is_local {
258            if self.in_transaction() {
259                let prior = self.session_param(&pname).map(String::from);
260                self.local_guc_saves.push((pname.clone(), prior));
261                self.set_session_param(pname, spg_sql::ast::SetValue::String(pval.clone()));
262            }
263        } else {
264            self.set_session_param(pname, spg_sql::ast::SetValue::String(pval.clone()));
265        }
266        Ok(Some(single(Value::text(pval))))
267    }
268
269    /// v4.5 — write path with cooperative cancellation. Same dispatch
270    /// as `execute_in_with_cancel(sql, IMPLICIT_TX, cancel)`. Kept as
271    /// a separate entry point for backward-compat with the v4.5
272    /// public API.
273    pub fn execute_with_cancel(
274        &mut self,
275        sql: &str,
276        cancel: CancelToken<'_>,
277    ) -> Result<QueryResult, EngineError> {
278        self.execute_in_with_cancel(sql, IMPLICIT_TX, cancel)
279    }
280
281    /// v4.41.1 multi-slot write entry. Routes `sql` through the TX
282    /// slot identified by `tx_id` so spg-server dispatch can scope
283    /// each implicit-wrap BEGIN..stmt..COMMIT to its own slot in
284    /// `tx_catalogs`. `IMPLICIT_TX` is the legacy single-slot path
285    /// every other caller (engine self-tests, replay, spg-embedded)
286    /// implicitly takes via `execute()` / `execute_with_cancel()`.
287    pub fn execute_in(&mut self, sql: &str, tx_id: TxId) -> Result<QueryResult, EngineError> {
288        self.execute_in_with_cancel(sql, tx_id, CancelToken::none())
289    }
290
291    /// v4.41.1 write path with cooperative cancellation + explicit TX
292    /// scope. Sets `self.current_tx` for the duration of the call so
293    /// every `exec_*` helper transparently sees its TX's shadow
294    /// catalog and savepoint stack; restores on exit so the field is
295    /// only valid mid-call (no leakage across calls).
296    pub fn execute_in_with_cancel(
297        &mut self,
298        sql: &str,
299        tx_id: TxId,
300        cancel: CancelToken<'_>,
301    ) -> Result<QueryResult, EngineError> {
302        // v7.38 P0 元机制 A — establish the per-engine injection
303        // scope for the duration of this execute. The guard pops
304        // the store on drop so nested or sibling engines don't see
305        // ours. No-op in release builds (feature off).
306        let _inj = self.enter_injection_scope();
307        // v7.39 (read01 round 46) — NOTICEs are per-statement: clear the
308        // buffer here so one statement's "…, skipping" can never leak into
309        // the next one's NoticeResponse batch.
310        self.pending_notices.clear();
311        let saved = self.current_tx;
312        self.current_tx = Some(tx_id);
313        // v7.37.15 (Epic W slice 2) — memoized autocommit writer version
314        // is scoped to one statement. Save + reset like `current_tx` so
315        // a re-entrant execute (e.g. deferred trigger SQL) can't leak its
316        // version into ours, and ours never leaks to the next statement.
317        let saved_stmt_wv = self.stmt_writer_version;
318        self.stmt_writer_version = None;
319        // v7.34 (crash-recovery P0 #2) — row-level redo capture. Arm the
320        // active catalog before dispatch; on success drain the physical
321        // changes into `last_redo` for the embedding layer's WAL, on
322        // failure discard them (a failed statement leaves no redo; the
323        // drain clears the tables' capture buffers either way).
324        // v7.39 (round 736, S14/B3) — a delta-maintainable materialized
325        // view needs the same physical change stream the WAL reads, so
326        // its presence enables capture too (the fan-out below copies;
327        // `last_redo` stays the embedding layer's alone).
328        let matview_capture = !self.matview_maintainable.is_empty();
329        if self.redo_capture || matview_capture {
330            self.active_catalog_mut().enable_redo_all();
331        }
332        // v7.38 Epic P (panic isolation) — run statement execution
333        // behind a catch_unwind firewall so a panic in query
334        // processing surfaces as an ordinary `EngineError` (after
335        // rolling the in-flight tx back) instead of unwinding through
336        // the server's engine `RwLock` write guard (which would poison
337        // it) or aborting the process. NO-OP under the release
338        // `panic = "abort"` profile — the process aborts before any
339        // unwind reaches here; active in dev/test (`panic = "unwind"`)
340        // and once a later slice flips the release profile.
341        let pre_in_tx = self.in_transaction();
342        let result = self.execute_inner_catching(sql, cancel);
343        // v7.39 (round 426) — MySQL's ROW_COUNT() reads what the LAST
344        // statement did. Measured on MariaDB 11: a DML statement leaves the
345        // number of rows it changed (0 when it matched none), a
346        // row-returning statement leaves -1, and DDL leaves 0. One place,
347        // because every statement funnels through here — and it must be
348        // AFTER the dispatch, so ROW_COUNT()'s own SELECT is what sets -1
349        // for the call after it (as MariaDB does).
350        //
351        // A failed statement leaves the previous value alone: MariaDB keeps
352        // the last successful statement's count through an error.
353        if let Ok(res) = &result {
354            self.row_count = match res {
355                QueryResult::CommandOk { affected, .. } => i64::try_from(*affected).unwrap_or(-1),
356                QueryResult::Rows { .. } => -1,
357            };
358        }
359        // v7.39 (pg_stat knife A) — PG counts every statement outside a
360        // transaction block as one implicit xact (commit on success,
361        // rollback on error). Statements INSIDE a block are counted
362        // once, by exec_commit / exec_rollback; BEGIN itself (state
363        // flips outside -> inside) and the block-closers (inside ->
364        // outside, counted in their exec fns) are skipped here.
365        if !pre_in_tx && !self.in_transaction() {
366            let ctr = if result.is_ok() {
367                &self.xact_commit
368            } else {
369                &self.xact_rollback
370            };
371            ctr.fetch_add(1, core::sync::atomic::Ordering::Relaxed);
372        }
373        // r196 — a statement that did NOT run inside its own open tx
374        // slot (autocommit, or a COMMIT/ROLLBACK that just closed its
375        // slot) may have moved the committed base; bump the epoch so
376        // OTHER open txs know their next RC rebase is real. The test
377        // must be per-statement (`tx_catalogs` membership of THIS
378        // call's tx_id), not the global `in_transaction()` — a
379        // concurrent autocommit while some tx is open is exactly the
380        // case the rebase exists for (the first cut used the global
381        // check and 10 isolation pins caught the missed bumps).
382        // Deliberately over-approximate (reads bump too — an extra
383        // rebase is only slower, never wrong).
384        if !self.tx_catalogs.contains_key(&tx_id) {
385            self.commit_epoch = self.commit_epoch.wrapping_add(1);
386            // v7.39 (round 306) — large-object descriptors live only as
387            // long as the transaction that opened them, so this is
388            // exactly where they die: an autocommit statement (the
389            // implicit transaction just ended) or the COMMIT / ROLLBACK
390            // that closed the slot. Numbering restarts from 0, as PG's
391            // does. Same per-slot witness as the epoch bump above —
392            // another connection's open transaction must not keep this
393            // one's descriptors alive.
394            self.lo_descriptors.clear();
395            self.lo_next_fd = 0;
396        }
397        if self.redo_capture || matview_capture {
398            let mut drained = self.active_catalog_mut().drain_redo();
399            if result.is_ok() {
400                if matview_capture {
401                    self.fan_out_matview_deltas(&drained);
402                }
403                // v7.37.15 (Epic W slice 2) — stamp the real committing
404                // writer version onto every change this statement
405                // produced. All changes from one statement share the one
406                // version (the statement's xmin/xmax): in autocommit it's
407                // the memoized value the writes already used; inside an
408                // explicit tx it's the deterministic tx entry. Purely
409                // additive metadata — replay still resolves by physical
410                // position and ignores `writer_version` (later slice).
411                if !drained.is_empty() {
412                    let v = self.writer_version_for_current_stmt();
413                    for change in &mut drained {
414                        change.set_writer_version(v);
415                    }
416                }
417                if self.redo_capture {
418                    self.last_redo = drained;
419                }
420            }
421        }
422        self.current_tx = saved;
423        self.stmt_writer_version = saved_stmt_wv;
424        result
425    }
426
427    /// v6.1.1 — parse and pre-process a SQL string ONCE so the
428    /// resulting [`Statement`] can be cached and re-executed via
429    /// [`Engine::execute_prepared`]. Returns the same `Statement`
430    /// the simple-query path would synthesise internally (clock
431    /// rewrites + ORDER BY position-ref resolution applied at
432    /// prepare time, since both are session-independent). The
433    /// `$N` placeholders in the SQL stay as `Expr::Placeholder(n)`
434    /// nodes; they're resolved to concrete values per-call by
435    /// `execute_prepared`'s substitution walk.
436    ///
437    /// Pgwire's `Parse` (P) message lands here.
438    pub fn prepare(&self, sql: &str) -> Result<Statement, ParseError> {
439        let mut stmt = parser::parse_statement_with(sql, self.backslash_escapes)?;
440        self.preprocess(&mut stmt);
441        Ok(stmt)
442    }
443
444    /// r1043 — every pre-pass a parsed statement gets before execution,
445    /// in one place.
446    ///
447    /// There were two copies. `prepare` had clock rewrites, `GROUP BY
448    /// ALL` expansion, ORDER BY position resolution and the JOIN reorder;
449    /// `execute_readonly_with_cancel` — the path EVERY autocommit SELECT
450    /// takes over the wire — had the same list minus the `GROUP BY ALL`
451    /// expansion, and then r1042 added constant folding to one of them.
452    ///
453    /// The result was a plan that `EXPLAIN` described and the wire did not
454    /// run: `WHERE b = decode(lpad(to_hex(7),16,'0'),'hex')` planned as an
455    /// index scan and took 194 ms, against 0.009 ms for the same statement
456    /// through the embedded API, on the same build and the same 400,000
457    /// rows. EXPLAIN went through `prepare`; the query did not.
458    ///
459    /// One function, both callers. A pass added here reaches every route
460    /// by construction rather than by remembering.
461    pub(crate) fn preprocess(&self, stmt: &mut Statement) {
462        let now_micros = self.clock.map(|f| f());
463        rewrite_clock_calls(
464            stmt,
465            now_micros,
466            self.backslash_escapes,
467            now_micros.map_or(0, |n| self.session_tz_offset_at(n)),
468        );
469        // r1042 — evaluate the constant parts of every predicate once,
470        // here, instead of once per row. A cast on a literal is the
471        // common case and it was costing an index seek: `WHERE id = 7`
472        // sought and `WHERE id = 7::int` scanned, 23x apart at 400k rows.
473        crate::constfold::fold_statement(stmt);
474        if let Statement::Select(s) = stmt {
475            // v6.4.1 — expand `GROUP BY ALL` to every non-aggregate
476            // SELECT-list item BEFORE position / alias resolution so
477            // downstream passes see the explicit list.
478            expand_group_by_all(s);
479            resolve_order_by_position(s);
480            // v6.2.3 — cost-based JOIN reorder. No-op for
481            // single-table FROMs or any non-INNER join shape.
482            // v7.38 元机制 D — `SPG_TEST_PLAN_DETERMINISTIC=1` gates
483            // this so regression tests pin a stable join order.
484            reorder::reorder_joins_with(
485                s,
486                &self.catalog,
487                &self.statistics,
488                self.env_cfg.plan_deterministic,
489            );
490        }
491    }
492
493    /// v6.3.0 — cached prepare. Returns a cloned `Statement` from
494    /// the plan cache on hit, runs the full `prepare()` path on miss
495    /// and inserts the resulting plan before returning. Skipping the
496    /// parse + JOIN-reorder pipeline on hit is the dominant win for
497    /// JDBC / sqlx / pgx clients that reuse the same SQL string.
498    ///
499    /// Returns a cloned `Statement` (not a borrow) because the
500    /// pgwire layer owns its `PreparedStmt` map per-session and the
501    /// engine-level cache must stay available for other sessions.
502    /// Clone cost on a 5-table JOIN AST is well under the parse cost
503    /// it replaces.
504    /// v7.39 (round 192) — bump the engine-side per-table DML
505    /// counters (pg_stat_user_tables n_tup_*). Non-transactional by
506    /// design, like PG's stats collector.
507    pub(crate) fn note_table_write(&mut self, table: &str, ins: u64, upd: u64, del: u64) {
508        let e = self
509            .table_write_stats
510            .entry(alloc::string::String::from(table))
511            .or_insert((0, 0, 0));
512        e.0 = e.0.saturating_add(ins);
513        e.1 = e.1.saturating_add(upd);
514        e.2 = e.2.saturating_add(del);
515    }
516
517    pub fn prepare_cached(&mut self, sql: &str) -> Result<Statement, ParseError> {
518        // v7.39 (round 200) — don't cache LARGE statements. A 24 KB
519        // multi-row VALUES INSERT paid a full AST deep-clone (~640 µs)
520        // just to enter the plan cache, where a unique bulk statement
521        // is never reused — and at that size a cache hit would only
522        // save the ~190 µs re-parse anyway. The threshold keeps every
523        // ORM-shaped statement (small, repeated) on the cached path.
524        const PLAN_CACHE_MAX_SQL_BYTES: usize = 4096;
525        if sql.len() > PLAN_CACHE_MAX_SQL_BYTES {
526            return self.prepare(sql);
527        }
528        // v6.3.1 — version-aware lookup. If the cached plan was
529        // prepared before the most recent ANALYZE, evict and replan.
530        let current_version = self.statistics.version();
531        if let Some(plan) = self.plan_cache.get(sql) {
532            if plan.statistics_version == current_version {
533                return Ok(plan.stmt.clone());
534            }
535            // Stale entry — fall through to evict + re-prepare.
536        }
537        self.plan_cache.evict(sql);
538        let stmt = self.prepare(sql)?;
539        let source_tables = plan_cache::collect_source_tables(&stmt);
540        let plan = plan_cache::PreparedPlan {
541            stmt: stmt.clone(),
542            statistics_version: current_version,
543            source_tables,
544            describe_columns: alloc::vec::Vec::new(),
545        };
546        self.plan_cache.insert(String::from(sql), plan);
547        Ok(stmt)
548    }
549
550    /// v6.3.0 — read-only accessor for tests and v6.3.1 invalidation.
551    pub fn plan_cache(&self) -> &plan_cache::PlanCache {
552        &self.plan_cache
553    }
554
555    /// v7.38 (mailrs prod 7.35 pool-exhaustion incident) — boot-time
556    /// plan-IR cache warm-up. Walks `sqls`, calls `prepare_cached`
557    /// on each one. Each successful prepare leaves the parsed +
558    /// reordered + clock-rewritten `Statement` in the engine-wide
559    /// plan cache; subsequent `Engine::execute` / `execute_prepared`
560    /// for the same SQL skips parse + JOIN reorder. Returns the
561    /// count of successfully cached statements.
562    ///
563    /// The mailrs `Database::new` boot path is the expected caller:
564    /// pre-warm the top-N query shapes (inbox listing, contacts
565    /// search, stats) so the first user-facing request doesn't
566    /// pay the 2-3 s first-fire cost on the readonly-blocking
567    /// sqlx pool — which (under prod concurrency) exhausts the
568    /// pool and stalls the whole UI.
569    pub fn warm_up_plan_cache(&mut self, sqls: &[&str]) -> usize {
570        let mut warmed = 0;
571        for sql in sqls {
572            if self.prepare_cached(sql).is_ok() {
573                warmed += 1;
574            }
575        }
576        warmed
577    }
578
579    /// v7.38 (mailrs prod 7.35 pool-exhaustion incident) — boot-time
580    /// cold-tier OS page-cache warm-up. Walks every table in the
581    /// active catalog, iterates the cold rows via the existing
582    /// BTree-driven `iter_cold_rows_of_table`, drops the rows on
583    /// the floor. The walk's side effect is that every cold
584    /// segment file gets mmap-read once — the OS page cache then
585    /// serves subsequent queries without disk I/O.
586    ///
587    /// Returns the total cold rows touched across all tables.
588    /// On a hot-only catalog (no `cold_segments` populated) the
589    /// call is a near-no-op.
590    pub fn warm_up_cold_tier(&self) -> usize {
591        let catalog = self.active_catalog();
592        let mut total = 0;
593        for name in catalog.table_names() {
594            if let Some(table) = catalog.get(&name) {
595                let rows = self.iter_cold_rows_of_table(table);
596                total += rows.len();
597            }
598        }
599        total
600    }
601
602    /// v6.3.0 — mutable accessor for v6.3.1 invalidation hooks.
603    pub fn plan_cache_mut(&mut self) -> &mut plan_cache::PlanCache {
604        &mut self.plan_cache
605    }
606
607    /// v6.3.3 — Describe a prepared `Statement` without executing.
608    /// Returns `(parameter_oids, output_columns)`. Empty
609    /// `output_columns` means the statement has no row-producing shape
610    /// we could resolve here — the pgwire layer maps that to `NoData`.
611    ///
612    /// v7.39 (round 462) — a SELECT over a system catalog view resolves
613    /// against the same materialised catalog execution builds, so the
614    /// two paths cannot disagree about what a system view looks like.
615    pub fn describe_prepared(&self, stmt: &Statement) -> (Vec<u32>, Vec<ColumnSchema>) {
616        if let Statement::Select(s) = stmt {
617            if crate::system_catalog::select_references_meta_view(s)
618                && let Ok(catalog) = self.meta_view_catalog(s)
619            {
620                return describe::describe_prepared(stmt, &catalog);
621            }
622            if let Some(catalog) = self.admin_view_catalog(s) {
623                return describe::describe_prepared(stmt, &catalog);
624            }
625        }
626        describe::describe_prepared(stmt, self.active_catalog())
627    }
628
629    /// v6.1.1 — execute a [`Statement`] previously returned by
630    /// [`Engine::prepare`], substituting `Expr::Placeholder(n)`
631    /// nodes for the corresponding [`Value`] in `params` (1-based
632    /// per PG: `$1` → `params[0]`). Bind-time string parameters
633    /// are decoded into typed `Value`s by the pgwire layer before
634    /// this call so the resulting AST hits the same execution
635    /// path as a simple query — no SQL re-parse.
636    ///
637    /// Pgwire's `Execute` (E) message after a `Bind` (B) lands here.
638    pub fn execute_prepared(
639        &mut self,
640        stmt: Statement,
641        params: &[Value<'static>],
642    ) -> Result<QueryResult, EngineError> {
643        self.execute_prepared_with_cancel(stmt, params, CancelToken::none())
644    }
645
646    /// v7.37 (SPGS small-query bar) — borrow-based SELECT entry for
647    /// the pgwire `Execute` hot path when the portal has no bound
648    /// parameters. Skips both the AST clone the prepared path used
649    /// to do at the pgwire call site AND the `substitute_
650    /// placeholders` walk (a no-op when params are empty). Caller
651    /// must already hold the engine write lock — read would be
652    /// cleaner, but `current_tx` mutation keeps it `&mut`.
653    pub fn execute_prepared_select_no_params(
654        &mut self,
655        stmt: &spg_sql::ast::SelectStatement,
656        cancel: CancelToken<'_>,
657    ) -> Result<QueryResult, EngineError> {
658        let saved = self.current_tx;
659        self.current_tx = Some(IMPLICIT_TX);
660        // v7.38 Epic P (panic isolation) — Slice 3: route this read-only
661        // prepared-SELECT hot path (pgwire `Execute` with no bound params)
662        // through the SAME `catch_unwind` firewall as the write paths. A
663        // panic in `exec_select_cancel` is caught inside the engine and
664        // returned as `EngineError::Internal`, so it never unwinds through
665        // the caller's engine `RwLock` write guard (poisoning it) or aborts
666        // the process. This path is read-only, so the firewall's
667        // `discard_tx_on_panic` is a no-op (no shadow / writer version to
668        // drop) — exactly right: nothing to roll back, just catch + survive.
669        // `exec_select_cancel` materialises its `QueryResult` synchronously,
670        // so the whole result is produced inside the catch (statement
671        // boundary only — no per-row cost).
672        #[cfg(feature = "std")]
673        let result = self.catch_stmt_panic(|s| s.exec_select_cancel(stmt, cancel));
674        #[cfg(not(feature = "std"))]
675        let result = self.exec_select_cancel(stmt, cancel);
676        self.current_tx = saved;
677        result
678    }
679
680    /// v7.37 — streaming SELECT for the pgwire `Execute` hot path.
681    /// Emits one `StreamItem::Header(cols)` then one
682    /// `StreamItem::Row(&[&Value])` per surviving row. Returns the
683    /// total row count for the `CommandComplete` tag.
684    ///
685    /// For shapes where the engine can stream directly (non-aggregate
686    /// join projection of bound columns, no ORDER BY / DISTINCT / etc.)
687    /// no `Vec<Row<'static>>` is materialised — cell references come straight
688    /// out of the source tables. For non-streamable shapes the engine
689    /// runs the full `exec_select_cancel`, then walks the materialised
690    /// `Vec<Row<'static>>` driving the same emit callback (no engine-side win,
691    /// but pgwire dispatches every Execute through one path).
692    pub fn execute_prepared_select_streaming<F>(
693        &mut self,
694        stmt: &spg_sql::ast::SelectStatement,
695        cancel: CancelToken<'_>,
696        mut emit: F,
697    ) -> Result<usize, EngineError>
698    where
699        F: FnMut(StreamItem<'_>) -> Result<(), EngineError>,
700    {
701        let saved = self.current_tx;
702        self.current_tx = Some(IMPLICIT_TX);
703        // v7.38 Epic P (panic isolation) — Slice 3: route the streaming
704        // read-only SELECT hot path through the SAME `catch_unwind` firewall.
705        //
706        // Catch SCOPE (verified): `exec_select_streaming` uses a *push*
707        // model — it drives the caller's `emit` callback synchronously via
708        // `?` for the header and every row (both the true-streaming
709        // `try_exec_joined_streaming` fast path and the materialising
710        // fall-back), and only returns once the whole result has been
711        // emitted. It does NOT hand a lazy iterator back to the wire layer to
712        // pull rows from later. Therefore a panic in the per-row streaming
713        // phase unwinds *inside* this call and IS caught by wrapping the one
714        // `exec_select_streaming` call — the entire streaming phase is
715        // covered, not just setup. This is a single statement-boundary catch
716        // (the `catch_unwind` landing pad is armed once, the whole emit loop
717        // runs inside it) — NOT a per-row catch, so there is no hot-path cost.
718        // Read-only, so `discard_tx_on_panic` is a no-op (correct: nothing to
719        // roll back). A panic caught mid-stream (after some rows were encoded
720        // into the wire buffer) leaves the same partial-`wbuf` + `Err` state
721        // the wire layer already handles when `emit` itself returns `Err`
722        // mid-stream, so no new torn-state concern is introduced.
723        #[cfg(feature = "std")]
724        let inner = self.catch_stmt_panic(|s| s.exec_select_streaming(stmt, cancel, &mut emit));
725        #[cfg(not(feature = "std"))]
726        let inner = self.exec_select_streaming(stmt, cancel, &mut emit);
727        self.current_tx = saved;
728        inner
729    }
730
731    /// v7.37 — internal streaming dispatcher. Phase 1: fall-back path
732    /// only — runs the materialising `exec_select_cancel`, then drives
733    /// the emit callback from the resulting `Vec<Row<'static>>`. Phase 2 will
734    /// add a true streaming path for the joined-projection shape.
735    fn exec_select_streaming<F>(
736        &mut self,
737        stmt: &spg_sql::ast::SelectStatement,
738        cancel: CancelToken<'_>,
739        emit: &mut F,
740    ) -> Result<usize, EngineError>
741    where
742        F: FnMut(StreamItem<'_>) -> Result<(), EngineError>,
743    {
744        // v7.37 — true-streaming fast path for joined-non-aggregate
745        // projection of bound columns. Skips `Vec<Row<'static>>` + per-cell
746        // `.cloned()` (about 4 ms saved on the 25 k-row PROJ shape).
747        // Unresolved subqueries / pull-up shapes / non-streamable
748        // structure (ORDER BY, DISTINCT, …) fall through to the
749        // materialising path.
750        if !crate::subquery::expr_tree_has_subquery(stmt) {
751            if let Some(n) = self.try_exec_joined_streaming(stmt, cancel, emit)? {
752                return Ok(n);
753            }
754        }
755        // Fall-back: materialise then iterate.
756        let QueryResult::Rows { columns, rows } = self.exec_select_cancel(stmt, cancel)? else {
757            return Err(EngineError::Unsupported(alloc::string::String::from(
758                "streaming SELECT got a non-Rows result",
759            )));
760        };
761        emit_materialised(&columns, &rows, cancel, emit)
762    }
763}
764
765/// Hand an already-materialised result to a streaming consumer: one
766/// `Header`, then every row, checking for cancellation as it goes.
767///
768/// v7.37 (round 824) — this loop existed three times, in
769/// `exec_select_streaming` and twice in the read-only entry points, and
770/// none of the three checked cancellation. A `statement_timeout` — and
771/// `CancelRequest`, which shares the token — therefore did not bound any
772/// shape the streaming path declines: arithmetic and function
773/// projections, `ORDER BY`, `DISTINCT`. Measured over 200k rows of 200
774/// bytes under a 120ms timeout, every one of them ran to completion,
775/// all 200000 rows, no error.
776///
777/// The loop reads like the cheap half of the work, since the rows
778/// already exist. It is not: handing them to `emit` is what encodes them
779/// and pushes them at the socket, and that is most of the elapsed time
780/// (first row out at 30ms of 400ms). So the interruption a client asked
781/// for never happened, and it never happened for the shapes most likely
782/// to need it.
783///
784/// It is one function now so that the next copy cannot go missing the
785/// check — which is how all three came to be missing it.
786pub(crate) fn emit_materialised<F>(
787    columns: &[ColumnSchema],
788    rows: &[spg_storage::Row<'static>],
789    cancel: CancelToken<'_>,
790    emit: &mut F,
791) -> Result<usize, EngineError>
792where
793    F: FnMut(StreamItem<'_>) -> Result<(), EngineError>,
794{
795    emit(StreamItem::Header(columns))?;
796    let mut cell_refs: Vec<&Value> = Vec::with_capacity(columns.len());
797    for (i, row) in rows.iter().enumerate() {
798        // Same cadence as the streaming path's own check.
799        if i.is_multiple_of(256) {
800            cancel.check()?;
801        }
802        cell_refs.clear();
803        for v in &row.values {
804            cell_refs.push(v);
805        }
806        emit(StreamItem::Row(RowCells::Refs(&cell_refs)))?;
807    }
808    Ok(rows.len())
809}
810
811/// One row's cells, in whichever shape the producer already holds them.
812///
813/// The channel used to be `&[&Value]` only, which cost a `Vec<&Value>`
814/// per row at the two producers that build their cells into a
815/// contiguous buffer: they had a `&[Value]` in hand and collected a
816/// second vector of pointers into it purely to satisfy the type. That
817/// is one heap allocation and one free per row — measured at 400k rows
818/// (round 957) as **9 ns/row**, which on a narrow scan was 54-56% of
819/// the whole walk and on a wide one 8-19%.
820///
821/// The reason it could not simply reuse one buffer is that the values
822/// buffer is refilled each row, so any pointers into it die at the top
823/// of the next iteration; only an owner of the storage (the
824/// materialising path, whose rows outlive the loop) can hoist the
825/// pointer vector out. Handing the contiguous slice over directly
826/// removes the question instead of answering it.
827///
828/// `Refs` stays for producers whose cells really are scattered (a join
829/// projecting out of several rows).
830#[derive(Debug, Clone, Copy)]
831pub enum RowCells<'a> {
832    Refs(&'a [&'a Value<'static>]),
833    Values(&'a [Value<'static>]),
834}
835
836impl<'a> RowCells<'a> {
837    pub fn len(&self) -> usize {
838        match self {
839            RowCells::Refs(v) => v.len(),
840            RowCells::Values(v) => v.len(),
841        }
842    }
843
844    pub fn is_empty(&self) -> bool {
845        self.len() == 0
846    }
847
848    pub fn get(&self, i: usize) -> Option<&'a Value<'static>> {
849        match self {
850            RowCells::Refs(v) => v.get(i).copied(),
851            RowCells::Values(v) => v.get(i),
852        }
853    }
854}
855
856/// v7.37 — one item in the streaming SELECT emit channel. The
857/// engine yields exactly one `Header` (before any row) then zero
858/// or more `Row`s. Pgwire (or any other consumer) decides how to
859/// turn those into wire bytes.
860#[derive(Debug)]
861pub enum StreamItem<'a> {
862    Header(&'a [ColumnSchema]),
863    Row(RowCells<'a>),
864}
865
866impl Engine {
867    /// v7.17.0 Phase 2.3 — prepared-statement entry that honors a
868    /// caller-supplied `CancelToken`. Mirrors `execute_prepared`'s
869    /// `current_tx` save/restore so the extended-query path stays
870    /// transactionally consistent with the simple-query path.
871    /// v7.39 (round 280) — `CREATE STATISTICS`.
872    fn exec_create_statistics(
873        &mut self,
874        name: String,
875        if_not_exists: bool,
876        kinds: alloc::vec::Vec<String>,
877        columns: alloc::vec::Vec<String>,
878        table: String,
879    ) -> Result<QueryResult, EngineError> {
880        if columns.len() < 2 {
881            return Err(EngineError::Unsupported(String::from(
882                "extended statistics require at least 2 columns",
883            )));
884        }
885        if self.active_catalog().get(&table).is_none() {
886            return Err(EngineError::Unsupported(alloc::format!(
887                "relation \"{table}\" does not exist"
888            )));
889        }
890        // PG's default kind set is all three.
891        let kinds = if kinds.is_empty() {
892            alloc::vec![String::from("d"), String::from("f"), String::from("m")]
893        } else {
894            kinds
895        };
896        let def = spg_storage::StatisticsExtDef {
897            name: name.clone(),
898            table,
899            kinds,
900            columns,
901        };
902        let cat = self.active_catalog_mut();
903        if let Err(taken) = cat.create_statistics_ext(def) {
904            if if_not_exists {
905                return Ok(QueryResult::CommandOk {
906                    affected: 0,
907                    modified_catalog: false,
908                });
909            }
910            return Err(EngineError::Unsupported(alloc::format!(
911                "statistics object \"{taken}\" already exists"
912            )));
913        }
914        Ok(QueryResult::CommandOk {
915            affected: 0,
916            modified_catalog: true,
917        })
918    }
919
920    /// v7.39 (round 280) — `DROP STATISTICS`.
921    fn exec_drop_statistics(
922        &mut self,
923        name: &str,
924        if_exists: bool,
925    ) -> Result<QueryResult, EngineError> {
926        let dropped = self.active_catalog_mut().drop_statistics_ext(name);
927        if !dropped && !if_exists {
928            return Err(EngineError::Unsupported(alloc::format!(
929                "statistics object \"{name}\" does not exist"
930            )));
931        }
932        Ok(QueryResult::CommandOk {
933            affected: 0,
934            modified_catalog: dropped,
935        })
936    }
937
938    /// v7.39 (round 277) — `PREPARE`. Session-scoped, and a duplicate
939    /// name is an error in PG rather than a silent replace.
940    fn exec_prepare(
941        &mut self,
942        name: String,
943        param_types: alloc::vec::Vec<String>,
944        body: Statement,
945        source: String,
946    ) -> Result<QueryResult, EngineError> {
947        if self.prepared_statements.contains_key(&name) {
948            return Err(EngineError::Unsupported(alloc::format!(
949                "prepared statement \"{name}\" already exists"
950            )));
951        }
952        self.prepared_statements.insert(
953            name,
954            crate::PreparedSqlStatement {
955                body,
956                param_types,
957                source,
958            },
959        );
960        Ok(QueryResult::CommandOk {
961            affected: 0,
962            modified_catalog: false,
963        })
964    }
965
966    /// v7.39 (round 277) — `EXECUTE`. The arguments evaluate as
967    /// constants and splice into the body's `$N` placeholders through
968    /// the same `execute_prepared_with_cancel` the extended-query path
969    /// uses, so a SQL EXECUTE and a wire Bind take the identical route.
970    fn exec_execute(
971        &mut self,
972        name: &str,
973        args: &[spg_sql::ast::Expr],
974        cancel: CancelToken<'_>,
975    ) -> Result<QueryResult, EngineError> {
976        let Some(entry) = self.prepared_statements.get(name) else {
977            return Err(EngineError::Unsupported(alloc::format!(
978                "prepared statement \"{name}\" does not exist"
979            )));
980        };
981        let body = entry.body.clone();
982        let empty: alloc::vec::Vec<spg_storage::ColumnSchema> = alloc::vec::Vec::new();
983        let ctx = self.ev_ctx(&empty, None);
984        let blank = spg_storage::Row::new(alloc::vec::Vec::new());
985        let mut params: alloc::vec::Vec<spg_storage::Value<'static>> =
986            alloc::vec::Vec::with_capacity(args.len());
987        for a in args {
988            params.push(crate::eval::eval_expr(a, &blank, &ctx).map_err(EngineError::Eval)?);
989        }
990        self.execute_prepared_with_cancel(body, &params, cancel)
991    }
992
993    /// v7.39 (round 277) — `DEALLOCATE <name>` / `DEALLOCATE ALL`.
994    /// Dropping a name that does not exist is an error in PG; ALL is
995    /// unconditional.
996    fn exec_deallocate(&mut self, name: Option<&str>) -> Result<QueryResult, EngineError> {
997        match name {
998            None => {
999                self.prepared_statements.clear();
1000                Ok(QueryResult::CommandOk {
1001                    affected: 0,
1002                    modified_catalog: false,
1003                })
1004            }
1005            Some(n) => {
1006                if self.prepared_statements.remove(n).is_none() {
1007                    return Err(EngineError::Unsupported(alloc::format!(
1008                        "prepared statement \"{n}\" does not exist"
1009                    )));
1010                }
1011                Ok(QueryResult::CommandOk {
1012                    affected: 0,
1013                    modified_catalog: false,
1014                })
1015            }
1016        }
1017    }
1018
1019    pub fn execute_prepared_with_cancel(
1020        &mut self,
1021        stmt: Statement,
1022        params: &[Value<'static>],
1023        cancel: CancelToken<'_>,
1024    ) -> Result<QueryResult, EngineError> {
1025        self.execute_prepared_in_with_cancel(stmt, params, IMPLICIT_TX, cancel)
1026    }
1027
1028    /// v7.39 (round 303, V22) — like [`Self::execute_prepared_with_cancel`]
1029    /// but binds the statement to an explicit transaction slot instead of
1030    /// the implicit one. The mysql-wire binary-protocol path uses this so a
1031    /// prepared INSERT/UPDATE lands in the connection's own `BEGIN`-opened
1032    /// transaction (and never collides with another connection on slot 0),
1033    /// mirroring what pgwire's `Bind`+`Execute` achieves by rendering
1034    /// bind-final SQL through [`Self::execute_in`].
1035    pub fn execute_prepared_in(
1036        &mut self,
1037        stmt: Statement,
1038        params: &[Value<'static>],
1039        tx_id: TxId,
1040    ) -> Result<QueryResult, EngineError> {
1041        self.execute_prepared_in_with_cancel(stmt, params, tx_id, CancelToken::none())
1042    }
1043
1044    pub fn execute_prepared_in_with_cancel(
1045        &mut self,
1046        mut stmt: Statement,
1047        params: &[Value<'static>],
1048        tx_id: TxId,
1049        cancel: CancelToken<'_>,
1050    ) -> Result<QueryResult, EngineError> {
1051        substitute_placeholders(&mut stmt, params)?;
1052        // v7.16.0 — set `current_tx` for the duration of the
1053        // dispatch so the `exec_*` helpers see the right TX
1054        // slot (matches what `execute_in_with_cancel` does for
1055        // simple-query). Pre-v7.16 the simple-query path
1056        // worked because every public entry point routed
1057        // through `execute_in_with_cancel`; the prepared path
1058        // skipped the wrap and so its INSERTs/UPDATEs landed
1059        // in the no-tx default slot, silently invisible to a
1060        // BEGIN/COMMIT-bracketed flow. Caught by spg-sqlx's
1061        // first transaction-visibility test.
1062        let saved = self.current_tx;
1063        self.current_tx = Some(tx_id);
1064        // v7.38 Epic P (panic isolation) — Slice 2: route the
1065        // prepared / extended-query path (the one sqlx / asyncpg / most
1066        // drivers actually use via pgwire `Bind`+`Execute`) through the
1067        // SAME `catch_unwind` firewall as the simple-query path (Slice 1,
1068        // `execute_inner_catching`). A panic in an extended-protocol
1069        // statement is caught inside the engine, the in-flight tx is
1070        // rolled back (shared `discard_tx_on_panic`), and the caller sees
1071        // an ordinary `EngineError::Internal` — never a poisoned write
1072        // guard or an aborted process. `current_tx` is `Some(IMPLICIT_TX)`
1073        // for the duration, so a caught panic rolls back the right tx; the
1074        // `saved` restore below still runs because the catch converts the
1075        // unwind into a normal `Result` return.
1076        let result = self.execute_stmt_catching(stmt, cancel);
1077        self.current_tx = saved;
1078        result
1079    }
1080
1081    /// v7.38 Epic P (panic isolation) — shared `catch_unwind` firewall
1082    /// (hosted `std` builds) used by every engine statement entry path: the
1083    /// simple-query ([`Self::execute_inner_catching`]), the prepared /
1084    /// extended-query ([`Self::execute_stmt_catching`]), and the read-only
1085    /// prepared-SELECT hot paths ([`Self::execute_prepared_select_no_params`]
1086    /// / [`Self::execute_prepared_select_streaming`]). Runs `run` under
1087    /// `catch_unwind`; a panic that unwinds out of statement execution is
1088    /// caught here and converted to [`EngineError::Internal`] after
1089    /// discarding the in-flight tx's shadow, so the caller sees a normal SQL
1090    /// error and the engine stays alive. This is the single place the
1091    /// rollback-on-panic policy lives — neither entry path reimplements it.
1092    ///
1093    /// **Why the post-catch engine state is consistent (COW shadow argument):**
1094    /// every uncommitted write of the panicked statement lives in
1095    /// `tx_catalogs[current_tx].catalog` — a per-tx *shadow* catalog that is
1096    /// only merged into the committed `self.catalog` at COMMIT (see
1097    /// `exec_commit`). The committed catalog is therefore never touched
1098    /// mid-statement, so dropping the shadow (mirroring `exec_rollback`)
1099    /// discards all half-applied work and leaves `self.catalog` exactly as it
1100    /// was before the statement. Redo-capture buffers live inside the
1101    /// shadow's tables and die with it, so no partial `RowChange` leaks into
1102    /// `last_redo` either (the caller publishes `last_redo` only on `Ok`).
1103    ///
1104    /// The `catch_unwind` closure holds `&mut self`; wrapping it in
1105    /// `AssertUnwindSafe` is sound precisely because of the above — the only
1106    /// caller-visible state a caught panic can leave behind is the discarded
1107    /// shadow, which is the correct rollback outcome, not a torn invariant.
1108    ///
1109    /// Generic over the closure's success type `T` so the read-only
1110    /// prepared-SELECT paths (which return a row count `usize`, not a
1111    /// `QueryResult`) reuse the *same* firewall — no second `catch_unwind`
1112    /// site. On those read-only paths `discard_tx_on_panic` is a no-op (a
1113    /// SELECT opens no shadow / writer version), which is the correct outcome:
1114    /// nothing to roll back, the point is purely to catch the unwind, return
1115    /// `Internal`, and leave the caller's write guard un-poisoned.
1116    #[cfg(feature = "std")]
1117    fn catch_stmt_panic<T>(
1118        &mut self,
1119        run: impl FnOnce(&mut Self) -> Result<T, EngineError>,
1120    ) -> Result<T, EngineError> {
1121        extern crate std;
1122        let caught = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| run(self)));
1123        match caught {
1124            Ok(result) => result,
1125            Err(payload) => {
1126                // The panic unwound past every `?`-return in the executor.
1127                // `current_tx` is Some here (set by the caller); roll that tx
1128                // back by discarding its shadow. A statement that panicked in
1129                // autocommit before any shadow was opened simply has nothing
1130                // to drop (`discard_tx_on_panic` is infallible).
1131                let tx_id = self.current_tx.unwrap_or(IMPLICIT_TX);
1132                self.discard_tx_on_panic(tx_id);
1133                Err(panic_payload_to_engine_error(payload.as_ref()))
1134            }
1135        }
1136    }
1137
1138    /// v7.38 Epic P (panic isolation) — simple-query path wrapper: run
1139    /// [`Self::execute_inner_with_cancel`] behind the shared
1140    /// [`Self::catch_stmt_panic`] firewall.
1141    #[cfg(feature = "std")]
1142    fn execute_inner_catching(
1143        &mut self,
1144        sql: &str,
1145        cancel: CancelToken<'_>,
1146    ) -> Result<QueryResult, EngineError> {
1147        self.catch_stmt_panic(|s| s.execute_inner_with_cancel(sql, cancel))
1148    }
1149
1150    /// `no_std` variant — there is no unwinding runtime, so statement
1151    /// execution runs directly with no catch.
1152    #[cfg(not(feature = "std"))]
1153    fn execute_inner_catching(
1154        &mut self,
1155        sql: &str,
1156        cancel: CancelToken<'_>,
1157    ) -> Result<QueryResult, EngineError> {
1158        self.execute_inner_with_cancel(sql, cancel)
1159    }
1160
1161    /// v7.38 Epic P (panic isolation) — Slice 2: prepared / extended-query
1162    /// path wrapper. The extended-protocol path already holds a resolved
1163    /// [`Statement`] (no re-parse), so it cannot reuse the `&str`-taking
1164    /// [`Self::execute_inner_catching`]; instead it runs
1165    /// [`Self::execute_stmt_with_cancel`] behind the SAME shared
1166    /// [`Self::catch_stmt_panic`] firewall — identical rollback + error
1167    /// semantics, zero duplicated policy.
1168    #[cfg(feature = "std")]
1169    fn execute_stmt_catching(
1170        &mut self,
1171        stmt: Statement,
1172        cancel: CancelToken<'_>,
1173    ) -> Result<QueryResult, EngineError> {
1174        self.catch_stmt_panic(|s| s.execute_stmt_with_cancel(stmt, cancel))
1175    }
1176
1177    /// `no_std` variant — there is no unwinding runtime, so statement
1178    /// execution runs directly with no catch.
1179    #[cfg(not(feature = "std"))]
1180    fn execute_stmt_catching(
1181        &mut self,
1182        stmt: Statement,
1183        cancel: CancelToken<'_>,
1184    ) -> Result<QueryResult, EngineError> {
1185        self.execute_stmt_with_cancel(stmt, cancel)
1186    }
1187
1188    /// v7.38 Epic P — discard an in-flight tx's shadow after a caught panic,
1189    /// mirroring the state cleanup of [`Engine::exec_rollback`] but
1190    /// infallibly. Drops the shadow catalog, marks the tx's writer version
1191    /// aborted, and releases its row locks. Leaves the committed
1192    /// `self.catalog` untouched (the COW model kept every uncommitted change
1193    /// inside the shadow), so this is a full rollback of the panicked
1194    /// statement's work.
1195    #[cfg(feature = "std")]
1196    fn discard_tx_on_panic(&mut self, tx_id: TxId) {
1197        self.tx_catalogs.remove(&tx_id);
1198        if let Some(v) = self.tx_writer_versions.remove(&tx_id) {
1199            self.abort_writer_version(v);
1200            self.release_tx_locks(v);
1201        }
1202        // Per-statement scratch: reset so no stale writer version leaks into
1203        // the next statement (the caller also restores the saved value).
1204        self.stmt_writer_version = None;
1205    }
1206
1207    fn execute_inner_with_cancel(
1208        &mut self,
1209        sql: &str,
1210        cancel: CancelToken<'_>,
1211    ) -> Result<QueryResult, EngineError> {
1212        cancel.check()?;
1213        let stmt = self.prepare(sql)?;
1214        // v6.5.1 — wrap the executor with a wall-clock window so we
1215        // can record into spg_stat_query. Skip when the engine has
1216        // no clock attached (no_std embedded callers).
1217        let start_us = self.clock.map(|f| f());
1218        let result = self.execute_stmt_with_cancel(stmt, cancel);
1219        if let (Some(t0), Ok(ok)) = (start_us, &result) {
1220            let now = self.clock.map_or(t0, |f| f());
1221            let elapsed = now.saturating_sub(t0).max(0) as u64;
1222            // v7.37.22 (22.9) — count rows produced (SELECT) or
1223            // affected (INSERT/UPDATE/DELETE) so pg_stat_statements'
1224            // `rows` column populates accurately.
1225            let row_count: u64 = match ok {
1226                QueryResult::Rows { rows, .. } => rows.len() as u64,
1227                QueryResult::CommandOk { affected, .. } => *affected as u64,
1228            };
1229            self.query_stats
1230                .record_with_rows(sql, elapsed, now as u64, row_count);
1231            // v6.5.6 — slow-query log: fire callback when elapsed
1232            // exceeds the configured floor.
1233            if let (Some(threshold), Some(logger)) =
1234                (self.slow_query_threshold_us, self.slow_query_logger)
1235                && elapsed >= threshold
1236            {
1237                logger(sql, elapsed);
1238            }
1239        }
1240        result
1241    }
1242
1243    /// v7.38 (read01 P3.26) — transaction-abort firewall around the raw
1244    /// statement dispatch. After a statement fails inside an explicit
1245    /// transaction PG aborts the whole block: every later statement except
1246    /// COMMIT / ROLLBACK / ROLLBACK TO SAVEPOINT is rejected, and a COMMIT
1247    /// is downgraded to a ROLLBACK so no partial work slips through. We
1248    /// mirror that here so both the embedded engine and the wire server
1249    /// enforce it uniformly.
1250    pub(crate) fn execute_stmt_with_cancel(
1251        &mut self,
1252        stmt: Statement,
1253        cancel: CancelToken<'_>,
1254    ) -> Result<QueryResult, EngineError> {
1255        // v7.39 (round 298) — ask THIS transaction, not "is any
1256        // transaction anywhere aborted".
1257        if self.current_tx_aborted() {
1258            match stmt {
1259                Statement::Rollback | Statement::RollbackToSavepoint(_) => {}
1260                // PG performs a ROLLBACK for a COMMIT in an aborted tx.
1261                Statement::Commit => {
1262                    let r = self.dispatch_stmt_inner(Statement::Rollback, cancel);
1263                    self.set_current_tx_aborted(false);
1264                    return r;
1265                }
1266                _ => return Err(EngineError::InFailedTransaction),
1267            }
1268        }
1269        let is_rollback_to_savepoint = matches!(stmt, Statement::RollbackToSavepoint(_));
1270        // v7.37.17 (Phase E2) — READ COMMITTED per-statement visibility:
1271        // classify (the statement moves into dispatch below), rebase the
1272        // open RC tx's shadow onto the latest committed catalog, then
1273        // record the statement's targets afterwards. Both calls are
1274        // no-ops outside an explicit transaction.
1275        let tx_class = crate::classify_stmt_for_tx(&stmt);
1276        if !matches!(tx_class, crate::TxStmtClass::TxControl) {
1277            // v7.37.17 (E4 r3) — a unique-key collision found while
1278            // rebasing fails THIS statement with 40001 (the tx aborts
1279            // via the standard failed-statement path below, like PG's
1280            // in-statement 23505 after the lock wait).
1281            self.maybe_rc_rebase()?;
1282        }
1283        // v7.39 (round 552) — what a SERIALIZABLE tx READ, taken before
1284        // the statement is consumed, recorded after it succeeds.
1285        let read_tables = crate::transaction::read_tables_of(&stmt);
1286        let result = self.dispatch_stmt_inner(stmt, cancel);
1287        if result.is_ok() {
1288            self.record_tx_stmt(&tx_class);
1289            self.record_tx_reads(read_tables);
1290        }
1291        // v7.39 (round 298) — the witness is THIS connection's slot.
1292        // `in_transaction()` is true whenever ANY connection holds a
1293        // transaction, so an autocommit failure used to abort a block
1294        // that belonged to somebody else.
1295        let mine_open = self.current_tx.is_some_and(|tx| self.is_tx_open(tx));
1296        if !mine_open {
1297            // The tx ended (COMMIT / ROLLBACK) or we were in autocommit;
1298            // either way there is no aborted block to remember.
1299            self.set_current_tx_aborted(false);
1300        } else if result.is_ok() && is_rollback_to_savepoint {
1301            // Rolling back to a savepoint recovers the transaction.
1302            self.set_current_tx_aborted(false);
1303        } else if matches!(result, Err(EngineError::LockWouldBlock)) {
1304            // v7.39 (round 300) — NOT a failure: the server drops the
1305            // engine lock and retries. Marking the block aborted here
1306            // made the FIRST block poison the transaction, so the
1307            // retry hit the abort firewall and the waiter lost a
1308            // deadlock it should have won.
1309        } else if result.is_err() {
1310            // A failure inside an open transaction aborts the whole block.
1311            self.set_current_tx_aborted(true);
1312        }
1313        result
1314    }
1315
1316    pub(crate) fn dispatch_stmt_inner(
1317        &mut self,
1318        stmt: Statement,
1319        cancel: CancelToken<'_>,
1320    ) -> Result<QueryResult, EngineError> {
1321        cancel.check()?;
1322        // v7.17.0 Phase 1.1 — pre-resolve nextval / currval /
1323        // setval calls in the statement tree. Walks SELECT
1324        // projection, INSERT VALUES, UPDATE SET, DELETE WHERE,
1325        // and DEFAULT exprs; replaces sequence FunctionCall
1326        // nodes with concrete Literal values minted against the
1327        // catalog. This is the only place that mutates sequence
1328        // state from a SELECT-shaped path (exec_select_cancel is
1329        // `&self` and can't reach the catalog mutably).
1330        //
1331        // Fast-path: when no sequences exist anywhere in the
1332        // catalog (the typical hot-path INSERT load), skip the
1333        // walker entirely. Single map-emptiness check on the
1334        // catalog beats walking every expression on every call.
1335        let mut stmt = stmt;
1336        // v7.17 dump-compat — the fast-path check
1337        // `sequences().is_empty()` skips pre-resolve when no
1338        // sequence exists in the *currently active* catalog
1339        // snapshot. The committed catalog or the implicit-TX
1340        // catalog may legitimately disagree on this between
1341        // CREATE SEQUENCE and a later setval(): always run the
1342        // resolver — the walk is O(expr-count) and dwarfed by
1343        // the parse cost we just paid.
1344        self.pre_resolve_sequence_calls_in_statement(&mut stmt)?;
1345        // v7.39 (round 305, V23) — evaluate any non-constant LIMIT /
1346        // OFFSET down to a literal row count. It belongs here, at the
1347        // one point both the simple-query and the prepared path pass
1348        // through, because every executor reads the row count as
1349        // `Option<u32>` and takes `None` for "no limit": an expression
1350        // that reached execution would silently widen the result to the
1351        // whole table rather than fail.
1352        self.resolve_limit_exprs_in_statement(&mut stmt, cancel)?;
1353        // v7.39 (read01 round 57) — the table-privilege gate. A superuser
1354        // session (the default login, or `SET ROLE admin`) skips it entirely,
1355        // so nothing changes for a customer who never assumes another role.
1356        self.acl_check_statement(&stmt)?;
1357        // v7.39 (round 435) — MySQL commits an open transaction BEFORE it
1358        // runs DDL (and before a nested START TRANSACTION), where PG keeps
1359        // the DDL inside the transaction. A MySQL client that writes rows,
1360        // runs DDL and then rolls back keeps those rows on MySQL and lost
1361        // them on SPG — silently, since nothing errors. This is the one
1362        // point every path (simple query, prepared, extended) passes
1363        // through, so the commit cannot be skipped by a spelling.
1364        //
1365        // v7.39 (round 444) — the witness is THIS connection's slot, not
1366        // `in_transaction()`. That predicate is true whenever ANY connection
1367        // holds a transaction, so a second client's `BEGIN` tried to commit a
1368        // slot of its own that held nothing and answered an error instead —
1369        // caught by `two_mysql_connections_can_each_hold_a_transaction`, which
1370        // had been failing since round 435 introduced this hook. Same
1371        // global-vs-slot confusion rounds 279 / 283 / 298 / 304 each fixed
1372        // elsewhere; `current_tx` is the connection's own slot here, set by
1373        // `execute_in_with_cancel` before dispatch.
1374        let in_own_tx = self.current_tx.is_some_and(|t| self.is_tx_open(t));
1375        if self.backslash_escapes && in_own_tx && stmt.mysql_implicit_commit() {
1376            self.exec_commit()?;
1377        }
1378        let result = match stmt {
1379            // v7.39 (round 547) — `ALTER ROLE … SET/RESET` and
1380            // `ALTER DATABASE … SET/RESET`. These reported success and
1381            // changed nothing: both fell into the parser's pg_dump
1382            // no-op tail, so a DBA setting a per-role default got no
1383            // effect and no error.
1384            Statement::SetDbRoleSetting(st) => {
1385                // PG refuses a scope that names something absent.
1386                // v7.39 (round 696) — one predicate. This wrote its own
1387                // (`users.any(…) || postgres`), `acl_check_role_exists`
1388                // wrote a third, and round 652 already recorded what
1389                // happens when a role predicate and the catalog it reflects
1390                // disagree. `role_exists` is the one that answers.
1391                if let Some(role) = &st.role
1392                    && !self.role_exists(role)
1393                {
1394                    return Err(EngineError::Unsupported(alloc::format!(
1395                        "role \"{role}\" does not exist"
1396                    )));
1397                }
1398                if let Some(db) = &st.database {
1399                    let current = self
1400                        .session_params
1401                        .get("spg.database")
1402                        .cloned()
1403                        .unwrap_or_else(|| alloc::string::String::from("spg"));
1404                    if !db.eq_ignore_ascii_case(&current) {
1405                        return Err(EngineError::Unsupported(alloc::format!(
1406                            "database \"{db}\" does not exist"
1407                        )));
1408                    }
1409                }
1410                let db = st.database.clone().unwrap_or_default();
1411                let role = st.role.clone().unwrap_or_default();
1412                let cat = self.active_catalog_mut();
1413                match (&st.param, &st.value) {
1414                    (None, _) => cat.reset_db_role_settings(&db, &role),
1415                    (Some(p), v) => cat.set_db_role_setting(&db, &role, p, v.as_deref()),
1416                }
1417                Ok(QueryResult::CommandOk {
1418                    affected: 0,
1419                    modified_catalog: true,
1420                })
1421            }
1422            // v7.39 (round 430) — MySQL USER variables. The value is an
1423            // arbitrary expression, evaluated against an empty row (a
1424            // user-variable assignment is a statement, not a per-row thing),
1425            // and stored in the session's own namespace.
1426            //
1427            // Every right-hand side sees the state as it was BEFORE the
1428            // statement — the assignments do NOT become visible to each
1429            // other. Measured on MariaDB 11: with both fresh,
1430            // `SET @p = 1, @q = @p + 1` leaves @q NULL; with @r already 100,
1431            // `SET @r = 1, @s = @r + 1` leaves @s at 101, i.e. @r's OLD
1432            // value. (Separate statements do chain, as you would expect.)
1433            // So: evaluate them all, THEN apply them all.
1434            Statement::SetUserVars(assigns, settings) => {
1435                let mut resolved: Vec<(String, spg_storage::Value<'static>)> =
1436                    Vec::with_capacity(assigns.len());
1437                for (name, mut expr) in assigns {
1438                    // `SET @total = (SELECT SUM(v) FROM t)` is ordinary MySQL,
1439                    // so the scalar subqueries have to be materialised the way
1440                    // every other statement's do — eval_expr itself refuses to
1441                    // meet one.
1442                    self.resolve_expr_subqueries(&mut expr, cancel)?;
1443                    let cols: Vec<ColumnSchema> = Vec::new();
1444                    let value = {
1445                        let ctx = self.ev_ctx(&cols, None);
1446                        let empty = spg_storage::Row::new(Vec::new());
1447                        crate::eval::eval_expr(&expr, &empty, &ctx).map_err(EngineError::Eval)?
1448                    };
1449                    resolved.push((name, value.into_owned()));
1450                }
1451                for (name, value) in resolved {
1452                    self.user_vars.insert(name, value);
1453                }
1454                // v7.39 (round 554) — the session settings written in
1455                // the same statement, applied after the saves. Routed
1456                // through the ordinary SET path so `SQL_MODE` still
1457                // flips strictness and the rest land where a plain
1458                // `SET x = y` puts them.
1459                for (name, value) in settings {
1460                    let rendered = match crate::conversions::literal_expr_to_value_in(
1461                        value.clone(),
1462                        Some(self.active_catalog()),
1463                    ) {
1464                        Ok(v) => crate::eval::value_to_text(&v),
1465                        Err(_) => alloc::format!("{value}"),
1466                    };
1467                    let _ = self.execute(&alloc::format!("SET {name} = '{rendered}'"));
1468                }
1469                Ok(QueryResult::CommandOk {
1470                    affected: 0,
1471                    modified_catalog: false,
1472                })
1473            }
1474            // v7.39 (round 277) — SQL-level prepared statements.
1475            Statement::Prepare {
1476                name,
1477                param_types,
1478                body,
1479                source,
1480            } => self.exec_prepare(name, param_types, *body, source),
1481            Statement::Execute { name, args } => self.exec_execute(&name, &args, cancel),
1482            Statement::Deallocate(name) => self.exec_deallocate(name.as_deref()),
1483            // v7.39 (round 278) — both were accepted and dropped. They
1484            // are reported as MISSING OBJECTS rather than as syntax
1485            // errors, because the SQL parses fine; what is absent is a
1486            // procedure catalog and a prepared-transaction registry.
1487            // v7.39 (round 280) — extended statistics as a real
1488            // catalog object. The planner does not consult them yet;
1489            // recording them is what makes a pg_dump restore and
1490            // reflection honest, instead of the statement vanishing.
1491            Statement::CreateStatistics {
1492                name,
1493                if_not_exists,
1494                kinds,
1495                columns,
1496                table,
1497            } => self.exec_create_statistics(name, if_not_exists, kinds, columns, table),
1498            Statement::DropStatistics { name, if_exists } => {
1499                self.exec_drop_statistics(&name, if_exists)
1500            }
1501            Statement::Call(name) => Err(EngineError::Unsupported(alloc::format!(
1502                "procedure {name}() does not exist HINT: No procedure matches the given name \
1503                 and argument types. You might need to add explicit type casts."
1504            ))),
1505            Statement::PrepareTransaction(_) => Err(EngineError::Unsupported(String::from(
1506                "prepared transactions are disabled HINT: Set \"max_prepared_transactions\" \
1507                 to a nonzero value.",
1508            ))),
1509            Statement::CreateTable(s) => self.exec_create_table(s),
1510            // v7.39 (round 218) — server-side cursors.
1511            Statement::DeclareCursor {
1512                name,
1513                scroll,
1514                hold,
1515                query,
1516            } => self.exec_declare_cursor(name, scroll, hold, *query),
1517            Statement::FetchCursor { name, direction } => self.exec_fetch_cursor(&name, direction),
1518            Statement::MoveCursor { name, direction } => self.exec_move_cursor(&name, direction),
1519            Statement::CloseCursor { name } => self.exec_close_cursor(name.as_deref()),
1520            // v7.39 (round 222) — LISTEN/NOTIFY with real delivery.
1521            Statement::Listen(ch) => self.exec_listen(ch),
1522            Statement::Notify { channel, payload } => self.exec_notify(channel, payload),
1523            Statement::Unlisten(ch) => self.exec_unlisten(ch),
1524            // v7.9.15 — CREATE EXTENSION is a no-op on SPG. Returns
1525            // CommandOk with affected=0; modified_catalog=false so
1526            // the WAL doesn't grow a useless entry. mailrs F3.
1527            Statement::CreateExtension(_) => Ok(QueryResult::CommandOk {
1528                affected: 0,
1529                modified_catalog: false,
1530            }),
1531            // v7.16.2 — DO $$ ... $$ block. mailrs round-10 A.2
1532            // — the pre-v7.9.27 no-op SILENTLY swallowed every
1533            // mailrs migrate-038/-040/-042 idempotent rename
1534            // (the IF EXISTS … THEN ALTER … END block never
1535            // ran). v7.16.2 dispatches to exec_do_block which
1536            // runs the PlPgSqlBlock at top level via the same
1537            // execute_stmts machinery the trigger executor
1538            // uses (NEW=None, OLD=None — DO blocks have no
1539            // row context).
1540            Statement::DoBlock(body) => self.exec_do_block(body),
1541            // v7.14.0 — empty-statement no-op for pg_dump /
1542            // mysqldump preamble lines that collapse to nothing
1543            // after comment-stripping.
1544            Statement::Empty => Ok(QueryResult::CommandOk {
1545                affected: 0,
1546                modified_catalog: false,
1547            }),
1548            // v7.39 (round 695) — `ALTER SYSTEM SET|RESET <name>`. SPG has
1549            // no postgresql.auto.conf to write, so nothing is APPLIED; what
1550            // changed is that a name PG18 does not know is now refused
1551            // instead of accepted. It reuses the session's own GUC check —
1552            // one place decides what a parameter name means, so `SET` and
1553            // `ALTER SYSTEM` cannot drift apart in what they accept.
1554            //
1555            // The F31 audit found this: the test was called
1556            // `alter_system_set_no_op` and set `work_mem`, a name that
1557            // exists, so it could never have caught a name that does not.
1558            // v7.39 (round 696) — the four statements the F31 sweep found
1559            // accepting a name that does not exist. SPG still performs
1560            // nothing for any of them; what changed is that it no longer
1561            // says "understood" about an object that is not there.
1562            // v7.39 (round 707) — see Statement::DropAggregate. Existence
1563            // first across the whole list (PG's order, measured), canonical
1564            // type names in the signature, and every SPG aggregate is a
1565            // built-in, so a name that exists is undroppable.
1566            Statement::DropAggregate { if_exists, items } => {
1567                let render = |name: &str, args: &Option<Vec<String>>| -> alloc::string::String {
1568                    match args {
1569                        None => alloc::format!("{name}(*)"),
1570                        Some(a) => {
1571                            let canon: Vec<alloc::string::String> = a
1572                                .iter()
1573                                .map(|t| {
1574                                    crate::conversions::type_name_to_data_type(t).map_or_else(
1575                                        || t.clone(),
1576                                        crate::conversions::pg_type_name_for_error,
1577                                    )
1578                                })
1579                                .collect();
1580                            alloc::format!("{name}({})", canon.join(", "))
1581                        }
1582                    }
1583                };
1584                for (name, args) in &items {
1585                    if !crate::aggregate::is_aggregate_name(name.as_str()) {
1586                        if if_exists {
1587                            continue;
1588                        }
1589                        return Err(EngineError::Unsupported(alloc::format!(
1590                            "aggregate {} does not exist",
1591                            render(name, args)
1592                        )));
1593                    }
1594                }
1595                if let Some((name, args)) = items
1596                    .iter()
1597                    .find(|(n, _)| crate::aggregate::is_aggregate_name(n.as_str()))
1598                {
1599                    return Err(EngineError::Unsupported(alloc::format!(
1600                        "cannot drop function {} because it is required by the database system",
1601                        render(name, args)
1602                    )));
1603                }
1604                Ok(QueryResult::CommandOk {
1605                    affected: 0,
1606                    modified_catalog: false,
1607                })
1608            }
1609            // v7.39 (round 750) — `ALTER ROLE … PASSWORD` really rotates
1610            // the credential now (it was a recorded no-op — ledgered as a
1611            // security defect in round 710: `ALTER USER x PASSWORD 'new'`
1612            // answered ALTER ROLE and the OLD password kept working).
1613            Statement::AlterRolePassword { name, password } => {
1614                if !self.role_exists(name.as_str()) {
1615                    return Err(EngineError::Unsupported(alloc::format!(
1616                        "role \"{name}\" does not exist"
1617                    )));
1618                }
1619                self.alter_user_password(&name, password.as_deref())
1620                    .map_err(|e| EngineError::Unsupported(alloc::format!("ALTER ROLE: {e}")))?;
1621                Ok(QueryResult::CommandOk {
1622                    affected: 0,
1623                    modified_catalog: self.catalog_change_is_committed(),
1624                })
1625            }
1626            Statement::ValidateOnly { kind, names } => {
1627                use spg_sql::ast::ValidateOnlyKind as K;
1628                match kind {
1629                    K::LockTable => {
1630                        for n in names {
1631                            if self.catalog.get(n.as_str()).is_none() {
1632                                return Err(EngineError::Storage(
1633                                    spg_storage::StorageError::TableNotFound { name: n.clone() },
1634                                ));
1635                            }
1636                        }
1637                    }
1638                    K::RoleName => {
1639                        for n in names {
1640                            if !self.role_exists(n.as_str()) {
1641                                return Err(EngineError::Unsupported(alloc::format!(
1642                                    "role \"{n}\" does not exist"
1643                                )));
1644                            }
1645                        }
1646                    }
1647                    // PG18 refuses this whatever it names, because no label
1648                    // provider is loaded — and SPG has none either, so the
1649                    // refusal is the honest answer rather than a stand-in.
1650                    // v7.39 (round 697) — one list answers both, which is
1651                    // why these and `pg_extension` cannot disagree.
1652                    //
1653                    // A WARNING, not an error, and that is a deliberate
1654                    // departure from PG. PG can error because an extension
1655                    // can be installed there; SPG cannot be installed into,
1656                    // so refusing would turn a customer dump that restores
1657                    // today into one that needs editing. Saying nothing was
1658                    // the actual defect: `CREATE EXTENSION hstore` reported
1659                    // success and nothing hstore-shaped worked afterwards.
1660                    K::ExtensionAvailable | K::ExtensionInstalled => {
1661                        for n in names {
1662                            if !crate::system_catalog::INSTALLED_EXTENSIONS
1663                                .iter()
1664                                .any(|(e, _)| e.eq_ignore_ascii_case(n.as_str()))
1665                            {
1666                                self.warning(alloc::format!(
1667                                    "extension \"{n}\" is not provided by this build; SPG \
1668                                     accepts the statement so a dump restores, but nothing \
1669                                     that extension supplies will be available"
1670                                ));
1671                            }
1672                        }
1673                    }
1674                    // v7.39 (round 708) — ALTER TYPE's no-op forms validate
1675                    // the name against the three user-type catalogs.
1676                    K::TypeName => {
1677                        for n in &names {
1678                            let cat = self.active_catalog();
1679                            if !cat.enum_types().contains_key(n)
1680                                && !cat.domain_types().contains_key(n)
1681                                && !cat.composite_types().contains_key(n)
1682                            {
1683                                return Err(EngineError::Unsupported(alloc::format!(
1684                                    "type \"{n}\" does not exist"
1685                                )));
1686                            }
1687                        }
1688                    }
1689                    // v7.39 (round 708) — names[0] = aggregate, rest = arg
1690                    // type names; existence by name (round 707's residual on
1691                    // overloads applies here too).
1692                    K::AggregateName => {
1693                        let Some(name) = names.first() else {
1694                            return Ok(QueryResult::CommandOk {
1695                                affected: 0,
1696                                modified_catalog: false,
1697                            });
1698                        };
1699                        if !crate::aggregate::is_aggregate_name(name.as_str()) {
1700                            let canon: Vec<alloc::string::String> = names[1..]
1701                                .iter()
1702                                .map(|t| {
1703                                    if t == "*" {
1704                                        alloc::string::String::from("*")
1705                                    } else {
1706                                        crate::conversions::type_name_to_data_type(t).map_or_else(
1707                                            || t.clone(),
1708                                            crate::conversions::pg_type_name_for_error,
1709                                        )
1710                                    }
1711                                })
1712                                .collect();
1713                            return Err(EngineError::Unsupported(alloc::format!(
1714                                "aggregate {name}({}) does not exist",
1715                                canon.join(", ")
1716                            )));
1717                        }
1718                    }
1719                    // v7.39 (round 708) — SPG ships no conversions at all,
1720                    // so PG's not-found answer is total here.
1721                    K::ConversionName => {
1722                        if let Some(n) = names.first() {
1723                            return Err(EngineError::Unsupported(alloc::format!(
1724                                "conversion \"{n}\" does not exist"
1725                            )));
1726                        }
1727                    }
1728                    // v7.39 (round 708) — the shipped languages are
1729                    // required; anything else does not exist. Both wordings
1730                    // are PG18 measurements.
1731                    K::LanguageName => {
1732                        // One name per statement; PG errors on the first
1733                        // either way, so `first` says what the loop only
1734                        // implied (clippy: never actually loops).
1735                        if let Some(n) = names.first() {
1736                            let lc = n.to_ascii_lowercase();
1737                            return Err(EngineError::Unsupported(match lc.as_str() {
1738                                "plpgsql" => alloc::format!(
1739                                    "cannot drop language {lc} because extension {lc} requires it"
1740                                ),
1741                                "sql" | "internal" | "c" => alloc::format!(
1742                                    "cannot drop language {lc} because it is required by the database system"
1743                                ),
1744                                _ => alloc::format!("language \"{n}\" does not exist"),
1745                            }));
1746                        }
1747                    }
1748                    // v7.39 (round 709) — batch-2 name checks, each wording
1749                    // a PG18 measurement.
1750                    K::CollationName => {
1751                        for n in &names {
1752                            if !crate::collate::is_supported(n) {
1753                                return Err(EngineError::Unsupported(alloc::format!(
1754                                    "collation \"{n}\" for encoding \"UTF8\" does not exist"
1755                                )));
1756                            }
1757                        }
1758                    }
1759                    K::TsConfigName => {
1760                        for n in &names {
1761                            // One list with the pg_ts_config synth: SPG
1762                            // ships `simple` and `english`.
1763                            if !n.eq_ignore_ascii_case("simple")
1764                                && !n.eq_ignore_ascii_case("english")
1765                            {
1766                                return Err(EngineError::Unsupported(alloc::format!(
1767                                    "text search configuration \"{n}\" does not exist"
1768                                )));
1769                            }
1770                        }
1771                    }
1772                    K::EventTriggerName => {
1773                        if let Some(n) = names.first() {
1774                            return Err(EngineError::Unsupported(alloc::format!(
1775                                "event trigger \"{n}\" does not exist"
1776                            )));
1777                        }
1778                    }
1779                    K::TablespaceName => {
1780                        if let Some(n) = names.first() {
1781                            return Err(EngineError::Unsupported(
1782                                if n.eq_ignore_ascii_case("pg_default")
1783                                    || n.eq_ignore_ascii_case("pg_global")
1784                                {
1785                                    alloc::format!("permission denied for tablespace {n}")
1786                                } else {
1787                                    alloc::format!("tablespace \"{n}\" does not exist")
1788                                },
1789                            ));
1790                        }
1791                    }
1792                    K::LargeObjectOid => {
1793                        if let Some(n) = names.first() {
1794                            let oid: u32 = n.parse().unwrap_or(0);
1795                            if !self.active_catalog().large_objects().contains_key(&oid) {
1796                                return Err(EngineError::Unsupported(alloc::format!(
1797                                    "large object {n} does not exist"
1798                                )));
1799                            }
1800                        }
1801                    }
1802                    // v7.39 (round 706) — see ValidateOnlyKind::ForeignInfra
1803                    // for why this warns instead of copying PG's refusal.
1804                    K::ForeignInfra => {
1805                        self.warning(alloc::string::String::from(
1806                            "foreign-data infrastructure is not provided by this build; \
1807                             SPG accepts the statement so a dump restores, but no foreign \
1808                             server, wrapper or table it defines will function",
1809                        ));
1810                    }
1811                    K::SecurityLabel => {
1812                        return Err(EngineError::Unsupported(alloc::string::String::from(
1813                            "no security label providers have been loaded",
1814                        )));
1815                    }
1816                }
1817                Ok(QueryResult::CommandOk {
1818                    affected: 0,
1819                    modified_catalog: false,
1820                })
1821            }
1822            Statement::DropDatabase { name, if_exists } => {
1823                // PG refuses this inside a transaction block; so does
1824                // CREATE DATABASE, and both go through the same guard.
1825                self.require_no_transaction_block("DROP DATABASE")?;
1826                // SPG serves one database, so the name is either the one
1827                // this session is connected to or a name that does not
1828                // exist here. PG has wording for both and never lets
1829                // either succeed, which is the whole behaviour.
1830                let is_current = self
1831                    .session_param("spg.database")
1832                    .unwrap_or("spg")
1833                    .eq_ignore_ascii_case(&name);
1834                if is_current {
1835                    return Err(EngineError::Unsupported(alloc::string::String::from(
1836                        "cannot drop the currently open database",
1837                    )));
1838                }
1839                if if_exists {
1840                    self.notice(alloc::format!(
1841                        "database \"{name}\" does not exist, skipping"
1842                    ));
1843                    return Ok(QueryResult::CommandOk {
1844                        affected: 0,
1845                        modified_catalog: false,
1846                    });
1847                }
1848                Err(EngineError::Unsupported(alloc::format!(
1849                    "database \"{name}\" does not exist"
1850                )))
1851            }
1852            Statement::NoOpPreventedInTransaction { what } => {
1853                self.require_no_transaction_block(&what)?;
1854                Ok(QueryResult::CommandOk {
1855                    affected: 0,
1856                    modified_catalog: false,
1857                })
1858            }
1859            Statement::AlterSystem { parameter } => {
1860                // PG refuses this inside a transaction block (25001): it
1861                // edits postgresql.auto.conf, which no rollback undoes.
1862                self.require_no_transaction_block("ALTER SYSTEM")?;
1863                if let Some(name) = parameter
1864                    && let Some(msg) = self.reject_unsettable_guc(name.as_str())
1865                {
1866                    return Err(EngineError::Unsupported(msg));
1867                }
1868                Ok(QueryResult::CommandOk {
1869                    affected: 0,
1870                    modified_catalog: false,
1871                })
1872            }
1873            Statement::DropTable { names, if_exists } => self.exec_drop_table(names, if_exists),
1874            Statement::DropIndex { name, if_exists } => self.exec_drop_index(name, if_exists),
1875            Statement::CreateIndex(s) => {
1876                // PG bars only the CONCURRENTLY form inside a transaction
1877                // block (25001); a plain CREATE INDEX there is fine.
1878                if s.concurrently {
1879                    self.require_no_transaction_block("CREATE INDEX CONCURRENTLY")?;
1880                }
1881                self.exec_create_index(s)
1882            }
1883            Statement::Insert(s) => {
1884                // v7.39 (pg_stat knife A) — per-table n_tup_ins. Charged
1885                // to the statement's target (a partition-routed insert
1886                // charges the parent; ON CONFLICT updates count here
1887                // too — split is a recorded residual).
1888                let stat_table = s.table.clone();
1889                let r = self.exec_insert(s)?;
1890                if let QueryResult::CommandOk { affected, .. } = &r {
1891                    self.stat_tup_inserted =
1892                        self.stat_tup_inserted.saturating_add(*affected as u64);
1893                    // r192 — engine-side, non-transactional (see
1894                    // table_write_stats): in-tx bumps used to land on
1895                    // the shadow table and vanish in the RC rebase.
1896                    self.note_table_write(&stat_table, *affected as u64, 0, 0);
1897                }
1898                Ok(r)
1899            }
1900            Statement::Update(mut s) => {
1901                // Materialise uncorrelated subqueries in SET / WHERE
1902                // before the row walk — the SELECT path has done this
1903                // since v4.10; UPDATE gained it for mailrs's
1904                // `UPDATE … WHERE id IN (SELECT … FOR UPDATE SKIP
1905                // LOCKED)` claim pattern (embed round-12).
1906                // v7.39 (round 157) — NOT with a WITH clause: the CTE
1907                // temps aren't installed yet here, so a subquery reading
1908                // a CTE either failed ("relation does not exist") or —
1909                // when a same-named real table existed — silently read
1910                // THAT. exec_update_with_ctes resolves after the temps
1911                // install instead.
1912                if s.ctes.is_empty() {
1913                    for (_, e) in &mut s.assignments {
1914                        self.resolve_expr_subqueries(e, cancel)?;
1915                    }
1916                    if let Some(w) = &mut s.where_ {
1917                        self.resolve_expr_subqueries(w, cancel)?;
1918                    }
1919                }
1920                let r = self.exec_update_cancel(&s, cancel)?;
1921                if let QueryResult::CommandOk { affected, .. } = &r {
1922                    self.stat_tup_updated = self.stat_tup_updated.saturating_add(*affected as u64);
1923                    self.note_table_write(&s.table, 0, *affected as u64, 0);
1924                }
1925                Ok(r)
1926            }
1927            Statement::Delete(mut s) => {
1928                // v7.39 (round 157) — see the Update arm: with a WITH
1929                // clause the resolve runs after the CTE temps install.
1930                if s.ctes.is_empty()
1931                    && let Some(w) = &mut s.where_
1932                {
1933                    self.resolve_expr_subqueries(w, cancel)?;
1934                }
1935                let r = self.exec_delete_cancel(&s, cancel)?;
1936                if let QueryResult::CommandOk { affected, .. } = &r {
1937                    self.stat_tup_deleted = self.stat_tup_deleted.saturating_add(*affected as u64);
1938                    self.note_table_write(&s.table, 0, 0, *affected as u64);
1939                }
1940                Ok(r)
1941            }
1942            Statement::Merge(s) => self.exec_merge_cancel(&s, cancel),
1943            // v7.39 (round 295, E3 Phase 1b) — a locking SELECT takes its
1944            // locks in a `&mut self` pre-pass that respects LIMIT, then
1945            // runs the ordinary read path with the rows another
1946            // transaction holds excluded.
1947            Statement::Select(ref sel) if sel.locking.is_some() => {
1948                let sel = sel.clone();
1949                self.lock_skip_rows = None;
1950                let pre = self.run_locking_prepass(&sel);
1951                if let Err(e) = pre {
1952                    self.lock_skip_rows = None;
1953                    return Err(e);
1954                }
1955                let out = self.exec_select_cancel(&sel, cancel);
1956                self.lock_skip_rows = None;
1957                out
1958            }
1959            Statement::Select(s) => {
1960                // v7.38 (read01 P3.20) — `SELECT set_config(name, value,
1961                // is_local)` is the writing sibling of SHOW / current_setting;
1962                // apply it to the session store (respecting is_local) so the
1963                // four GUC surfaces stay unified. pg_dump's
1964                // `SELECT set_config('search_path', '', false)` relies on this.
1965                if let Some(r) = self.try_exec_set_config(&s)? {
1966                    return Ok(r);
1967                }
1968                if s.ctes.iter().any(|c| c.body.is_modifying()) {
1969                    self.exec_select_with_modifying_ctes(s, cancel)
1970                } else {
1971                    self.exec_select_cancel(&s, cancel)
1972                }
1973            }
1974            // v7.39 (round 249) — the engine is no_std: the HOST reads the
1975            // file and calls `copy_from_buffer`. Reaching this arm means a
1976            // host that hasn't wired the file endpoint.
1977            Statement::CopyFromFile { path, .. } => Err(EngineError::Unsupported(alloc::format!(
1978                "COPY FROM file: the host must read {path:?} and call copy_from_buffer"
1979            ))),
1980            Statement::CopyTo {
1981                table,
1982                columns,
1983                query,
1984                options,
1985            } => self.exec_copy_to(
1986                &table,
1987                columns.as_deref(),
1988                query.as_deref(),
1989                &options,
1990                cancel,
1991            ),
1992            // v7.39 (round 252) — the engine is no_std: the HOST renders
1993            // via `copy_to_buffer` and writes the file itself.
1994            Statement::CopyToFile { path, .. } => Err(EngineError::Unsupported(alloc::format!(
1995                "COPY TO file: the host must render via copy_to_buffer and write {path:?}"
1996            ))),
1997            // v7.39 (round 475) — a redundant BEGIN inside a transaction.
1998            //
1999            // SPG raised "a transaction is already open" AND left the
2000            // transaction in the aborted state, so the next statement failed
2001            // with "current transaction is aborted" and the whole block was
2002            // lost. A connection pooler or a framework that wraps its own
2003            // BEGIN around one the caller already opened does this routinely.
2004            //
2005            // The two oracles genuinely differ, and both were measured:
2006            //   PG18       WARNING: there is already a transaction in
2007            //              progress — the BEGIN is a no-op and the existing
2008            //              transaction continues (a later ROLLBACK undoes
2009            //              everything, both rows in the probe).
2010            //   MariaDB 11 START TRANSACTION implicitly COMMITS the open one
2011            //              and begins a new one (the first row survives the
2012            //              rollback, the second does not).
2013            // The predicate is THIS connection's slot, not the engine-global
2014            // `in_transaction()`: the server shares one Engine, so the global
2015            // form makes connection B's BEGIN see connection A's transaction
2016            // (rounds 279 / 283 / 298 / 304 / 443 / 444 are the same trap).
2017            Statement::Begin(_)
2018                if self.current_tx.is_some_and(|t| self.is_tx_open(t))
2019                    && !self.backslash_escapes =>
2020            {
2021                self.warning(alloc::string::String::from(
2022                    "there is already a transaction in progress",
2023                ));
2024                Ok(QueryResult::CommandOk {
2025                    affected: 0,
2026                    modified_catalog: false,
2027                })
2028            }
2029            Statement::Begin(isolation) if self.current_tx.is_some_and(|t| self.is_tx_open(t)) => {
2030                // MySQL dialect: commit what is open, then start fresh.
2031                self.exec_commit()?;
2032                self.exec_begin(isolation)
2033            }
2034            Statement::Begin(isolation) => self.exec_begin(isolation),
2035            // v7.39 (round 435) — a bare COMMIT / ROLLBACK outside a
2036            // transaction is a no-op that SUCCEEDS. Measured on both
2037            // oracles: PG18 answers `WARNING: there is no transaction in
2038            // progress` and still reports COMMIT / ROLLBACK; MariaDB 11
2039            // succeeds silently. SPG answered "no active transaction" as an
2040            // ERROR to both dialects — a divergence from each of them.
2041            // It moved onto the hot path with the implicit-commit rule
2042            // above, which leaves a client's trailing ROLLBACK with nothing
2043            // to roll back.
2044            Statement::Commit | Statement::Rollback if !self.in_transaction() => {
2045                if !self.backslash_escapes {
2046                    self.warning(alloc::string::String::from(
2047                        "there is no transaction in progress",
2048                    ));
2049                }
2050                Ok(QueryResult::CommandOk {
2051                    affected: 0,
2052                    modified_catalog: false,
2053                })
2054            }
2055            Statement::Commit => self.exec_commit(),
2056            Statement::Rollback => self.exec_rollback(),
2057            Statement::Savepoint(name) => self.exec_savepoint(name),
2058            Statement::RollbackToSavepoint(name) => self.exec_rollback_to_savepoint(&name),
2059            Statement::ReleaseSavepoint(name) => self.exec_release_savepoint(&name),
2060            Statement::ShowTables => Ok(self.exec_show_tables()),
2061            Statement::ShowDatabases => Ok(self.exec_show_databases()),
2062            Statement::ShowCreateTable(name) => self.exec_show_create_table(&name),
2063            Statement::ShowIndexes(name) => self.exec_show_indexes(&name),
2064            Statement::ShowStatus => Ok(self.exec_show_status()),
2065            Statement::ShowVariables => Ok(self.exec_show_variables()),
2066            Statement::ShowProcesslist => Ok(self.exec_show_processlist()),
2067            Statement::Kill { query_only, id } => self.exec_kill(query_only, &id),
2068            Statement::Discard(target) => self.exec_discard(target),
2069            Statement::ShowColumns(table) => self.exec_show_columns(&table),
2070            Statement::ShowUsers => Ok(self.exec_show_users()),
2071            Statement::ShowPublications => Ok(self.exec_show_publications()),
2072            Statement::ShowSubscriptions => Ok(self.exec_show_subscriptions()),
2073            Statement::CreateUser(s) => self.exec_create_user(&s),
2074            Statement::DropUser { name, if_exists } => self.exec_drop_user(&name, if_exists),
2075            Statement::SetRole(role) => {
2076                match role {
2077                    Some(name) => {
2078                        // v7.39 (read01 round 58) — PG rejects a SET ROLE to a
2079                        // role that does not exist. Before roles were real
2080                        // there was nothing to check against, so any name was
2081                        // accepted — and a typo silently put the session into
2082                        // a role that held nothing.
2083                        self.acl_check_role_exists(&name)?;
2084                        self.session_params.insert(
2085                            alloc::string::String::from(crate::session::CURRENT_ROLE_KEY),
2086                            name,
2087                        );
2088                    }
2089                    None => {
2090                        self.session_params.remove(crate::session::CURRENT_ROLE_KEY);
2091                    }
2092                }
2093                Ok(QueryResult::CommandOk {
2094                    affected: 0,
2095                    modified_catalog: false,
2096                })
2097            }
2098            Statement::Grant(g) => self.exec_grant(&g, true),
2099            Statement::Revoke(g) => self.exec_grant(&g, false),
2100            Statement::CreatePolicy(s) => self.exec_create_policy(s),
2101            Statement::AlterPolicy(s) => self.exec_alter_policy(s),
2102            Statement::DropPolicy(s) => self.exec_drop_policy(s),
2103            // v7.39 (round 286) — ANALYZE over DML really executes, so it
2104            // needs the `&mut self` sibling. Everything else (including
2105            // plain EXPLAIN of a write) stays on the read-only renderer.
2106            // v7.39 (round 288) — SET CONSTRAINTS sets the timing for the
2107            // rest of the transaction. IMMEDIATE also runs everything the
2108            // transaction has postponed, right here — PG raises the
2109            // violation at this statement, not at COMMIT.
2110            Statement::SetConstraints { names, deferred } => {
2111                self.exec_set_constraints(&names, deferred)
2112            }
2113            Statement::Explain(e)
2114                if e.analyze
2115                    && !e.suggest
2116                    && matches!(
2117                        &*e.inner,
2118                        Statement::Insert(_) | Statement::Update(_) | Statement::Delete(_)
2119                    ) =>
2120            {
2121                self.exec_explain_analyze_dml(&e, cancel)
2122            }
2123            Statement::Explain(e) => self.exec_explain(&e, cancel),
2124            Statement::AlterIndex(s) => self.exec_alter_index(s),
2125            Statement::AlterTable(s) => self.exec_alter_table(s),
2126            Statement::CreatePublication(s) => self.exec_create_publication(s),
2127            Statement::DropPublication { name, if_exists } => {
2128                self.exec_drop_publication(&name, if_exists)
2129            }
2130            Statement::CreateSubscription(s) => self.exec_create_subscription(s),
2131            Statement::DropSubscription { name, if_exists } => {
2132                self.exec_drop_subscription(&name, if_exists)
2133            }
2134            // v6.1.7 — WAIT FOR WAL POSITION needs `lag_state`,
2135            // which lives in spg-server's ServerState. The engine
2136            // surfaces a clear error; the server-layer dispatch
2137            // intercepts the SQL before it reaches the engine on
2138            // a server build, so this arm only fires for
2139            // engine-only callers (spg-embedded, lib tests).
2140            Statement::WaitForWalPosition { .. } => Err(EngineError::Unsupported(
2141                "WAIT FOR WAL POSITION must be handled by the server layer".into(),
2142            )),
2143            // v6.2.0 — ANALYZE recomputes per-column histograms.
2144            Statement::Analyze(target) => self.exec_analyze(target.as_deref()),
2145            // v7.39 (round 535) — REINDEX / CLUSTER. SPG has neither index
2146            // bloat to rebuild nor a clustering order to impose, so the
2147            // work is a no-op — but PG VALIDATES the target, and both
2148            // statements were swallowed at parse time AND intercepted at
2149            // the wire, so `REINDEX TABLE typo` answered `REINDEX`. A
2150            // maintenance script that misspells a table was told it
2151            // succeeded.
2152            Statement::Maintain {
2153                kind,
2154                concurrently,
2155                target,
2156            } => {
2157                use spg_sql::ast::MaintainKind;
2158                if concurrently {
2159                    self.require_no_transaction_block(match kind {
2160                        MaintainKind::ClusterRelation => "CLUSTER",
2161                        _ => "REINDEX CONCURRENTLY",
2162                    })?;
2163                }
2164                match (kind, target.as_deref()) {
2165                    (MaintainKind::ReindexRelation | MaintainKind::ClusterRelation, Some(t)) => {
2166                        // An INDEX is a relation too — `REINDEX INDEX ix`
2167                        // names one, and looking only at tables refused a
2168                        // name that is right there.
2169                        let is_index = self
2170                            .active_catalog()
2171                            .table_names()
2172                            .iter()
2173                            .filter_map(|n| self.active_catalog().get(n))
2174                            .any(|tbl| {
2175                                tbl.indices().iter().any(|i| i.name.eq_ignore_ascii_case(t))
2176                            });
2177                        if !is_index && self.active_catalog().get(t).is_none() {
2178                            return Err(EngineError::Storage(
2179                                spg_storage::StorageError::TableNotFound { name: t.into() },
2180                            ));
2181                        }
2182                    }
2183                    (MaintainKind::ReindexSchema, Some(t)) => {
2184                        if !spg_storage::is_builtin_schema(t)
2185                            && !self.active_catalog().schema_exists(t)
2186                        {
2187                            return Err(EngineError::Unsupported(alloc::format!(
2188                                "schema \"{t}\" does not exist"
2189                            )));
2190                        }
2191                    }
2192                    // `REINDEX SYSTEM` / `REINDEX DATABASE` / a bare
2193                    // `CLUSTER` name nothing to check.
2194                    _ => {}
2195                }
2196                Ok(QueryResult::CommandOk {
2197                    affected: 0,
2198                    modified_catalog: false,
2199                })
2200            }
2201            // v7.39 (round 169) — VACUUM does real work under the MVCC
2202            // gate (tombstoned versions are actual bloat); the pre-MVCC
2203            // parse-time no-op silently ignored a customer's manual
2204            // reclaim. Gate-off stays a provable no-op inside vacuum.
2205            Statement::Vacuum { table, analyze } => {
2206                // PG 18.4, measured: every VACUUM form — bare, with a
2207                // table, and VACUUM ANALYZE — is refused inside a
2208                // transaction block with 25001, while a plain ANALYZE is
2209                // allowed. Reclaiming storage cannot be rolled back, so
2210                // it must not be able to join a transaction that can.
2211                self.require_no_transaction_block("VACUUM")?;
2212                match &table {
2213                    Some(t) => {
2214                        // v7.39 (round 535) — PG refuses a VACUUM whose
2215                        // relation does not exist; `vacuum_one_table`
2216                        // simply found nothing to do and said nothing,
2217                        // so a typo'd table reported success.
2218                        if self.active_catalog().get(t).is_none() {
2219                            return Err(EngineError::Storage(
2220                                spg_storage::StorageError::TableNotFound { name: t.clone() },
2221                            ));
2222                        }
2223                        self.vacuum_one_table(t);
2224                    }
2225                    None => {
2226                        let _ = self.vacuum_pass(false);
2227                    }
2228                }
2229                if analyze {
2230                    self.exec_analyze(table.as_deref())?;
2231                }
2232                Ok(QueryResult::CommandOk {
2233                    affected: 0,
2234                    modified_catalog: false,
2235                })
2236            }
2237            // v7.37.17 (17.6 sibling) — TRUNCATE [TABLE] <t>[, ...]
2238            // [RESTART IDENTITY] [CASCADE]. Clears every row from
2239            // each named table. CASCADE currently accepts the syntax
2240            // + records the flag; the FK-referring cascade walk lands
2241            // when FK-cascade delete surface gets extended to
2242            // multi-relation batching (v7.38).
2243            Statement::Truncate {
2244                tables,
2245                restart_identity,
2246                cascade: _,
2247                only,
2248            } => {
2249                for t in &tables {
2250                    self.bump_table_change(t);
2251                }
2252                self.exec_truncate(tables.as_slice(), restart_identity, only)
2253            }
2254            // v6.7.3 — COMPACT COLD SEGMENTS.
2255            Statement::CompactColdSegments => self.exec_compact_cold_segments(),
2256            // v7.12.1 — SET / RESET session parameter. Engine
2257            // tracks the value in `session_params`; FTS dispatcher
2258            // reads `default_text_search_config`. Everything else
2259            // is a recorded no-op (PG dump compat).
2260            Statement::SetParameter { name, value, local } => {
2261                // v7.39 (round 501) — a name PG18 does not know, or one a
2262                // session cannot change, is an error there and was
2263                // silently accepted here (round 500).
2264                if let Some(msg) = self.reject_unsettable_guc(&name) {
2265                    return Err(EngineError::Unsupported(msg));
2266                }
2267                // v7.38 (read01) — SPG serves the wire as UTF8, so a
2268                // non-UTF8 client_encoding can't be honoured (the bytes
2269                // stay UTF8). Reject it rather than silently store a value
2270                // that would mislabel the stream; an unusable name is
2271                // rejected the way PG rejects an invalid one.
2272                if name.eq_ignore_ascii_case("client_encoding") {
2273                    let v: &str = match &value {
2274                        spg_sql::ast::SetValue::String(s)
2275                        | spg_sql::ast::SetValue::Ident(s)
2276                        | spg_sql::ast::SetValue::Number(s) => s.as_str(),
2277                        spg_sql::ast::SetValue::Default => "UTF8",
2278                    };
2279                    let norm: alloc::string::String = v
2280                        .trim()
2281                        .to_ascii_uppercase()
2282                        .chars()
2283                        .filter(|c| *c != '-' && *c != '_')
2284                        .collect();
2285                    if !matches!(norm.as_str(), "UTF8" | "UNICODE") {
2286                        return Err(EngineError::Unsupported(alloc::format!(
2287                            "invalid value for parameter \"client_encoding\": \"{v}\" \
2288                             (SPG serves UTF8 only)"
2289                        )));
2290                    }
2291                }
2292                // v7.38 (read01 P3.17) — reject a clearly-invalid value for
2293                // a handful of well-known typed GUCs (`SET work_mem =
2294                // 'bogus'` errors like PG). Unknown GUCs stay accept-and-
2295                // record for pg_dump compat.
2296                if let spg_sql::ast::SetValue::String(s)
2297                | spg_sql::ast::SetValue::Ident(s)
2298                | spg_sql::ast::SetValue::Number(s) = &value
2299                {
2300                    validate_known_guc(&name, s)?;
2301                    // v7.39 (tz epic) — timezone accepts UTC / fixed
2302                    // offsets / abbreviations (resolve_zone_offset) and
2303                    // IANA names (host tzdb); anything else is PG's
2304                    // invalid-parameter error. Named zones store their
2305                    // canonical spelling (SHOW returns 'Asia/Tokyo'
2306                    // after SET 'asia/tokyo').
2307                    if name.eq_ignore_ascii_case("timezone")
2308                        || name.eq_ignore_ascii_case("time zone")
2309                    {
2310                        let canon = self.canonicalize_timezone(s)?;
2311                        let local = local;
2312                        if local {
2313                            if self.in_transaction() {
2314                                let prior = self.session_param("timezone").map(String::from);
2315                                self.local_guc_saves.push(("timezone".into(), prior));
2316                                self.set_session_param(
2317                                    "timezone".into(),
2318                                    spg_sql::ast::SetValue::String(canon),
2319                                );
2320                            }
2321                        } else {
2322                            self.set_session_param(
2323                                "timezone".into(),
2324                                spg_sql::ast::SetValue::String(canon),
2325                            );
2326                        }
2327                        return Ok(QueryResult::CommandOk {
2328                            affected: 0,
2329                            modified_catalog: false,
2330                        });
2331                    }
2332                }
2333                // v7.38 (read01 P3.19) — `SET LOCAL` scopes the change to
2334                // the current transaction: record the prior value in the
2335                // undo log so COMMIT / ROLLBACK (and ROLLBACK TO) restore
2336                // it. Outside a transaction block it has no lasting effect
2337                // (PG scopes it to the implicit single-statement txn), so
2338                // it is dropped rather than persisted to the session.
2339                if local {
2340                    if self.in_transaction() {
2341                        let prior = self.session_param(&name).map(String::from);
2342                        self.local_guc_saves.push((name.clone(), prior));
2343                        self.set_session_param(name, value);
2344                    }
2345                } else {
2346                    self.set_session_param(name, value);
2347                }
2348                Ok(QueryResult::CommandOk {
2349                    affected: 0,
2350                    modified_catalog: false,
2351                })
2352            }
2353            // v7.38 轴 4 — `SET TRANSACTION ISOLATION LEVEL …`. The
2354            // surface is recorded on `Engine::current_isolation_level`
2355            // and visible via `SHOW transaction_isolation`. Behavioural
2356            // implementation (REPEATABLE READ snapshot / SERIALIZABLE
2357            // SSI) lands separately; today every level reads as
2358            // effective READ COMMITTED (same as PG's silent upgrade
2359            // of READ UNCOMMITTED).
2360            Statement::SetTransaction { isolation } => {
2361                // v7.37.17 (Phase E3) — PG rejects an isolation switch
2362                // after the transaction's first query (SQLSTATE 25001);
2363                // silently applying it to the remaining statements would
2364                // give a tx that is half one level, half another.
2365                if let Some(tx_id) = self.current_tx
2366                    && self
2367                        .tx_catalogs
2368                        .get(&tx_id)
2369                        .is_some_and(|st| st.stmts_run > 0)
2370                {
2371                    return Err(EngineError::Unsupported(
2372                        "SET TRANSACTION ISOLATION LEVEL must be called before any query".into(),
2373                    ));
2374                }
2375                self.current_isolation_level = isolation;
2376                // v7.37.17 (Phase E2) — inside an open tx, switching to
2377                // RR/SER BEFORE the first query freezes the tx's view by
2378                // caching a snapshot now (PG allows the switch until the
2379                // first query; the RC rebase keys off cached_snapshot).
2380                // Switching (back) to RC/RU clears it so the rebase
2381                // resumes.
2382                if let Some(tx_id) = self.current_tx
2383                    && self.tx_catalogs.contains_key(&tx_id)
2384                {
2385                    let cache = match isolation {
2386                        spg_sql::ast::IsolationLevel::RepeatableRead
2387                        | spg_sql::ast::IsolationLevel::Serializable => {
2388                            Some(self.current_snapshot())
2389                        }
2390                        spg_sql::ast::IsolationLevel::ReadUncommitted
2391                        | spg_sql::ast::IsolationLevel::ReadCommitted => None,
2392                    };
2393                    if let Some(st) = self.tx_catalogs.get_mut(&tx_id) {
2394                        st.cached_snapshot = cache;
2395                    }
2396                }
2397                Ok(QueryResult::CommandOk {
2398                    affected: 0,
2399                    modified_catalog: false,
2400                })
2401            }
2402            // v7.38 轴 4 surface expansion — `SHOW <parameter>`
2403            // returns a 1-row 1-column TEXT result (the PG psql
2404            // wire shape). The handler dispatches per-name:
2405            //
2406            // 1. transaction_isolation — direct read of
2407            //    current_isolation_level (the v7.38 axis-4 surface).
2408            // 2. PG preset / engine-tracked params — values mirror
2409            //    pg_catalog.pg_settings to keep ORM /
2410            //    driver-connect probes happy (sqlx asks
2411            //    server_version + standard_conforming_strings +
2412            //    client_encoding; npgsql asks application_name;
2413            //    asyncpg asks search_path). Any
2414            //    SET-tracked override on self.session_params wins.
2415            // 3. Anything else — error with a list-pointer to
2416            //    pg_settings (which lists every recognised name).
2417            Statement::ShowParameter(name) => {
2418                use spg_storage::{ColumnSchema, DataType, Row, Value};
2419                // v7.37.17 (17.6 sibling) — `SHOW ALL` returns a
2420                // (name, setting, description) triple for every
2421                // parameter SPG knows about. PG's shape is the same.
2422                // Emitting a fixed curated inventory here keeps the
2423                // client shape stable without wire-tapping every
2424                // per-session parameter.
2425                // v7.38 (read01 P3.20/P3.23) — SHOW reads the same canonical
2426                // GUC inventory as pg_settings, so `SHOW <name>` / `SHOW ALL`
2427                // and pg_settings never disagree on which params exist.
2428                let canon = crate::system_catalog::canonical_gucs();
2429                let effective = |n: &str, boot: &str| -> alloc::string::String {
2430                    self.session_params
2431                        .iter()
2432                        .find(|(k, _)| k.eq_ignore_ascii_case(n))
2433                        .map(|(_, v)| v.clone())
2434                        .unwrap_or_else(|| boot.into())
2435                };
2436                if name.eq_ignore_ascii_case("all") {
2437                    let cols = alloc::vec![
2438                        ColumnSchema::new("name", DataType::Text, false),
2439                        ColumnSchema::new("setting", DataType::Text, false),
2440                        ColumnSchema::new("description", DataType::Text, false),
2441                    ];
2442                    let mut rows: Vec<Row> = Vec::new();
2443                    // Dynamic params outside the static canonical table.
2444                    rows.push(Row::new(alloc::vec![
2445                        Value::text(alloc::string::String::from("transaction_isolation")),
2446                        Value::text(alloc::string::String::from(
2447                            self.current_isolation_level.as_pg_str(),
2448                        )),
2449                        Value::text(alloc::string::String::from(
2450                            "Shows the current transaction's isolation level.",
2451                        )),
2452                    ]));
2453                    rows.push(Row::new(alloc::vec![
2454                        Value::text(alloc::string::String::from("is_superuser")),
2455                        Value::text(alloc::string::String::from("on")),
2456                        Value::text(alloc::string::String::from("Reports superuser status.")),
2457                    ]));
2458                    for (n, boot, cat, _, _) in canon {
2459                        rows.push(Row::new(alloc::vec![
2460                            Value::text(alloc::string::String::from(*n)),
2461                            Value::text(effective(n, boot)),
2462                            Value::text(alloc::string::String::from(*cat)),
2463                        ]));
2464                    }
2465                    return Ok(QueryResult::Rows {
2466                        columns: cols,
2467                        rows,
2468                    });
2469                }
2470                let value: alloc::string::String = match name.to_ascii_lowercase().as_str() {
2471                    "transaction_isolation" => {
2472                        alloc::string::String::from(self.current_isolation_level.as_pg_str())
2473                    }
2474                    "is_superuser" => alloc::string::String::from("on"),
2475                    _ => {
2476                        // Canonical GUC? report the session override or its
2477                        // boot default. Otherwise a user-set custom GUC, or a
2478                        // recognised-name error pointing at pg_settings.
2479                        if let Some((_, boot, ..)) =
2480                            canon.iter().find(|(n, ..)| n.eq_ignore_ascii_case(&name))
2481                        {
2482                            effective(&name, boot)
2483                        } else if let Some(v) = self.session_param(&name) {
2484                            alloc::string::String::from(v)
2485                        } else if let Some(boot) = crate::guc_catalog::guc_boot_value(&name) {
2486                            // v7.39 (round 534) — a parameter PG18 knows but
2487                            // SPG does not model reports its compiled-in
2488                            // default. `SHOW random_page_cost` printed
2489                            // nothing at all before, and `SHOW fsync` with
2490                            // it.
2491                            alloc::string::String::from(boot)
2492                        } else {
2493                            return Err(EngineError::Unsupported(alloc::format!(
2494                                "SHOW {name:?}: parameter not recognised; \
2495                                 see `SELECT name, setting FROM pg_settings` for \
2496                                 the full inventory"
2497                            )));
2498                        }
2499                    }
2500                };
2501                Ok(QueryResult::Rows {
2502                    columns: alloc::vec![ColumnSchema::new(name, DataType::Text, false)],
2503                    rows: alloc::vec![Row::new(alloc::vec![Value::text(value)])],
2504                })
2505            }
2506            // v7.14.0 — MySQL multi-assignment SET. Each pair runs
2507            // through `set_session_param` so engine-known params
2508            // (FOREIGN_KEY_CHECKS, session_replication_role, …) take
2509            // effect; unknown pairs (including `@VAR` LHS from the
2510            // mysqldump preamble) are recorded then ignored.
2511            Statement::SetParameterList(pairs) => {
2512                // Same validation as the single form (round 501).
2513                for (name, _) in &pairs {
2514                    if let Some(msg) = self.reject_unsettable_guc(name) {
2515                        return Err(EngineError::Unsupported(msg));
2516                    }
2517                }
2518                for (name, value) in pairs {
2519                    self.set_session_param(name, value);
2520                }
2521                Ok(QueryResult::CommandOk {
2522                    affected: 0,
2523                    modified_catalog: false,
2524                })
2525            }
2526            // v7.12.4 — CREATE FUNCTION / CREATE TRIGGER / DROP …
2527            // for the PL/pgSQL trigger surface. exec_* methods are
2528            // defined alongside the existing CREATE handlers below.
2529            Statement::CreateFunction(s) => self.exec_create_function(s),
2530            Statement::CreateTrigger(s) => self.exec_create_trigger(s),
2531            Statement::DropTrigger {
2532                name,
2533                table,
2534                if_exists,
2535            } => self.exec_drop_trigger(&name, &table, if_exists),
2536            Statement::CreateRule(s) => self.exec_create_rule(s),
2537            Statement::DropRule {
2538                name,
2539                table,
2540                if_exists,
2541            } => self.exec_drop_rule(&name, &table, if_exists),
2542            Statement::DropFunction {
2543                name,
2544                args,
2545                if_exists,
2546            } => self.exec_drop_function(&name, args.as_deref(), if_exists),
2547            Statement::CreateSequence(s) => self.exec_create_sequence(s),
2548            Statement::AlterSequence(s) => self.exec_alter_sequence(s),
2549            Statement::DropSequence { names, if_exists } => {
2550                self.exec_drop_sequence(&names, if_exists)
2551            }
2552            Statement::CreateView(s) => self.exec_create_view(s),
2553            Statement::DropView { names, if_exists } => self.exec_drop_view(&names, if_exists),
2554            Statement::CreateMaterializedView(s) => self.exec_create_materialized_view(s),
2555            Statement::RefreshMaterializedView { name, with_data } => {
2556                self.exec_refresh_materialized_view(&name, with_data)
2557            }
2558            Statement::DropMaterializedView { names, if_exists } => {
2559                self.exec_drop_materialized_view(&names, if_exists)
2560            }
2561            Statement::CreateType(s) => self.exec_create_type(s),
2562            Statement::CommentOn {
2563                kind,
2564                name,
2565                comment,
2566            } => self.exec_comment_on(&kind, &name, comment.as_deref()),
2567            Statement::AlterTypeRenameValue {
2568                type_name,
2569                old,
2570                new,
2571            } => {
2572                self.active_catalog_mut()
2573                    .rename_enum_value(&type_name, &old, &new)
2574                    .map_err(EngineError::Storage)?;
2575                Ok(QueryResult::CommandOk {
2576                    affected: 0,
2577                    modified_catalog: self.catalog_change_is_committed(),
2578                })
2579            }
2580            Statement::AlterTypeAddValue {
2581                type_name,
2582                label,
2583                if_not_exists,
2584                position,
2585            } => {
2586                let added = self
2587                    .active_catalog_mut()
2588                    .add_enum_value(&type_name, &label, if_not_exists, position)
2589                    .map_err(EngineError::Storage)?;
2590                Ok(QueryResult::CommandOk {
2591                    affected: 0,
2592                    modified_catalog: added,
2593                })
2594            }
2595            Statement::DropType { names, if_exists } => self.exec_drop_type(&names, if_exists),
2596            Statement::CreateDomain(s) => self.exec_create_domain(s),
2597            Statement::AlterDomain { name, action } => self.exec_alter_domain(&name, action),
2598            Statement::DropDomain { names, if_exists } => self.exec_drop_domain(&names, if_exists),
2599            Statement::CreateSchema {
2600                name,
2601                if_not_exists,
2602            } => self.exec_create_schema(name, if_not_exists),
2603            Statement::DropSchema { names, if_exists } => self.exec_drop_schema(&names, if_exists),
2604            Statement::ResetParameter(target) => {
2605                match target {
2606                    // v7.39 (round 320, V53) — RESET ALL resets GUCs. It
2607                    // must NOT throw away the two internal keys the server
2608                    // parks in the same map: the connection's login
2609                    // identity and its database. PG has no way to reset
2610                    // those with RESET ALL (they are not GUCs), and
2611                    // clearing them here made `current_user` fall back to
2612                    // the admin default mid-session.
2613                    None => self.reset_all_gucs(),
2614                    Some(name) => {
2615                        self.session_params.remove(&name.to_ascii_lowercase());
2616                    }
2617                }
2618                self.refresh_render_style();
2619                Ok(QueryResult::CommandOk {
2620                    affected: 0,
2621                    modified_catalog: false,
2622                })
2623            }
2624        };
2625        self.enforce_row_limit(result)
2626    }
2627}
2628
2629impl Engine {
2630    /// v7.39 (round 247) — resolve the CSV-only extras. QUOTE / ESCAPE /
2631    /// FORCE_QUOTE outside CSV mode are PG's 0A000 refusals (SPG used to
2632    /// ignore a text-mode QUOTE silently); the returned mask marks the
2633    /// force-quoted columns of `column_names`.
2634    fn resolve_copy_csv_extras(
2635        options: &spg_sql::ast::CopyOptions,
2636        is_csv: bool,
2637        quote: char,
2638        column_names: &[alloc::string::String],
2639    ) -> Result<(char, Option<alloc::vec::Vec<bool>>), EngineError> {
2640        if !is_csv {
2641            if options.quote.is_some() {
2642                return Err(EngineError::Unsupported(
2643                    "COPY QUOTE requires CSV mode".into(),
2644                ));
2645            }
2646            if options.escape.is_some() {
2647                return Err(EngineError::Unsupported(
2648                    "COPY ESCAPE requires CSV mode".into(),
2649                ));
2650            }
2651        }
2652        // v7.39 (round 265) — the direction-dependent rules (FORCE_QUOTE is
2653        // TO-only, FORCE_NOT_NULL / FORCE_NULL are FROM-only), sharing one
2654        // validator with the FROM path.
2655        crate::copy::validate_copy_option_direction(options, true)?;
2656        let escape = options.escape.unwrap_or(quote);
2657        let force = match &options.force_quote {
2658            None => None,
2659            Some(cols) if cols.is_empty() => Some(alloc::vec![true; column_names.len()]),
2660            Some(cols) => {
2661                let mut mask = alloc::vec![false; column_names.len()];
2662                for c in cols {
2663                    let pos = column_names
2664                        .iter()
2665                        .position(|n| n.eq_ignore_ascii_case(c))
2666                        .ok_or_else(|| {
2667                            EngineError::Unsupported(alloc::format!(
2668                                "column \"{c}\" does not exist"
2669                            ))
2670                        })?;
2671                    mask[pos] = true;
2672                }
2673                Some(mask)
2674            }
2675        };
2676        Ok((escape, force))
2677    }
2678
2679    /// v7.39 (round 249) — resolve the effective COPY FROM target column
2680    /// list, running PG's pre-file checks in PG's order: the relation
2681    /// must exist, an explicit column must exist on it, and no column
2682    /// may appear twice — all before a single data row is looked at.
2683    ///
2684    /// # Errors
2685    /// `relation "t" does not exist`, `column "x" of relation "t" does
2686    /// not exist` (42703), `column "x" specified more than once` (42701).
2687    /// v7.39 (round 343, V40) — store a file the host just read as a
2688    /// large object. The host does the IO (the engine is `no_std`); the
2689    /// catalog side is the same `create_large_object` the rest of the
2690    /// lo_* family uses, so an imported object is indistinguishable from
2691    /// one built with `lo_from_bytea`.
2692    pub fn lo_import_bytes(
2693        &mut self,
2694        want_oid: u32,
2695        data: alloc::vec::Vec<u8>,
2696    ) -> Result<u32, EngineError> {
2697        self.active_catalog_mut()
2698            .create_large_object(want_oid, data)
2699            .map_err(EngineError::Unsupported)
2700    }
2701
2702    /// v7.39 (round 343, V40) — the bytes the host is about to write out.
2703    /// PG's message for a missing object, verbatim.
2704    pub fn lo_export_bytes(&self, oid: u32) -> Result<alloc::vec::Vec<u8>, EngineError> {
2705        self.active_catalog()
2706            .large_object(oid)
2707            .map(<[u8]>::to_vec)
2708            .ok_or_else(|| {
2709                EngineError::Unsupported(alloc::format!("large object {oid} does not exist"))
2710            })
2711    }
2712
2713    pub fn copy_target_columns(
2714        &self,
2715        table: &str,
2716        columns: Option<&[alloc::string::String]>,
2717    ) -> Result<alloc::vec::Vec<alloc::string::String>, EngineError> {
2718        let table_ref = self.active_catalog().get(table).ok_or_else(|| {
2719            EngineError::Storage(spg_storage::StorageError::TableNotFound {
2720                name: alloc::string::String::from(table),
2721            })
2722        })?;
2723        let schema_cols = &table_ref.schema().columns;
2724        match columns {
2725            None => Ok(schema_cols.iter().map(|c| c.name.clone()).collect()),
2726            Some(cols) => {
2727                for (i, name) in cols.iter().enumerate() {
2728                    if !schema_cols
2729                        .iter()
2730                        .any(|c| c.name.eq_ignore_ascii_case(name))
2731                    {
2732                        return Err(EngineError::Unsupported(alloc::format!(
2733                            "column \"{name}\" of relation \"{table}\" does not exist"
2734                        )));
2735                    }
2736                    if cols[..i].iter().any(|p| p.eq_ignore_ascii_case(name)) {
2737                        return Err(EngineError::Unsupported(alloc::format!(
2738                            "column \"{name}\" specified more than once"
2739                        )));
2740                    }
2741                }
2742                Ok(cols.to_vec())
2743            }
2744        }
2745    }
2746
2747    /// v7.39 (round 249) — execute a parsed `COPY … FROM '<file>'` whose
2748    /// file contents the HOST has already read (the engine is no_std and
2749    /// performs no I/O). Lowers to per-row INSERTs via
2750    /// [`crate::copy::copy_buffer_inserts`]; outside an explicit
2751    /// transaction the rows are wrapped in one, so a bad row aborts the
2752    /// whole COPY exactly as in PG.
2753    ///
2754    /// # Errors
2755    /// The failing row's INSERT error propagates (after rollback).
2756    pub fn copy_from_buffer(
2757        &mut self,
2758        table: &str,
2759        columns: Option<&[alloc::string::String]>,
2760        options: &spg_sql::ast::CopyOptions,
2761        data: &str,
2762    ) -> Result<QueryResult, EngineError> {
2763        let target = self.copy_target_columns(table, columns)?;
2764        let inserts = crate::copy::copy_buffer_inserts(table, columns, &target, options, data)?;
2765        let wrap = !self.in_transaction();
2766        if wrap {
2767            self.execute("BEGIN")?;
2768        }
2769        let mut affected: usize = 0;
2770        for insert in &inserts {
2771            match self.execute(insert) {
2772                Ok(QueryResult::CommandOk { affected: n, .. }) => affected += n,
2773                Ok(_) => affected += 1,
2774                Err(e) => {
2775                    if wrap {
2776                        let _ = self.execute("ROLLBACK");
2777                    }
2778                    return Err(e);
2779                }
2780            }
2781        }
2782        if wrap {
2783            self.execute("COMMIT")?;
2784        }
2785        Ok(QueryResult::CommandOk {
2786            affected,
2787            modified_catalog: false,
2788        })
2789    }
2790
2791    /// v7.39 (round 252) — render a `COPY … TO '<file>'` payload for the
2792    /// HOST to write (the engine is no_std and performs no I/O). Returns
2793    /// the encoded bytes (one line per record, trailing newline) and the
2794    /// DATA row count for the `COPY n` tag — the HEADER line, when
2795    /// present, is part of the payload but not of the count.
2796    ///
2797    /// # Errors
2798    /// Same surface as `COPY … TO STDOUT` (missing relation / column,
2799    /// CSV-mode option refusals).
2800    pub fn copy_to_buffer(
2801        &mut self,
2802        table: &str,
2803        columns: Option<&[alloc::string::String]>,
2804        query: Option<&Statement>,
2805        options: &spg_sql::ast::CopyOptions,
2806    ) -> Result<(alloc::string::String, usize), EngineError> {
2807        let result = self.exec_copy_to(table, columns, query, options, CancelToken::none())?;
2808        let QueryResult::Rows { rows, .. } = result else {
2809            return Err(EngineError::Unsupported(
2810                "COPY TO rendered a non-row result".into(),
2811            ));
2812        };
2813        let mut payload = alloc::string::String::new();
2814        for row in &rows {
2815            if let Some(Value::Text(line)) = row.values.first() {
2816                payload.push_str(line);
2817            }
2818            payload.push('\n');
2819        }
2820        let data_rows = rows.len().saturating_sub(usize::from(options.header));
2821        Ok((payload, data_rows))
2822    }
2823
2824    /// `COPY table [(cols)] TO STDOUT` — render the visible rows
2825    /// in COPY text format (tab-separated, `\N` nulls, backslash
2826    /// escapes) as a single-text-column result set. Embedded
2827    /// consumers read the lines directly; the wire layer streams
2828    /// CopyData frames from them.
2829    fn exec_copy_to(
2830        &mut self,
2831        table_name: &str,
2832        columns: Option<&[String]>,
2833        query: Option<&Statement>,
2834        options: &spg_sql::ast::CopyOptions,
2835        cancel: CancelToken<'_>,
2836    ) -> Result<QueryResult, EngineError> {
2837        use spg_sql::ast::CopyFormat;
2838        // v7.39 (read01 round 94) — `COPY (<query>) TO STDOUT`: run the inner
2839        // statement and render its result set with the same per-format cell
2840        // encoder the table form uses. Kept as an early branch so the
2841        // battle-tested table path below is untouched.
2842        if let Some(q) = query {
2843            return self.exec_copy_to_query(q, options, cancel);
2844        }
2845        let table = self.active_catalog().get(table_name).ok_or_else(|| {
2846            EngineError::Storage(spg_storage::StorageError::TableNotFound {
2847                name: alloc::string::String::from(table_name),
2848            })
2849        })?;
2850        let schema_cols = table.schema().columns.clone();
2851        let positions: alloc::vec::Vec<usize> = match columns {
2852            Some(cols) => cols
2853                .iter()
2854                .map(|c| {
2855                    schema_cols
2856                        .iter()
2857                        .position(|s| s.name.eq_ignore_ascii_case(c))
2858                        .ok_or_else(|| {
2859                            EngineError::Eval(crate::eval::EvalError::ColumnNotFound {
2860                                name: c.clone(),
2861                            })
2862                        })
2863                })
2864                .collect::<Result<_, _>>()?,
2865            None => (0..schema_cols.len()).collect(),
2866        };
2867        // Per-format defaults: text = tab / `\N`; csv = comma / `` / `"`.
2868        let is_csv = options.format == CopyFormat::Csv;
2869        let delimiter = options.delimiter.unwrap_or(if is_csv { ',' } else { '\t' });
2870        let quote = options.quote.unwrap_or('"');
2871        let null_str = options
2872            .null_str
2873            .clone()
2874            .unwrap_or_else(|| alloc::string::String::from(if is_csv { "" } else { "\\N" }));
2875        // v7.39 (round 247) — the FORCE_QUOTE mask follows the emitted
2876        // column order (the projection), not the table order.
2877        let out_names: alloc::vec::Vec<alloc::string::String> = positions
2878            .iter()
2879            .filter_map(|&p| schema_cols.get(p).map(|c| c.name.clone()))
2880            .collect();
2881        let (escape, force_mask) =
2882            Self::resolve_copy_csv_extras(options, is_csv, quote, &out_names)?;
2883        let encode_cells = |cells: &[Option<alloc::string::String>]| -> alloc::string::String {
2884            if is_csv {
2885                crate::copy::encode_copy_csv_cells_opts(
2886                    cells,
2887                    delimiter,
2888                    quote,
2889                    escape,
2890                    force_mask.as_deref(),
2891                    &null_str,
2892                )
2893            } else {
2894                crate::copy::encode_copy_text_cells_opts(cells, delimiter, &null_str)
2895            }
2896        };
2897        let snap = self.current_snapshot();
2898        let mut out_rows: alloc::vec::Vec<spg_storage::Row<'static>> = alloc::vec::Vec::new();
2899        // HEADER: the selected column names as the first line, encoded
2900        // per the same format rules (a name is never NULL).
2901        if options.header {
2902            let names: alloc::vec::Vec<Option<alloc::string::String>> = positions
2903                .iter()
2904                .map(|&p| Some(schema_cols[p].name.clone()))
2905                .collect();
2906            out_rows.push(spg_storage::Row::new(alloc::vec![Value::text(
2907                encode_cells(&names)
2908            )]));
2909        }
2910        // COPY renders each value with its type's output function, the
2911        // same as the wire — notably bool as `t` / `f`, not the engine's
2912        // debug-ish `true` / `false`.
2913        // v7.38 (T-tstz Phase 1) — `ty` is the column's declared type, needed
2914        // only to tell timestamptz from timestamp: PG's COPY renders the former
2915        // with its offset. Everything else renders identically either way.
2916        let cell_text = |v: &Value, ty: spg_storage::DataType| -> Option<alloc::string::String> {
2917            match v {
2918                Value::Null => None,
2919                Value::Bool(b) => Some(alloc::string::String::from(if *b { "t" } else { "f" })),
2920                Value::Timestamp(t) if matches!(ty, spg_storage::DataType::Timestamptz) => {
2921                    Some(crate::eval::format_timestamptz(*t))
2922                }
2923                other => Some(crate::eval::values::value_to_text(other)),
2924            }
2925        };
2926        let encode = |row: &spg_storage::Row<'static>| {
2927            let cells: alloc::vec::Vec<Option<alloc::string::String>> = positions
2928                .iter()
2929                .map(|&p| {
2930                    row.values
2931                        .get(p)
2932                        .and_then(|v| cell_text(v, schema_cols[p].ty))
2933                })
2934                .collect();
2935            encode_cells(&cells)
2936        };
2937        for (_, row) in table.scan_visible(&snap) {
2938            cancel.check()?;
2939            out_rows.push(spg_storage::Row::new(alloc::vec![Value::text(encode(row))]));
2940        }
2941        for row in self.iter_cold_rows_of_table(table) {
2942            cancel.check()?;
2943            out_rows.push(spg_storage::Row::new(alloc::vec![Value::text(encode(
2944                &row
2945            ))]));
2946        }
2947        Ok(QueryResult::Rows {
2948            columns: alloc::vec![spg_storage::ColumnSchema::new(
2949                alloc::string::String::from("copy"),
2950                spg_storage::DataType::Text,
2951                false,
2952            )],
2953            rows: out_rows,
2954        })
2955    }
2956
2957    /// v7.39 (read01 round 94) — the `COPY (<query>) TO STDOUT` renderer.
2958    /// Executes the inner statement and encodes its result set into a single
2959    /// `copy` text column (one row per COPY line, header first when asked),
2960    /// exactly like the table form's tail — the difference is only where the
2961    /// rows and their column types come from.
2962    fn exec_copy_to_query(
2963        &mut self,
2964        query: &Statement,
2965        options: &spg_sql::ast::CopyOptions,
2966        cancel: CancelToken<'_>,
2967    ) -> Result<QueryResult, EngineError> {
2968        use spg_sql::ast::CopyFormat;
2969        let (result_cols, result_rows) = match self.dispatch_stmt_inner(query.clone(), cancel)? {
2970            QueryResult::Rows { columns, rows } => (columns, rows),
2971            _ => {
2972                return Err(EngineError::Unsupported(
2973                    "COPY (query) source did not produce a result set".into(),
2974                ));
2975            }
2976        };
2977        let is_csv = options.format == CopyFormat::Csv;
2978        let delimiter = options.delimiter.unwrap_or(if is_csv { ',' } else { '\t' });
2979        let quote = options.quote.unwrap_or('"');
2980        let null_str = options
2981            .null_str
2982            .clone()
2983            .unwrap_or_else(|| alloc::string::String::from(if is_csv { "" } else { "\\N" }));
2984        let out_names: alloc::vec::Vec<alloc::string::String> =
2985            result_cols.iter().map(|c| c.name.clone()).collect();
2986        let (escape, force_mask) =
2987            Self::resolve_copy_csv_extras(options, is_csv, quote, &out_names)?;
2988        let encode_cells = |cells: &[Option<alloc::string::String>]| -> alloc::string::String {
2989            if is_csv {
2990                crate::copy::encode_copy_csv_cells_opts(
2991                    cells,
2992                    delimiter,
2993                    quote,
2994                    escape,
2995                    force_mask.as_deref(),
2996                    &null_str,
2997                )
2998            } else {
2999                crate::copy::encode_copy_text_cells_opts(cells, delimiter, &null_str)
3000            }
3001        };
3002        let cell_text = |v: &Value, ty: spg_storage::DataType| -> Option<alloc::string::String> {
3003            match v {
3004                Value::Null => None,
3005                Value::Bool(b) => Some(alloc::string::String::from(if *b { "t" } else { "f" })),
3006                Value::Timestamp(t) if matches!(ty, spg_storage::DataType::Timestamptz) => {
3007                    Some(crate::eval::format_timestamptz(*t))
3008                }
3009                other => Some(crate::eval::values::value_to_text(other)),
3010            }
3011        };
3012        let mut out_rows: alloc::vec::Vec<spg_storage::Row<'static>> = alloc::vec::Vec::new();
3013        if options.header {
3014            let names: alloc::vec::Vec<Option<alloc::string::String>> =
3015                result_cols.iter().map(|c| Some(c.name.clone())).collect();
3016            out_rows.push(spg_storage::Row::new(alloc::vec![Value::text(
3017                encode_cells(&names)
3018            )]));
3019        }
3020        for row in &result_rows {
3021            cancel.check()?;
3022            let cells: alloc::vec::Vec<Option<alloc::string::String>> = result_cols
3023                .iter()
3024                .enumerate()
3025                .map(|(p, c)| row.values.get(p).and_then(|v| cell_text(v, c.ty)))
3026                .collect();
3027            out_rows.push(spg_storage::Row::new(alloc::vec![Value::text(
3028                encode_cells(&cells)
3029            )]));
3030        }
3031        Ok(QueryResult::Rows {
3032            columns: alloc::vec![spg_storage::ColumnSchema::new(
3033                alloc::string::String::from("copy"),
3034                spg_storage::DataType::Text,
3035                false,
3036            )],
3037            rows: out_rows,
3038        })
3039    }
3040}
3041
3042impl Engine {
3043    /// PG's `PreventInTransactionBlock`: statements whose effect no
3044    /// rollback can undo are refused inside an explicit transaction with
3045    /// 25001, naming themselves in the message.
3046    ///
3047    /// The witness is THIS connection's slot, not the global
3048    /// `in_transaction()`: the engine is shared, so a global check would
3049    /// refuse an autocommit VACUUM merely because a different connection
3050    /// had a transaction open. Same predicate `DISCARD ALL` already uses.
3051    /// Whether a catalog change this statement made is already committed,
3052    /// i.e. THIS connection is not inside an explicit transaction block.
3053    ///
3054    /// It rides out on `QueryResult::modified_catalog`, and the server
3055    /// takes it as "persist and audit this now": in no-WAL mode it drives
3056    /// the snapshot write, and it gates the audit append in every mode.
3057    ///
3058    /// The witness has to be this connection's slot. Asking the
3059    /// engine-wide `in_transaction()` — true while ANY connection holds a
3060    /// transaction — reported an autocommit DDL as uncommitted, and both
3061    /// consequences were measured in round 795: the statement was missing
3062    /// from the audit log entirely, and after `kill -9` plus a restart the
3063    /// table it created was gone, having been acked to the client. A
3064    /// second connection idling inside a BEGIN was the whole cause.
3065    pub(crate) fn catalog_change_is_committed(&self) -> bool {
3066        !self.current_tx.is_some_and(|tx| self.is_tx_open(tx))
3067    }
3068
3069    pub(crate) fn require_no_transaction_block(&self, what: &str) -> Result<(), EngineError> {
3070        if self.current_tx.is_some_and(|tx| self.is_tx_open(tx)) {
3071            return Err(EngineError::Unsupported(alloc::format!(
3072                "{what} cannot run inside a transaction block"
3073            )));
3074        }
3075        Ok(())
3076    }
3077}