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