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
26impl Engine {
27    pub fn execute(&mut self, sql: &str) -> Result<QueryResult, EngineError> {
28        self.execute_in_with_cancel(sql, IMPLICIT_TX, CancelToken::none())
29    }
30
31    /// v4.5 — write path with cooperative cancellation. Same dispatch
32    /// as `execute_in_with_cancel(sql, IMPLICIT_TX, cancel)`. Kept as
33    /// a separate entry point for backward-compat with the v4.5
34    /// public API.
35    pub fn execute_with_cancel(
36        &mut self,
37        sql: &str,
38        cancel: CancelToken<'_>,
39    ) -> Result<QueryResult, EngineError> {
40        self.execute_in_with_cancel(sql, IMPLICIT_TX, cancel)
41    }
42
43    /// v4.41.1 multi-slot write entry. Routes `sql` through the TX
44    /// slot identified by `tx_id` so spg-server dispatch can scope
45    /// each implicit-wrap BEGIN..stmt..COMMIT to its own slot in
46    /// `tx_catalogs`. `IMPLICIT_TX` is the legacy single-slot path
47    /// every other caller (engine self-tests, replay, spg-embedded)
48    /// implicitly takes via `execute()` / `execute_with_cancel()`.
49    pub fn execute_in(&mut self, sql: &str, tx_id: TxId) -> Result<QueryResult, EngineError> {
50        self.execute_in_with_cancel(sql, tx_id, CancelToken::none())
51    }
52
53    /// v4.41.1 write path with cooperative cancellation + explicit TX
54    /// scope. Sets `self.current_tx` for the duration of the call so
55    /// every `exec_*` helper transparently sees its TX's shadow
56    /// catalog and savepoint stack; restores on exit so the field is
57    /// only valid mid-call (no leakage across calls).
58    pub fn execute_in_with_cancel(
59        &mut self,
60        sql: &str,
61        tx_id: TxId,
62        cancel: CancelToken<'_>,
63    ) -> Result<QueryResult, EngineError> {
64        // v7.38 P0 元机制 A — establish the per-engine injection
65        // scope for the duration of this execute. The guard pops
66        // the store on drop so nested or sibling engines don't see
67        // ours. No-op in release builds (feature off).
68        let _inj = self.enter_injection_scope();
69        let saved = self.current_tx;
70        self.current_tx = Some(tx_id);
71        // v7.34 (crash-recovery P0 #2) — row-level redo capture. Arm the
72        // active catalog before dispatch; on success drain the physical
73        // changes into `last_redo` for the embedding layer's WAL, on
74        // failure discard them (a failed statement leaves no redo; the
75        // drain clears the tables' capture buffers either way).
76        if self.redo_capture {
77            self.active_catalog_mut().enable_redo_all();
78        }
79        let result = self.execute_inner_with_cancel(sql, cancel);
80        if self.redo_capture {
81            let drained = self.active_catalog_mut().drain_redo();
82            if result.is_ok() {
83                self.last_redo = drained;
84            }
85        }
86        self.current_tx = saved;
87        result
88    }
89
90    /// v6.1.1 — parse and pre-process a SQL string ONCE so the
91    /// resulting [`Statement`] can be cached and re-executed via
92    /// [`Engine::execute_prepared`]. Returns the same `Statement`
93    /// the simple-query path would synthesise internally (clock
94    /// rewrites + ORDER BY position-ref resolution applied at
95    /// prepare time, since both are session-independent). The
96    /// `$N` placeholders in the SQL stay as `Expr::Placeholder(n)`
97    /// nodes; they're resolved to concrete values per-call by
98    /// `execute_prepared`'s substitution walk.
99    ///
100    /// Pgwire's `Parse` (P) message lands here.
101    pub fn prepare(&self, sql: &str) -> Result<Statement, ParseError> {
102        let mut stmt = parser::parse_statement_with(sql, self.backslash_escapes)?;
103        let now_micros = self.clock.map(|f| f());
104        rewrite_clock_calls(&mut stmt, now_micros);
105        if let Statement::Select(s) = &mut stmt {
106            // v6.4.1 — expand `GROUP BY ALL` to every non-aggregate
107            // SELECT-list item BEFORE position / alias resolution so
108            // downstream passes see the explicit list.
109            expand_group_by_all(s);
110            resolve_order_by_position(s);
111            // v6.2.3 — cost-based JOIN reorder. No-op for
112            // single-table FROMs or any non-INNER join shape.
113            // v7.38 元机制 D — `SPG_TEST_PLAN_DETERMINISTIC=1` gates
114            // this so regression tests pin a stable join order.
115            reorder::reorder_joins_with(
116                s,
117                &self.catalog,
118                &self.statistics,
119                self.env_cfg.plan_deterministic,
120            );
121        }
122        Ok(stmt)
123    }
124
125    /// v6.3.0 — cached prepare. Returns a cloned `Statement` from
126    /// the plan cache on hit, runs the full `prepare()` path on miss
127    /// and inserts the resulting plan before returning. Skipping the
128    /// parse + JOIN-reorder pipeline on hit is the dominant win for
129    /// JDBC / sqlx / pgx clients that reuse the same SQL string.
130    ///
131    /// Returns a cloned `Statement` (not a borrow) because the
132    /// pgwire layer owns its `PreparedStmt` map per-session and the
133    /// engine-level cache must stay available for other sessions.
134    /// Clone cost on a 5-table JOIN AST is well under the parse cost
135    /// it replaces.
136    pub fn prepare_cached(&mut self, sql: &str) -> Result<Statement, ParseError> {
137        // v6.3.1 — version-aware lookup. If the cached plan was
138        // prepared before the most recent ANALYZE, evict and replan.
139        let current_version = self.statistics.version();
140        if let Some(plan) = self.plan_cache.get(sql) {
141            if plan.statistics_version == current_version {
142                return Ok(plan.stmt.clone());
143            }
144            // Stale entry — fall through to evict + re-prepare.
145        }
146        self.plan_cache.evict(sql);
147        let stmt = self.prepare(sql)?;
148        let source_tables = plan_cache::collect_source_tables(&stmt);
149        let plan = plan_cache::PreparedPlan {
150            stmt: stmt.clone(),
151            statistics_version: current_version,
152            source_tables,
153            describe_columns: alloc::vec::Vec::new(),
154        };
155        self.plan_cache.insert(String::from(sql), plan);
156        Ok(stmt)
157    }
158
159    /// v6.3.0 — read-only accessor for tests and v6.3.1 invalidation.
160    pub fn plan_cache(&self) -> &plan_cache::PlanCache {
161        &self.plan_cache
162    }
163
164    /// v7.38 (mailrs prod 7.35 pool-exhaustion incident) — boot-time
165    /// plan-IR cache warm-up. Walks `sqls`, calls `prepare_cached`
166    /// on each one. Each successful prepare leaves the parsed +
167    /// reordered + clock-rewritten `Statement` in the engine-wide
168    /// plan cache; subsequent `Engine::execute` / `execute_prepared`
169    /// for the same SQL skips parse + JOIN reorder. Returns the
170    /// count of successfully cached statements.
171    ///
172    /// The mailrs `Database::new` boot path is the expected caller:
173    /// pre-warm the top-N query shapes (inbox listing, contacts
174    /// search, stats) so the first user-facing request doesn't
175    /// pay the 2-3 s first-fire cost on the readonly-blocking
176    /// sqlx pool — which (under prod concurrency) exhausts the
177    /// pool and stalls the whole UI.
178    pub fn warm_up_plan_cache(&mut self, sqls: &[&str]) -> usize {
179        let mut warmed = 0;
180        for sql in sqls {
181            if self.prepare_cached(sql).is_ok() {
182                warmed += 1;
183            }
184        }
185        warmed
186    }
187
188    /// v7.38 (mailrs prod 7.35 pool-exhaustion incident) — boot-time
189    /// cold-tier OS page-cache warm-up. Walks every table in the
190    /// active catalog, iterates the cold rows via the existing
191    /// BTree-driven `iter_cold_rows_of_table`, drops the rows on
192    /// the floor. The walk's side effect is that every cold
193    /// segment file gets mmap-read once — the OS page cache then
194    /// serves subsequent queries without disk I/O.
195    ///
196    /// Returns the total cold rows touched across all tables.
197    /// On a hot-only catalog (no `cold_segments` populated) the
198    /// call is a near-no-op.
199    pub fn warm_up_cold_tier(&self) -> usize {
200        let catalog = self.active_catalog();
201        let mut total = 0;
202        for name in catalog.table_names() {
203            if let Some(table) = catalog.get(&name) {
204                let rows = self.iter_cold_rows_of_table(table);
205                total += rows.len();
206            }
207        }
208        total
209    }
210
211    /// v6.3.0 — mutable accessor for v6.3.1 invalidation hooks.
212    pub fn plan_cache_mut(&mut self) -> &mut plan_cache::PlanCache {
213        &mut self.plan_cache
214    }
215
216    /// v6.3.3 — Describe a prepared `Statement` without executing.
217    /// Returns `(parameter_oids, output_columns)`. Empty
218    /// `output_columns` means the statement has no row-producing
219    /// shape we could resolve here (JOIN, subquery, non-SELECT, …)
220    /// — pgwire layer maps that to a `NoData` reply.
221    pub fn describe_prepared(&self, stmt: &Statement) -> (Vec<u32>, Vec<ColumnSchema>) {
222        describe::describe_prepared(stmt, self.active_catalog())
223    }
224
225    /// v6.1.1 — execute a [`Statement`] previously returned by
226    /// [`Engine::prepare`], substituting `Expr::Placeholder(n)`
227    /// nodes for the corresponding [`Value`] in `params` (1-based
228    /// per PG: `$1` → `params[0]`). Bind-time string parameters
229    /// are decoded into typed `Value`s by the pgwire layer before
230    /// this call so the resulting AST hits the same execution
231    /// path as a simple query — no SQL re-parse.
232    ///
233    /// Pgwire's `Execute` (E) message after a `Bind` (B) lands here.
234    pub fn execute_prepared(
235        &mut self,
236        stmt: Statement,
237        params: &[Value<'static>],
238    ) -> Result<QueryResult, EngineError> {
239        self.execute_prepared_with_cancel(stmt, params, CancelToken::none())
240    }
241
242    /// v7.37 (SPGS small-query bar) — borrow-based SELECT entry for
243    /// the pgwire `Execute` hot path when the portal has no bound
244    /// parameters. Skips both the AST clone the prepared path used
245    /// to do at the pgwire call site AND the `substitute_
246    /// placeholders` walk (a no-op when params are empty). Caller
247    /// must already hold the engine write lock — read would be
248    /// cleaner, but `current_tx` mutation keeps it `&mut`.
249    pub fn execute_prepared_select_no_params(
250        &mut self,
251        stmt: &spg_sql::ast::SelectStatement,
252        cancel: CancelToken<'_>,
253    ) -> Result<QueryResult, EngineError> {
254        let saved = self.current_tx;
255        self.current_tx = Some(IMPLICIT_TX);
256        let result = self.exec_select_cancel(stmt, cancel);
257        self.current_tx = saved;
258        result
259    }
260
261    /// v7.37 — streaming SELECT for the pgwire `Execute` hot path.
262    /// Emits one `StreamItem::Header(cols)` then one
263    /// `StreamItem::Row(&[&Value])` per surviving row. Returns the
264    /// total row count for the `CommandComplete` tag.
265    ///
266    /// For shapes where the engine can stream directly (non-aggregate
267    /// join projection of bound columns, no ORDER BY / DISTINCT / etc.)
268    /// no `Vec<Row<'static>>` is materialised — cell references come straight
269    /// out of the source tables. For non-streamable shapes the engine
270    /// runs the full `exec_select_cancel`, then walks the materialised
271    /// `Vec<Row<'static>>` driving the same emit callback (no engine-side win,
272    /// but pgwire dispatches every Execute through one path).
273    pub fn execute_prepared_select_streaming<F>(
274        &mut self,
275        stmt: &spg_sql::ast::SelectStatement,
276        cancel: CancelToken<'_>,
277        mut emit: F,
278    ) -> Result<usize, EngineError>
279    where
280        F: FnMut(StreamItem<'_>) -> Result<(), EngineError>,
281    {
282        let saved = self.current_tx;
283        self.current_tx = Some(IMPLICIT_TX);
284        let inner = self.exec_select_streaming(stmt, cancel, &mut emit);
285        self.current_tx = saved;
286        inner
287    }
288
289    /// v7.37 — internal streaming dispatcher. Phase 1: fall-back path
290    /// only — runs the materialising `exec_select_cancel`, then drives
291    /// the emit callback from the resulting `Vec<Row<'static>>`. Phase 2 will
292    /// add a true streaming path for the joined-projection shape.
293    fn exec_select_streaming<F>(
294        &mut self,
295        stmt: &spg_sql::ast::SelectStatement,
296        cancel: CancelToken<'_>,
297        emit: &mut F,
298    ) -> Result<usize, EngineError>
299    where
300        F: FnMut(StreamItem<'_>) -> Result<(), EngineError>,
301    {
302        // v7.37 — true-streaming fast path for joined-non-aggregate
303        // projection of bound columns. Skips `Vec<Row<'static>>` + per-cell
304        // `.cloned()` (about 4 ms saved on the 25 k-row PROJ shape).
305        // Unresolved subqueries / pull-up shapes / non-streamable
306        // structure (ORDER BY, DISTINCT, …) fall through to the
307        // materialising path.
308        if !crate::subquery::expr_tree_has_subquery(stmt) {
309            if let Some(n) = self.try_exec_joined_streaming(stmt, cancel, emit)? {
310                return Ok(n);
311            }
312        }
313        // Fall-back: materialise then iterate.
314        let QueryResult::Rows { columns, rows } = self.exec_select_cancel(stmt, cancel)? else {
315            return Err(EngineError::Unsupported(alloc::string::String::from(
316                "streaming SELECT got a non-Rows result",
317            )));
318        };
319        emit(StreamItem::Header(&columns))?;
320        let mut cell_refs: Vec<&Value> = Vec::with_capacity(columns.len());
321        for row in &rows {
322            cell_refs.clear();
323            for v in &row.values {
324                cell_refs.push(v);
325            }
326            emit(StreamItem::Row(&cell_refs))?;
327        }
328        Ok(rows.len())
329    }
330}
331
332/// v7.37 — one item in the streaming SELECT emit channel. The
333/// engine yields exactly one `Header` (before any row) then zero
334/// or more `Row`s. Pgwire (or any other consumer) decides how to
335/// turn those into wire bytes.
336#[derive(Debug)]
337pub enum StreamItem<'a> {
338    Header(&'a [ColumnSchema]),
339    Row(&'a [&'a Value<'static>]),
340}
341
342impl Engine {
343    /// v7.17.0 Phase 2.3 — prepared-statement entry that honors a
344    /// caller-supplied `CancelToken`. Mirrors `execute_prepared`'s
345    /// `current_tx` save/restore so the extended-query path stays
346    /// transactionally consistent with the simple-query path.
347    pub fn execute_prepared_with_cancel(
348        &mut self,
349        mut stmt: Statement,
350        params: &[Value<'static>],
351        cancel: CancelToken<'_>,
352    ) -> Result<QueryResult, EngineError> {
353        substitute_placeholders(&mut stmt, params)?;
354        // v7.16.0 — set `current_tx` for the duration of the
355        // dispatch so the `exec_*` helpers see the right TX
356        // slot (matches what `execute_in_with_cancel` does for
357        // simple-query). Pre-v7.16 the simple-query path
358        // worked because every public entry point routed
359        // through `execute_in_with_cancel`; the prepared path
360        // skipped the wrap and so its INSERTs/UPDATEs landed
361        // in the no-tx default slot, silently invisible to a
362        // BEGIN/COMMIT-bracketed flow. Caught by spg-sqlx's
363        // first transaction-visibility test.
364        let saved = self.current_tx;
365        self.current_tx = Some(IMPLICIT_TX);
366        let result = self.execute_stmt_with_cancel(stmt, cancel);
367        self.current_tx = saved;
368        result
369    }
370
371    fn execute_inner_with_cancel(
372        &mut self,
373        sql: &str,
374        cancel: CancelToken<'_>,
375    ) -> Result<QueryResult, EngineError> {
376        cancel.check()?;
377        let stmt = self.prepare(sql)?;
378        // v6.5.1 — wrap the executor with a wall-clock window so we
379        // can record into spg_stat_query. Skip when the engine has
380        // no clock attached (no_std embedded callers).
381        let start_us = self.clock.map(|f| f());
382        let result = self.execute_stmt_with_cancel(stmt, cancel);
383        if let (Some(t0), Ok(_)) = (start_us, &result) {
384            let now = self.clock.map_or(t0, |f| f());
385            let elapsed = now.saturating_sub(t0).max(0) as u64;
386            self.query_stats.record(sql, elapsed, now as u64);
387            // v6.5.6 — slow-query log: fire callback when elapsed
388            // exceeds the configured floor.
389            if let (Some(threshold), Some(logger)) =
390                (self.slow_query_threshold_us, self.slow_query_logger)
391                && elapsed >= threshold
392            {
393                logger(sql, elapsed);
394            }
395        }
396        result
397    }
398
399    pub(crate) fn execute_stmt_with_cancel(
400        &mut self,
401        stmt: Statement,
402        cancel: CancelToken<'_>,
403    ) -> Result<QueryResult, EngineError> {
404        cancel.check()?;
405        // v7.17.0 Phase 1.1 — pre-resolve nextval / currval /
406        // setval calls in the statement tree. Walks SELECT
407        // projection, INSERT VALUES, UPDATE SET, DELETE WHERE,
408        // and DEFAULT exprs; replaces sequence FunctionCall
409        // nodes with concrete Literal values minted against the
410        // catalog. This is the only place that mutates sequence
411        // state from a SELECT-shaped path (exec_select_cancel is
412        // `&self` and can't reach the catalog mutably).
413        //
414        // Fast-path: when no sequences exist anywhere in the
415        // catalog (the typical hot-path INSERT load), skip the
416        // walker entirely. Single map-emptiness check on the
417        // catalog beats walking every expression on every call.
418        let mut stmt = stmt;
419        // v7.17 dump-compat — the fast-path check
420        // `sequences().is_empty()` skips pre-resolve when no
421        // sequence exists in the *currently active* catalog
422        // snapshot. The committed catalog or the implicit-TX
423        // catalog may legitimately disagree on this between
424        // CREATE SEQUENCE and a later setval(): always run the
425        // resolver — the walk is O(expr-count) and dwarfed by
426        // the parse cost we just paid.
427        self.pre_resolve_sequence_calls_in_statement(&mut stmt)?;
428        let result = match stmt {
429            Statement::CreateTable(s) => self.exec_create_table(s),
430            // v7.9.15 — CREATE EXTENSION is a no-op on SPG. Returns
431            // CommandOk with affected=0; modified_catalog=false so
432            // the WAL doesn't grow a useless entry. mailrs F3.
433            Statement::CreateExtension(_) => Ok(QueryResult::CommandOk {
434                affected: 0,
435                modified_catalog: false,
436            }),
437            // v7.16.2 — DO $$ ... $$ block. mailrs round-10 A.2
438            // — the pre-v7.9.27 no-op SILENTLY swallowed every
439            // mailrs migrate-038/-040/-042 idempotent rename
440            // (the IF EXISTS … THEN ALTER … END block never
441            // ran). v7.16.2 dispatches to exec_do_block which
442            // runs the PlPgSqlBlock at top level via the same
443            // execute_stmts machinery the trigger executor
444            // uses (NEW=None, OLD=None — DO blocks have no
445            // row context).
446            Statement::DoBlock(body) => self.exec_do_block(body),
447            // v7.14.0 — empty-statement no-op for pg_dump /
448            // mysqldump preamble lines that collapse to nothing
449            // after comment-stripping.
450            Statement::Empty => Ok(QueryResult::CommandOk {
451                affected: 0,
452                modified_catalog: false,
453            }),
454            Statement::DropTable { names, if_exists } => self.exec_drop_table(names, if_exists),
455            Statement::DropIndex { name, if_exists } => self.exec_drop_index(name, if_exists),
456            Statement::CreateIndex(s) => self.exec_create_index(s),
457            Statement::Insert(s) => self.exec_insert(s),
458            Statement::Update(mut s) => {
459                // Materialise uncorrelated subqueries in SET / WHERE
460                // before the row walk — the SELECT path has done this
461                // since v4.10; UPDATE gained it for mailrs's
462                // `UPDATE … WHERE id IN (SELECT … FOR UPDATE SKIP
463                // LOCKED)` claim pattern (embed round-12).
464                for (_, e) in &mut s.assignments {
465                    self.resolve_expr_subqueries(e, cancel)?;
466                }
467                if let Some(w) = &mut s.where_ {
468                    self.resolve_expr_subqueries(w, cancel)?;
469                }
470                self.exec_update_cancel(&s, cancel)
471            }
472            Statement::Delete(mut s) => {
473                if let Some(w) = &mut s.where_ {
474                    self.resolve_expr_subqueries(w, cancel)?;
475                }
476                self.exec_delete_cancel(&s, cancel)
477            }
478            Statement::Merge(s) => self.exec_merge_cancel(&s, cancel),
479            Statement::Select(s) => self.exec_select_cancel(&s, cancel),
480            Statement::Begin => self.exec_begin(),
481            Statement::Commit => self.exec_commit(),
482            Statement::Rollback => self.exec_rollback(),
483            Statement::Savepoint(name) => self.exec_savepoint(name),
484            Statement::RollbackToSavepoint(name) => self.exec_rollback_to_savepoint(&name),
485            Statement::ReleaseSavepoint(name) => self.exec_release_savepoint(&name),
486            Statement::ShowTables => Ok(self.exec_show_tables()),
487            Statement::ShowDatabases => Ok(self.exec_show_databases()),
488            Statement::ShowCreateTable(name) => self.exec_show_create_table(&name),
489            Statement::ShowIndexes(name) => self.exec_show_indexes(&name),
490            Statement::ShowStatus => Ok(self.exec_show_status()),
491            Statement::ShowVariables => Ok(self.exec_show_variables()),
492            Statement::ShowProcesslist => Ok(self.exec_show_processlist()),
493            Statement::ShowColumns(table) => self.exec_show_columns(&table),
494            Statement::ShowUsers => Ok(self.exec_show_users()),
495            Statement::ShowPublications => Ok(self.exec_show_publications()),
496            Statement::ShowSubscriptions => Ok(self.exec_show_subscriptions()),
497            Statement::CreateUser(s) => self.exec_create_user(&s),
498            Statement::DropUser(name) => self.exec_drop_user(&name),
499            Statement::Explain(e) => self.exec_explain(&e, cancel),
500            Statement::AlterIndex(s) => self.exec_alter_index(s),
501            Statement::AlterTable(s) => self.exec_alter_table(s),
502            Statement::CreatePublication(s) => self.exec_create_publication(s),
503            Statement::DropPublication(name) => self.exec_drop_publication(&name),
504            Statement::CreateSubscription(s) => self.exec_create_subscription(s),
505            Statement::DropSubscription(name) => self.exec_drop_subscription(&name),
506            // v6.1.7 — WAIT FOR WAL POSITION needs `lag_state`,
507            // which lives in spg-server's ServerState. The engine
508            // surfaces a clear error; the server-layer dispatch
509            // intercepts the SQL before it reaches the engine on
510            // a server build, so this arm only fires for
511            // engine-only callers (spg-embedded, lib tests).
512            Statement::WaitForWalPosition { .. } => Err(EngineError::Unsupported(
513                "WAIT FOR WAL POSITION must be handled by the server layer".into(),
514            )),
515            // v6.2.0 — ANALYZE recomputes per-column histograms.
516            Statement::Analyze(target) => self.exec_analyze(target.as_deref()),
517            // v6.7.3 — COMPACT COLD SEGMENTS.
518            Statement::CompactColdSegments => self.exec_compact_cold_segments(),
519            // v7.12.1 — SET / RESET session parameter. Engine
520            // tracks the value in `session_params`; FTS dispatcher
521            // reads `default_text_search_config`. Everything else
522            // is a recorded no-op (PG dump compat).
523            Statement::SetParameter { name, value } => {
524                self.set_session_param(name, value);
525                Ok(QueryResult::CommandOk {
526                    affected: 0,
527                    modified_catalog: false,
528                })
529            }
530            // v7.38 轴 4 — `SET TRANSACTION ISOLATION LEVEL …`. The
531            // surface is recorded on `Engine::current_isolation_level`
532            // and visible via `SHOW transaction_isolation`. Behavioural
533            // implementation (REPEATABLE READ snapshot / SERIALIZABLE
534            // SSI) lands separately; today every level reads as
535            // effective READ COMMITTED (same as PG's silent upgrade
536            // of READ UNCOMMITTED).
537            Statement::SetTransaction { isolation } => {
538                self.current_isolation_level = isolation;
539                Ok(QueryResult::CommandOk {
540                    affected: 0,
541                    modified_catalog: false,
542                })
543            }
544            // v7.38 轴 4 surface expansion — `SHOW <parameter>`
545            // returns a 1-row 1-column TEXT result (the PG psql
546            // wire shape). The handler dispatches per-name:
547            //
548            // 1. transaction_isolation — direct read of
549            //    current_isolation_level (the v7.38 axis-4 surface).
550            // 2. PG preset / engine-tracked params — values mirror
551            //    pg_catalog.pg_settings to keep ORM /
552            //    driver-connect probes happy (sqlx asks
553            //    server_version + standard_conforming_strings +
554            //    client_encoding; npgsql asks application_name;
555            //    asyncpg asks search_path). Any
556            //    SET-tracked override on self.session_params wins.
557            // 3. Anything else — error with a list-pointer to
558            //    pg_settings (which lists every recognised name).
559            Statement::ShowParameter(name) => {
560                use spg_storage::{ColumnSchema, DataType, Row, Value};
561                let owned;
562                let value: &str = match name.as_str() {
563                    "transaction_isolation" => self.current_isolation_level.as_pg_str(),
564                    "server_version" => "16.0 (spg)",
565                    "server_encoding" => "UTF8",
566                    "is_superuser" => "on",
567                    "TimeZone" | "timezone" => self
568                        .session_param("TimeZone")
569                        .or_else(|| self.session_param("timezone"))
570                        .unwrap_or("UTC"),
571                    "DateStyle" | "datestyle" => self
572                        .session_param("DateStyle")
573                        .or_else(|| self.session_param("datestyle"))
574                        .unwrap_or("ISO, MDY"),
575                    "client_encoding" => {
576                        self.session_param("client_encoding").unwrap_or("UTF8")
577                    }
578                    "standard_conforming_strings" => self
579                        .session_param("standard_conforming_strings")
580                        .unwrap_or("on"),
581                    "search_path" => self.session_param("search_path").unwrap_or("\"$user\", public"),
582                    "application_name" => self.session_param("application_name").unwrap_or(""),
583                    "statement_timeout" => {
584                        self.session_param("statement_timeout").unwrap_or("0")
585                    }
586                    "default_transaction_isolation" => self
587                        .session_param("default_transaction_isolation")
588                        .unwrap_or("read committed"),
589                    "intervalstyle" | "IntervalStyle" => self
590                        .session_param("IntervalStyle")
591                        .or_else(|| self.session_param("intervalstyle"))
592                        .unwrap_or("postgres"),
593                    other => {
594                        // Fall through to session_params for any user-set
595                        // override that didn't fall into a named bucket.
596                        if let Some(v) = self.session_param(other) {
597                            owned = alloc::string::String::from(v);
598                            &owned
599                        } else {
600                            return Err(EngineError::Unsupported(alloc::format!(
601                                "SHOW {other:?}: parameter not recognised; \
602                                 see `SELECT name, setting FROM pg_settings` for \
603                                 the full inventory"
604                            )));
605                        }
606                    }
607                };
608                Ok(QueryResult::Rows {
609                    columns: alloc::vec![ColumnSchema::new(name, DataType::Text, false)],
610                    rows: alloc::vec![Row::new(alloc::vec![Value::text(value)])],
611                })
612            }
613            // v7.14.0 — MySQL multi-assignment SET. Each pair runs
614            // through `set_session_param` so engine-known params
615            // (FOREIGN_KEY_CHECKS, session_replication_role, …) take
616            // effect; unknown pairs (including `@VAR` LHS from the
617            // mysqldump preamble) are recorded then ignored.
618            Statement::SetParameterList(pairs) => {
619                for (name, value) in pairs {
620                    self.set_session_param(name, value);
621                }
622                Ok(QueryResult::CommandOk {
623                    affected: 0,
624                    modified_catalog: false,
625                })
626            }
627            // v7.12.4 — CREATE FUNCTION / CREATE TRIGGER / DROP …
628            // for the PL/pgSQL trigger surface. exec_* methods are
629            // defined alongside the existing CREATE handlers below.
630            Statement::CreateFunction(s) => self.exec_create_function(s),
631            Statement::CreateTrigger(s) => self.exec_create_trigger(s),
632            Statement::DropTrigger {
633                name,
634                table,
635                if_exists,
636            } => self.exec_drop_trigger(&name, &table, if_exists),
637            Statement::DropFunction { name, if_exists } => {
638                self.exec_drop_function(&name, if_exists)
639            }
640            Statement::CreateSequence(s) => self.exec_create_sequence(s),
641            Statement::AlterSequence(s) => self.exec_alter_sequence(s),
642            Statement::DropSequence { names, if_exists } => {
643                self.exec_drop_sequence(&names, if_exists)
644            }
645            Statement::CreateView(s) => self.exec_create_view(s),
646            Statement::DropView { names, if_exists } => self.exec_drop_view(&names, if_exists),
647            Statement::CreateMaterializedView(s) => self.exec_create_materialized_view(s),
648            Statement::RefreshMaterializedView { name, with_data } => {
649                self.exec_refresh_materialized_view(&name, with_data)
650            }
651            Statement::DropMaterializedView { names, if_exists } => {
652                self.exec_drop_materialized_view(&names, if_exists)
653            }
654            Statement::CreateType(s) => self.exec_create_type(s),
655            Statement::DropType { names, if_exists } => self.exec_drop_type(&names, if_exists),
656            Statement::CreateDomain(s) => self.exec_create_domain(s),
657            Statement::DropDomain { names, if_exists } => self.exec_drop_domain(&names, if_exists),
658            Statement::CreateSchema {
659                name,
660                if_not_exists,
661            } => self.exec_create_schema(name, if_not_exists),
662            Statement::DropSchema { names, if_exists } => self.exec_drop_schema(&names, if_exists),
663            Statement::ResetParameter(target) => {
664                match target {
665                    None => self.session_params.clear(),
666                    Some(name) => {
667                        self.session_params.remove(&name.to_ascii_lowercase());
668                    }
669                }
670                Ok(QueryResult::CommandOk {
671                    affected: 0,
672                    modified_catalog: false,
673                })
674            }
675        };
676        self.enforce_row_limit(result)
677    }
678}