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