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