#![allow(
clippy::cast_lossless,
clippy::cast_possible_truncation,
clippy::cast_possible_wrap,
clippy::cast_sign_loss,
clippy::doc_markdown,
clippy::similar_names,
clippy::too_many_lines,
clippy::uninlined_format_args,
clippy::unreadable_literal,
clippy::assigning_clones
)]
use std::cell::RefCell;
use std::collections::VecDeque;
use std::io::{Read, Write};
use std::net::{TcpListener, TcpStream};
use std::sync::Arc;
use std::thread;
use spg_engine::{CancelToken, EngineError, MonotonicNowFn, QueryResult, Role};
use spg_storage::{ColumnSchema, DataType, Row, Value};
use crate::ServerState;
use crate::mysqlwire::ReadWrite;
const PGWIRE_PARSE_CACHE_CAP: usize = 64;
thread_local! {
static PGWIRE_PARSE_CACHE: RefCell<
VecDeque<(String, Arc<spg_engine::SelectStatement>)>,
> = const { RefCell::new(VecDeque::new()) };
}
fn pgwire_parse_cache_get(sql: &str) -> Option<Arc<spg_engine::SelectStatement>> {
PGWIRE_PARSE_CACHE.with(|c| {
let c = c.borrow();
for (k, s) in c.iter() {
if k == sql {
return Some(Arc::clone(s));
}
}
None
})
}
fn pgwire_parse_cache_put(sql: &str, stmt: Arc<spg_engine::SelectStatement>) {
PGWIRE_PARSE_CACHE.with(|c| {
let mut c = c.borrow_mut();
c.retain(|(k, _)| k != sql);
if c.len() >= PGWIRE_PARSE_CACHE_CAP {
c.pop_back();
}
c.push_front((sql.to_string(), stmt));
});
}
const PROTOCOL_V3: u32 = 196608;
pub fn spawn_listener(
addr: &str,
state: Arc<ServerState>,
) -> std::io::Result<std::net::SocketAddr> {
let listener = TcpListener::bind(addr)?;
let local = listener.local_addr()?;
let _ = thread::Builder::new()
.name("spg-pgwire-listen".into())
.spawn(move || {
for stream in listener.incoming() {
let Ok(stream) = stream else {
continue;
};
let state = Arc::clone(&state);
let _ = thread::Builder::new()
.name("spg-pgwire-conn".into())
.spawn(move || {
if let Err(e) = handle_conn(stream, &state) {
eprintln!("spg-server: pg-wire conn error: {e}");
}
});
}
});
Ok(local)
}
pub(crate) fn split_top_level_statements(body: &[u8]) -> Vec<&[u8]> {
let mut out: Vec<&[u8]> = Vec::new();
let mut start = 0usize;
let mut i = 0usize;
let n = body.len();
while i < n {
let b = body[i];
match b {
b'\'' => {
i += 1;
while i < n {
if body[i] == b'\'' {
if i + 1 < n && body[i + 1] == b'\'' {
i += 2;
continue;
}
i += 1;
break;
}
i += 1;
}
}
b'"' => {
i += 1;
while i < n {
if body[i] == b'"' {
if i + 1 < n && body[i + 1] == b'"' {
i += 2;
continue;
}
i += 1;
break;
}
i += 1;
}
}
b'-' if i + 1 < n && body[i + 1] == b'-' => {
i += 2;
while i < n && body[i] != b'\n' {
i += 1;
}
}
b'/' if i + 1 < n && body[i + 1] == b'*' => {
i += 2;
let mut depth = 1u32;
while i < n && depth > 0 {
if i + 1 < n && body[i] == b'/' && body[i + 1] == b'*' {
depth += 1;
i += 2;
} else if i + 1 < n && body[i] == b'*' && body[i + 1] == b'/' {
depth -= 1;
i += 2;
} else {
i += 1;
}
}
}
b'$' => {
let tag_start = i + 1;
let mut j = tag_start;
let mut valid_tag = true;
while j < n && body[j] != b'$' {
let c = body[j];
let ok = if j == tag_start {
c.is_ascii_alphabetic() || c == b'_'
} else {
c.is_ascii_alphanumeric() || c == b'_'
};
if !ok {
valid_tag = false;
break;
}
j += 1;
}
if !valid_tag || j >= n {
i += 1;
continue;
}
let close_len = j - i + 1;
let close_start = i;
let close_end = j + 1;
i = j + 1;
while i + close_len <= n {
if body[i..i + close_len] == body[close_start..close_end] {
i += close_len;
break;
}
i += 1;
}
if i + close_len > n {
i = n;
}
}
b';' => {
let slice = &body[start..i];
if !slice.iter().all(|c| c.is_ascii_whitespace()) {
out.push(slice);
}
i += 1;
start = i;
}
_ => {
i += 1;
}
}
}
if start < n {
let slice = &body[start..n];
if !slice.iter().all(|c| c.is_ascii_whitespace()) {
out.push(slice);
}
}
out
}
#[allow(clippy::too_many_arguments)]
fn handle_pg_simple_query(
stream: &mut dyn ReadWrite,
body: &[u8],
state: &Arc<ServerState>,
conn_state: &Arc<crate::ConnState>,
role: Role,
tx_state: &mut u8,
settings: &mut std::collections::HashMap<String, String>,
wbuf: &mut Vec<u8>,
) -> std::io::Result<()> {
let sql_bytes = body.strip_suffix(b"\0").unwrap_or(body);
{
let stmts = split_top_level_statements(sql_bytes);
if stmts.len() > 1 {
return dispatch_pg_simple_query_multi(
stream, &stmts, state, conn_state, role, tx_state, settings, wbuf,
);
}
}
if *tx_state == b'E' {
let mut vb = trim_ascii(sql_bytes);
if vb.last() == Some(&b';') {
vb = trim_ascii(&vb[..vb.len() - 1]);
}
let verb = vb
.split(|&b| b == b' ' || b == b'\t' || b == b'\n' || b == b'\r')
.next()
.unwrap_or(b"");
let is_tx_control = ci_eq(verb, b"commit")
|| ci_eq(verb, b"rollback")
|| ci_eq(verb, b"end")
|| ci_eq(verb, b"abort");
if !is_tx_control {
send_error(
wbuf,
"25P02",
"current transaction is aborted, commands ignored until end of transaction block",
)?;
send_ready_for_query(wbuf, *tx_state)?;
stream.write_all(wbuf)?;
wbuf.clear();
return Ok(());
}
}
{
let trimmed_bytes = trim_ascii(sql_bytes);
let trimmed_bytes = if trimmed_bytes.last() == Some(&b';') {
trim_ascii(&trimmed_bytes[..trimmed_bytes.len() - 1])
} else {
trimmed_bytes
};
if let Some(rest) = ci_strip_prefix(trimmed_bytes, b"select ") {
let rest_trim = trim_ascii(rest);
if !rest_trim.is_empty()
&& rest_trim
.iter()
.all(|c| c.is_ascii_digit() || *c == b'-' || *c == b'+')
&& let Ok(s) = core::str::from_utf8(rest_trim)
&& let Ok(n) = s.parse::<i64>()
{
let mine: Vec<(String, String)> = conn_state
.notify_queue
.lock()
.map(|mut q| core::mem::take(&mut *q))
.unwrap_or_default();
for (channel, payload) in &mine {
let mut body = Vec::with_capacity(channel.len() + payload.len() + 8);
body.extend_from_slice(&conn_state.pid.to_be_bytes());
body.extend_from_slice(channel.as_bytes());
body.push(0);
body.extend_from_slice(payload.as_bytes());
body.push(0);
send_msg(wbuf, b'A', &body)?;
}
encode_select_int_response(wbuf, n, *tx_state)?;
stream.write_all(wbuf)?;
wbuf.clear();
return Ok(());
}
}
}
if let Ok(mut e) = state.engine.write() {
e.set_current_session(conn_state.pid);
}
let now_us = wallclock_unix_micros();
conn_state
.last_query_start_us
.store(now_us, std::sync::atomic::Ordering::Relaxed);
if let Ok(mut s) = conn_state.current_sql.write() {
s.clear();
match std::str::from_utf8(sql_bytes) {
Ok(valid) => s.push_str(valid),
Err(_) => s.push_str(&String::from_utf8_lossy(sql_bytes)),
}
}
let Ok(sql_str) = std::str::from_utf8(sql_bytes) else {
send_error(wbuf, "22021", "invalid UTF-8 in query")?;
send_ready_for_query(wbuf, *tx_state)?;
stream.write_all(wbuf)?;
wbuf.clear();
return Ok(());
};
let sql: &str = sql_str.trim_end_matches(';').trim();
if let Some((name, value)) = parse_set_statement(sql) {
let name_lc = name.to_ascii_lowercase();
settings.insert(name_lc.clone(), value.clone());
if name_lc == "application_name" {
if let Ok(mut g) = conn_state.application_name.write() {
*g = value.clone();
}
}
let _ = &value;
}
{
let t = sql.trim();
let b = t.as_bytes();
let restore_app_name = |settings: &mut std::collections::HashMap<String, String>| {
if !conn_state.startup_app_name.is_empty() {
settings.insert(
"application_name".to_string(),
conn_state.startup_app_name.clone(),
);
}
if let Ok(mut g) = conn_state.application_name.write() {
g.clone_from(&conn_state.startup_app_name);
}
};
if ci_eq(b, b"reset all") || ci_starts_with(b, b"discard all") {
settings.clear();
restore_app_name(settings);
} else if ci_starts_with(b, b"reset ") {
let name = t[6..]
.trim()
.trim_end_matches(';')
.trim()
.to_ascii_lowercase();
settings.remove(&name);
if name == "application_name" {
restore_app_name(settings);
}
}
}
if let Some(name) = parse_show_statement(sql) {
let engine_val: Option<String> = if name != "all" {
state
.engine
.read()
.ok()
.and_then(|e| e.session_param(&name).map(str::to_string))
} else {
None
};
let resp = match engine_val {
Some(v) => CannedResponse::Rows {
columns: vec![ColumnSchema::new(name.clone(), DataType::Text, false)],
rows: vec![Row::new(vec![Value::text(v)])],
},
None => render_show(&name, settings),
};
send_canned(wbuf, &resp)?;
send_ready_for_query(wbuf, *tx_state)?;
stream.write_all(wbuf)?;
wbuf.clear();
return Ok(());
}
if let Some(call) = spg_engine::largeobject::parse_lo_file_call(sql) {
handle_lo_file_call(wbuf, state, role, &call)?;
send_ready_for_query(wbuf, *tx_state)?;
stream.write_all(wbuf)?;
wbuf.clear();
return Ok(());
}
if let Some(copy) = parse_copy_intent(sql) {
if !wbuf.is_empty() {
stream.write_all(wbuf)?;
wbuf.clear();
}
match copy {
CopyIntent::From(table, cols, opts) => {
handle_copy_from_stdin(
stream,
state,
role,
&table,
cols.as_deref(),
&opts,
tx_state,
conn_state.tx_id,
)?;
}
CopyIntent::FromFile(spec) => {
handle_copy_from_file(stream, state, role, &spec, tx_state, conn_state.tx_id)?;
}
CopyIntent::ToFile(spec) => {
handle_copy_to_file(stream, state, role, &spec)?;
}
CopyIntent::BadOption(name) => {
send_error(
stream,
"42601",
&format!("option \"{name}\" not recognized"),
)?;
send_ready_for_query(stream, *tx_state)?;
}
CopyIntent::To(table, opts) => {
let sql = format!("SELECT * FROM {table}");
handle_copy_to_stdout(
stream,
state,
role,
&sql,
&opts,
tx_state,
conn_state.tx_id,
)?;
}
CopyIntent::ToQuery(query, opts) => {
handle_copy_to_stdout(
stream,
state,
role,
&query,
&opts,
tx_state,
conn_state.tx_id,
)?;
}
}
send_ready_for_query(wbuf, *tx_state)?;
stream.write_all(wbuf)?;
wbuf.clear();
return Ok(());
}
if let Some(canned) = canned_response(sql, state) {
send_canned(wbuf, &canned)?;
send_ready_for_query(wbuf, *tx_state)?;
stream.write_all(wbuf)?;
wbuf.clear();
return Ok(());
}
conn_state
.wait_event
.store(1, std::sync::atomic::Ordering::Relaxed);
conn_state
.cancel_flag
.store(false, std::sync::atomic::Ordering::Relaxed);
let cancel = statement_cancel(settings, &conn_state.cancel_flag);
let streaming_select_attempt = {
let trimmed_start = sql.trim_start();
let first = trimmed_start.split_ascii_whitespace().next().unwrap_or("");
if first.eq_ignore_ascii_case("select") {
let b = sql.as_bytes();
let wants_locks = ci_contains(b, b" for update")
|| ci_contains(b, b" for share")
|| ci_contains(b, b" for no key update")
|| ci_contains(b, b" for key share");
if !sql_has_sequence_mutator(b) && !wants_locks {
Some(())
} else {
None
}
} else {
None
}
};
let conn_in_tx = matches!(*tx_state, b'T' | b'E');
if streaming_select_attempt.is_some() && !conn_in_tx {
let engine_lock = state
.engine
.read()
.map_err(|_| std::io::Error::other("engine rwlock poisoned"))?;
let pre_len = wbuf.len();
let wire_style = engine_lock.render_style();
let wire_tz = engine_lock.session_tz();
let mut cols_storage: Vec<ColumnSchema> = Vec::new();
let mut wrote_header = false;
let mut first_row_size: Option<usize> = None;
let wire_arena = bumpalo::Bump::new();
let sql_b = sql.as_bytes();
let cache_eligible = !sql_has_clock_function(sql_b);
let cached_stmt = if cache_eligible {
pgwire_parse_cache_get(sql)
} else {
None
};
let prepared_stmt = if let Some(s) = cached_stmt {
Some(s)
} else if let Ok(s) = engine_lock.prepare_select_streaming(sql) {
let arc = Arc::new(s);
if cache_eligible {
pgwire_parse_cache_put(sql, Arc::clone(&arc));
}
Some(arc)
} else {
None
};
const WBUF_FLUSH_WATERMARK: usize = 1 << 20;
let mut emit = |item: spg_engine::StreamItem<'_>| -> Result<(), spg_engine::EngineError> {
match item {
spg_engine::StreamItem::Header(cols) => {
cols_storage.extend_from_slice(cols);
let r = send_row_description(wbuf, cols);
if r.is_ok() {
wrote_header = true;
}
r.map_err(|e| spg_engine::EngineError::Unsupported(e.to_string()))
}
spg_engine::StreamItem::Row(cells) => {
if first_row_size.is_none() {
let before = wbuf.len();
let r = encode_data_row_cells(
wbuf,
&cols_storage,
cells,
&wire_arena,
&wire_style,
&wire_tz,
);
first_row_size = Some(wbuf.len() - before);
return r.map_err(|e| spg_engine::EngineError::Unsupported(e.to_string()));
}
encode_data_row_cells(
wbuf,
&cols_storage,
cells,
&wire_arena,
&wire_style,
&wire_tz,
)
.map_err(|e| spg_engine::EngineError::Unsupported(e.to_string()))?;
if wbuf.len() >= WBUF_FLUSH_WATERMARK {
stream
.write_all(wbuf)
.map_err(|e| spg_engine::EngineError::Unsupported(e.to_string()))?;
wbuf.clear();
}
Ok(())
}
}
};
let take_scalarsq_streaming = prepared_stmt.as_ref().is_some_and(|s| {
spg_engine::scalarsq_streaming::is_scalarsq_streaming_shape(s.as_ref())
});
let take_materialised_path = !take_scalarsq_streaming
&& prepared_stmt.as_ref().is_some_and(|s| {
spg_engine::expr_tree_has_subquery(s.as_ref())
|| spg_engine::aggregate::uses_aggregate(s.as_ref())
});
let stream_result: Result<usize, spg_engine::EngineError> = if take_scalarsq_streaming {
let s = prepared_stmt
.as_ref()
.expect("guarded by is_some_and above");
let arena = bumpalo::Bump::new();
(|| -> Result<usize, spg_engine::EngineError> {
let mut header_written = false;
let emit_row = |columns: &[spg_storage::ColumnSchema],
values: &[spg_storage::Value<'_>]|
-> Result<(), spg_engine::EngineError> {
if !header_written {
cols_storage.extend_from_slice(columns);
send_row_description(wbuf, columns)
.map_err(|e| spg_engine::EngineError::Unsupported(e.to_string()))?;
header_written = true;
wrote_header = true;
}
let before = wbuf.len();
encode_data_row_from_values(
wbuf,
&cols_storage,
values,
&arena,
&wire_style,
&wire_tz,
)
.map_err(|e| spg_engine::EngineError::Unsupported(e.to_string()))?;
if first_row_size.is_none() {
first_row_size = Some(wbuf.len() - before);
}
if wbuf.len() >= WBUF_FLUSH_WATERMARK {
stream
.write_all(wbuf)
.map_err(|e| spg_engine::EngineError::Unsupported(e.to_string()))?;
wbuf.clear();
}
Ok(())
};
let (columns, n) = engine_lock.execute_readonly_select_with_arena(
s.as_ref(),
cancel,
&arena,
emit_row,
)?;
if !header_written {
cols_storage.extend_from_slice(&columns);
send_row_description(wbuf, &columns)
.map_err(|e| spg_engine::EngineError::Unsupported(e.to_string()))?;
}
wrote_header = true;
Ok(n)
})()
} else if take_materialised_path {
let s = prepared_stmt
.as_ref()
.expect("guarded by is_some_and above");
(|| -> Result<usize, spg_engine::EngineError> {
match engine_lock.execute_readonly_select_prepared(s.as_ref(), cancel)? {
spg_engine::QueryResult::Rows { columns, rows } => {
cols_storage.extend_from_slice(&columns);
send_row_description(wbuf, &columns)
.map_err(|e| spg_engine::EngineError::Unsupported(e.to_string()))?;
wrote_header = true;
for (i, row) in rows.iter().enumerate() {
if i.is_multiple_of(256) {
cancel.check()?;
}
let before = wbuf.len();
encode_data_row(
wbuf,
&cols_storage,
row,
&wire_arena,
&wire_style,
&wire_tz,
)
.map_err(|e| spg_engine::EngineError::Unsupported(e.to_string()))?;
if first_row_size.is_none() {
first_row_size = Some(wbuf.len() - before);
}
if wbuf.len() >= WBUF_FLUSH_WATERMARK {
stream.write_all(wbuf).map_err(|e| {
spg_engine::EngineError::Unsupported(e.to_string())
})?;
wbuf.clear();
}
}
Ok(rows.len())
}
_ => Err(spg_engine::EngineError::Unsupported(
"select returned non-Rows".into(),
)),
}
})()
} else if let Some(s) = prepared_stmt.as_ref() {
engine_lock.execute_readonly_select_streaming_prepared(s.as_ref(), cancel, &mut emit)
} else {
engine_lock.execute_readonly_select_streaming(sql, cancel, &mut emit)
};
drop(prepared_stmt);
drop(engine_lock);
conn_state
.wait_event
.store(0, std::sync::atomic::Ordering::Relaxed);
match stream_result {
Ok(n) => {
drain_notices(state, wbuf)?;
drain_notifications(state, wbuf, conn_state)?;
send_command_complete_select_count(wbuf, n)?;
send_ready_for_query(wbuf, *tx_state)?;
stream.write_all(wbuf)?;
wbuf.clear();
return Ok(());
}
Err(_e) if !wrote_header => {
wbuf.truncate(pre_len);
}
Err(e) => {
wbuf.truncate(pre_len);
let (sqlstate, msg) = engine_error_to_wire_conn(&e, conn_state);
send_error_pos(wbuf, sqlstate, &msg, parse_error_position(&e, sql))?;
send_ready_for_query(wbuf, *tx_state)?;
stream.write_all(wbuf)?;
wbuf.clear();
return Ok(());
}
}
conn_state
.wait_event
.store(1, std::sync::atomic::Ordering::Relaxed);
}
let was_aborted = *tx_state == b'E';
let (result, queue_persisted) = match try_queue_plain_dml(state, sql, role, *tx_state, settings)
{
Some(r) => (r, true),
None => (
execute_with_role(
state,
sql,
role,
cancel,
matches!(*tx_state, b'T' | b'E'),
conn_state.tx_id,
settings,
),
false,
),
};
conn_state
.wait_event
.store(0, std::sync::atomic::Ordering::Relaxed);
drain_notices(state, wbuf)?;
drain_notifications(state, wbuf, conn_state)?;
let result = if queue_persisted {
result
} else {
match persist_wire_write(state, sql, &result, conn_state.tx_id) {
Ok(()) => result,
Err(e) => Err(EngineError::Unsupported(format!(
"durability append failed: {e}"
))),
}
};
match result {
Ok(QueryResult::Rows { columns, rows }) => {
send_row_description(wbuf, &columns)?;
let n = rows.len();
let mat_arena = bumpalo::Bump::new();
let (wire_style, wire_tz) = state
.engine
.read()
.map(|e| (e.render_style(), e.session_tz()))
.unwrap_or((Default::default(), spg_engine::SessionTz::Utc));
if let Some((first, rest)) = rows.split_first() {
let before = wbuf.len();
encode_data_row(wbuf, &columns, first, &mat_arena, &wire_style, &wire_tz)?;
let first_size = wbuf.len() - before;
if rest.len() > 0 {
wbuf.reserve(first_size.saturating_mul(rest.len()).saturating_add(32));
}
for row in rest {
encode_data_row(wbuf, &columns, row, &mat_arena, &wire_style, &wire_tz)?;
}
}
send_command_complete(wbuf, &command_tag_for_rows_sql(sql, n))?;
}
Ok(QueryResult::CommandOk { affected, .. }) => {
let verb = sql.split_ascii_whitespace().next().unwrap_or("");
let tag = if was_aborted
&& (verb.eq_ignore_ascii_case("commit") || verb.eq_ignore_ascii_case("end"))
{
"ROLLBACK".to_string()
} else {
command_tag(sql, affected)
};
send_command_complete(wbuf, &tag)?;
*tx_state = if state
.engine
.read()
.is_ok_and(|e| e.is_tx_open(conn_state.tx_id))
{
b'T'
} else {
b'I'
};
}
Err(e) => {
let (sqlstate, msg) = engine_error_to_wire_conn(&e, conn_state);
send_error_pos(wbuf, sqlstate, &msg, parse_error_position(&e, sql))?;
*tx_state = if state
.engine
.read()
.is_ok_and(|e| e.is_tx_open(conn_state.tx_id))
{
b'E'
} else {
b'I'
};
}
Ok(_) => {
send_error(wbuf, "XX000", "unexpected QueryResult variant")?;
}
}
send_ready_for_query(wbuf, *tx_state)?;
stream.write_all(wbuf)?;
wbuf.clear();
Ok(())
}
#[allow(clippy::too_many_arguments)]
fn handle_pg_simple_query_one_into_wbuf(
sql_bytes: &[u8],
state: &Arc<ServerState>,
conn_state: &Arc<crate::ConnState>,
role: Role,
tx_state: &mut u8,
settings: &mut std::collections::HashMap<String, String>,
wbuf: &mut Vec<u8>,
) -> std::io::Result<()> {
let now_us = wallclock_unix_micros();
conn_state
.last_query_start_us
.store(now_us, std::sync::atomic::Ordering::Relaxed);
if let Ok(mut s) = conn_state.current_sql.write() {
s.clear();
match std::str::from_utf8(sql_bytes) {
Ok(valid) => s.push_str(valid),
Err(_) => s.push_str(&String::from_utf8_lossy(sql_bytes)),
}
}
let Ok(sql_str) = std::str::from_utf8(sql_bytes) else {
send_error(wbuf, "22021", "invalid UTF-8 in query")?;
return Ok(());
};
let sql: &str = sql_str.trim_end_matches(';').trim();
if sql.is_empty() {
send_command_complete(wbuf, "")?;
return Ok(());
}
if let Some((name, value)) = parse_set_statement(sql) {
let name_lc = name.to_ascii_lowercase();
settings.insert(name_lc.clone(), value.clone());
if name_lc == "application_name" {
if let Ok(mut g) = conn_state.application_name.write() {
*g = value.clone();
}
}
}
{
let t = sql.trim();
let b = t.as_bytes();
let restore_app_name = |settings: &mut std::collections::HashMap<String, String>| {
if !conn_state.startup_app_name.is_empty() {
settings.insert(
"application_name".to_string(),
conn_state.startup_app_name.clone(),
);
}
if let Ok(mut g) = conn_state.application_name.write() {
g.clone_from(&conn_state.startup_app_name);
}
};
if ci_eq(b, b"reset all") || ci_starts_with(b, b"discard all") {
settings.clear();
restore_app_name(settings);
} else if ci_starts_with(b, b"reset ") {
let name = t[6..]
.trim()
.trim_end_matches(';')
.trim()
.to_ascii_lowercase();
settings.remove(&name);
if name == "application_name" {
restore_app_name(settings);
}
}
}
if let Some(name) = parse_show_statement(sql) {
let engine_val: Option<String> = if name != "all" {
state
.engine
.read()
.ok()
.and_then(|e| e.session_param(&name).map(str::to_string))
} else {
None
};
let resp = match engine_val {
Some(v) => CannedResponse::Rows {
columns: vec![ColumnSchema::new(name.clone(), DataType::Text, false)],
rows: vec![Row::new(vec![Value::text(v)])],
},
None => render_show(&name, settings),
};
send_canned(wbuf, &resp)?;
return Ok(());
}
if parse_copy_intent(sql).is_some() {
send_error(
wbuf,
"0A000",
"COPY is not supported within a multi-statement simple-query script; \
send COPY as its own Query message",
)?;
return Ok(());
}
if let Some(canned) = canned_response(sql, state) {
send_canned(wbuf, &canned)?;
return Ok(());
}
conn_state
.cancel_flag
.store(false, std::sync::atomic::Ordering::Relaxed);
let cancel = statement_cancel(settings, &conn_state.cancel_flag);
conn_state
.wait_event
.store(1, std::sync::atomic::Ordering::Relaxed);
let (result, queue_persisted) = match try_queue_plain_dml(state, sql, role, *tx_state, settings)
{
Some(r) => (r, true),
None => (
execute_with_role(
state,
sql,
role,
cancel,
matches!(*tx_state, b'T' | b'E'),
conn_state.tx_id,
settings,
),
false,
),
};
conn_state
.wait_event
.store(0, std::sync::atomic::Ordering::Relaxed);
drain_notices(state, wbuf)?;
drain_notifications(state, wbuf, conn_state)?;
let result = if queue_persisted {
result
} else {
match persist_wire_write(state, sql, &result, conn_state.tx_id) {
Ok(()) => result,
Err(e) => Err(EngineError::Unsupported(format!(
"durability append failed: {e}"
))),
}
};
match result {
Ok(QueryResult::Rows { columns, rows }) => {
send_row_description(wbuf, &columns)?;
let mat_arena = bumpalo::Bump::new();
let (wire_style, wire_tz) = state
.engine
.read()
.map(|e| (e.render_style(), e.session_tz()))
.unwrap_or((Default::default(), spg_engine::SessionTz::Utc));
for row in &rows {
encode_data_row(wbuf, &columns, row, &mat_arena, &wire_style, &wire_tz)?;
}
send_command_complete(wbuf, &command_tag_for_rows_sql(sql, rows.len()))?;
}
Ok(QueryResult::CommandOk { affected, .. }) => {
let tag = command_tag(sql, affected);
send_command_complete(wbuf, &tag)?;
*tx_state = if state
.engine
.read()
.is_ok_and(|e| e.is_tx_open(conn_state.tx_id))
{
b'T'
} else {
b'I'
};
}
Err(e) => {
let (sqlstate, msg) = engine_error_to_wire_conn(&e, conn_state);
send_error_pos(wbuf, sqlstate, &msg, parse_error_position(&e, sql))?;
*tx_state = if state
.engine
.read()
.is_ok_and(|e| e.is_tx_open(conn_state.tx_id))
{
b'E'
} else {
b'I'
};
}
Ok(_) => {
send_error(wbuf, "XX000", "unexpected QueryResult variant")?;
}
}
Ok(())
}
#[allow(clippy::too_many_arguments)]
fn dispatch_pg_simple_query_multi(
stream: &mut dyn ReadWrite,
stmts: &[&[u8]],
state: &Arc<ServerState>,
conn_state: &Arc<crate::ConnState>,
role: Role,
tx_state: &mut u8,
settings: &mut std::collections::HashMap<String, String>,
wbuf: &mut Vec<u8>,
) -> std::io::Result<()> {
let script = stmts.len() > 1;
let mut implicit_tx = false;
for (i, stmt) in stmts.iter().enumerate() {
if script && !implicit_tx && *tx_state == b'I' && i + 1 < stmts.len() {
let mut discard = Vec::new();
handle_pg_simple_query_one_into_wbuf(
b"BEGIN",
state,
conn_state,
role,
tx_state,
settings,
&mut discard,
)?;
implicit_tx = true;
}
let pre_len = wbuf.len();
handle_pg_simple_query_one_into_wbuf(
stmt, state, conn_state, role, tx_state, settings, wbuf,
)?;
if implicit_tx && *tx_state == b'I' {
implicit_tx = false;
}
if *tx_state == b'E' {
break;
}
if let Some(&first_byte_of_last_msg) = wbuf.get(pre_len) {
if first_byte_of_last_msg == b'E' {
break;
}
}
}
if implicit_tx && matches!(*tx_state, b'T' | b'E') {
let closing: &[u8] = if *tx_state == b'E' {
b"ROLLBACK"
} else {
b"COMMIT"
};
let mut discard = Vec::new();
handle_pg_simple_query_one_into_wbuf(
closing,
state,
conn_state,
role,
tx_state,
settings,
&mut discard,
)?;
}
send_ready_for_query(wbuf, *tx_state)?;
stream.write_all(wbuf)?;
wbuf.clear();
Ok(())
}
fn handle_conn(mut stream: TcpStream, state: &Arc<ServerState>) -> std::io::Result<()> {
let _ = stream.set_nodelay(true);
let sock = stream.try_clone().ok();
loop {
match peek_startup_proto(&stream)? {
Some(80877103) => {
let mut hdr = [0u8; 8];
stream.read_exact(&mut hdr)?;
stream.write_all(b"S")?;
let mut tls_conn =
crate::mysqlwire::build_server_connection().map_err(std::io::Error::other)?;
let mut tls = rustls::Stream::new(&mut tls_conn, &mut stream);
return run_pg_session(&mut tls, state, true, sock);
}
Some(80877104) => {
let mut hdr = [0u8; 8];
stream.read_exact(&mut hdr)?;
stream.write_all(b"N")?;
}
Some(80877102) => {
let mut pkt = [0u8; 16];
stream.read_exact(&mut pkt)?;
let pid = u32::from_be_bytes([pkt[8], pkt[9], pkt[10], pkt[11]]);
let secret = u32::from_be_bytes([pkt[12], pkt[13], pkt[14], pkt[15]]);
if let Ok(conns) = state.connections.read()
&& let Some(c) = conns.iter().find(|c| c.pid == pid)
&& c.cancel_secret == secret
{
c.cancel_flag
.store(true, std::sync::atomic::Ordering::Relaxed);
}
return Ok(());
}
_ => return run_pg_session(&mut stream, state, false, sock),
}
}
}
fn require_tls() -> bool {
static REQUIRE: std::sync::OnceLock<bool> = std::sync::OnceLock::new();
*REQUIRE.get_or_init(|| {
std::env::var("SPG_REQUIRE_TLS")
.map(|v| {
matches!(
v.trim().to_ascii_lowercase().as_str(),
"1" | "true" | "yes" | "on"
)
})
.unwrap_or(false)
})
}
fn peek_startup_proto(stream: &TcpStream) -> std::io::Result<Option<u32>> {
let mut buf = [0u8; 8];
let n = stream.peek(&mut buf)?;
if n < 8 {
return Ok(None);
}
Ok(Some(u32::from_be_bytes([buf[4], buf[5], buf[6], buf[7]])))
}
fn run_pg_session(
stream: &mut dyn ReadWrite,
state: &Arc<ServerState>,
secure: bool,
sock: Option<TcpStream>,
) -> std::io::Result<()> {
let (user, params) = read_startup(stream)?;
if !secure && require_tls() {
send_error(
stream,
"08P01",
"SSL/TLS connection required (SPG_REQUIRE_TLS is set)",
)?;
return Ok(());
}
let startup_app_name = params
.iter()
.find_map(|(k, v)| (k == "application_name").then(|| v.clone()))
.unwrap_or_default();
let startup_db = params
.iter()
.find_map(|(k, v)| (k == "database").then(|| v.clone()))
.filter(|db| {
!db.is_empty()
&& db
.chars()
.all(|c| c.is_ascii_alphanumeric() || c == '_' || c == '-')
})
.unwrap_or_default();
let conn_tx_id = match state.engine.write() {
Ok(mut e) => e.alloc_tx_id(),
Err(_) => spg_engine::IMPLICIT_TX,
};
let conn_state = Arc::new(crate::ConnState {
tx_id: conn_tx_id,
pid: crate::alloc_conn_id(),
user: user.clone(),
started_at_us: std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_micros() as i64)
.unwrap_or(0),
current_sql: std::sync::RwLock::new(String::new()),
wait_event: std::sync::atomic::AtomicU8::new(0),
last_query_start_us: std::sync::atomic::AtomicI64::new(0),
in_transaction: std::sync::atomic::AtomicBool::new(false),
application_name: std::sync::RwLock::new(startup_app_name.clone()),
startup_app_name: startup_app_name.clone(),
cancel_secret: new_cancel_secret(),
cancel_flag: std::sync::atomic::AtomicBool::new(false),
terminate: std::sync::atomic::AtomicBool::new(false),
client_addr: sock.as_ref().and_then(|s| s.peer_addr().ok()),
database: std::sync::RwLock::new(startup_db.clone()),
sock,
notify_queue: std::sync::Mutex::new(Vec::new()),
});
if let Ok(mut e) = state.engine.write() {
e.set_current_session(conn_state.pid);
if !user.is_empty() {
e.set_session_user(&user);
}
if !startup_db.is_empty() {
let _ = e.execute(&format!("SET spg.database = '{startup_db}'"));
}
e.apply_db_role_settings(&startup_db, &user);
}
crate::set_conn_pid(conn_state.pid);
if let Ok(mut conns) = state.connections.write() {
conns.push(Arc::clone(&conn_state));
}
crate::backend_count_incr();
struct ConnGuard {
state: Arc<ServerState>,
conn: Arc<crate::ConnState>,
}
impl Drop for ConnGuard {
fn drop(&mut self) {
if let Ok(mut conns) = self.state.connections.write() {
conns.retain(|x| !Arc::ptr_eq(x, &self.conn));
}
if let Ok(mut e) = self.state.engine.write() {
if e.is_tx_open(self.conn.tx_id) {
let _ = e.execute_in("ROLLBACK", self.conn.tx_id);
}
e.end_session(self.conn.pid);
}
crate::backend_count_decr();
}
}
let _conn_guard = ConnGuard {
state: Arc::clone(state),
conn: Arc::clone(&conn_state),
};
let has_users = state
.engine
.read()
.is_ok_and(|e| e.users().iter().any(|(_, r)| r.has_credentials()));
let role = if has_users {
let user_has_scram = state
.engine
.read()
.ok()
.and_then(|e| {
e.users()
.iter()
.find_map(|(n, r)| (n == user).then(|| r.scram().is_some()))
})
.unwrap_or(false);
let outcome = if user_has_scram {
scram_auth(stream, state, &user, secure)?
} else {
cleartext_auth(stream, state, &user)?
};
match outcome {
Some(r) => r,
None => return Ok(()), }
} else {
Role::Admin
};
if has_users && !user.is_empty() {
if let Ok(mut e) = state.engine.write() {
e.set_current_session(conn_state.pid);
e.set_session_authenticated();
}
}
send_msg(stream, b'R', &0u32.to_be_bytes())?;
send_parameter_status(stream, "server_version", "18.4 (spg-4.3)")?;
send_parameter_status(stream, "client_encoding", "UTF8")?;
send_parameter_status(stream, "DateStyle", "ISO, MDY")?;
send_parameter_status(stream, "integer_datetimes", "on")?;
send_parameter_status(stream, "standard_conforming_strings", "on")?;
let mut bkd = Vec::with_capacity(8);
bkd.extend_from_slice(&conn_state.pid.to_be_bytes());
bkd.extend_from_slice(&conn_state.cancel_secret.to_be_bytes());
send_msg(stream, b'K', &bkd)?;
send_ready_for_query(stream, b'I')?;
let mut tx_state = b'I'; let mut prepared: std::collections::HashMap<String, PreparedStmt> =
std::collections::HashMap::default();
let mut portals: std::collections::HashMap<String, Portal> =
std::collections::HashMap::default();
let mut settings: std::collections::HashMap<String, String> =
std::collections::HashMap::default();
if !startup_app_name.is_empty() {
settings.insert("application_name".to_string(), startup_app_name.clone());
}
const PIPELINE_FLUSH_BYTES: usize = 4096;
let mut wbuf: Vec<u8> = Vec::with_capacity(8192);
let mut rbuf: Vec<u8> = Vec::with_capacity(8192);
let mut peek_buf = [0u8; 256];
let mut peek_have: usize = 0;
loop {
if terminated(&conn_state) {
let _ = stream.write_all(&wbuf);
wbuf.clear();
return send_fatal_terminated(stream);
}
while peek_have < 5 {
let n = match stream.read(&mut peek_buf[peek_have..]) {
Ok(n) => n,
Err(e) => {
if e.kind() == std::io::ErrorKind::UnexpectedEof {
return if terminated(&conn_state) {
send_fatal_terminated(stream)
} else {
Ok(())
};
}
return Err(e);
}
};
if n == 0 {
return if terminated(&conn_state) {
send_fatal_terminated(stream)
} else {
Ok(())
};
}
peek_have += n;
}
let msg_type = peek_buf[0];
let len = u32::from_be_bytes([peek_buf[1], peek_buf[2], peek_buf[3], peek_buf[4]]) as usize;
let body_len = len.saturating_sub(4);
let in_peek = peek_have - 5;
if body_len <= in_peek {
rbuf.resize(body_len, 0);
rbuf[..body_len].copy_from_slice(&peek_buf[5..5 + body_len]);
let leftover = peek_have - 5 - body_len;
if leftover > 0 {
peek_buf.copy_within(5 + body_len..peek_have, 0);
}
peek_have = leftover;
} else {
rbuf.resize(body_len, 0);
rbuf[..in_peek].copy_from_slice(&peek_buf[5..peek_have]);
stream.read_exact(&mut rbuf[in_peek..])?;
peek_have = 0;
}
let body: &[u8] = &rbuf;
trace_frontend_message(msg_type, body, tx_state);
let timing_start = timing_enabled().then(std::time::Instant::now);
match msg_type {
b'Q' => handle_pg_simple_query(
stream,
body,
state,
&conn_state,
role,
&mut tx_state,
&mut settings,
&mut wbuf,
)?,
b'X' => {
if !wbuf.is_empty() {
let _ = stream.write_all(&wbuf);
}
return Ok(());
}
b'P' => {
if let Err(msg) = handle_parse(body, &mut prepared, state) {
send_error(&mut wbuf, "42601", &msg)?;
} else {
send_msg(&mut wbuf, b'1', &[])?;
}
}
b'B' => {
match handle_bind(body, &prepared) {
Ok(portal) => {
portals.insert(portal.0.clone(), portal.1);
send_msg(&mut wbuf, b'2', &[])?; }
Err(msg) => send_error(&mut wbuf, "42601", &msg)?,
}
}
b'D' => {
if !body.is_empty() {
let kind = body[0];
let name = cstring_at(body, 1).unwrap_or_default();
let (param_oids, columns): (Vec<u32>, Vec<ColumnSchema>) = if kind == b'S' {
if let Some(stmt) = prepared.get(&name) {
let eng = state
.engine
.read()
.map_err(|_| std::io::Error::other("engine lock poisoned"))?;
eng.describe_prepared(&stmt.ast)
} else {
(Vec::new(), Vec::new())
}
} else if kind == b'P' {
let cols = if let Some(portal) = portals.get(&name) {
if let Some(stmt) = prepared.get(&portal.stmt_name) {
let eng = state
.engine
.read()
.map_err(|_| std::io::Error::other("engine lock poisoned"))?;
let (_, c) = eng.describe_prepared(&stmt.ast);
c
} else {
Vec::new()
}
} else {
Vec::new()
};
(Vec::new(), cols)
} else {
(Vec::new(), Vec::new())
};
if kind == b'S' {
let n = u16::try_from(param_oids.len())
.map_err(|_| std::io::Error::other("too many parameters"))?;
let mut pd = Vec::with_capacity(2 + param_oids.len() * 4);
pd.extend_from_slice(&n.to_be_bytes());
for oid in ¶m_oids {
pd.extend_from_slice(&oid.to_be_bytes());
}
send_msg(&mut wbuf, b't', &pd)?;
}
if columns.is_empty() {
send_msg(&mut wbuf, b'n', &[])?; } else {
send_row_description(&mut wbuf, &columns)?;
}
}
}
b'E' => {
if let Err((sqlstate, msg)) = handle_execute(
body,
&mut portals,
&prepared,
&settings,
&mut wbuf,
state,
role,
&mut tx_state,
&conn_state,
) {
send_error(&mut wbuf, sqlstate, &msg)?;
}
}
b'C' => {
if body.len() >= 2 {
let kind = body[0];
let name = cstring_at(body, 1).unwrap_or_default();
if kind == b'S' {
prepared.remove(&name);
} else if kind == b'P' {
portals.remove(&name);
}
}
send_msg(&mut wbuf, b'3', &[])?; }
b'H' => {
if !wbuf.is_empty() {
stream.write_all(&wbuf)?;
wbuf.clear();
}
}
b'S' => {
send_ready_for_query(&mut wbuf, tx_state)?;
stream.write_all(&wbuf)?;
wbuf.clear();
}
b'd' | b'c' | b'f' => {
send_error(
&mut wbuf,
"08P01",
"unexpected CopyData/Done/Fail outside COPY mode",
)?;
send_ready_for_query(&mut wbuf, tx_state)?;
stream.write_all(&wbuf)?;
wbuf.clear();
}
_ => {
send_error(
&mut wbuf,
"08P01",
&format!("unknown frontend message type: 0x{msg_type:02x}"),
)?;
send_ready_for_query(&mut wbuf, tx_state)?;
stream.write_all(&wbuf)?;
wbuf.clear();
}
}
timing_record(timing_start);
if wbuf.len() >= PIPELINE_FLUSH_BYTES {
stream.write_all(&wbuf)?;
wbuf.clear();
}
}
}
fn try_queue_plain_dml(
state: &Arc<ServerState>,
sql: &str,
role: Role,
tx_state: u8,
settings: &std::collections::HashMap<String, String>,
) -> Option<Result<QueryResult, EngineError>> {
if tx_state != b'I' || state.wal.is_none() {
return None;
}
let verb = sql
.trim_start()
.split_ascii_whitespace()
.next()
.unwrap_or("")
.to_ascii_lowercase();
if !matches!(verb.as_str(), "insert" | "update" | "delete" | "merge") {
return None;
}
let b = sql.as_bytes();
if ci_contains(b, b"returning") {
return None;
}
if sql.trim_end().trim_end_matches(';').contains(';') {
return None;
}
let timeout_set = settings
.get("statement_timeout")
.map(String::as_str)
.and_then(parse_timeout_ms)
.unwrap_or(0)
> 0;
if timeout_set {
return None;
}
if !role.can_write() {
return Some(Err(EngineError::Unsupported(
"permission denied: write requires admin or readwrite role".into(),
)));
}
let queue_flag = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
let (result, wal_outcome) = crate::commit_queue_execute(state, sql.to_string(), &queue_flag);
if let Err(e) = wal_outcome {
return Some(Err(EngineError::Unsupported(format!(
"durability append failed: {e}"
))));
}
if matches!(&result, Ok(QueryResult::CommandOk { .. }))
&& state.audit_path.is_some()
&& let Err(e) = crate::append_audit_pub(state, sql)
{
return Some(Err(EngineError::Unsupported(format!(
"audit append failed: {e}"
))));
}
Some(result)
}
fn execute_with_role(
state: &Arc<ServerState>,
sql: &str,
role: Role,
cancel: CancelToken<'_>,
conn_in_tx: bool,
tx_id: spg_engine::TxId,
settings: &std::collections::HashMap<String, String>,
) -> Result<QueryResult, EngineError> {
crate::try_lazy_preload_cold(state);
let lower_first = sql
.trim_start()
.split_ascii_whitespace()
.next()
.unwrap_or("")
.to_ascii_lowercase();
let mut is_read = matches!(lower_first.as_str(), "select" | "show");
if is_read {
let b = sql.as_bytes();
if spg_engine::MUTATING_CALL_NEEDLES
.iter()
.any(|n| ci_contains(b, n))
|| ci_contains(b, b" for update")
|| ci_contains(b, b" for share")
|| ci_contains(b, b" for no key update")
|| ci_contains(b, b" for key share")
{
is_read = false;
}
}
if !is_read && !role.can_write() {
return Err(EngineError::Unsupported(
"permission denied: write requires admin or readwrite role".into(),
));
}
let is_user_mgmt = (lower_first == "create" || lower_first == "drop")
&& sql
.split_ascii_whitespace()
.nth(1)
.is_some_and(|w| w.eq_ignore_ascii_case("user"));
if is_user_mgmt && !role.can_manage_users() {
return Err(EngineError::Unsupported(
"permission denied: user management requires admin role".into(),
));
}
if is_read && !conn_in_tx {
let engine = state
.engine
.read()
.map_err(|_| EngineError::Unsupported("engine rwlock poisoned".into()))?;
engine.execute_readonly_with_cancel(sql, cancel)
} else {
let deadline = lock_wait_deadline(settings);
loop {
let attempt = {
let mut engine = state
.engine
.write()
.map_err(|_| EngineError::Unsupported("engine rwlock poisoned".into()))?;
let r = engine.execute_in_with_cancel(sql, tx_id, cancel);
if std::env::var("SPG_MATVIEW_TRACE").is_ok() {
use core::sync::atomic::Ordering;
eprintln!(
"spg-server: matview-trace sql={:?} fanout={} applied={} bailed={}",
&sql[..sql.len().min(60)],
spg_engine::MATVIEW_FANOUT_BUFFERED.load(Ordering::Relaxed),
spg_engine::MATVIEW_DELTA_APPLIED.load(Ordering::Relaxed),
spg_engine::MATVIEW_DELTA_BAILED.load(Ordering::Relaxed),
);
}
r
}; match attempt {
Err(EngineError::LockWouldBlock) => {
if let Some(d) = deadline
&& std::time::Instant::now() >= d
{
return Err(EngineError::Unsupported(
"canceling statement due to lock timeout".into(),
));
}
std::thread::sleep(std::time::Duration::from_millis(5));
}
other => return other,
}
}
}
}
fn lock_wait_deadline(
settings: &std::collections::HashMap<String, String>,
) -> Option<std::time::Instant> {
let raw = settings.get("lock_timeout")?;
let ms = parse_timeout_ms(raw)?;
(ms > 0).then(|| std::time::Instant::now() + std::time::Duration::from_millis(ms))
}
fn timing_enabled() -> bool {
static ON: std::sync::OnceLock<bool> = std::sync::OnceLock::new();
*ON.get_or_init(|| std::env::var("SPG_PGWIRE_TIMING").is_ok_and(|v| v != "0"))
}
fn timing_record(start: Option<std::time::Instant>) {
let Some(t0) = start else { return };
static N: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
static TOT_NS: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
let ns = u64::try_from(t0.elapsed().as_nanos()).unwrap_or(u64::MAX);
let tot = TOT_NS.fetch_add(ns, std::sync::atomic::Ordering::Relaxed) + ns;
let n = N.fetch_add(1, std::sync::atomic::Ordering::Relaxed) + 1;
if n.is_multiple_of(1000) {
eprintln!(
"[pgwire-timing] n={n} mean_handle_us={:.1}",
tot as f64 / n as f64 / 1000.0
);
}
}
fn trace_frontend_message(msg_type: u8, body: &[u8], tx_state: u8) {
static ON: std::sync::OnceLock<bool> = std::sync::OnceLock::new();
if !*ON.get_or_init(|| std::env::var("SPG_PGWIRE_TRACE").is_ok_and(|v| v != "0")) {
return;
}
let field = |n: usize| -> String {
body.split(|b| *b == 0)
.nth(n)
.unwrap_or(&[])
.iter()
.copied()
.filter(|b| b.is_ascii_graphic() || *b == b' ')
.map(char::from)
.take(70)
.collect()
};
let head = if msg_type == b'P' {
format!("{} | {}", field(0), field(1))
} else {
field(0)
};
eprintln!(
"[pgwire-trace] tx={} {} {head}",
tx_state as char, msg_type as char
);
}
pub(crate) fn persist_wire_write(
state: &Arc<ServerState>,
sql: &str,
result: &Result<QueryResult, EngineError>,
tx_id: spg_engine::TxId,
) -> std::io::Result<()> {
let modified_catalog = match result {
Ok(QueryResult::CommandOk {
modified_catalog, ..
}) => *modified_catalog,
Ok(QueryResult::Rows { .. }) if crate::sql_is_dmlish(sql) => true,
_ => return Ok(()), };
let modified_catalog = &modified_catalog;
if state.wal.is_some() {
let in_tx = state.engine.read().is_ok_and(|e| e.is_tx_open(tx_id));
crate::append_wal(state, sql, crate::session_sync_commit(state) && !in_tx)?;
} else if *modified_catalog && state.db_path.is_some() {
let bytes = state
.engine
.read()
.map_err(|_| std::io::Error::other("engine rwlock poisoned"))?
.snapshot();
if let Some(path) = state.db_path.as_deref() {
crate::write_atomic(path, &bytes)?;
}
}
if *modified_catalog && state.audit_path.is_some() {
crate::append_audit_pub(state, sql)?;
}
Ok(())
}
fn command_tag(sql: &str, affected: usize) -> String {
let first = sql
.trim_start()
.split_ascii_whitespace()
.next()
.unwrap_or("")
.to_ascii_uppercase();
match first.as_str() {
"INSERT" => format!("INSERT 0 {affected}"),
"UPDATE" => format!("UPDATE {affected}"),
"DELETE" => format!("DELETE {affected}"),
"BEGIN" => "BEGIN".to_string(),
"DISCARD" => {
let target = sql
.trim_start()
.split_ascii_whitespace()
.nth(1)
.unwrap_or("")
.trim_end_matches(';')
.to_ascii_uppercase();
if target.is_empty() {
"DISCARD".to_string()
} else {
format!("DISCARD {target}")
}
}
"COMMIT" => "COMMIT".to_string(),
"ROLLBACK" => "ROLLBACK".to_string(),
"DECLARE" => "DECLARE CURSOR".to_string(),
"MOVE" => format!("MOVE {affected}"),
"CLOSE" => {
let second = sql
.trim_start()
.split_ascii_whitespace()
.nth(1)
.unwrap_or("");
if second.eq_ignore_ascii_case("all") {
"CLOSE CURSOR ALL".to_string()
} else {
"CLOSE CURSOR".to_string()
}
}
"WITH" | "MERGE" => spg_sql::parser::parse_statement(sql)
.map(|stmt| command_tag_for_ast(&stmt, affected))
.unwrap_or(first),
first @ ("CREATE" | "ALTER" | "DROP") => {
let object = sql.trim_start().split_ascii_whitespace().skip(1).find(|w| {
!w.eq_ignore_ascii_case("unique")
&& !w.eq_ignore_ascii_case("or")
&& !w.eq_ignore_ascii_case("replace")
&& !w.eq_ignore_ascii_case("temp")
&& !w.eq_ignore_ascii_case("temporary")
&& !w.eq_ignore_ascii_case("unlogged")
&& !w.eq_ignore_ascii_case("global")
&& !w.eq_ignore_ascii_case("local")
&& !w.eq_ignore_ascii_case("recursive")
&& !w.eq_ignore_ascii_case("concurrently")
&& !w.eq_ignore_ascii_case("constraint")
});
const OBJECTS: &[&str] = &[
"TABLE",
"INDEX",
"VIEW",
"SEQUENCE",
"SCHEMA",
"TYPE",
"EXTENSION",
"DOMAIN",
"TRIGGER",
"FUNCTION",
"DATABASE",
"PUBLICATION",
"SUBSCRIPTION",
"POLICY",
"ROLE",
];
match object {
Some(w) if w.eq_ignore_ascii_case("user") => format!("{first} ROLE"),
Some(w) if w.eq_ignore_ascii_case("materialized") && first != "CREATE" => {
format!("{first} MATERIALIZED VIEW")
}
Some(w) => match OBJECTS.iter().find(|o| w.eq_ignore_ascii_case(o)) {
Some(o) => format!("{first} {o}"),
None => first.to_string(),
},
None => first.to_string(),
}
}
"TRUNCATE" => "TRUNCATE TABLE".to_string(),
"REFRESH" => "REFRESH MATERIALIZED VIEW".to_string(),
other => other.to_string(),
}
}
fn ci_starts_with(b: &[u8], prefix: &[u8]) -> bool {
b.len() >= prefix.len() && b[..prefix.len()].eq_ignore_ascii_case(prefix)
}
fn ci_eq(b: &[u8], target: &[u8]) -> bool {
b.eq_ignore_ascii_case(target)
}
fn ci_strip_prefix<'a>(b: &'a [u8], prefix: &[u8]) -> Option<&'a [u8]> {
if ci_starts_with(b, prefix) {
Some(&b[prefix.len()..])
} else {
None
}
}
fn trim_ascii(b: &[u8]) -> &[u8] {
let mut start = 0;
while start < b.len() && b[start].is_ascii_whitespace() {
start += 1;
}
let mut end = b.len();
while end > start && b[end - 1].is_ascii_whitespace() {
end -= 1;
}
&b[start..end]
}
fn canned_response(sql: &str, state: &Arc<ServerState>) -> Option<CannedResponse> {
let trimmed = sql.trim();
let b = trimmed.as_bytes();
if let Some(rest) = ci_strip_prefix(b, b"select ") {
let rest_trim = trim_ascii(rest);
if !rest_trim.is_empty()
&& rest_trim
.iter()
.all(|c| c.is_ascii_digit() || *c == b'-' || *c == b'+')
&& let Ok(s) = core::str::from_utf8(rest_trim)
&& let Ok(n) = s.parse::<i64>()
{
return Some(match i32::try_from(n) {
Ok(small) => CannedResponse::Rows {
columns: vec![ColumnSchema::new("?column?", DataType::Int, false)],
rows: vec![Row::new(vec![Value::Int(small)])],
},
Err(_) => CannedResponse::Rows {
columns: vec![ColumnSchema::new("?column?", DataType::BigInt, false)],
rows: vec![Row::new(vec![Value::BigInt(n)])],
},
});
}
if ci_eq(rest_trim, b"null") {
return Some(CannedResponse::Rows {
columns: vec![ColumnSchema::new("?column?", DataType::Text, true)],
rows: vec![Row::new(vec![Value::Null])],
});
}
}
None
}
fn ci_contains(b: &[u8], needle: &[u8]) -> bool {
if needle.is_empty() {
return true;
}
if b.len() < needle.len() {
return false;
}
let n = needle.len();
for i in 0..=b.len() - n {
if b[i..i + n].eq_ignore_ascii_case(needle) {
return true;
}
}
false
}
fn sql_has_sequence_mutator(b: &[u8]) -> bool {
static ANCHORS: std::sync::OnceLock<[bool; 256]> = std::sync::OnceLock::new();
let anchors = ANCHORS.get_or_init(|| {
let mut t = [false; 256];
for n in spg_engine::MUTATING_CALL_NEEDLES {
if let Some(&first) = n.first() {
t[(first | 0x20) as usize] = true;
t[(first & !0x20) as usize] = true;
}
}
t
});
let shortest = spg_engine::MUTATING_CALL_NEEDLES
.iter()
.map(|n| n.len())
.min()
.unwrap_or(usize::MAX);
if b.len() < shortest {
return false;
}
for i in 0..b.len() {
if !anchors[b[i] as usize] {
continue;
}
for needle in spg_engine::MUTATING_CALL_NEEDLES {
let n = needle.len();
if i + n <= b.len() && b[i..i + n].eq_ignore_ascii_case(needle) {
return true;
}
}
}
false
}
fn sql_has_clock_function(b: &[u8]) -> bool {
if b.len() < 4 {
return false;
}
let needles: &[&[u8]] = &[
b"current_timestamp",
b"current_time",
b"current_date",
b"clock_timestamp",
b"transaction_timestamp",
b"now(",
];
for i in 0..b.len() {
let c = b[i] | 0x20; match c {
b'c' | b't' | b'n' => {
for needle in needles {
let n = needle.len();
if i + n <= b.len() && b[i..i + n].eq_ignore_ascii_case(needle) {
return true;
}
}
}
_ => {}
}
}
false
}
enum CannedResponse {
Rows {
columns: Vec<ColumnSchema>,
rows: Vec<Row<'static>>,
},
Tag(&'static str),
}
impl CannedResponse {
fn single_text(col: &'static str, val: &'static str) -> Self {
Self::Rows {
columns: vec![ColumnSchema::new(col, DataType::Text, false)],
rows: vec![Row::new(vec![Value::text(val)])],
}
}
}
fn send_canned(stream: &mut dyn Write, c: &CannedResponse) -> std::io::Result<()> {
match c {
CannedResponse::Rows { columns, rows } => {
send_row_description(stream, columns)?;
for row in rows {
send_data_row(stream, columns, row)?;
}
send_command_complete(stream, &format!("SELECT {}", rows.len()))?;
}
CannedResponse::Tag(tag) => {
send_command_complete(stream, tag)?;
}
}
Ok(())
}
#[derive(Debug, Clone)]
struct PreparedStmt {
ast: spg_sql::ast::Statement,
placeholder_count: u16,
param_type_oids: Vec<u32>,
row_desc_body: Option<Vec<u8>>,
sql: String,
}
#[derive(Debug, Clone)]
struct Portal {
stmt_name: String,
params: Vec<spg_storage::Value<'static>>,
suspended: Option<SuspendedRows>,
result_formats: Vec<i16>,
}
fn col_is_binary(formats: &[i16], i: usize) -> bool {
match formats.len() {
0 => false,
1 => formats[0] == 1,
_ => formats.get(i).copied().unwrap_or(0) == 1,
}
}
#[derive(Debug, Clone)]
struct SuspendedRows {
columns: Vec<ColumnSchema>,
rows: Vec<Row<'static>>,
cursor: usize,
formats: Vec<i16>,
tag: String,
}
fn cstring_at(body: &[u8], pos: usize) -> Option<String> {
let null_off = body[pos..].iter().position(|&b| b == 0)?;
let bytes = &body[pos..pos + null_off];
std::str::from_utf8(bytes).ok().map(str::to_string)
}
fn read_cstring<'a>(body: &'a [u8], cursor: &mut usize) -> Option<&'a str> {
let null_off = body[*cursor..].iter().position(|&b| b == 0)?;
let bytes = &body[*cursor..*cursor + null_off];
*cursor += null_off + 1;
std::str::from_utf8(bytes).ok()
}
fn handle_parse(
body: &[u8],
prepared: &mut std::collections::HashMap<String, PreparedStmt>,
state: &Arc<ServerState>,
) -> Result<(), String> {
let mut cur = 0;
let name = read_cstring(body, &mut cur)
.ok_or("Parse: name not null-terminated UTF-8")?
.to_string();
let sql = read_cstring(body, &mut cur)
.ok_or("Parse: SQL not null-terminated UTF-8")?
.trim_end_matches(';')
.trim()
.to_string();
if cur + 2 > body.len() {
return Err("Parse: missing parameter type count".into());
}
let oid_count = u16::from_be_bytes([body[cur], body[cur + 1]]) as usize;
cur += 2;
if cur + oid_count * 4 > body.len() {
return Err("Parse: truncated parameter OIDs".into());
}
let mut param_type_oids: Vec<u32> = Vec::with_capacity(oid_count);
for _ in 0..oid_count {
let oid = u32::from_be_bytes([body[cur], body[cur + 1], body[cur + 2], body[cur + 3]]);
param_type_oids.push(oid);
cur += 4;
}
let _ = cur; let mut eng = state
.engine
.write()
.map_err(|_| "Parse: engine lock poisoned".to_string())?;
let ast = eng
.prepare_cached(&sql)
.map_err(|e| format!("Parse: {e}"))?;
let (inferred_oids, columns) = eng.describe_prepared(&ast);
drop(eng);
let row_desc_body: Option<Vec<u8>> = if columns.is_empty() {
None
} else {
Some(encode_row_description_body(&columns))
};
let placeholder_count = count_placeholders(&sql);
let mut param_type_oids = param_type_oids;
if param_type_oids.len() < inferred_oids.len() {
param_type_oids.resize(inferred_oids.len(), 0);
}
for (slot, inferred) in param_type_oids.iter_mut().zip(inferred_oids.iter()) {
if *slot == 0 {
*slot = *inferred;
}
}
prepared.insert(
name,
PreparedStmt {
ast,
placeholder_count,
param_type_oids,
row_desc_body,
sql,
},
);
Ok(())
}
fn count_placeholders(sql: &str) -> u16 {
let bytes = sql.as_bytes();
let mut max: u32 = 0;
let mut i = 0;
while i + 1 < bytes.len() {
if bytes[i] == b'$' && bytes[i + 1].is_ascii_digit() {
let mut j = i + 1;
let mut n: u32 = 0;
while j < bytes.len() && bytes[j].is_ascii_digit() {
n = n * 10 + u32::from(bytes[j] - b'0');
j += 1;
}
if n > max {
max = n;
}
i = j;
} else {
i += 1;
}
}
u16::try_from(max).unwrap_or(u16::MAX)
}
fn handle_bind(
body: &[u8],
prepared: &std::collections::HashMap<String, PreparedStmt>,
) -> Result<(String, Portal), String> {
let mut cur = 0;
let portal_name = read_cstring(body, &mut cur)
.ok_or("Bind: portal name not UTF-8")?
.to_string();
let stmt_name = read_cstring(body, &mut cur)
.ok_or("Bind: statement name not UTF-8")?
.to_string();
let stmt = prepared
.get(&stmt_name)
.ok_or_else(|| format!("Bind: prepared statement {stmt_name:?} not found"))?;
if cur + 2 > body.len() {
return Err("Bind: truncated format-code count".into());
}
let fmt_count = u16::from_be_bytes([body[cur], body[cur + 1]]) as usize;
cur += 2;
if cur + fmt_count * 2 > body.len() {
return Err("Bind: truncated format codes".into());
}
let mut formats = Vec::with_capacity(fmt_count);
for _ in 0..fmt_count {
formats.push(u16::from_be_bytes([body[cur], body[cur + 1]]));
cur += 2;
}
if cur + 2 > body.len() {
return Err("Bind: truncated parameter count".into());
}
let param_count = u16::from_be_bytes([body[cur], body[cur + 1]]) as usize;
cur += 2;
if usize::from(stmt.placeholder_count) != param_count {
return Err(format!(
"Bind: parameter count mismatch (SQL has {}, Bind has {param_count})",
stmt.placeholder_count
));
}
let mut params: Vec<spg_storage::Value<'static>> = Vec::with_capacity(param_count);
for i in 0..param_count {
if cur + 4 > body.len() {
return Err("Bind: truncated parameter length".into());
}
let len = i32::from_be_bytes([body[cur], body[cur + 1], body[cur + 2], body[cur + 3]]);
cur += 4;
if len < 0 {
params.push(spg_storage::Value::Null);
continue;
}
let len = len as usize;
if cur + len > body.len() {
return Err("Bind: parameter value truncated".into());
}
let fmt = match formats.len() {
0 => 0,
1 => formats[0],
_ => formats.get(i).copied().unwrap_or(0),
};
if fmt == 1 {
let oid = stmt.param_type_oids.get(i).copied().unwrap_or(0);
let v = decode_binary_param(oid, &body[cur..cur + len])?;
params.push(v);
cur += len;
continue;
}
if fmt != 0 {
return Err(format!("Bind: unsupported parameter format code {fmt}"));
}
let s = std::str::from_utf8(&body[cur..cur + len])
.map_err(|_| "Bind: text parameter not valid UTF-8".to_string())?;
params.push(text_param_to_value(s));
cur += len;
}
let mut result_formats: Vec<i16> = Vec::new();
if cur + 2 <= body.len() {
let rf_count = u16::from_be_bytes([body[cur], body[cur + 1]]) as usize;
cur += 2;
if cur + rf_count * 2 <= body.len() {
for _ in 0..rf_count {
result_formats.push(i16::from_be_bytes([body[cur], body[cur + 1]]));
cur += 2;
}
}
}
Ok((
portal_name,
Portal {
stmt_name,
params,
suspended: None,
result_formats,
},
))
}
fn text_param_to_value(s: &str) -> spg_storage::Value<'static> {
let trimmed = s.trim();
if trimmed.eq_ignore_ascii_case("true") {
return spg_storage::Value::Bool(true);
}
if trimmed.eq_ignore_ascii_case("false") {
return spg_storage::Value::Bool(false);
}
if let Ok(n) = trimmed.parse::<i32>() {
return spg_storage::Value::Int(n);
}
if let Ok(n) = trimmed.parse::<i64>() {
return spg_storage::Value::BigInt(n);
}
if let Ok(x) = trimmed.parse::<f64>() {
return spg_storage::Value::Float(x);
}
if let Some(v) = parse_vector_text(trimmed) {
return spg_storage::Value::vector(v);
}
spg_storage::Value::text(s)
}
fn decode_binary_param(oid: u32, bytes: &[u8]) -> Result<spg_storage::Value<'static>, String> {
use spg_storage::Value;
match oid {
16 => {
if bytes.len() != 1 {
return Err(format!(
"Bind binary BOOL must be 1 byte, got {}",
bytes.len()
));
}
Ok(Value::Bool(bytes[0] != 0))
}
17 | 25 | 1043 => {
if oid == 17 {
let s =
bytes
.iter()
.fold(String::with_capacity(2 + bytes.len() * 2), |mut acc, b| {
if acc.is_empty() {
acc.push('\\');
acc.push('x');
}
acc.push_str(&format!("{b:02x}"));
acc
});
Ok(Value::text(if s.is_empty() {
"\\x".to_string()
} else {
s
}))
} else {
let s = std::str::from_utf8(bytes)
.map_err(|_| "Bind binary TEXT/VARCHAR: invalid UTF-8".to_string())?;
Ok(Value::text(s))
}
}
20 => {
if bytes.len() != 8 {
return Err(format!(
"Bind binary BIGINT must be 8 bytes, got {}",
bytes.len()
));
}
let n = i64::from_be_bytes(bytes.try_into().unwrap());
Ok(Value::BigInt(n))
}
21 => {
if bytes.len() != 2 {
return Err(format!(
"Bind binary INT2 must be 2 bytes, got {}",
bytes.len()
));
}
let n = i16::from_be_bytes(bytes.try_into().unwrap());
Ok(Value::SmallInt(n))
}
23 => {
if bytes.len() != 4 {
return Err(format!(
"Bind binary INT must be 4 bytes, got {}",
bytes.len()
));
}
let n = i32::from_be_bytes(bytes.try_into().unwrap());
Ok(Value::Int(n))
}
700 => {
if bytes.len() != 4 {
return Err(format!(
"Bind binary REAL must be 4 bytes, got {}",
bytes.len()
));
}
let f = f32::from_be_bytes(bytes.try_into().unwrap()) as f64;
Ok(Value::Float(f))
}
701 => {
if bytes.len() != 8 {
return Err(format!(
"Bind binary DOUBLE must be 8 bytes, got {}",
bytes.len()
));
}
let f = f64::from_be_bytes(bytes.try_into().unwrap());
Ok(Value::Float(f))
}
1082 => {
if bytes.len() != 4 {
return Err(format!(
"Bind binary DATE must be 4 bytes, got {}",
bytes.len()
));
}
const PG_EPOCH_DAYS_FROM_UNIX: i32 = 10957;
let pg_days = i32::from_be_bytes(bytes.try_into().unwrap());
Ok(Value::Date(pg_days + PG_EPOCH_DAYS_FROM_UNIX))
}
1114 | 1184 => {
if bytes.len() != 8 {
return Err(format!(
"Bind binary TIMESTAMP must be 8 bytes, got {}",
bytes.len()
));
}
const PG_EPOCH_MICROS_FROM_UNIX: i64 = 946_684_800_000_000;
let pg_micros = i64::from_be_bytes(bytes.try_into().unwrap());
Ok(Value::Timestamp(pg_micros + PG_EPOCH_MICROS_FROM_UNIX))
}
1700 => decode_binary_numeric(bytes),
1186 => {
if bytes.len() != 16 {
return Err(format!(
"Bind binary INTERVAL must be 16 bytes, got {}",
bytes.len()
));
}
let micros = i64::from_be_bytes(bytes[0..8].try_into().unwrap());
let days = i32::from_be_bytes(bytes[8..12].try_into().unwrap());
let months = i32::from_be_bytes(bytes[12..16].try_into().unwrap());
Ok(Value::Interval {
months,
days,
micros,
})
}
2950 => {
if bytes.len() != 16 {
return Err(format!(
"Bind binary UUID must be 16 bytes, got {}",
bytes.len()
));
}
let mut b = [0u8; 16];
b.copy_from_slice(bytes);
Ok(Value::Uuid(b))
}
0 => Err(
"Bind: binary format requires the parameter OID to be declared in Parse \
(got OID=0 meaning unknown)"
.into(),
),
_ => Err(format!(
"Bind: binary format for OID {oid} not supported in v6.3.4"
)),
}
}
fn decode_binary_numeric(bytes: &[u8]) -> Result<spg_storage::Value<'static>, String> {
if bytes.len() < 8 {
return Err("Bind binary NUMERIC: header truncated".into());
}
let ndigits = u16::from_be_bytes([bytes[0], bytes[1]]) as usize;
let weight = i16::from_be_bytes([bytes[2], bytes[3]]);
let sign = u16::from_be_bytes([bytes[4], bytes[5]]);
let dscale = u16::from_be_bytes([bytes[6], bytes[7]]);
if bytes.len() != 8 + ndigits * 2 {
return Err(format!(
"Bind binary NUMERIC: declared ndigits={ndigits} but body has {} bytes",
bytes.len()
));
}
if sign == 0xC000 {
return Err("Bind binary NUMERIC: NaN sign not supported".into());
}
let mut digits: Vec<u16> = Vec::with_capacity(ndigits);
for i in 0..ndigits {
let off = 8 + i * 2;
let d = u16::from_be_bytes([bytes[off], bytes[off + 1]]);
digits.push(d);
}
let mut unscaled: i128 = 0;
let total_digits_after_weight = ndigits as i32 - 1 - weight as i32;
for (k, d) in digits.iter().enumerate() {
let exp = (weight as i32 - k as i32) * 4;
let final_exp = exp + dscale as i32;
if final_exp >= 0 {
let pow = 10i128.pow(final_exp as u32);
unscaled = unscaled
.checked_add((*d as i128).checked_mul(pow).ok_or("NUMERIC overflow")?)
.ok_or("NUMERIC overflow")?;
} else {
let shift = (-final_exp) as u32;
let pow = 10i128.pow(shift);
unscaled = unscaled
.checked_add((*d as i128) / pow)
.ok_or("NUMERIC overflow")?;
}
}
let _ = total_digits_after_weight; let final_value = if sign == 0x4000 { -unscaled } else { unscaled };
let scale = dscale;
Ok(spg_storage::Value::Numeric {
scaled: final_value,
scale,
kind: spg_storage::NumericKind::Finite,
})
}
fn parse_vector_text(s: &str) -> Option<Vec<f32>> {
let bytes = s.as_bytes();
if bytes.len() < 2 || bytes[0] != b'[' || bytes[bytes.len() - 1] != b']' {
return None;
}
let inner = &s[1..s.len() - 1];
if inner.trim().is_empty() {
return Some(Vec::new());
}
let mut out = Vec::with_capacity(inner.split(',').count());
for tok in inner.split(',') {
let t = tok.trim();
let f: f32 = t.parse().ok()?;
if !f.is_finite() {
return None;
}
out.push(f);
}
Some(out)
}
#[allow(clippy::too_many_arguments)]
fn handle_execute(
body: &[u8],
portals: &mut std::collections::HashMap<String, Portal>,
prepared: &std::collections::HashMap<String, PreparedStmt>,
settings: &std::collections::HashMap<String, String>,
out: &mut Vec<u8>,
state: &Arc<ServerState>,
role: Role,
tx_state: &mut u8,
conn_state: &Arc<crate::ConnState>,
) -> Result<(), (&'static str, String)> {
let stream: &mut Vec<u8> = out;
let proto = |m: String| ("42000", m);
let mut cur = 0;
let portal_name = read_cstring(body, &mut cur)
.ok_or_else(|| proto("Execute: portal name not UTF-8".to_string()))?;
if cur + 4 > body.len() {
return Err(proto("Execute: missing max-rows".to_string()));
}
let max_rows =
u32::from_be_bytes([body[cur], body[cur + 1], body[cur + 2], body[cur + 3]]) as usize;
let portal_key = portal_name.to_string();
if let Some(p) = portals.get_mut(&portal_key)
&& let Some(susp) = p.suspended.as_mut()
{
let stream: &mut Vec<u8> = out;
let end = if max_rows == 0 {
susp.rows.len()
} else {
(susp.cursor + max_rows).min(susp.rows.len())
};
let row_arena = bumpalo::Bump::new();
let any_binary = susp.formats.iter().any(|&f| f == 1);
let (wire_style, wire_tz) = state
.engine
.read()
.map(|e| (e.render_style(), e.session_tz()))
.unwrap_or((Default::default(), spg_engine::SessionTz::Utc));
for row in &susp.rows[susp.cursor..end] {
if any_binary {
encode_data_row_formats(
stream,
&susp.columns,
row,
&susp.formats,
&row_arena,
&wire_style,
&wire_tz,
)
.map_err(|e| proto(e.to_string()))?;
} else {
encode_data_row(
stream,
&susp.columns,
row,
&row_arena,
&wire_style,
&wire_tz,
)
.map_err(|e| proto(e.to_string()))?;
}
}
susp.cursor = end;
if end < susp.rows.len() {
send_msg(stream, b's', &[]).map_err(|e| proto(e.to_string()))?;
} else {
let tag = susp.tag.clone();
p.suspended = None;
send_command_complete(stream, &tag).map_err(|e| proto(e.to_string()))?;
}
return Ok(());
}
let portal = portals
.get(portal_name)
.ok_or_else(|| proto(format!("Execute: portal {portal_name:?} not found")))?;
let stmt = prepared.get(&portal.stmt_name).ok_or_else(|| {
proto(format!(
"Execute: prepared statement {:?} dropped while a portal held a reference",
portal.stmt_name
))
})?;
conn_state
.cancel_flag
.store(false, std::sync::atomic::Ordering::Relaxed);
let cancel = statement_cancel(settings, &conn_state.cancel_flag);
let needs_write = !matches!(&stmt.ast, spg_sql::ast::Statement::Select(_));
let cached_row_desc = stmt.row_desc_body.clone();
let wants_binary = portal.result_formats.iter().any(|&f| f == 1);
if let (spg_sql::ast::Statement::Select(s), true, 0, false) =
(&stmt.ast, portal.params.is_empty(), max_rows, wants_binary)
{
let mut eng = state
.engine
.write()
.map_err(|_| proto("Execute: engine lock poisoned".to_string()))?;
if matches!(role, Role::ReadOnly) {
let _ = needs_write;
}
let mut cols_storage: Vec<ColumnSchema> = Vec::new();
let wire_style = eng.render_style();
let wire_tz = eng.session_tz();
let ext_arena = bumpalo::Bump::new();
let stream_emit_result =
eng.execute_prepared_select_streaming(s, cancel, |item| match item {
spg_engine::StreamItem::Header(cols) => {
cols_storage.extend_from_slice(cols);
let _ = &cached_row_desc;
Ok(())
}
spg_engine::StreamItem::Row(cells) => encode_data_row_cells(
stream,
&cols_storage,
cells,
&ext_arena,
&wire_style,
&wire_tz,
)
.map_err(|e| spg_engine::EngineError::Unsupported(e.to_string())),
});
drop(eng);
let row_count = match stream_emit_result {
Ok(n) => n,
Err(e) => {
let (sqlstate, msg) = engine_error_to_wire_conn(&e, conn_state);
return Err((sqlstate, msg));
}
};
send_command_complete(stream, &format!("SELECT {row_count}"))
.map_err(|e| proto(e.to_string()))?;
return Ok(());
}
let plain_dml = {
use spg_sql::ast::Statement as S;
match &stmt.ast {
S::Insert(i) => i.returning.is_none(),
S::Update(u) => u.returning.is_none(),
S::Delete(d) => d.returning.is_none(),
S::Merge(m) => m.returning.is_none(),
_ => false,
}
};
let timeout_set = settings
.get("statement_timeout")
.map(String::as_str)
.and_then(parse_timeout_ms)
.unwrap_or(0)
> 0;
let (result, queue_persisted) =
if *tx_state == b'I' && state.wal.is_some() && plain_dml && !timeout_set {
if matches!(role, Role::ReadOnly) {
return Err(proto("permission denied: readonly role".to_string()));
}
let bind_sql = if portal.params.is_empty() {
stmt.sql.clone()
} else {
let mut bind_ast = stmt.ast.clone();
spg_engine::substitute_placeholders(&mut bind_ast, &portal.params)
.map_err(|e| proto(format!("Execute: bind-final render failed: {e}")))?;
bind_ast.to_string()
};
let queue_flag = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
let (result, wal_outcome) =
crate::commit_queue_execute(state, bind_sql.clone(), &queue_flag);
if let Err(e) = wal_outcome {
return Err(proto(format!("Execute: durability append failed: {e}")));
}
if matches!(&result, Ok(QueryResult::CommandOk { .. }))
&& state.audit_path.is_some()
&& let Err(e) = crate::append_audit_pub(state, &bind_sql)
{
return Err(proto(format!("Execute: audit append failed: {e}")));
}
(result, true)
} else {
let result = {
let mut eng = state
.engine
.write()
.map_err(|_| proto("Execute: engine lock poisoned".to_string()))?;
if needs_write && matches!(role, Role::ReadOnly) {
return Err(proto("permission denied: readonly role".to_string()));
}
eng.execute_prepared_in_with_cancel(
stmt.ast.clone(),
&portal.params,
conn_state.tx_id,
cancel,
)
};
(result, false)
};
if !queue_persisted
&& needs_write
&& matches!(
&result,
Ok(QueryResult::CommandOk { .. }) | Ok(QueryResult::Rows { .. })
)
{
let mut bind_ast = stmt.ast.clone();
spg_engine::substitute_placeholders(&mut bind_ast, &portal.params)
.map_err(|e| proto(format!("Execute: bind-final render failed: {e}")))?;
persist_wire_write(state, &bind_ast.to_string(), &result, conn_state.tx_id)
.map_err(|e| proto(format!("Execute: durability append failed: {e}")))?;
}
let (wire_style, wire_tz) = state
.engine
.read()
.map(|e| (e.render_style(), e.session_tz()))
.unwrap_or((Default::default(), spg_engine::SessionTz::Utc));
match result {
Ok(QueryResult::Rows { columns, rows }) => {
let n = rows.len();
let row_arena = bumpalo::Bump::new();
let emit_end = if max_rows > 0 { max_rows.min(n) } else { n };
let formats = portals
.get(&portal_key)
.map(|p| p.result_formats.clone())
.unwrap_or_default();
for row in &rows[..emit_end] {
if wants_binary {
encode_data_row_formats(
stream,
&columns,
row,
&formats,
&row_arena,
&wire_style,
&wire_tz,
)
.map_err(|e| proto(e.to_string()))?;
} else {
encode_data_row(stream, &columns, row, &row_arena, &wire_style, &wire_tz)
.map_err(|e| proto(e.to_string()))?;
}
}
let tag = command_tag_for_rows_ast(&stmt.ast, n);
if emit_end < n {
send_msg(stream, b's', &[]).map_err(|e| proto(e.to_string()))?;
if let Some(p) = portals.get_mut(&portal_key) {
p.suspended = Some(SuspendedRows {
columns,
rows,
cursor: emit_end,
formats,
tag,
});
}
} else {
send_command_complete(stream, &tag).map_err(|e| proto(e.to_string()))?;
}
}
Ok(QueryResult::CommandOk { affected, .. }) => {
let tag = command_tag_for_ast(&stmt.ast, affected);
send_command_complete(stream, &tag).map_err(|e| proto(e.to_string()))?;
*tx_state = if state
.engine
.read()
.is_ok_and(|e| e.is_tx_open(conn_state.tx_id))
{
b'T'
} else {
b'I'
};
}
Err(e) => {
let (sqlstate, msg) = engine_error_to_wire_conn(&e, conn_state);
return Err((sqlstate, msg));
}
Ok(_) => return Err(proto("unexpected QueryResult variant".to_string())),
}
Ok(())
}
fn command_tag_for_rows_ast(stmt: &spg_sql::ast::Statement, n: usize) -> String {
use spg_sql::ast::Statement;
match stmt {
Statement::Insert(_) => format!("INSERT 0 {n}"),
Statement::Update(_) => format!("UPDATE {n}"),
Statement::Delete(_) => format!("DELETE {n}"),
Statement::Merge(_) => format!("MERGE {n}"),
_ => format!("SELECT {n}"),
}
}
fn command_tag_for_rows_sql(sql: &str, n: usize) -> String {
let first = sql
.trim_start()
.split_ascii_whitespace()
.next()
.unwrap_or("")
.to_ascii_uppercase();
match first.as_str() {
"INSERT" => format!("INSERT 0 {n}"),
"UPDATE" => format!("UPDATE {n}"),
"DELETE" => format!("DELETE {n}"),
"MERGE" => format!("MERGE {n}"),
"FETCH" => format!("FETCH {n}"),
"WITH" => spg_sql::parser::parse_statement(sql)
.map(|s| command_tag_for_rows_ast(&s, n))
.unwrap_or_else(|_| format!("SELECT {n}")),
_ => format!("SELECT {n}"),
}
}
fn command_tag_for_ast(stmt: &spg_sql::ast::Statement, affected: usize) -> String {
use spg_sql::ast::Statement;
match stmt {
Statement::Insert(_) => format!("INSERT 0 {affected}"),
Statement::Update(_) => format!("UPDATE {affected}"),
Statement::Delete(_) => format!("DELETE {affected}"),
Statement::Merge(_) => format!("MERGE {affected}"),
Statement::CreateTable(_) => "CREATE TABLE".to_string(),
Statement::DropTable { .. } => "DROP TABLE".to_string(),
Statement::AlterTable(_) => "ALTER TABLE".to_string(),
Statement::CreateIndex(_) => "CREATE INDEX".to_string(),
Statement::AlterIndex(_) => "ALTER INDEX".to_string(),
Statement::CreateView(_) => "CREATE VIEW".to_string(),
Statement::DropView { .. } => "DROP VIEW".to_string(),
Statement::CreateSequence(_) => "CREATE SEQUENCE".to_string(),
Statement::Truncate { .. } => "TRUNCATE TABLE".to_string(),
Statement::Begin(_) => "BEGIN".to_string(),
Statement::Commit => "COMMIT".to_string(),
Statement::Rollback => "ROLLBACK".to_string(),
Statement::Savepoint(_) => "SAVEPOINT".to_string(),
Statement::RollbackToSavepoint(_) => "ROLLBACK".to_string(),
Statement::ReleaseSavepoint(_) => "RELEASE".to_string(),
Statement::CreateType(_) => "CREATE TYPE".to_string(),
Statement::AlterTypeAddValue { .. } => "ALTER TYPE".to_string(),
Statement::DropType { .. } => "DROP TYPE".to_string(),
Statement::CreateUser(_) => "CREATE ROLE".to_string(),
Statement::DropUser { .. } => "DROP ROLE".to_string(),
Statement::CreatePublication(_) => "CREATE PUBLICATION".to_string(),
Statement::DropPublication { .. } => "DROP PUBLICATION".to_string(),
Statement::CreateSubscription(_) => "CREATE SUBSCRIPTION".to_string(),
Statement::DropSubscription { .. } => "DROP SUBSCRIPTION".to_string(),
_ => "OK".to_string(),
}
}
fn parse_set_statement(sql: &str) -> Option<(String, String)> {
let trimmed = sql.trim();
if !ci_starts_with(trimmed.as_bytes(), b"set ") {
return None;
}
let lower = trimmed.to_ascii_lowercase();
let rest = lower.strip_prefix("set ")?;
if rest.starts_with("local ") {
return None;
}
let rest = rest.strip_prefix("session ").unwrap_or(rest);
let (name, value_part) = if let Some(idx) = rest.find('=') {
(rest[..idx].trim().to_string(), rest[idx + 1..].trim())
} else {
let idx = rest.find(" to ")?;
(rest[..idx].trim().to_string(), rest[idx + 4..].trim())
};
if name.is_empty() {
return None;
}
let value = value_part.trim_matches('\'').trim_matches('"').to_string();
Some((name, value))
}
fn parse_show_statement(sql: &str) -> Option<String> {
let trimmed = sql.trim();
if !ci_starts_with(trimmed.as_bytes(), b"show ") {
return None;
}
let lower = trimmed.to_ascii_lowercase();
let rest = lower.strip_prefix("show ")?;
if rest.trim().trim_end_matches(';').trim() == "time zone" {
return Some("timezone".to_string());
}
if rest.trim().trim_end_matches(';').trim() == "transaction isolation level" {
return Some("transaction_isolation".to_string());
}
let name = rest.split_ascii_whitespace().next()?.to_string();
Some(name)
}
fn render_show(name: &str, settings: &std::collections::HashMap<String, String>) -> CannedResponse {
if name == "all" {
let mut entries: Vec<(String, String)> = known_defaults()
.iter()
.map(|(k, v)| (k.to_string(), v.to_string()))
.collect();
for (k, v) in settings {
if let Some(pos) = entries.iter().position(|(name, _)| name == k) {
entries[pos].1.clone_from(v);
} else {
entries.push((k.clone(), v.clone()));
}
}
entries.sort();
let columns = vec![
ColumnSchema::new("name", DataType::Text, false),
ColumnSchema::new("setting", DataType::Text, false),
ColumnSchema::new("description", DataType::Text, true),
];
let rows: Vec<Row<'static>> = entries
.into_iter()
.map(|(n, v)| Row::new(vec![Value::text(n), Value::text(v), Value::Null]))
.collect();
return CannedResponse::Rows { columns, rows };
}
let value = settings
.get(name)
.cloned()
.or_else(|| {
known_defaults()
.iter()
.find(|(k, _)| *k == name)
.map(|(_, v)| (*v).to_string())
})
.or_else(|| spg_engine::pg_guc_boot_value(name).map(str::to_string))
.unwrap_or_default();
let columns = vec![ColumnSchema::new(name.to_string(), DataType::Text, false)];
CannedResponse::Rows {
columns,
rows: vec![Row::new(vec![Value::text(value)])],
}
}
fn known_defaults() -> &'static [(&'static str, &'static str)] {
&[
("application_name", ""),
("client_encoding", "UTF8"),
("datestyle", "ISO, MDY"),
("default_text_search_config", "pg_catalog.english"),
("default_transaction_isolation", "read committed"),
("default_transaction_read_only", "off"),
("work_mem", "4MB"),
("maintenance_work_mem", "64MB"),
("shared_buffers", "128MB"),
("effective_cache_size", "4GB"),
("client_min_messages", "notice"),
("intervalstyle", "postgres"),
("search_path", "\"$user\", public"),
("server_encoding", "UTF8"),
("server_version", "18.4 (spg-4.19)"),
("server_version_num", "180004"),
("standard_conforming_strings", "on"),
("statement_timeout", "0"),
("timezone", "UTC"),
("transaction_isolation", "read committed"),
("transaction_read_only", "off"),
]
}
fn monotonic_now_us() -> u64 {
use std::sync::OnceLock;
use std::time::Instant;
static ORIGIN: OnceLock<Instant> = OnceLock::new();
let origin = ORIGIN.get_or_init(Instant::now);
let micros = origin.elapsed().as_micros();
u64::try_from(micros).unwrap_or(u64::MAX)
}
fn wallclock_unix_micros() -> i64 {
use std::sync::OnceLock;
use std::time::{Instant, SystemTime, UNIX_EPOCH};
static ORIGIN: OnceLock<(Instant, i64)> = OnceLock::new();
let (boot_instant, boot_unix_us) = *ORIGIN.get_or_init(|| {
let i = Instant::now();
let u = SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|d| d.as_micros() as i64)
.unwrap_or(0);
(i, u)
});
let elapsed_us = boot_instant.elapsed().as_micros() as i64;
boot_unix_us.saturating_add(elapsed_us)
}
fn parse_timeout_ms(s: &str) -> Option<u64> {
let t = s.trim();
if t.is_empty() {
return None;
}
if let Ok(n) = t.parse::<u64>() {
return Some(n);
}
let split_at = t.find(|c: char| c.is_ascii_alphabetic()).unwrap_or(t.len());
let (num_part, unit) = t.split_at(split_at);
let num: u64 = num_part.trim().parse().ok()?;
let mult: u64 = match unit.trim().to_ascii_lowercase().as_str() {
"us" => return Some(num / 1000),
"ms" => 1,
"s" => 1_000,
"min" | "m" => 60_000,
"h" => 3_600_000,
"d" => 86_400_000,
_ => return None,
};
num.checked_mul(mult)
}
fn statement_cancel<'a>(
settings: &std::collections::HashMap<String, String>,
flag: &'a std::sync::atomic::AtomicBool,
) -> CancelToken<'a> {
let base = CancelToken::from_flag(flag);
let raw = settings
.get("statement_timeout")
.map(String::as_str)
.unwrap_or("0");
let Some(ms) = parse_timeout_ms(raw) else {
return base;
};
if ms == 0 {
return base;
}
let now_fn: MonotonicNowFn = monotonic_now_us;
let deadline_us = monotonic_now_us().saturating_add(ms.saturating_mul(1_000));
base.with_deadline(now_fn, deadline_us)
}
pub(crate) fn new_cancel_secret() -> u32 {
use std::hash::{BuildHasher, Hasher};
let mut h = std::collections::hash_map::RandomState::new().build_hasher();
h.write_u64(
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_nanos() as u64)
.unwrap_or(0),
);
(h.finish() >> 16) as u32
}
fn engine_error_to_wire_conn(
e: &EngineError,
conn_state: &crate::ConnState,
) -> (&'static str, String) {
if matches!(e, EngineError::Cancelled)
&& conn_state
.cancel_flag
.load(std::sync::atomic::Ordering::Relaxed)
{
return (
"57014",
"canceling statement due to user request".to_string(),
);
}
engine_error_to_wire(e)
}
fn strip_error_class(msg: &str) -> String {
const CLASSES: &[&str] = &[
"eval: type mismatch: ",
"eval: ",
"unsupported: ",
"parse: lex: ",
"lex: ",
"parse: ",
"storage: ",
];
for c in CLASSES {
if let Some(rest) = msg.strip_prefix(c) {
return rest.to_string();
}
}
msg.to_string()
}
fn window_sqlstate(msg: &str) -> Option<&'static str> {
if msg.contains("window functions are not allowed in ")
|| msg.contains("frame start cannot be ")
|| msg.contains("frame end cannot be ")
|| msg.contains("frame starting from ")
|| msg.contains("cannot override PARTITION BY clause of window ")
|| msg.contains("cannot override ORDER BY clause of window ")
|| msg.contains("because it has a frame clause")
|| msg.contains("RANGE with offset PRECEDING/FOLLOWING ")
|| (msg.contains("window \"") && msg.contains("\" is already defined"))
{
return Some("42P20");
}
if msg.contains("window \"") && msg.contains("\" does not exist") {
return Some("42704");
}
if msg.contains("is not implemented for window functions") {
return Some("0A000");
}
None
}
fn parse_error_position(e: &EngineError, sql: &str) -> Option<usize> {
match e {
EngineError::Parse(pe) => spg_sql::parser::syntax_error_position(sql, false, pe.token_pos),
_ => None,
}
}
pub(crate) fn engine_error_to_wire(e: &EngineError) -> (&'static str, String) {
if let EngineError::Cancelled = e {
return (
"57014",
"canceling statement due to statement timeout".to_string(),
);
}
if let EngineError::InFailedTransaction = e {
return ("25P02", e.to_string());
}
if let EngineError::CardinalityViolation = e {
return ("21000", e.to_string());
}
if let EngineError::SerializationFailure(_) = e {
return ("40001", e.to_string());
}
if let EngineError::LockDeadlock = e {
return ("40P01", e.to_string());
}
{
let msg = e.to_string();
if msg.contains("canceling statement due to lock timeout") {
return ("55P03", msg);
}
}
{
let msg = e.to_string();
if msg.contains("could not obtain lock on row in relation") {
return ("55P03", msg);
}
}
{
let msg = e.to_string();
if msg.contains("is not in select list")
|| msg.contains("must appear in select list")
|| msg.contains("must match initial ORDER BY expressions")
{
return ("42P10", msg);
}
if msg.contains("cannot affect row a second time") {
return ("21000", msg);
}
if msg.contains("missing FROM-clause entry for table") {
return ("42P01", msg);
}
if msg.contains("arguments to GROUPING must be grouping expressions") {
return ("42803", msg);
}
if msg.contains("must appear in the GROUP BY clause") {
return ("42803", msg);
}
if msg.contains("does not exist") && msg.starts_with("type \"") {
return ("42704", msg);
}
if msg.contains("is out of bounds for sequence") {
return ("22003", msg);
}
if msg.contains("cannot be less than MINVALUE")
|| msg.contains("cannot be greater than MAXVALUE")
|| msg.contains("INCREMENT must not be zero")
{
return ("22023", msg);
}
if msg.contains("ON CONFLICT DO UPDATE requires inference specification") {
return ("42601", msg);
}
if msg.contains("query must have the same number of columns") {
return ("42601", msg);
}
if msg.contains("LIMIT must not be negative") {
return ("2201W", msg);
}
if msg.contains("OFFSET must not be negative") {
return ("2201X", msg);
}
if msg.contains("invalid input syntax for type bigint") {
return ("22P02", msg);
}
if msg.contains(" types ") && msg.contains(" cannot be matched") {
return ("42804", msg);
}
}
if let Some(code) = window_sqlstate(&e.to_string()) {
return (code, e.to_string());
}
if let EngineError::Parse(_) = e {
return ("42601", e.to_string());
}
let msg = e.to_string();
let code =
if msg.contains("cannot drop the currently open database") {
"55006"
} else if msg.contains("database \"") && msg.ends_with("does not exist") {
"3D000"
} else if msg.contains("must be called before any query")
|| msg.contains("cannot run inside a transaction block")
{
"25001"
} else if msg.contains("value too long for type") {
"22001"
} else if msg.contains("date/time field value out of range")
|| msg.contains("timestamp out of range")
|| msg.contains("interval out of range")
{
"22008"
} else if msg.contains("invalid input syntax for type date")
|| msg.contains("invalid input syntax for type timestamp")
{
"22007"
} else if msg.contains("invalid input syntax for type")
|| msg.contains("invalid Roman numeral")
|| msg.contains("invalid cidr value")
{
"22P02"
} else if msg.contains("input is out of range")
|| msg.contains("integer out of range")
|| msg.contains("smallint out of range")
|| msg.contains("bigint out of range")
|| msg.contains("value overflows numeric format")
|| msg.contains("OID out of range")
|| msg.contains("is out of range for type double precision")
|| msg.contains("is out of range for type real")
|| msg.contains("is not between 0 and 1")
|| msg.contains("numeric field overflow")
|| msg.contains("BIGINT UNSIGNED value is out of range")
{
"22003"
} else if msg.contains("doesn't have a default value") {
"HY000"
} else if msg.contains("division by zero") {
"22012"
} else if msg.contains("range lower bound must be less than or equal")
|| msg.contains("result of range difference would not be contiguous")
|| msg.contains("result of range union would not be contiguous")
{
"22000"
} else if msg.contains("malformed range literal")
|| msg.contains("is not a valid binary digit")
{
"22P02"
} else if msg.contains("bit strings of different sizes") {
"22026"
} else if msg.contains("negative substring length not allowed") {
"22011"
} else if msg.contains("out of valid range, 0..") {
"22003"
} else if msg.contains("invalid regular expression") {
"2201B"
} else if msg.contains("more than one function named")
|| msg.contains("more than one operator named")
{
"42725"
} else if (msg.contains("type \"") && msg.contains("\" does not exist"))
|| msg.contains("text search configuration \"")
|| msg.contains("text search dictionary \"")
|| (msg.contains("index \"") && msg.contains("\" does not exist"))
{
"42704"
} else if msg.contains("\" specified more than once") {
"42701"
} else if (msg.contains("function \"") && msg.contains("\" does not exist"))
|| msg.contains("operator does not exist:")
|| msg.contains("does not support named arguments")
|| msg.contains("has no argument named")
{
"42883"
} else if msg.contains("multiple primary keys for table") {
"42P16"
} else if msg.contains("cannot insert a non-DEFAULT value into column") {
"428C9"
} else if msg.contains("column \"") && msg.contains("already exists") {
"42701"
} else if msg.contains("column \"") && msg.contains("does not exist") {
"42703"
} else if msg.contains("constraint \"")
&& (msg.contains("\" for relation \"") || msg.contains("\" for table \""))
&& msg.contains("already exists")
{
"42710"
} else if msg.contains("constraint \"") && msg.contains("does not exist") {
"42704"
} else if msg.contains("type \"") && msg.contains("already exists") {
"42710"
} else if msg.contains("enum label \"") && msg.contains("already exists") {
"42710"
} else if msg.contains("JSON object does not contain key")
|| msg.contains("jsonpath member accessor can only be applied")
{
"2203A"
} else if msg.contains("jsonpath array subscript is out of bounds") {
"22033"
} else if msg.contains("jsonpath wildcard array accessor can only be applied")
|| msg.contains("jsonpath array accessor can only be applied")
{
"22039"
} else if msg.contains("is not an existing enum label")
|| msg.contains("cannot delete from scalar")
|| msg.contains("cannot delete path in scalar")
|| msg.contains("cannot set path in scalar")
|| msg.contains("cannot delete from object using integer index")
{
"22023"
} else if msg.contains("is not an identity column") {
"42703"
} else if msg.contains("relation \"") && msg.contains("already exists") {
"42P07"
} else if (msg.contains("table \"")
|| msg.contains("relation \"")
|| msg.contains("view \"")
|| msg.contains("sequence \""))
&& msg.contains("does not exist")
{
"42P01"
} else if msg.contains("cannot take logarithm of") {
"2201E"
} else if msg.contains("zero raised to a negative power is undefined")
|| msg.contains("a negative number raised to a non-integer power")
|| msg.contains("cannot take square root of a negative number")
{
"2201F"
} else if msg.contains("duplicate JSON object key value") {
"22030"
} else if msg.contains("LIKE pattern must not end with escape")
|| msg.contains("invalid escape string")
{
"22025"
} else if msg.contains("null character not permitted")
|| msg.contains("requested character too large for encoding")
{
"54000"
} else if msg.contains("invalid value for parameter")
|| msg.contains("is outside the valid range for parameter")
|| msg.contains("sample size must be between")
|| msg.contains("unrecognized headline parameter")
|| msg.contains("MinWords must be")
|| msg.contains("ShortWord must be")
|| msg.contains("MaxFragments must be")
|| msg.contains("step size cannot equal zero")
|| msg.contains("field position must not be zero")
|| msg.contains("start value cannot be")
|| msg.contains("stop value cannot be")
|| msg.contains("step size cannot be")
|| msg.contains("cannot get array length of")
|| msg.contains("cannot call json_object_keys")
|| msg.contains("cannot call jsonb_object_keys")
|| msg.contains("string is not a valid identifier")
|| msg.contains("unrecognized privilege type")
|| (msg.contains("unit \"") && msg.contains("\" not recognized for type"))
{
"22023"
} else if msg.contains("invalid parameter list format")
|| msg.contains("of jsonpath input")
|| msg.contains("INSERT has more expressions than target columns")
|| msg.contains("INSERT has more target columns than expressions")
{
"42601"
} else if msg.contains("searching for elements in multidimensional arrays")
|| msg.contains("encoding conversion from UTF8 to ASCII")
|| msg.contains("cannot accept a value of type")
|| msg.contains("does not have a RETURNING clause")
|| msg.contains("must be at the top level")
|| msg.contains("must not contain data-modifying statements in WITH")
|| msg.contains("must not use data-modifying statements in WITH")
|| msg.contains("View columns that are not columns of their base relation")
|| msg.contains("requires CSV mode")
|| msg.contains("must be a single one-byte character")
|| (msg.contains("unit \"") && msg.contains("\" not supported for type"))
{
"0A000"
} else if msg.contains("duplicate key value violates unique constraint")
|| (msg.contains("violation") && (msg.contains("UNIQUE") || msg.contains("PRIMARY KEY")))
|| msg.contains("could not create unique index")
{
"23505"
} else if msg.contains("violates exclusion constraint") {
"23P01"
} else if msg.contains("nextval: reached") {
"2200H"
} else if msg.contains("violates foreign key constraint")
|| msg.contains("FOREIGN KEY violation")
{
"23503"
} else if msg.contains("violates check option") {
"44000"
} else if msg.contains("violates check constraint")
|| msg.contains("CHECK constraint violation")
{
"23514"
} else if msg.contains("violates not-null constraint")
|| msg.contains("NOT NULL column")
|| msg.contains("contains null values")
{
"23502"
} else if msg.contains("durability append failed") || msg.contains("could not write") {
let lower = msg.to_ascii_lowercase();
if lower.contains("no space left")
|| lower.contains("quota")
|| lower.contains("storage full")
|| lower.contains("below water-mark")
{
"53100"
} else if lower.contains("out of memory") {
"53200"
} else if lower.contains("permission denied") {
"42501"
} else if lower.contains("no such file") {
"58P01"
} else {
"58030"
}
} else if msg.contains("permission denied for table")
|| msg.contains("permission denied for sequence")
|| msg.contains("permission denied for schema")
|| msg.contains("permission denied for function")
|| msg.contains("must be owner of table")
{
"42501"
} else if msg.contains("role \"") && msg.contains("does not exist") {
"42704"
} else if msg.contains("cannot be dropped because some objects depend on it") {
"2BP01"
} else if msg.contains("is not unique") {
"42725"
} else if msg.contains("function") && msg.contains("does not exist") {
"42883"
} else if msg.contains("column reference") && msg.contains("is ambiguous") {
"42702"
} else if matches!(
e,
EngineError::Eval(spg_engine::eval::EvalError::TypeMismatch { .. })
) {
if msg.contains("cannot cast jsonb") {
"22023"
} else if msg.contains("cannot cast") {
"42846"
} else {
"42883"
}
} else {
"42000"
};
let msg = strip_error_class(&msg);
(code, msg)
}
#[cfg(test)]
mod engine_error_sqlstate_tests {
use super::engine_error_to_wire;
use spg_engine::EngineError;
fn code(msg: &str) -> &'static str {
engine_error_to_wire(&EngineError::Unsupported(msg.to_string())).0
}
fn tm_code(detail: &str) -> &'static str {
engine_error_to_wire(&EngineError::Eval(
spg_engine::eval::EvalError::TypeMismatch {
detail: detail.to_string(),
},
))
.0
}
#[test]
fn round622_wrong_argument_type_is_42883_not_the_class_code() {
assert_eq!(tm_code("upper() needs text, got integer"), "42883");
assert_eq!(tm_code("sum/avg need numeric, got text"), "42883");
assert_eq!(tm_code("LIKE requires text operands, got integer"), "42883");
assert_eq!(
tm_code("unnest() expects an array argument, got integer"),
"42883"
);
assert_eq!(
tm_code("date_trunc() needs DATE or TIMESTAMP, got integer"),
"42883"
);
assert_eq!(tm_code("cannot cast integer to inet"), "42846");
assert_eq!(tm_code("cannot cast integer[] to int"), "42846");
assert_eq!(tm_code("cannot cast jsonb object to type integer"), "22023");
assert_eq!(code("function length(boolean) does not exist"), "42883");
assert_eq!(code("operator does not exist: text + integer"), "42883");
assert_eq!(code("syntax error near \"FROM\""), "42000");
}
#[test]
fn constraint_violations_map_to_class_23() {
assert_eq!(
code(
"duplicate key value violates unique constraint \"t_pkey\" on table \"t\" \
DETAIL: Key (id)=(1) already exists."
),
"23505"
);
assert_eq!(
code(
"PRIMARY KEY violation on \"t\" columns [\"id\"]: row #0 duplicates an existing key"
),
"23505"
);
assert_eq!(
code("UNIQUE INDEX \"i\" violation on \"t\": row #0 duplicates"),
"23505"
);
assert_eq!(
code("FOREIGN KEY violation: no parent row in \"t\" where id = Int(9)"),
"23503"
);
assert_eq!(
code(
"conflicting key value violates exclusion constraint \"ov_during_excl\" \
on table \"ov\" DETAIL: Key (during)=([3,7)) conflicts with existing \
key (during)=([1,5))."
),
"23P01"
);
assert_eq!(
code("CHECK constraint violation on \"t\" (row #0): \"(y > 0)\""),
"23514"
);
assert_eq!(
code("new row violates check option for view \"vv\""),
"44000"
);
assert_eq!(
code("storage: NULL value in NOT NULL column \"x\""),
"23502"
);
assert_eq!(
code("cannot add UNIQUE constraint to column with duplicate data"),
"42000"
);
assert_eq!(code("syntax error near \"FROM\""), "42000");
assert_eq!(
code("durability append failed: No space left on device (os error 28)"),
"53100"
);
assert_eq!(
code("durability append failed: WAL append hit storage full"),
"53100"
);
assert_eq!(
code("durability append failed: Permission denied (os error 13)"),
"42501"
);
assert_eq!(
code("durability append failed: No such file or directory (os error 2)"),
"58P01"
);
assert_eq!(code("durability append failed: broken pipe"), "58030");
assert_eq!(
code("multiple primary keys for table \"t3\" are not allowed"),
"42P16"
);
assert_eq!(
code("column \"a\" of relation \"t2\" already exists"),
"42701"
);
assert_eq!(
code("column \"nonexist\" of relation \"t2\" does not exist"),
"42703"
);
assert_eq!(code("table \"nonexist_tbl\" does not exist"), "42P01");
assert_eq!(
code(
"cannot insert a non-DEFAULT value into column \"a\" \
DETAIL: Column \"a\" is an identity column defined as GENERATED ALWAYS.\n\
HINT: Use OVERRIDING SYSTEM VALUE to override."
),
"428C9"
);
assert_eq!(code("relation \"r1\" already exists"), "42P07");
assert_eq!(code("relation \"nope_tbl\" does not exist"), "42P01");
assert_eq!(code("sequence \"nope_seq\" does not exist"), "42P01");
assert_eq!(
crate::strip_layer_prefixes_keeping_unsupported(
"storage: relation \"nope\" does not exist"
),
"relation \"nope\" does not exist"
);
assert_eq!(
crate::strip_layer_prefixes_keeping_unsupported(
"corrupt on-disk format: trigger \"t\" for table \"x\" does not exist"
),
"trigger \"t\" for table \"x\" does not exist"
);
assert_eq!(
crate::strip_layer_prefixes_keeping_unsupported("unsupported: whatever"),
"unsupported: whatever"
);
assert_eq!(
crate::strip_internal_error_prefixes("unsupported: whatever"),
"whatever"
);
assert_eq!(
code("corrupt on-disk format: sequence \"nope_seq\" does not exist"),
"42P01"
);
assert_eq!(code("type \"r_enum\" already exists"), "42710");
assert_eq!(
code("constraint \"c1\" for relation \"r1\" already exists"),
"42710"
);
assert_eq!(
code("constraint \"nope\" of relation \"r1\" does not exist"),
"42704"
);
assert_eq!(code("column \"nope\" does not exist"), "42703");
assert_eq!(
code(
"duplicate key value violates unique constraint \"t_pkey\" on table \"t\" \
DETAIL: Key (id)=(1) already exists."
),
"23505"
);
}
}
#[derive(Debug)]
enum CopyIntent {
From(String, Option<Vec<String>>, CopyOptions),
To(String, CopyOptions),
ToQuery(String, CopyOptions),
FromFile(spg_engine::copy::CopyFromFileSpec),
ToFile(spg_engine::copy::CopyToFileSpec),
BadOption(String),
}
#[derive(Debug, Clone, Default)]
struct CopyOptions {
pub skip: u64,
pub on_error_set_null: bool,
pub format_json: bool,
pub format_csv: bool,
pub csv_delimiter: Option<char>,
pub csv_quote: Option<char>,
pub null_string: Option<String>,
pub header: bool,
}
fn parse_copy_intent(sql: &str) -> Option<CopyIntent> {
let trimmed = sql.trim();
if !ci_starts_with(trimmed.as_bytes(), b"copy ") {
return None;
}
let lower = trimmed.to_ascii_lowercase();
let rest = lower.strip_prefix("copy ")?;
let bytes = rest.as_bytes();
let mut i = skip_ws_bytes(bytes, 0);
if bytes.get(i) == Some(&b'(') {
return parse_copy_query_intent(trimmed, rest, i);
}
let table_start = i;
while i < bytes.len() {
let c = bytes[i] as char;
if c.is_ascii_whitespace() || c == '(' {
break;
}
i += 1;
}
if i == table_start {
return None;
}
let raw = &rest[table_start..i];
let table = match raw.rsplit_once('.') {
Some((_, bare)) => bare.to_string(),
None => raw.to_string(),
};
i = skip_ws_bytes(bytes, i);
let mut column_list: Option<Vec<String>> = None;
if bytes.get(i) == Some(&b'(') {
let list_start = i + 1;
let mut depth = 1usize;
i += 1;
while i < bytes.len() && depth > 0 {
match bytes[i] {
b'(' => depth += 1,
b')' => depth -= 1,
_ => {}
}
i += 1;
}
let prefix = trimmed.len() - rest.len();
let names: Vec<String> = trimmed[prefix + list_start..prefix + i - 1]
.split(',')
.map(|c| c.trim().trim_matches('"').to_string())
.filter(|c| !c.is_empty())
.collect();
if !names.is_empty() {
column_list = Some(names);
}
i = skip_ws_bytes(bytes, i);
}
let dir_start = i;
while i < bytes.len() && !(bytes[i] as char).is_ascii_whitespace() {
i += 1;
}
if i == dir_start {
return None;
}
let dir = &rest[dir_start..i];
i = skip_ws_bytes(bytes, i);
let ep_start = i;
while i < bytes.len() && !(bytes[i] as char).is_ascii_whitespace() && bytes[i] != b';' {
i += 1;
}
if i == ep_start {
return None;
}
let endpoint = &rest[ep_start..i];
if dir == "from" && endpoint.starts_with('\'') {
return spg_engine::copy::parse_copy_from_file(trimmed).map(CopyIntent::FromFile);
}
if dir == "to" && endpoint.starts_with('\'') {
return spg_engine::copy::parse_copy_to_file(trimmed).map(CopyIntent::ToFile);
}
match (dir, endpoint) {
("from", "stdin") => {
match parse_copy_options_checked(trimmed) {
Ok(opts) => Some(CopyIntent::From(table, column_list, opts)),
Err(bad) => Some(CopyIntent::BadOption(bad)),
}
}
("to", "stdout") => match parse_copy_options_checked(trimmed) {
Ok(opts) => Some(CopyIntent::To(table, opts)),
Err(bad) => Some(CopyIntent::BadOption(bad)),
},
_ => None,
}
}
fn parse_copy_query_intent(trimmed: &str, rest: &str, lparen: usize) -> Option<CopyIntent> {
let bytes = rest.as_bytes();
let mut depth = 0usize;
let mut j = lparen;
let mut close = None;
while j < bytes.len() {
match bytes[j] {
b'(' => depth += 1,
b')' => {
depth -= 1;
if depth == 0 {
close = Some(j);
break;
}
}
_ => {}
}
j += 1;
}
let close = close?;
let prefix = trimmed.len() - rest.len();
let query = trimmed[prefix + lparen + 1..prefix + close]
.trim()
.to_string();
if query.is_empty() {
return None;
}
let after = &rest[close + 1..];
let lower_after = after.trim_start();
let mut it = lower_after.split_ascii_whitespace();
if !matches!(it.next(), Some("to")) {
return None;
}
let ep = it.next().unwrap_or("");
if ep.starts_with('\'') {
return spg_engine::copy::parse_copy_to_file(trimmed).map(CopyIntent::ToFile);
}
if ep.trim_end_matches(';') != "stdout" {
return None;
}
let opts = parse_copy_options(&trimmed[prefix + close + 1..]);
Some(CopyIntent::ToQuery(query, opts))
}
fn skip_ws_bytes(bytes: &[u8], mut i: usize) -> usize {
while i < bytes.len() && (bytes[i] as char).is_ascii_whitespace() {
i += 1;
}
i
}
fn parse_copy_options(sql: &str) -> CopyOptions {
parse_copy_options_checked(sql).unwrap_or_else(|_| CopyOptions::default())
}
fn parse_copy_options_checked(sql: &str) -> Result<CopyOptions, String> {
let mut opts = CopyOptions::default();
let Some(search_from) = sql.to_ascii_lowercase().rfind("with") else {
return Ok(opts);
};
let Some(open) = sql[search_from..].find('(').map(|p| search_from + p) else {
return Ok(opts);
};
let Some(close) = sql[open..].rfind(')').map(|p| open + p) else {
return Ok(opts);
};
let inner = &sql[open + 1..close];
for pair in inner.split(',') {
let pair = pair.trim();
if pair.is_empty() {
continue;
}
let mut it = pair.split_ascii_whitespace();
let key = it.next().unwrap_or("").to_ascii_lowercase();
let val_raw = it.next().unwrap_or("");
let val_lc = val_raw.to_ascii_lowercase();
let (key, val) = (key.as_str(), val_lc.as_str());
match key {
"skip" => {
opts.skip = val.parse().unwrap_or(0);
}
"on_error" => {
if val == "set_null" {
opts.on_error_set_null = true;
}
}
"format" => match val {
"json" => opts.format_json = true,
"csv" => opts.format_csv = true,
_ => {}
},
"header" => {
if val.is_empty() || val == "true" || val == "on" {
opts.skip = opts.skip.max(1);
opts.header = true;
}
}
"delimiter" => {
opts.csv_delimiter = unquote_copy_char(val_raw);
}
"quote" => {
opts.csv_quote = unquote_copy_char(val_raw);
}
"null" => {
let t = val_raw.trim_matches(|c| c == '\'' || c == '"');
if !t.is_empty() {
opts.null_string = Some(t.to_string());
}
}
other => {
return Err(other.to_ascii_lowercase());
}
}
}
Ok(opts)
}
fn unquote_copy_char(val: &str) -> Option<char> {
let s = val.trim_matches(|c| c == '\'' || c == '"');
let mut chars = s.chars();
let c = chars.next()?;
if chars.next().is_some() {
return None;
}
Some(c)
}
fn drain_copy_in_frames(stream: &mut dyn ReadWrite) -> std::io::Result<()> {
loop {
let mut header = [0u8; 5];
if stream.read_exact(&mut header).is_err() {
return Ok(());
}
let ty = header[0];
let len = u32::from_be_bytes([header[1], header[2], header[3], header[4]]) as usize;
let body_len = len.saturating_sub(4);
if body_len > 0 {
let mut body = vec![0u8; body_len];
if stream.read_exact(&mut body).is_err() {
return Ok(());
}
}
if ty == b'c' || ty == b'f' {
return Ok(());
}
}
}
fn handle_copy_to_file(
stream: &mut dyn ReadWrite,
state: &Arc<ServerState>,
role: Role,
spec: &spg_engine::copy::CopyToFileSpec,
) -> std::io::Result<()> {
if role != Role::Admin {
send_error(
stream,
"42501",
"permission denied to COPY to a file DETAIL: Only roles with privileges of the \
\"pg_write_server_files\" role may COPY to a file.\nHINT: Anyone can COPY to \
stdout or from stdin. psql's \\copy command also works for anyone.",
)?;
return Ok(());
}
let rendered = state
.engine
.write()
.map_err(|_| std::io::Error::other("engine rwlock poisoned"))
.map(|mut e| {
e.copy_to_buffer(
&spec.table,
spec.columns.as_deref(),
spec.query.as_deref(),
&spec.options,
)
})?;
let (payload, n) = match rendered {
Ok(r) => r,
Err(e) => {
let msg = format!("{e}");
let code = if msg.contains("relation") && msg.contains("does not exist") {
"42P01"
} else if msg.contains("does not exist") || msg.contains("column") {
"42703"
} else {
"0A000"
};
send_error(stream, code, &msg)?;
return Ok(());
}
};
if let Err(e) = std::fs::write(&spec.path, payload) {
let os = e.to_string();
let os = os.split(" (os error").next().unwrap_or(&os).to_string();
let code = if e.kind() == std::io::ErrorKind::PermissionDenied {
"42501"
} else {
"58P01"
};
send_error(
stream,
code,
&format!(
"could not open file \"{path}\" for writing: {os}\nHINT: COPY TO instructs \
the PostgreSQL server process to write a file. You may want a client-side \
facility such as psql's \\copy.",
path = spec.path
),
)?;
return Ok(());
}
send_command_complete(stream, &format!("COPY {n}"))?;
Ok(())
}
fn handle_lo_file_call(
wbuf: &mut Vec<u8>,
state: &Arc<ServerState>,
role: Role,
call: &spg_engine::largeobject::LoFileCall,
) -> std::io::Result<()> {
use spg_engine::largeobject::LoFileCall;
if role != Role::Admin {
return send_error(
wbuf,
"42501",
&spg_engine::largeobject::permission_denied(call),
);
}
let value = match call {
LoFileCall::Import { path, oid } => {
let data = match std::fs::read(path) {
Ok(d) => d,
Err(e) => {
return send_error(
wbuf,
"58P01",
&spg_engine::largeobject::could_not_open(path, &e.to_string()),
);
}
};
let mut eng = state
.engine
.write()
.map_err(|_| std::io::Error::other("engine rwlock poisoned"))?;
match eng.lo_import_bytes(oid.unwrap_or(0), data) {
Ok(new_oid) => i64::from(new_oid),
Err(e) => return send_error(wbuf, "58P01", &format!("{e}")),
}
}
LoFileCall::Export { oid, path } => {
let bytes = {
let eng = state
.engine
.read()
.map_err(|_| std::io::Error::other("engine rwlock poisoned"))?;
match eng.lo_export_bytes(*oid) {
Ok(b) => b,
Err(e) => return send_error(wbuf, "42704", &format!("{e}")),
}
};
if let Err(e) = std::fs::write(path, &bytes) {
return send_error(
wbuf,
"58P01",
&spg_engine::largeobject::could_not_create(path, &e.to_string()),
);
}
1
}
};
send_canned(
wbuf,
&CannedResponse::Rows {
columns: vec![ColumnSchema::new(
call.column_name().to_string(),
DataType::BigInt,
false,
)],
rows: vec![Row::new(vec![Value::BigInt(value)])],
},
)
}
fn handle_copy_from_file(
stream: &mut dyn ReadWrite,
state: &Arc<ServerState>,
role: Role,
spec: &spg_engine::copy::CopyFromFileSpec,
tx_state: &mut u8,
tx_id: spg_engine::TxId,
) -> std::io::Result<()> {
if role != Role::Admin {
send_error(
stream,
"42501",
"permission denied to COPY from a file DETAIL: Only roles with privileges of the \
\"pg_read_server_files\" role may COPY from a file.\nHINT: Anyone can COPY to \
stdout or from stdin. psql's \\copy command also works for anyone.",
)?;
return Ok(());
}
let target = match state
.engine
.read()
.map_err(|_| std::io::Error::other("engine rwlock poisoned"))
.map(|e| e.copy_target_columns(&spec.table, spec.columns.as_deref()))
{
Ok(Ok(t)) => t,
Ok(Err(e)) => {
let msg = format!("{e}");
let code = if msg.contains("does not exist")
&& msg.contains("relation")
&& !msg.contains("column")
{
"42P01"
} else if msg.contains("specified more than once") {
"42701"
} else {
"42703"
};
send_error(stream, code, &msg)?;
return Ok(());
}
Err(e) => return Err(e),
};
let data = match std::fs::read_to_string(&spec.path) {
Ok(d) => d,
Err(e) => {
let os = e.to_string();
let os = os.split(" (os error").next().unwrap_or(&os);
send_error(
stream,
"58P01",
&format!(
"could not open file \"{path}\" for reading: {os}\nHINT: COPY FROM \
instructs the PostgreSQL server process to read a file. You may want a \
client-side facility such as psql's \\copy.",
path = spec.path
),
)?;
return Ok(());
}
};
let inserts = match spg_engine::copy::copy_buffer_inserts(
&spec.table,
spec.columns.as_deref(),
&target,
&spec.options,
&data,
) {
Ok(i) => i,
Err(e) => {
let msg = format!("{e}");
let code = if msg.contains("missing data for column")
|| msg.contains("extra data after last expected column")
{
"22P04"
} else {
"22P02"
};
send_error(stream, code, &msg)?;
return Ok(());
}
};
let wrap = !state.engine.read().is_ok_and(|e| e.is_tx_open(tx_id));
let run = |state: &Arc<ServerState>, sql: &str| -> Result<(), String> {
state
.engine
.write()
.map_err(|_| "engine rwlock poisoned".to_string())
.and_then(|mut e| {
e.execute_in(sql, tx_id)
.map(|_| ())
.map_err(|err| format!("{err}"))
})
};
if wrap {
if let Err(e) = run(state, "BEGIN") {
send_error(stream, "XX000", &format!("COPY: {e}"))?;
return Ok(());
}
if let Err(e) = crate::append_wal(state, "BEGIN", false) {
let _ = run(state, "ROLLBACK");
send_error(stream, "53100", &format!("{e}"))?;
return Ok(());
}
}
let mut inserted: u64 = 0;
for insert in &inserts {
let step = run(state, insert)
.and_then(|()| crate::append_wal(state, insert, false).map_err(|e| format!("{e}")));
match step {
Ok(()) => inserted += 1,
Err(msg) => {
if wrap {
let _ = run(state, "ROLLBACK");
let _ = crate::append_wal(state, "ROLLBACK", false);
}
send_error(stream, "22P02", &msg)?;
return Ok(());
}
}
}
if wrap {
if let Err(e) = run(state, "COMMIT") {
let _ = crate::append_wal(state, "ROLLBACK", false);
send_error(stream, "XX000", &format!("COPY: {e}"))?;
return Ok(());
}
if let Err(e) = crate::append_wal(state, "COMMIT", crate::session_sync_commit(state)) {
send_error(stream, "53100", &format!("{e}"))?;
return Ok(());
}
}
send_command_complete(stream, &format!("COPY {inserted}"))?;
*tx_state = if state.engine.read().is_ok_and(|e| e.is_tx_open(tx_id)) {
b'T'
} else {
b'I'
};
Ok(())
}
fn handle_copy_from_stdin(
stream: &mut dyn ReadWrite,
state: &Arc<ServerState>,
role: Role,
table: &str,
column_list: Option<&[String]>,
opts: &CopyOptions,
tx_state: &mut u8,
tx_id: spg_engine::TxId,
) -> std::io::Result<()> {
if !role.can_write() {
send_error(
stream,
"42501",
"permission denied: COPY FROM requires admin or readwrite",
)?;
return Ok(());
}
let table_col_names: Vec<String> = state
.engine
.read()
.ok()
.and_then(|e| {
e.catalog()
.get(table)
.map(|t| t.schema().columns.iter().map(|c| c.name.clone()).collect())
})
.unwrap_or_default();
let expected_names: Vec<String> = match column_list {
Some(cols) => cols.to_vec(),
None => table_col_names.clone(),
};
let Some(col_count) = state
.engine
.read()
.ok()
.and_then(|e| e.catalog().get(table).map(|t| t.schema().columns.len()))
else {
send_error(
stream,
"42P01",
&format!("relation {table:?} does not exist"),
)?;
return Ok(());
};
let mut body = Vec::with_capacity(3 + col_count * 2);
body.push(0);
body.extend_from_slice(&u16::try_from(col_count).unwrap_or(0).to_be_bytes());
for _ in 0..col_count {
body.extend_from_slice(&0u16.to_be_bytes());
}
send_msg(stream, b'G', &body)?;
let wrap = !opts.on_error_set_null && !state.engine.read().is_ok_and(|e| e.is_tx_open(tx_id));
if wrap {
if let Err(e) = state
.engine
.write()
.map_err(|_| std::io::Error::other("engine rwlock poisoned"))
.and_then(|mut e| {
e.execute_in("BEGIN", tx_id)
.map(|_| ())
.map_err(|err| std::io::Error::other(format!("{err}")))
})
{
send_error(stream, "XX000", &format!("COPY: {e}"))?;
drain_copy_in_frames(stream)?;
return Ok(());
}
if let Err(e) = crate::append_wal(state, "BEGIN", false) {
let _ = state
.engine
.write()
.map(|mut en| en.execute_in("ROLLBACK", tx_id));
send_error(stream, "53100", &format!("{e}"))?;
drain_copy_in_frames(stream)?;
return Ok(());
}
}
let rollback_wrap = |state: &Arc<ServerState>| {
if wrap {
let _ = state
.engine
.write()
.map(|mut e| e.execute_in("ROLLBACK", tx_id));
let _ = crate::append_wal(state, "ROLLBACK", false);
}
};
let mut buf: Vec<u8> = Vec::new();
let mut inserted: u64 = 0;
let mut skipped: u64 = 0;
loop {
let mut header = [0u8; 5];
stream.read_exact(&mut header)?;
let ty = header[0];
let len = u32::from_be_bytes([header[1], header[2], header[3], header[4]]) as usize;
let body_len = len.saturating_sub(4);
let mut body = vec![0u8; body_len];
if body_len > 0 {
stream.read_exact(&mut body)?;
}
match ty {
b'd' => buf.extend_from_slice(&body),
b'c' => {
if !buf.is_empty() && !buf.ends_with(b"\n") {
buf.push(b'\n');
}
break;
}
b'f' => {
send_error(stream, "57014", "client aborted COPY")?;
return Ok(());
}
other => {
send_error(
stream,
"08P01",
&format!("unexpected frame 0x{other:02x} during COPY"),
)?;
return Ok(());
}
}
if let Err(msg) = process_copy_chunk(
state,
table,
column_list,
&expected_names,
&mut buf,
&mut inserted,
&mut skipped,
opts,
tx_id,
) {
let code = if msg.contains("missing data for column")
|| msg.contains("extra data after last expected column")
{
"22P04"
} else {
"22P02"
};
rollback_wrap(state);
send_error(stream, code, &msg)?;
drain_copy_in_frames(stream)?;
return Ok(());
}
}
if let Err(msg) = process_copy_chunk(
state,
table,
column_list,
&expected_names,
&mut buf,
&mut inserted,
&mut skipped,
opts,
tx_id,
) {
let code = if msg.contains("missing data for column")
|| msg.contains("extra data after last expected column")
{
"22P04"
} else {
"22P02"
};
rollback_wrap(state);
send_error(stream, code, &msg)?;
return Ok(());
}
if !wrap && crate::session_sync_commit(state) {
if opts.on_error_set_null
&& inserted > 0
&& let Err(e) = crate::wal_fsync_now(state)
{
send_error(stream, "53100", &format!("{e}"))?;
return Ok(());
}
}
if wrap {
if let Err(e) = state
.engine
.write()
.map_err(|_| std::io::Error::other("engine rwlock poisoned"))
.and_then(|mut e| {
e.execute_in("COMMIT", tx_id)
.map(|_| ())
.map_err(|err| std::io::Error::other(format!("{err}")))
})
{
let _ = crate::append_wal(state, "ROLLBACK", false);
send_error(stream, "XX000", &format!("COPY: {e}"))?;
return Ok(());
}
if let Err(e) = crate::append_wal(state, "COMMIT", crate::session_sync_commit(state)) {
send_error(stream, "53100", &format!("{e}"))?;
return Ok(());
}
}
send_command_complete(stream, &format!("COPY {inserted}"))?;
*tx_state = if state.engine.read().is_ok_and(|e| e.is_tx_open(tx_id)) {
b'T'
} else {
b'I'
};
Ok(())
}
fn process_copy_chunk(
state: &Arc<ServerState>,
table: &str,
column_list: Option<&[String]>,
expected_names: &[String],
buf: &mut Vec<u8>,
inserted: &mut u64,
skipped: &mut u64,
opts: &CopyOptions,
tx_id: spg_engine::TxId,
) -> Result<(), String> {
if opts.format_csv {
let delim = opts.csv_delimiter.unwrap_or(',') as u8;
let quote = opts.csv_quote.unwrap_or('"') as u8;
while let Some(end) = spg_engine::copy::csv_record_end(buf, delim, quote) {
let record: Vec<u8> = buf.drain(..end).collect();
let mut rec = &record[..record.len() - 1];
if rec.last() == Some(&b'\r') {
rec = &rec[..rec.len() - 1];
}
if rec == b"\\." {
return Ok(());
}
if rec.is_empty() {
continue;
}
let row_text =
std::str::from_utf8(rec).map_err(|_| "COPY row not valid UTF-8".to_string())?;
if *skipped < opts.skip {
*skipped += 1;
continue;
}
let values = spg_engine::copy::decode_copy_csv_record(
row_text,
delim as char,
quote as char,
opts.null_string.as_deref().unwrap_or(""),
);
if let Err(msg) = copy_row_arity(&values, expected_names) {
if opts.on_error_set_null {
continue;
}
return Err(msg);
}
let sql = spg_engine::copy::build_copy_insert(table, column_list, &values);
{
let mut engine = state
.engine
.write()
.map_err(|_| "engine rwlock poisoned".to_string())?;
match engine.execute_in(&sql, tx_id) {
Ok(_) => *inserted += 1,
Err(e) => {
if opts.on_error_set_null {
continue;
}
return Err(format!("{e}"));
}
}
}
crate::append_wal(state, &sql, false).map_err(|e| format!("{e}"))?;
}
return Ok(());
}
while let Some(nl) = buf.iter().position(|&b| b == b'\n') {
let line: Vec<u8> = buf.drain(..=nl).collect();
let line = &line[..line.len() - 1]; if line == b"\\." {
return Ok(());
}
if line.is_empty() {
continue;
}
let row_text =
std::str::from_utf8(line).map_err(|_| "COPY row not valid UTF-8".to_string())?;
if *skipped < opts.skip {
*skipped += 1;
continue;
}
let sql = if opts.format_json {
match build_copy_insert_from_json(state, table, row_text, opts.on_error_set_null) {
Ok(s) => s,
Err(e) => {
if opts.on_error_set_null {
continue;
}
return Err(format!("COPY FORMAT JSON: {e}"));
}
}
} else {
let values = decode_copy_text_row(row_text);
if let Err(msg) = copy_row_arity(&values, expected_names) {
if opts.on_error_set_null {
continue;
}
return Err(msg);
}
spg_engine::copy::build_copy_insert(table, column_list, &values)
};
{
let mut engine = state
.engine
.write()
.map_err(|_| "engine rwlock poisoned".to_string())?;
match engine.execute_in(&sql, tx_id) {
Ok(_) => *inserted += 1,
Err(e) => {
if opts.on_error_set_null {
continue;
}
return Err(format!("{e}"));
}
}
}
crate::append_wal(state, &sql, false).map_err(|e| format!("{e}"))?;
}
Ok(())
}
fn build_copy_insert_from_json(
state: &Arc<ServerState>,
table: &str,
line: &str,
_on_error: bool,
) -> Result<String, String> {
let cols: Vec<String> = state
.engine
.read()
.ok()
.and_then(|e| {
e.catalog()
.get(table)
.map(|t| t.schema().columns.iter().map(|c| c.name.clone()).collect())
})
.ok_or_else(|| format!("relation {table:?} does not exist"))?;
let pairs = parse_json_object_top_level(line)?;
let mut sql = format!("INSERT INTO {table} (");
for (i, c) in cols.iter().enumerate() {
if i > 0 {
sql.push(',');
}
sql.push_str(c);
}
sql.push_str(") VALUES (");
for (i, c) in cols.iter().enumerate() {
if i > 0 {
sql.push(',');
}
let val = pairs.iter().find(|(k, _)| k == c).map(|(_, v)| v.clone());
match val {
None => sql.push_str("NULL"),
Some(v) => sql.push_str(&v),
}
}
sql.push(')');
Ok(sql)
}
fn parse_json_object_top_level(s: &str) -> Result<Vec<(String, String)>, String> {
let trimmed = s.trim();
let body = trimmed
.strip_prefix('{')
.and_then(|s| s.strip_suffix('}'))
.ok_or_else(|| "expected JSON object {...}".to_string())?;
let mut out = Vec::new();
let mut chars = body.chars().peekable();
while chars.peek().is_some() {
skip_ws(&mut chars);
if chars.peek().is_none() {
break;
}
let key = read_json_string(&mut chars)?;
skip_ws(&mut chars);
if chars.next() != Some(':') {
return Err("expected ':' after key".into());
}
skip_ws(&mut chars);
let val_sql = read_json_value_as_sql(&mut chars)?;
out.push((key, val_sql));
skip_ws(&mut chars);
if chars.peek() == Some(&',') {
chars.next();
}
}
Ok(out)
}
fn skip_ws(chars: &mut std::iter::Peekable<std::str::Chars>) {
while let Some(&c) = chars.peek() {
if c.is_whitespace() {
chars.next();
} else {
break;
}
}
}
fn read_json_string(chars: &mut std::iter::Peekable<std::str::Chars>) -> Result<String, String> {
if chars.next() != Some('"') {
return Err("expected '\"' to start string".into());
}
let mut out = String::new();
loop {
match chars.next() {
None => return Err("unterminated JSON string".into()),
Some('"') => return Ok(out),
Some('\\') => {
let n = chars.next().ok_or("trailing escape")?;
out.push(match n {
'"' => '"',
'\\' => '\\',
'/' => '/',
'b' => '\u{08}',
'f' => '\u{0c}',
'n' => '\n',
'r' => '\r',
't' => '\t',
other => other,
});
}
Some(c) => out.push(c),
}
}
}
fn read_json_value_as_sql(
chars: &mut std::iter::Peekable<std::str::Chars>,
) -> Result<String, String> {
skip_ws(chars);
let Some(&first) = chars.peek() else {
return Err("expected value".into());
};
match first {
'"' => {
let s = read_json_string(chars)?;
Ok(format!("'{}'", s.replace('\'', "''")))
}
't' | 'f' => {
let mut s = String::new();
while let Some(&c) = chars.peek() {
if c.is_ascii_alphabetic() {
s.push(c);
chars.next();
} else {
break;
}
}
if s == "true" {
Ok("TRUE".to_string())
} else if s == "false" {
Ok("FALSE".to_string())
} else {
Err(format!("invalid bool token: {s}"))
}
}
'n' => {
for expected in ['n', 'u', 'l', 'l'] {
if chars.next() != Some(expected) {
return Err("invalid null token".into());
}
}
Ok("NULL".to_string())
}
c if c == '-' || c.is_ascii_digit() => {
let mut s = String::new();
while let Some(&c) = chars.peek() {
if c == '-' || c == '+' || c == '.' || c == 'e' || c == 'E' || c.is_ascii_digit() {
s.push(c);
chars.next();
} else {
break;
}
}
Ok(s)
}
other => Err(format!("unsupported JSON value start: {other:?}")),
}
}
fn decode_copy_text_row(line: &str) -> Vec<Option<String>> {
spg_engine::copy::decode_copy_text_row(line)
}
fn build_copy_insert(table: &str, values: &[Option<String>]) -> String {
spg_engine::copy::build_copy_insert(table, None, values)
}
fn copy_row_arity(values: &[Option<String>], expected_names: &[String]) -> Result<(), String> {
if values.len() < expected_names.len() {
let missing = &expected_names[values.len()];
return Err(format!("missing data for column \"{missing}\""));
}
if values.len() > expected_names.len() {
return Err("extra data after last expected column".to_string());
}
Ok(())
}
fn handle_copy_to_stdout(
stream: &mut dyn ReadWrite,
state: &Arc<ServerState>,
role: Role,
sql: &str,
opts: &CopyOptions,
tx_state: &mut u8,
tx_id: spg_engine::TxId,
) -> std::io::Result<()> {
let _ = role.can_read(); let result = execute_with_role(
state,
sql,
role,
CancelToken::none(),
matches!(*tx_state, b'T' | b'E'),
tx_id,
&std::collections::HashMap::new(),
);
let (columns, rows) = match result {
Ok(QueryResult::Rows { columns, rows }) => (columns, rows),
Ok(QueryResult::CommandOk { .. }) => {
send_error(stream, "42000", "COPY TO source produced no rows")?;
return Ok(());
}
Err(e) => {
send_error(stream, "42000", &e.to_string())?;
return Ok(());
}
Ok(_) => {
send_error(stream, "XX000", "unexpected QueryResult variant")?;
return Ok(());
}
};
let col_count = columns.len();
let mut body = Vec::with_capacity(3 + col_count * 2);
body.push(0);
body.extend_from_slice(&u16::try_from(col_count).unwrap_or(0).to_be_bytes());
for _ in 0..col_count {
body.extend_from_slice(&0u16.to_be_bytes());
}
send_msg(stream, b'H', &body)?;
let n = rows.len();
let (wire_style, wire_tz) = state
.engine
.read()
.map(|e| (e.render_style(), e.session_tz()))
.unwrap_or((Default::default(), spg_engine::SessionTz::Utc));
let is_csv = opts.format_csv;
let delimiter = opts
.csv_delimiter
.unwrap_or(if is_csv { ',' } else { '\t' });
let quote = opts.csv_quote.unwrap_or('"');
let null_str = opts.null_string.clone().unwrap_or_else(|| {
if is_csv {
String::new()
} else {
"\\N".to_string()
}
});
let encode_line = |cells: &[Option<String>]| -> String {
if is_csv {
spg_engine::copy::encode_copy_csv_cells(cells, delimiter, quote, &null_str)
} else {
spg_engine::copy::encode_copy_text_cells_opts(cells, delimiter, &null_str)
}
};
let mut send_line =
|stream: &mut dyn ReadWrite, cells: &[Option<String>]| -> std::io::Result<()> {
let mut line = encode_line(cells);
line.push('\n');
send_msg(stream, b'd', line.as_bytes())
};
if opts.header {
let names: Vec<Option<String>> = columns.iter().map(|c| Some(c.name.clone())).collect();
send_line(stream, &names)?;
}
for row in &rows {
let cells: Vec<Option<String>> = row
.values
.iter()
.enumerate()
.map(|(i, v)| copy_cell_raw(v, columns.get(i).map(|c| c.ty), &wire_style, &wire_tz))
.collect();
send_line(stream, &cells)?;
}
send_msg(stream, b'c', &[])?; send_command_complete(stream, &format!("COPY {n}"))?;
let _ = tx_state;
Ok(())
}
fn copy_cell_raw(
v: &spg_storage::Value,
ty: Option<spg_storage::DataType>,
style: &spg_engine::eval::RenderStyle,
tz: &spg_engine::SessionTz,
) -> Option<String> {
use spg_storage::Value;
let s = match v {
Value::Null => return None,
Value::Bool(b) => if *b { "t" } else { "f" }.to_string(),
Value::SmallInt(n) => n.to_string(),
Value::Int(n) => n.to_string(),
Value::BigInt(n) => n.to_string(),
Value::Float(x) => spg_engine::eval::format_float_styled(*x, style),
Value::Real(x) => spg_engine::eval::format_real_styled(*x, style),
Value::Text(s) | Value::Json(s) => s.to_string(),
Value::BpChar(s) => s.to_string(),
Value::TsVector(lexs) => spg_engine::eval::format_tsvector(lexs),
Value::TsQuery(ast) => spg_engine::eval::format_tsquery(ast),
Value::Numeric {
scaled,
scale,
kind,
} => spg_engine::eval::format_numeric_kind(*kind, *scaled, *scale),
Value::Date(d) => spg_engine::eval::format_date_styled(*d, style),
Value::Timestamp(t) => {
if matches!(ty, Some(DataType::Timestamptz)) {
let abbr = tz.abbrev_at(*t);
spg_engine::eval::format_timestamptz_tz(
*t,
style,
tz.offset_at(*t),
abbr.as_deref(),
)
} else {
spg_engine::eval::format_timestamp_styled(*t, style)
}
}
Value::Interval {
months,
days,
micros,
} => spg_engine::eval::format_interval_styled(*months, *days, *micros, style),
Value::Vector(v) => {
let parts: Vec<String> = v.iter().map(std::string::ToString::to_string).collect();
format!("[{}]", parts.join(","))
}
Value::Sq8Vector(q) => {
let parts: Vec<String> = spg_storage::quantize::dequantize(q)
.iter()
.map(std::string::ToString::to_string)
.collect();
format!("[{}]", parts.join(","))
}
Value::HalfVector(h) => {
let parts: Vec<String> = h
.to_f32_vec()
.iter()
.map(std::string::ToString::to_string)
.collect();
format!("[{}]", parts.join(","))
}
other => spg_engine::eval::value_to_text(other),
};
Some(s)
}
fn cleartext_auth(
stream: &mut dyn ReadWrite,
state: &Arc<ServerState>,
user: &str,
) -> std::io::Result<Option<Role>> {
send_msg(stream, b'R', &3u32.to_be_bytes())?;
let pwd = read_password_message(stream)?;
let verified = state
.engine
.read()
.ok()
.and_then(|e| e.verify_user(user, &pwd));
if let Some(r) = verified {
Ok(Some(r))
} else {
send_error(stream, "28P01", "password authentication failed")?;
Ok(None)
}
}
fn scram_auth(
stream: &mut dyn ReadWrite,
state: &Arc<ServerState>,
user: &str,
secure: bool,
) -> std::io::Result<Option<Role>> {
let cbind_hash = if secure {
crate::mysqlwire::tls_channel_binding_hash()
} else {
None
};
let advertise_plus = cbind_hash.is_some();
let mut sasl_body = Vec::new();
sasl_body.extend_from_slice(&10u32.to_be_bytes());
if advertise_plus {
sasl_body.extend_from_slice(b"SCRAM-SHA-256-PLUS\0SCRAM-SHA-256\0\0");
} else {
sasl_body.extend_from_slice(b"SCRAM-SHA-256\0\0");
}
send_msg(stream, b'R', &sasl_body)?;
let mut header = [0u8; 5];
stream.read_exact(&mut header)?;
if header[0] != b'p' {
send_error(stream, "28000", "expected SASLInitialResponse")?;
return Ok(None);
}
let len = u32::from_be_bytes([header[1], header[2], header[3], header[4]]) as usize;
let mut body = vec![0u8; len.saturating_sub(4)];
stream.read_exact(&mut body)?;
let Some(mech_end) = body.iter().position(|&b| b == 0) else {
send_error(
stream,
"28000",
"SASLInitial: mechanism name not null-terminated",
)?;
return Ok(None);
};
let mech = std::str::from_utf8(&body[..mech_end]).unwrap_or("");
let using_plus = match mech {
"SCRAM-SHA-256-PLUS" => {
if !advertise_plus {
send_error(
stream,
"28000",
"SCRAM-SHA-256-PLUS not available on this connection",
)?;
return Ok(None);
}
true
}
"SCRAM-SHA-256" => false,
_ => {
send_error(
stream,
"28000",
&format!("only SCRAM-SHA-256[-PLUS] is supported, got {mech:?}"),
)?;
return Ok(None);
}
};
let mut cur = mech_end + 1;
if cur + 4 > body.len() {
send_error(stream, "28000", "SASLInitial: missing client-first length")?;
return Ok(None);
}
let cf_len =
u32::from_be_bytes([body[cur], body[cur + 1], body[cur + 2], body[cur + 3]]) as usize;
cur += 4;
if cur + cf_len > body.len() {
send_error(stream, "28000", "SASLInitial: client-first truncated")?;
return Ok(None);
}
let Ok(client_first_msg) = std::str::from_utf8(&body[cur..cur + cf_len]).map(str::to_string)
else {
send_error(stream, "28000", "SASLInitial: client-first not UTF-8")?;
return Ok(None);
};
let client_first = match crate::scram::parse_client_first(&client_first_msg) {
Ok(c) => c,
Err(e) => {
send_error(stream, "28000", &e.to_string())?;
return Ok(None);
}
};
use crate::scram::Gs2CbindFlag;
match (&client_first.cbind_flag, using_plus) {
(Gs2CbindFlag::Required, true) => {}
(Gs2CbindFlag::Required, false) => {
send_error(
stream,
"28000",
"SCRAM: channel-binding flag set on a non-PLUS mechanism",
)?;
return Ok(None);
}
(Gs2CbindFlag::NotSupported | Gs2CbindFlag::SupportedNotUsed, true) => {
send_error(
stream,
"28000",
"SCRAM-SHA-256-PLUS requires the p=tls-server-end-point flag",
)?;
return Ok(None);
}
(Gs2CbindFlag::SupportedNotUsed, false) => {
if advertise_plus {
send_error(stream, "28000", "SCRAM: channel-binding downgrade detected")?;
return Ok(None);
}
}
(Gs2CbindFlag::NotSupported, false) => {}
}
let cbind_data: &[u8] = if using_plus {
cbind_hash.as_ref().map(|h| h.as_slice()).unwrap_or(&[])
} else {
&[]
};
let secrets = state
.engine
.read()
.ok()
.and_then(|e| {
e.users()
.iter()
.find(|(n, _)| *n == user)
.map(|(_, r)| r.scram().cloned())
})
.flatten();
let Some(secrets) = secrets else {
send_error(stream, "28P01", "user has no SCRAM verifier on file")?;
return Ok(None);
};
let server_nonce = match random_nonce_b64(18) {
Ok(n) => n,
Err(e) => {
send_error(stream, "58000", &format!("RNG failure: {e}"))?;
return Ok(None);
}
};
let combined_nonce = format!("{}{}", client_first.client_nonce, server_nonce);
let server_first = crate::scram::build_server_first(&combined_nonce, &secrets);
let mut cont_body = Vec::new();
cont_body.extend_from_slice(&11u32.to_be_bytes());
cont_body.extend_from_slice(server_first.as_bytes());
send_msg(stream, b'R', &cont_body)?;
let mut header = [0u8; 5];
stream.read_exact(&mut header)?;
if header[0] != b'p' {
send_error(stream, "28000", "expected SASLResponse")?;
return Ok(None);
}
let len = u32::from_be_bytes([header[1], header[2], header[3], header[4]]) as usize;
let mut body = vec![0u8; len.saturating_sub(4)];
stream.read_exact(&mut body)?;
let Ok(client_final_msg) = std::str::from_utf8(&body).map(str::to_string) else {
send_error(stream, "28000", "SASLResponse: client-final not UTF-8")?;
return Ok(None);
};
let client_final = match crate::scram::parse_client_final(&client_final_msg) {
Ok(f) => f,
Err(e) => {
send_error(stream, "28000", &e.to_string())?;
return Ok(None);
}
};
let expected_c = crate::scram::channel_binding_c_value(&client_first.gs2_header, cbind_data);
if client_final.channel_binding != expected_c {
send_error(stream, "28000", "SCRAM: channel binding mismatch")?;
return Ok(None);
}
if client_final.combined_nonce != combined_nonce {
send_error(stream, "28000", "SCRAM: nonce mismatch")?;
return Ok(None);
}
let server_signature = match crate::scram::verify_and_sign(
&secrets,
&client_first.bare,
&server_first,
&client_final.without_proof,
&client_final.client_proof,
) {
Ok(s) => s,
Err(e) => {
send_error(stream, "28P01", &e.to_string())?;
return Ok(None);
}
};
let mut final_body = Vec::new();
final_body.extend_from_slice(&12u32.to_be_bytes());
final_body.extend_from_slice(server_signature.as_bytes());
send_msg(stream, b'R', &final_body)?;
let role = state.engine.read().ok().and_then(|e| {
e.users()
.iter()
.find(|(n, _)| *n == user)
.map(|(_, r)| r.role)
});
Ok(role)
}
fn random_nonce_b64(byte_len: usize) -> std::io::Result<String> {
let mut buf = vec![0u8; byte_len];
std::fs::File::open("/dev/urandom")?.read_exact(&mut buf)?;
Ok(spg_crypto::base64::encode(&buf))
}
fn read_startup(stream: &mut dyn ReadWrite) -> std::io::Result<(String, Vec<(String, String)>)> {
loop {
let mut len_bytes = [0u8; 4];
stream.read_exact(&mut len_bytes)?;
let total = u32::from_be_bytes(len_bytes) as usize;
if total < 8 {
return Err(std::io::Error::other("startup message too short"));
}
let mut body = vec![0u8; total - 4];
stream.read_exact(&mut body)?;
let proto = u32::from_be_bytes([body[0], body[1], body[2], body[3]]);
if proto == 80877103 {
stream.write_all(b"N")?;
continue;
}
if proto == 80877104 {
stream.write_all(b"N")?;
continue;
}
if proto != PROTOCOL_V3 {
return Err(std::io::Error::other(format!(
"unsupported protocol version: {proto}"
)));
}
let mut params = Vec::new();
let mut user = String::new();
let mut p = 4;
while p < body.len() {
let k_end = body[p..]
.iter()
.position(|&b| b == 0)
.ok_or_else(|| std::io::Error::other("startup key not null-terminated"))?;
let key = std::str::from_utf8(&body[p..p + k_end])
.map_err(|_| std::io::Error::other("startup key not UTF-8"))?
.to_string();
p += k_end + 1;
if key.is_empty() {
break;
}
let v_end = body[p..]
.iter()
.position(|&b| b == 0)
.ok_or_else(|| std::io::Error::other("startup value not null-terminated"))?;
let value = std::str::from_utf8(&body[p..p + v_end])
.map_err(|_| std::io::Error::other("startup value not UTF-8"))?
.to_string();
p += v_end + 1;
if key == "user" {
user = value.clone();
}
params.push((key, value));
}
return Ok((user, params));
}
}
fn read_password_message(stream: &mut dyn ReadWrite) -> std::io::Result<String> {
let mut header = [0u8; 5];
stream.read_exact(&mut header)?;
if header[0] != b'p' {
return Err(std::io::Error::other("expected PasswordMessage"));
}
let len = u32::from_be_bytes([header[1], header[2], header[3], header[4]]) as usize;
let body_len = len.saturating_sub(4);
let mut body = vec![0u8; body_len];
stream.read_exact(&mut body)?;
let pw = body.strip_suffix(b"\0").unwrap_or(&body);
std::str::from_utf8(pw)
.map(str::to_string)
.map_err(|_| std::io::Error::other("password not UTF-8"))
}
fn send_msg(stream: &mut dyn Write, ty: u8, body: &[u8]) -> std::io::Result<()> {
let len = u32::try_from(body.len() + 4)
.map_err(|_| std::io::Error::other("PG message body too large"))?;
let mut out = Vec::with_capacity(5 + body.len());
out.push(ty);
out.extend_from_slice(&len.to_be_bytes());
out.extend_from_slice(body);
stream.write_all(&out)
}
fn send_parameter_status(stream: &mut dyn Write, key: &str, value: &str) -> std::io::Result<()> {
let mut body = Vec::with_capacity(key.len() + value.len() + 2);
body.extend_from_slice(key.as_bytes());
body.push(0);
body.extend_from_slice(value.as_bytes());
body.push(0);
send_msg(stream, b'S', &body)
}
fn send_ready_for_query(stream: &mut dyn Write, state: u8) -> std::io::Result<()> {
send_msg(stream, b'Z', &[state])
}
fn encode_select_int_response(out: &mut Vec<u8>, n: i64, tx_state: u8) -> std::io::Result<()> {
#[rustfmt::skip]
const ROW_DESC_INT8: [u8; 34] = [
b'T',
0, 0, 0, 33, 0, 1, b'?', b'c', b'o', b'l', b'u', b'm', b'n', b'?', 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 20, 0, 8, 255, 255, 255, 255, 0, 0, ];
#[rustfmt::skip]
const ROW_DESC_INT4: [u8; 34] = [
b'T',
0, 0, 0, 33,
0, 1,
b'?', b'c', b'o', b'l', b'u', b'm', b'n', b'?', 0,
0, 0, 0, 0,
0, 0,
0, 0, 0, 23, 0, 4, 255, 255, 255, 255,
0, 0,
];
out.extend_from_slice(if i32::try_from(n).is_ok() {
&ROW_DESC_INT4
} else {
&ROW_DESC_INT8
});
let mut digits = [0u8; 24];
let mut pos = digits.len();
let (mut x, negative) = if n < 0 {
((n as i128).unsigned_abs() as u64, true)
} else {
(n as u64, false)
};
loop {
pos -= 1;
digits[pos] = b'0' + (x % 10) as u8;
x /= 10;
if x == 0 {
break;
}
}
if negative {
pos -= 1;
digits[pos] = b'-';
}
let int_text = &digits[pos..];
let cell_len = int_text.len() as u32;
let frame_len = 4 + 2 + 4 + cell_len; out.push(b'D');
out.extend_from_slice(&frame_len.to_be_bytes());
out.extend_from_slice(&1u16.to_be_bytes()); out.extend_from_slice(&cell_len.to_be_bytes());
out.extend_from_slice(int_text);
#[rustfmt::skip]
const COMPLETE_FRAME: [u8; 14] = [
b'C',
0, 0, 0, 13,
b'S', b'E', b'L', b'E', b'C', b'T', b' ', b'1', 0,
];
out.extend_from_slice(&COMPLETE_FRAME);
out.push(b'Z');
out.extend_from_slice(&5u32.to_be_bytes());
out.push(tx_state);
Ok(())
}
fn send_command_complete(stream: &mut dyn Write, tag: &str) -> std::io::Result<()> {
let mut body = Vec::with_capacity(tag.len() + 1);
body.extend_from_slice(tag.as_bytes());
body.push(0);
send_msg(stream, b'C', &body)
}
fn send_command_complete_select_count(out: &mut Vec<u8>, n: usize) -> std::io::Result<()> {
let mut digits = [0u8; 24];
let mut pos = digits.len();
let mut x = n as u64;
loop {
pos -= 1;
digits[pos] = b'0' + (x % 10) as u8;
x /= 10;
if x == 0 {
break;
}
}
let int_text = &digits[pos..];
let body_len = 7 + int_text.len() + 1;
let frame_len = u32::try_from(4 + body_len)
.map_err(|_| std::io::Error::other("PG message body too large"))?;
out.reserve(1 + 4 + body_len);
out.push(b'C');
out.extend_from_slice(&frame_len.to_be_bytes());
out.extend_from_slice(b"SELECT ");
out.extend_from_slice(int_text);
out.push(0);
Ok(())
}
fn drain_notices(state: &ServerState, wbuf: &mut Vec<u8>) -> std::io::Result<()> {
let (notices, warning_reaches) = match state.engine.write() {
Ok(mut e) => {
let kept: Vec<_> = e
.take_notices()
.into_iter()
.filter(|n| e.notice_severity_reaches_client(n.severity))
.collect();
(
kept,
e.notice_severity_reaches_client(spg_engine::NoticeSeverity::Warning),
)
}
Err(_) => return Ok(()),
};
for n in ¬ices {
send_notice(wbuf, n.severity, &n.message)?;
}
for w in crate::take_host_warnings() {
if warning_reaches {
send_notice(wbuf, spg_engine::NoticeSeverity::Warning, &w)?;
}
}
Ok(())
}
fn drain_notifications(
state: &ServerState,
wbuf: &mut Vec<u8>,
conn_state: &crate::ConnState,
) -> std::io::Result<()> {
let notifies = match state.engine.write() {
Ok(mut e) => e.take_notifications(),
Err(_) => return Ok(()),
};
if !notifies.is_empty()
&& let Ok(conns) = state.connections.read()
{
for c in conns.iter() {
if let Ok(mut q) = c.notify_queue.lock() {
q.extend(notifies.iter().cloned());
}
}
}
let mine: Vec<(String, String)> = match conn_state.notify_queue.lock() {
Ok(mut q) => core::mem::take(&mut *q),
Err(_) => Vec::new(),
};
for (channel, payload) in &mine {
let mut body = Vec::with_capacity(channel.len() + payload.len() + 8);
body.extend_from_slice(&conn_state.pid.to_be_bytes());
body.extend_from_slice(channel.as_bytes());
body.push(0);
body.extend_from_slice(payload.as_bytes());
body.push(0);
send_msg(wbuf, b'A', &body)?;
}
Ok(())
}
fn send_notice(
stream: &mut dyn Write,
severity: spg_engine::NoticeSeverity,
msg: &str,
) -> std::io::Result<()> {
let mut body = Vec::new();
body.push(b'S');
body.extend_from_slice(severity.as_pg_str().as_bytes());
body.push(0);
body.push(b'V');
body.extend_from_slice(severity.as_pg_str().as_bytes());
body.push(0);
body.push(b'C');
body.extend_from_slice(b"00000");
body.push(0);
body.push(b'M');
body.extend_from_slice(msg.as_bytes());
body.push(0);
body.push(0);
stream.write_all(b"N")?;
stream.write_all(
&u32::try_from(body.len() + 4)
.unwrap_or(u32::MAX)
.to_be_bytes(),
)?;
stream.write_all(&body)
}
fn terminated(conn_state: &crate::ConnState) -> bool {
conn_state
.terminate
.load(std::sync::atomic::Ordering::Relaxed)
}
fn send_fatal_terminated(stream: &mut dyn Write) -> std::io::Result<()> {
let mut body = Vec::new();
for (code, val) in [
(b'S', "FATAL"),
(b'V', "FATAL"),
(b'C', "57P01"),
(b'M', "terminating connection due to administrator command"),
] {
body.push(code);
body.extend_from_slice(val.as_bytes());
body.push(0);
}
body.push(0);
stream.write_all(b"E")?;
stream.write_all(
&u32::try_from(body.len() + 4)
.unwrap_or(u32::MAX)
.to_be_bytes(),
)?;
stream.write_all(&body)
}
fn send_error(stream: &mut dyn Write, sqlstate: &str, msg: &str) -> std::io::Result<()> {
send_error_pos(stream, sqlstate, msg, None)
}
fn send_error_pos(
stream: &mut dyn Write,
sqlstate: &str,
msg: &str,
position: Option<usize>,
) -> std::io::Result<()> {
let mut body = Vec::new();
body.push(b'S');
body.extend_from_slice(b"ERROR");
body.push(0);
body.push(b'C');
body.extend_from_slice(sqlstate.as_bytes());
body.push(0);
let (msg, hint) = match msg.split_once("\nHINT: ") {
Some((m, h)) => (m, Some(h)),
None => (msg, None),
};
let (main, detail) = match msg.split_once(" DETAIL: ") {
Some((m, d)) => (m, Some(d)),
None => (msg, None),
};
let main_msg: &str = if sqlstate == "42000" {
crate::strip_layer_prefixes_keeping_unsupported(main)
} else {
crate::strip_internal_error_prefixes(main)
};
let main_msg: &str = match (sqlstate, main_msg.find(" on table \"")) {
("23505" | "23P01", Some(cut)) => &main_msg[..cut],
_ => main_msg,
};
body.push(b'M');
body.extend_from_slice(main_msg.as_bytes());
body.push(0);
if let Some(d) = detail {
body.push(b'D');
body.extend_from_slice(d.as_bytes());
body.push(0);
}
if let Some(h) = hint {
body.push(b'H');
body.extend_from_slice(h.as_bytes());
body.push(0);
}
let quoted_after = |marker: &str| -> Option<&str> {
let rest = &main[main.find(marker)? + marker.len()..];
rest.strip_prefix('"')?.split('"').next()
};
if let Some(con) = quoted_after("violates unique constraint ")
.or_else(|| quoted_after("violates foreign key constraint "))
.or_else(|| quoted_after("violates check constraint "))
.or_else(|| quoted_after("violates exclusion constraint "))
{
body.push(b'n');
body.extend_from_slice(con.as_bytes());
body.push(0);
}
if let Some(t) = quoted_after("on table ").or_else(|| quoted_after("of relation ")) {
body.push(b't');
body.extend_from_slice(t.as_bytes());
body.push(0);
body.push(b's');
body.extend_from_slice(b"public");
body.push(0);
}
if let Some(c) = quoted_after("null value in column ") {
body.push(b'c');
body.extend_from_slice(c.as_bytes());
body.push(0);
}
if let Some(p) = position {
body.push(b'P');
body.extend_from_slice(p.to_string().as_bytes());
body.push(0);
}
body.push(0);
send_msg(stream, b'E', &body)
}
fn send_row_description(stream: &mut dyn Write, cols: &[ColumnSchema]) -> std::io::Result<()> {
let body = encode_row_description_body(cols);
send_msg(stream, b'T', &body)
}
fn send_row_description_cached(out: &mut Vec<u8>, body: &[u8]) -> std::io::Result<()> {
send_msg(out, b'T', body)
}
fn encode_row_description_body(cols: &[ColumnSchema]) -> Vec<u8> {
let n = u16::try_from(cols.len()).unwrap_or(u16::MAX);
let mut body = Vec::with_capacity(2 + cols.len() * 24);
body.extend_from_slice(&n.to_be_bytes());
for c in cols {
body.extend_from_slice(c.name.as_bytes());
body.push(0);
body.extend_from_slice(&0u32.to_be_bytes()); body.extend_from_slice(&0u16.to_be_bytes()); body.extend_from_slice(&pg_type_oid(c.ty).to_be_bytes()); body.extend_from_slice(&pg_type_len(c.ty).to_be_bytes()); body.extend_from_slice(&(-1i32).to_be_bytes()); body.extend_from_slice(&0u16.to_be_bytes()); }
body
}
const PG_EPOCH_DAYS: i32 = 10_957;
const PG_EPOCH_MICROS: i64 = 946_684_800_000_000;
fn encode_binary_cell(out: &mut Vec<u8>, v: &Value, ty: DataType) -> Result<(), String> {
let mut put = |payload: &[u8]| {
out.extend_from_slice(&(payload.len() as i32).to_be_bytes());
out.extend_from_slice(payload);
};
match v {
Value::Null => out.extend_from_slice(&(-1i32).to_be_bytes()),
Value::Bool(b) => put(&[u8::from(*b)]),
Value::SmallInt(n) => put(&n.to_be_bytes()),
Value::Int(n) => put(&n.to_be_bytes()),
Value::BigInt(n) => put(&n.to_be_bytes()),
Value::Real(x) => put(&x.to_be_bytes()),
Value::Float(x) => put(&x.to_be_bytes()),
Value::Text(s) | Value::BpChar(s) => put(s.as_bytes()),
Value::Json(s) => {
if matches!(ty, DataType::Jsonb) {
let mut buf = Vec::with_capacity(s.len() + 1);
buf.push(1);
buf.extend_from_slice(s.as_bytes());
put(&buf);
} else {
put(s.as_bytes());
}
}
Value::Xml(s) => put(s.as_bytes()),
Value::Bytes(b) => put(b),
Value::Uuid(u) => put(&u[..]),
Value::Date(days) => put(&(days - PG_EPOCH_DAYS).to_be_bytes()),
Value::Timestamp(us) => put(&(us - PG_EPOCH_MICROS).to_be_bytes()),
Value::Time(us) => put(&us.to_be_bytes()),
Value::Interval {
months,
days,
micros,
} => {
let mut buf = Vec::with_capacity(16);
buf.extend_from_slice(µs.to_be_bytes());
buf.extend_from_slice(&days.to_be_bytes());
buf.extend_from_slice(&months.to_be_bytes());
put(&buf);
}
Value::Numeric { scaled, scale, .. } => {
put(&numeric_binary(*scaled, *scale));
}
Value::NumericBig(b) => {
let (scaled, scale) = decimal_str_to_scaled(&b.to_decimal_str())
.ok_or("binary numeric: value out of range")?;
put(&numeric_binary(scaled, scale));
}
Value::IntArray(items) => put(&binary_array(items, 23, |v, b| {
b.extend_from_slice(&v.to_be_bytes());
})),
Value::BigIntArray(items) => put(&binary_array(items, 20, |v, b| {
b.extend_from_slice(&v.to_be_bytes());
})),
Value::SmallIntArray(items) => put(&binary_array(items, 21, |v, b| {
b.extend_from_slice(&v.to_be_bytes());
})),
Value::FloatArray(items) => put(&binary_array(items, 701, |v, b| {
b.extend_from_slice(&v.to_be_bytes());
})),
Value::BoolArray(items) => put(&binary_array(items, 16, |v, b| {
b.push(u8::from(*v));
})),
Value::TextArray(items) => put(&binary_array(items, 25, |v, b| {
b.extend_from_slice(v.as_bytes());
})),
Value::UuidArray(items) => put(&binary_array(items, 2950, |v, b| {
b.extend_from_slice(&v[..]);
})),
other => {
return Err(format!(
"binary result format not implemented for {:?}",
other.data_type()
));
}
}
Ok(())
}
fn binary_array<T>(items: &[Option<T>], elem_oid: u32, enc: impl Fn(&T, &mut Vec<u8>)) -> Vec<u8> {
let has_null = items.iter().any(Option::is_none);
let mut buf = Vec::with_capacity(20 + items.len() * 8);
buf.extend_from_slice(&1i32.to_be_bytes()); buf.extend_from_slice(&i32::from(has_null).to_be_bytes());
buf.extend_from_slice(&elem_oid.to_be_bytes());
buf.extend_from_slice(&(items.len() as i32).to_be_bytes());
buf.extend_from_slice(&1i32.to_be_bytes()); for it in items {
match it {
None => buf.extend_from_slice(&(-1i32).to_be_bytes()),
Some(v) => {
let mut payload = Vec::new();
enc(v, &mut payload);
buf.extend_from_slice(&(payload.len() as i32).to_be_bytes());
buf.extend_from_slice(&payload);
}
}
}
buf
}
fn decimal_str_to_scaled(s: &str) -> Option<(i128, u16)> {
let (int_part, frac_part) = match s.split_once('.') {
Some((i, f)) => (i, f),
None => (s, ""),
};
let digits: String = format!("{int_part}{frac_part}");
let scaled: i128 = digits.parse().ok()?;
Some((scaled, u16::try_from(frac_part.len()).ok()?))
}
fn numeric_binary(scaled: i128, scale: u16) -> Vec<u8> {
let neg = scaled < 0;
let mut abs = scaled.unsigned_abs();
let scale_usize = scale as usize;
let frac_pad = (4 - (scale_usize % 4)) % 4;
for _ in 0..frac_pad {
abs *= 10;
}
let frac_groups = (scale_usize + frac_pad) / 4;
let mut groups: Vec<u16> = Vec::new();
if abs == 0 {
groups.push(0);
}
while abs > 0 {
groups.push((abs % 10_000) as u16);
abs /= 10_000;
}
while groups.len() <= frac_groups {
groups.push(0); }
let mut lo = 0;
while lo < frac_groups && groups[lo] == 0 {
lo += 1;
}
let mut hi = groups.len();
while hi > frac_groups + 1 && groups[hi - 1] == 0 {
hi -= 1;
}
let digits: Vec<u16> = groups[lo..hi].iter().rev().copied().collect();
let weight = (hi - frac_groups) as i32 - 1;
let all_zero = digits.iter().all(|&d| d == 0);
let (digits, weight) = if all_zero {
(Vec::new(), 0)
} else {
(digits, weight)
};
let mut buf = Vec::with_capacity(8 + digits.len() * 2);
buf.extend_from_slice(&(digits.len() as i16).to_be_bytes());
buf.extend_from_slice(&(weight as i16).to_be_bytes());
buf.extend_from_slice(&(if neg { 0x4000u16 } else { 0 }).to_be_bytes());
buf.extend_from_slice(&scale.to_be_bytes());
for d in &digits {
buf.extend_from_slice(&d.to_be_bytes());
}
buf
}
fn encode_data_row_formats(
out: &mut Vec<u8>,
cols: &[ColumnSchema],
row: &Row,
formats: &[i16],
arena: &bumpalo::Bump,
style: &spg_engine::eval::RenderStyle,
tz: &spg_engine::SessionTz,
) -> std::io::Result<()> {
let mut body: Vec<u8> = Vec::with_capacity(cols.len() * 12);
body.extend_from_slice(&(cols.len() as u16).to_be_bytes());
for (i, c) in cols.iter().enumerate() {
let v = row.values.get(i).unwrap_or(&Value::Null);
if col_is_binary(formats, i) {
encode_binary_cell(&mut body, v, c.ty).map_err(std::io::Error::other)?;
} else {
match value_to_pg_text(v, Some(c.ty), arena, style, tz) {
None => body.extend_from_slice(&(-1i32).to_be_bytes()),
Some(s) => {
body.extend_from_slice(&(s.len() as i32).to_be_bytes());
body.extend_from_slice(s.as_bytes());
}
}
}
}
send_msg(out, b'D', &body)
}
fn send_data_row(stream: &mut dyn Write, cols: &[ColumnSchema], row: &Row) -> std::io::Result<()> {
let arena = bumpalo::Bump::new();
let n = u16::try_from(row.values.len())
.map_err(|_| std::io::Error::other("DataRow: too many cells"))?;
let mut body = Vec::with_capacity(2 + row.values.len() * 8);
body.extend_from_slice(&n.to_be_bytes());
let style = spg_engine::eval::RenderStyle::default();
let tz = spg_engine::SessionTz::Utc;
for (i, v) in row.values.iter().enumerate() {
encode_pg_text_cell(&mut body, v, cols.get(i).map(|c| c.ty), &arena, &style, &tz)?;
}
send_msg(stream, b'D', &body)
}
fn encode_data_row_from_refs(
out: &mut Vec<u8>,
cols: &[ColumnSchema],
values: &[&spg_storage::Value<'_>],
arena: &bumpalo::Bump,
style: &spg_engine::eval::RenderStyle,
tz: &spg_engine::SessionTz,
) -> std::io::Result<()> {
let n = u16::try_from(values.len())
.map_err(|_| std::io::Error::other("DataRow: too many cells"))?;
let frame_start = out.len();
out.push(b'D');
out.extend_from_slice(&[0u8; 4]); out.extend_from_slice(&n.to_be_bytes());
for (i, v) in values.iter().enumerate() {
encode_pg_text_cell(out, v, cols.get(i).map(|c| c.ty), arena, style, tz)?;
}
let body_plus_len_field = out.len() - frame_start - 1;
let len = u32::try_from(body_plus_len_field)
.map_err(|_| std::io::Error::other("PG message body too large"))?;
out[frame_start + 1..frame_start + 5].copy_from_slice(&len.to_be_bytes());
Ok(())
}
fn encode_data_row_cells(
out: &mut Vec<u8>,
cols: &[ColumnSchema],
cells: spg_engine::RowCells<'_>,
arena: &bumpalo::Bump,
style: &spg_engine::eval::RenderStyle,
tz: &spg_engine::SessionTz,
) -> std::io::Result<()> {
match cells {
spg_engine::RowCells::Refs(v) => encode_data_row_from_refs(out, cols, v, arena, style, tz),
spg_engine::RowCells::Values(v) => {
encode_data_row_from_values(out, cols, v, arena, style, tz)
}
}
}
fn encode_data_row(
out: &mut Vec<u8>,
cols: &[ColumnSchema],
row: &Row,
arena: &bumpalo::Bump,
style: &spg_engine::eval::RenderStyle,
tz: &spg_engine::SessionTz,
) -> std::io::Result<()> {
encode_data_row_from_values(out, cols, &row.values, arena, style, tz)
}
fn encode_data_row_from_values(
out: &mut Vec<u8>,
cols: &[ColumnSchema],
values: &[Value<'_>],
arena: &bumpalo::Bump,
style: &spg_engine::eval::RenderStyle,
tz: &spg_engine::SessionTz,
) -> std::io::Result<()> {
let n = u16::try_from(values.len())
.map_err(|_| std::io::Error::other("DataRow: too many cells"))?;
let frame_start = out.len();
out.push(b'D');
out.extend_from_slice(&[0u8; 4]); out.extend_from_slice(&n.to_be_bytes());
for (i, v) in values.iter().enumerate() {
encode_pg_text_cell(out, v, cols.get(i).map(|c| c.ty), arena, style, tz)?;
}
let body_plus_len_field = out.len() - frame_start - 1;
let len = u32::try_from(body_plus_len_field)
.map_err(|_| std::io::Error::other("PG message body too large"))?;
out[frame_start + 1..frame_start + 5].copy_from_slice(&len.to_be_bytes());
Ok(())
}
fn encode_pg_text_cell(
out: &mut Vec<u8>,
v: &Value<'_>,
ty: Option<DataType>,
arena: &bumpalo::Bump,
style: &spg_engine::eval::RenderStyle,
tz: &spg_engine::SessionTz,
) -> std::io::Result<()> {
match v {
Value::Null => {
out.extend_from_slice(&(-1i32).to_be_bytes());
return Ok(());
}
Value::Bool(b) => return write_cell_bytes(out, if *b { b"t" } else { b"f" }),
Value::SmallInt(n) => return write_cell_int(out, i64::from(*n)),
Value::Int(n) => return write_cell_int(out, i64::from(*n)),
Value::BigInt(n) => return write_cell_int(out, *n),
Value::Text(s) | Value::Json(s) => return write_cell_bytes(out, s.as_bytes()),
Value::BpChar(s) => return write_cell_bytes(out, s.as_bytes()),
Value::Timestamp(micros)
if style.date_style == spg_engine::eval::DateStyleKind::Iso
&& *micros >= AD_FLOOR_DAYS * 86_400_000_000
&& *micros != i64::MAX
&& (tz.is_utc() || !matches!(ty, Some(DataType::Timestamptz))) =>
{
let with_tz = matches!(ty, Some(DataType::Timestamptz));
return write_cell_timestamp(out, *micros, with_tz);
}
Value::Date(days)
if style.date_style == spg_engine::eval::DateStyleKind::Iso
&& i64::from(*days) >= AD_FLOOR_DAYS
&& *days != i32::MAX =>
{
return write_cell_date(out, *days);
}
_ => {}
}
match value_to_pg_text(v, ty, arena, style, tz) {
None => out.extend_from_slice(&(-1i32).to_be_bytes()),
Some(s) => write_cell_bytes(out, s.as_bytes())?,
}
Ok(())
}
fn write_cell_bytes(out: &mut Vec<u8>, bytes: &[u8]) -> std::io::Result<()> {
let len =
i32::try_from(bytes.len()).map_err(|_| std::io::Error::other("cell value too large"))?;
let total = 4usize + bytes.len();
out.reserve(total);
let len_be = len.to_be_bytes();
#[allow(unsafe_code)]
unsafe {
let head = out.as_mut_ptr().add(out.len());
core::ptr::copy_nonoverlapping(len_be.as_ptr(), head, 4);
core::ptr::copy_nonoverlapping(bytes.as_ptr(), head.add(4), bytes.len());
out.set_len(out.len() + total);
}
Ok(())
}
#[inline]
fn write_pad2(buf: &mut [u8], p: &mut usize, n: u32) {
buf[*p] = b'0' + ((n / 10) % 10) as u8;
buf[*p + 1] = b'0' + (n % 10) as u8;
*p += 2;
}
#[inline]
fn write_pad4(buf: &mut [u8], p: &mut usize, n: u32) {
buf[*p] = b'0' + ((n / 1000) % 10) as u8;
buf[*p + 1] = b'0' + ((n / 100) % 10) as u8;
buf[*p + 2] = b'0' + ((n / 10) % 10) as u8;
buf[*p + 3] = b'0' + (n % 10) as u8;
*p += 4;
}
#[inline]
fn write_pad6(buf: &mut [u8], p: &mut usize, n: u32) {
buf[*p] = b'0' + ((n / 100_000) % 10) as u8;
buf[*p + 1] = b'0' + ((n / 10_000) % 10) as u8;
buf[*p + 2] = b'0' + ((n / 1_000) % 10) as u8;
buf[*p + 3] = b'0' + ((n / 100) % 10) as u8;
buf[*p + 4] = b'0' + ((n / 10) % 10) as u8;
buf[*p + 5] = b'0' + (n % 10) as u8;
*p += 6;
}
const AD_FLOOR_DAYS: i64 = -719_162;
fn write_cell_timestamp(out: &mut Vec<u8>, micros: i64, with_tz: bool) -> std::io::Result<()> {
const MICROS_PER_DAY: i64 = 86_400_000_000;
let days = micros.div_euclid(MICROS_PER_DAY);
let day_micros = micros.rem_euclid(MICROS_PER_DAY);
let secs = day_micros / 1_000_000;
let frac = (day_micros % 1_000_000) as u32;
let (y, m, d, _, _, _) = secs_to_ymdhms(days * 86_400);
if !(0..=9999).contains(&y) {
let s = if with_tz {
spg_engine::eval::format_timestamptz(micros)
} else {
spg_engine::eval::format_timestamp(micros)
};
return write_cell_bytes(out, s.as_bytes());
}
let hh = (secs / 3600) as u32;
let mm = ((secs / 60) % 60) as u32;
let ss = (secs % 60) as u32;
let mut buf = [0u8; 32];
let mut p = 0;
write_pad4(&mut buf, &mut p, y as u32);
buf[p] = b'-';
p += 1;
write_pad2(&mut buf, &mut p, m);
buf[p] = b'-';
p += 1;
write_pad2(&mut buf, &mut p, d);
buf[p] = b' ';
p += 1;
write_pad2(&mut buf, &mut p, hh);
buf[p] = b':';
p += 1;
write_pad2(&mut buf, &mut p, mm);
buf[p] = b':';
p += 1;
write_pad2(&mut buf, &mut p, ss);
if frac != 0 {
buf[p] = b'.';
p += 1;
let frac_start = p;
write_pad6(&mut buf, &mut p, frac);
while p > frac_start && buf[p - 1] == b'0' {
p -= 1;
}
}
if with_tz {
buf[p] = b'+';
p += 1;
buf[p] = b'0';
p += 1;
buf[p] = b'0';
p += 1;
}
write_cell_bytes(out, &buf[..p])
}
fn write_cell_date(out: &mut Vec<u8>, days: i32) -> std::io::Result<()> {
let secs = i64::from(days) * 86_400;
let (y, m, d, _, _, _) = secs_to_ymdhms(secs);
if !(0..=9999).contains(&y) {
return write_cell_bytes(out, format_date(days).as_bytes());
}
let mut buf = [0u8; 10];
let mut p = 0;
write_pad4(&mut buf, &mut p, y as u32);
buf[p] = b'-';
p += 1;
write_pad2(&mut buf, &mut p, m);
buf[p] = b'-';
p += 1;
write_pad2(&mut buf, &mut p, d);
write_cell_bytes(out, &buf[..p])
}
fn write_cell_int(out: &mut Vec<u8>, n: i64) -> std::io::Result<()> {
let mut buf = [0u8; 24];
let mut pos = buf.len();
let (mut x, negative) = if n < 0 {
((n as i128).unsigned_abs() as u64, true)
} else {
(n as u64, false)
};
loop {
pos -= 1;
buf[pos] = b'0' + (x % 10) as u8;
x /= 10;
if x == 0 {
break;
}
}
if negative {
pos -= 1;
buf[pos] = b'-';
}
write_cell_bytes(out, &buf[pos..])
}
const fn pg_type_oid(ty: DataType) -> u32 {
match ty {
DataType::Bool => 16,
DataType::Name => 19,
DataType::Xid => 28,
DataType::Xid8 => 5069,
DataType::Oid => 26,
DataType::SmallInt => 21,
DataType::Int => 23,
DataType::BigInt => 20,
DataType::Float => 701,
DataType::Real => 700,
DataType::Text | DataType::Varchar(_) | DataType::Char(_) | DataType::Vector { .. } => 25,
DataType::Timestamp => 1114,
DataType::Timestamptz => 1184, DataType::Date => 1082,
DataType::Interval => 1186,
DataType::Numeric { .. } => 1700,
DataType::Json => 114, DataType::Jsonb => 3802, DataType::Bytes => 17, DataType::TextArray => 1009, DataType::IntArray => 1007, DataType::BigIntArray => 1016, DataType::OidArray => 1028,
DataType::TsVector => 3614, DataType::TsQuery => 3615, DataType::Uuid => 2950, DataType::Time => 1083, DataType::Year => 23,
DataType::TimeTz => 1266,
DataType::Money => 790,
DataType::Range(k) => match k {
spg_storage::RangeKind::Int4 => 3904,
spg_storage::RangeKind::Int8 => 3926,
spg_storage::RangeKind::Num => 3906,
spg_storage::RangeKind::Ts => 3908,
spg_storage::RangeKind::TsTz => 3910,
spg_storage::RangeKind::Date => 3912,
},
DataType::Hstore => 25,
DataType::IntArray2D => 1007,
DataType::BigIntArray2D => 1016,
DataType::TextArray2D => 1009,
DataType::BoolArray2D => 1000,
DataType::IntervalArray => 1187,
DataType::BoolArray => 1000, DataType::SmallIntArray => 1005, DataType::FloatArray => 1022, DataType::NumericArray => 1231, DataType::DateArray => 1182, DataType::TimestampArray => 1115, DataType::TimestamptzArray => 1185, DataType::UuidArray => 2951, DataType::JsonArray => 199, DataType::JsonbArray => 3807, DataType::BytesArray => 1001, DataType::VarcharArray => 1015, DataType::CharArray => 1014, DataType::Point => 600,
DataType::Lseg => 601,
DataType::Path => 602,
DataType::PgBox => 603,
DataType::Polygon => 604,
DataType::Line => 628,
DataType::Circle => 718,
DataType::Inet => 869,
DataType::Cidr => 650,
DataType::Macaddr => 829,
DataType::Macaddr8 => 774,
DataType::PgLsn => 3220,
DataType::Bit(_) => 1560,
DataType::BitVarying(_) => 1562,
DataType::Xml => 142,
DataType::Char1 => 18,
DataType::MoneyArray => 791,
DataType::Multirange(k) => match k {
spg_storage::RangeKind::Int4 => 4451,
spg_storage::RangeKind::Int8 => 4537,
spg_storage::RangeKind::Num => 4536,
spg_storage::RangeKind::Ts => 4533,
spg_storage::RangeKind::TsTz => 4534,
spg_storage::RangeKind::Date => 4535,
},
}
}
const fn pg_type_len(ty: DataType) -> i16 {
match ty {
DataType::Bool => 1,
DataType::SmallInt => 2,
DataType::Int | DataType::Date | DataType::Real => 4,
DataType::BigInt | DataType::Float | DataType::Timestamp | DataType::Timestamptz => 8,
DataType::Interval => 16,
DataType::Uuid => 16,
DataType::Time => 8,
DataType::Year => 2,
DataType::TimeTz => 12,
DataType::Money => 8,
DataType::Range(_) => -1,
DataType::Hstore => -1,
DataType::IntArray2D | DataType::BigIntArray2D | DataType::TextArray2D => -1,
_ => -1, }
}
fn value_to_pg_text<'a>(
v: &Value<'_>,
ty: Option<DataType>,
arena: &'a bumpalo::Bump,
style: &spg_engine::eval::RenderStyle,
tz: &spg_engine::SessionTz,
) -> Option<&'a str> {
use bumpalo::collections::String as BumpString;
use core::fmt::Write;
let into_arena = |s: &str| -> &'a str { BumpString::from_str_in(s, arena).into_bump_str() };
let display_into_arena = |d: &dyn core::fmt::Display| -> &'a str {
let mut buf = BumpString::new_in(arena);
let _ = write!(&mut buf, "{d}");
buf.into_bump_str()
};
Some(match v {
Value::Null => return None,
Value::Bool(b) => {
if *b {
"t"
} else {
"f"
}
}
Value::SmallInt(n) => display_into_arena(n),
Value::Int(n) => display_into_arena(n),
Value::BigInt(n) => display_into_arena(n),
Value::Float(f) => into_arena(&spg_engine::eval::format_float_styled(*f, style)),
Value::Real(x) => into_arena(&spg_engine::eval::format_real_styled(*x, style)),
Value::Text(s) | Value::Json(s) => into_arena(s.as_ref()),
Value::BpChar(s) => into_arena(s.as_ref()),
Value::TsVector(lexs) => into_arena(&spg_engine::eval::format_tsvector(lexs)),
Value::TsQuery(ast) => into_arena(&spg_engine::eval::format_tsquery(ast)),
Value::Timestamp(micros) if matches!(ty, Some(DataType::Timestamptz)) => {
let off = tz.offset_at(*micros);
let abbr = tz.abbrev_at(*micros);
into_arena(&spg_engine::eval::format_timestamptz_tz(
*micros,
style,
off,
abbr.as_deref(),
))
}
Value::Timestamp(micros) => {
into_arena(&spg_engine::eval::format_timestamp_styled(*micros, style))
}
Value::Date(days) => into_arena(&spg_engine::eval::format_date_styled(*days, style)),
Value::Interval {
months,
days,
micros,
} => into_arena(&spg_engine::eval::format_interval_styled(
*months, *days, *micros, style,
)),
Value::Numeric {
scaled,
scale,
kind,
} => into_arena(&format_numeric_kind(*kind, *scaled, *scale)),
Value::Vector(vec) => {
let mut buf = BumpString::new_in(arena);
buf.push('[');
let mut first = true;
for x in vec.iter() {
if !first {
buf.push_str(",");
}
first = false;
use core::fmt::Write;
let _ = write!(&mut buf, "{x}");
}
buf.push(']');
buf.into_bump_str()
}
Value::Sq8Vector(q) => {
let dequant = spg_storage::quantize::dequantize(q);
let mut buf = BumpString::new_in(arena);
buf.push('[');
let mut first = true;
for x in dequant.iter() {
if !first {
buf.push_str(",");
}
first = false;
use core::fmt::Write;
let _ = write!(&mut buf, "{x}");
}
buf.push(']');
buf.into_bump_str()
}
Value::HalfVector(h) => {
let halfs = h.to_f32_vec();
let mut buf = BumpString::new_in(arena);
buf.push('[');
let mut first = true;
for x in halfs.iter() {
if !first {
buf.push_str(",");
}
first = false;
use core::fmt::Write;
let _ = write!(&mut buf, "{x}");
}
buf.push(']');
buf.into_bump_str()
}
Value::Uuid(b) => into_arena(&spg_storage::format_uuid(b)),
Value::Time(us) => into_arena(&spg_engine::eval::format_time(*us)),
Value::Year(y) => {
let mut buf = BumpString::new_in(arena);
let _ = write!(&mut buf, "{y:04}");
buf.into_bump_str()
}
Value::TimeTz { us, offset_secs } => {
into_arena(&spg_engine::eval::format_timetz(*us, *offset_secs))
}
Value::Money(c) => into_arena(&spg_engine::eval::format_money(*c)),
Value::Range { .. } => into_arena(&spg_engine::format_range_text(v)),
Value::Hstore(pairs) => into_arena(&spg_engine::format_hstore_text(pairs)),
Value::IntArray2D(rows) => into_arena(&spg_engine::format_int_2d_text_pub(rows)),
Value::BigIntArray2D(rows) => into_arena(&spg_engine::format_bigint_2d_text_pub(rows)),
Value::TextArray2D(rows) => into_arena(&spg_engine::format_text_2d_text_pub(rows)),
Value::TextArray(items) => into_arena(&spg_engine::eval::format_text_array(items)),
Value::IntArray(items) => into_arena(&spg_engine::eval::format_int_array(items)),
Value::BigIntArray(items) => into_arena(&spg_engine::eval::format_bigint_array(items)),
Value::BoolArray(items) => into_arena(&spg_engine::eval::format_bool_array(items)),
Value::SmallIntArray(items) => into_arena(&spg_engine::eval::format_smallint_array(items)),
Value::FloatArray(items) => {
into_arena(&spg_engine::eval::format_float_array_styled(items, style))
}
Value::NumericArray(items) => into_arena(&spg_engine::eval::format_numeric_array(items)),
Value::DateArray(items) => {
into_arena(&spg_engine::eval::format_date_array_styled(items, style))
}
Value::TimestampArray(items) => into_arena(
&spg_engine::eval::format_timestamp_array_styled(items, false, style),
),
Value::TimestamptzArray(items) => {
let mut out = String::from("{");
for (i, item) in items.iter().enumerate() {
if i > 0 {
out.push(',');
}
match item {
None => out.push_str("NULL"),
Some(t) => {
let abbr = tz.abbrev_at(*t);
out.push_str(&spg_engine::eval::format_timestamptz_tz(
*t,
style,
tz.offset_at(*t),
abbr.as_deref(),
));
}
}
}
out.push('}');
into_arena(&out)
}
Value::UuidArray(items) => into_arena(&spg_engine::eval::format_uuid_array(items)),
Value::IntervalArray(items) => into_arena(&spg_engine::eval::format_interval_array_styled(
items, style,
)),
other => into_arena(&spg_engine::eval::value_to_text_styled(other, style)),
})
}
fn format_timestamp(micros: i64) -> String {
let secs = micros.div_euclid(1_000_000);
let frac = micros.rem_euclid(1_000_000) as u32;
let (y, m, d, hh, mm, ss) = secs_to_ymdhms(secs);
if frac == 0 {
format!("{y:04}-{m:02}-{d:02} {hh:02}:{mm:02}:{ss:02}")
} else {
format!("{y:04}-{m:02}-{d:02} {hh:02}:{mm:02}:{ss:02}.{frac:06}")
}
}
fn format_date(days: i32) -> String {
let secs = i64::from(days) * 86_400;
let (y, m, d, _, _, _) = secs_to_ymdhms(secs);
format!("{y:04}-{m:02}-{d:02}")
}
#[allow(clippy::cast_possible_truncation, clippy::cast_sign_loss)]
fn secs_to_ymdhms(secs: i64) -> (i32, u32, u32, u32, u32, u32) {
let day = secs.div_euclid(86_400);
let tod = secs.rem_euclid(86_400) as u32;
let hh = tod / 3600;
let mm = (tod / 60) % 60;
let ss = tod % 60;
let z = day + 719_468;
let era = z.div_euclid(146_097);
let doe = (z - era * 146_097) as u32;
let yoe = (doe - doe / 1460 + doe / 36_524 - doe / 146_096) / 365;
let y_int = yoe as i32 + (era as i32) * 400;
let doy = doe - (365 * yoe + yoe / 4 - yoe / 100);
let mp = (5 * doy + 2) / 153;
let d = doy - (153 * mp + 2) / 5 + 1;
let m = if mp < 10 { mp + 3 } else { mp - 9 };
let y = if m <= 2 { y_int + 1 } else { y_int };
(y, m, d, hh, mm, ss)
}
fn format_numeric_kind(kind: spg_storage::NumericKind, scaled: i128, scale: u16) -> String {
use spg_storage::NumericKind;
match kind {
NumericKind::Finite => format_numeric(scaled, scale),
NumericKind::NaN => "NaN".to_string(),
NumericKind::PosInf => "Infinity".to_string(),
NumericKind::NegInf => "-Infinity".to_string(),
}
}
fn format_numeric(scaled: i128, scale: u16) -> String {
if scale == 0 {
return scaled.to_string();
}
let s = scaled.abs().to_string();
let scale = scale as usize;
let (int_part, frac_part) = if s.len() > scale {
let split = s.len() - scale;
(&s[..split], &s[split..])
} else {
("0", s.as_str())
};
let mut frac_pad = "0".repeat(scale.saturating_sub(frac_part.len()));
frac_pad.push_str(frac_part);
if scaled < 0 {
format!("-{int_part}.{frac_pad}")
} else {
format!("{int_part}.{frac_pad}")
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn command_tag_derives_verb_from_ast_not_first_word() {
assert_eq!(command_tag("INSERT INTO t VALUES (1)", 1), "INSERT 0 1");
assert_eq!(command_tag("UPDATE t SET x = 1", 4), "UPDATE 4");
assert_eq!(command_tag("DELETE FROM t", 2), "DELETE 2");
assert_eq!(
command_tag("WITH c AS (SELECT 1) INSERT INTO t SELECT * FROM c", 3),
"INSERT 0 3"
);
assert_eq!(
command_tag("WITH c AS (SELECT 1) UPDATE t SET x = 1", 5),
"UPDATE 5"
);
assert_eq!(
command_tag("WITH c AS (SELECT 1) DELETE FROM t WHERE id > 9", 0),
"DELETE 0"
);
assert_eq!(
command_tag("CREATE TYPE mood AS ENUM ('a')", 0),
"CREATE TYPE"
);
assert_eq!(
command_tag("ALTER TYPE mood ADD VALUE 'b'", 0),
"ALTER TYPE"
);
assert_eq!(command_tag("DROP TYPE IF EXISTS mood", 0), "DROP TYPE");
assert_eq!(command_tag("CREATE TABLE t (id INT)", 0), "CREATE TABLE");
assert_eq!(
command_tag("CREATE UNIQUE INDEX i ON t (id)", 0),
"CREATE INDEX"
);
assert_eq!(
command_tag("CREATE OR REPLACE VIEW v AS SELECT 1", 0),
"CREATE VIEW"
);
assert_eq!(
command_tag("CREATE TEMP TABLE t (id INT)", 0),
"CREATE TABLE"
);
assert_eq!(command_tag("DROP TABLE IF EXISTS t", 0), "DROP TABLE");
assert_eq!(
command_tag("ALTER TABLE t ADD COLUMN x INT", 0),
"ALTER TABLE"
);
assert_eq!(command_tag("TRUNCATE t", 0), "TRUNCATE TABLE");
assert_eq!(command_tag("CREATE USER u", 0), "CREATE ROLE");
assert_eq!(
command_tag("DROP MATERIALIZED VIEW m", 0),
"DROP MATERIALIZED VIEW"
);
assert_eq!(
command_tag("CREATE MATERIALIZED VIEW m AS SELECT 1", 0),
"CREATE"
);
assert_eq!(
command_tag("CREATE EXTENSION pgcrypto", 0),
"CREATE EXTENSION"
);
assert_eq!(
command_tag("REFRESH MATERIALIZED VIEW m", 0),
"REFRESH MATERIALIZED VIEW"
);
}
fn read_cell(buf: &[u8]) -> &[u8] {
let len = i32::from_be_bytes(buf[..4].try_into().unwrap());
assert!(len >= 0, "negative cell length");
let len = len as usize;
&buf[4..4 + len]
}
fn split_owned(body: &str) -> Vec<String> {
split_top_level_statements(body.as_bytes())
.into_iter()
.map(|s| String::from_utf8(s.to_vec()).unwrap())
.map(|s| s.trim().to_string())
.filter(|s| !s.is_empty())
.collect()
}
#[test]
fn splitter_single_statement_no_semicolon() {
assert_eq!(split_owned("SELECT 1"), vec!["SELECT 1".to_string()]);
}
#[test]
fn splitter_two_statements_semicolon_separated() {
let r = split_owned("CREATE TABLE a (); CREATE TABLE b ();");
assert_eq!(r, vec!["CREATE TABLE a ()", "CREATE TABLE b ()"]);
}
#[test]
fn splitter_ignores_semicolon_inside_single_quotes() {
let r = split_owned("INSERT INTO t VALUES ('a;b'); SELECT 1");
assert_eq!(r, vec!["INSERT INTO t VALUES ('a;b')", "SELECT 1"]);
}
#[test]
fn splitter_ignores_semicolon_inside_double_quoted_ident() {
let r = split_owned(r#"SELECT "a;b"; SELECT 1"#);
assert_eq!(r, vec![r#"SELECT "a;b""#, "SELECT 1"]);
}
#[test]
fn splitter_ignores_semicolon_inside_line_comment() {
let r = split_owned("SELECT 1 -- comment ; nope\n; SELECT 2");
assert_eq!(r, vec!["SELECT 1 -- comment ; nope", "SELECT 2"]);
}
#[test]
fn splitter_ignores_semicolon_inside_block_comment() {
let r = split_owned("SELECT 1 /* ; not a split */; SELECT 2");
assert_eq!(r, vec!["SELECT 1 /* ; not a split */", "SELECT 2"]);
}
#[test]
fn splitter_handles_dollar_quoted_do_block() {
let script = "DO $$ BEGIN \
IF NOT EXISTS (SELECT 1) THEN \
CREATE TYPE foo AS ENUM ('a'); \
END IF; \
END $$; \
CREATE TABLE t ()";
let r = split_owned(script);
assert_eq!(r.len(), 2, "want 2 stmts, got: {r:?}");
assert!(r[0].starts_with("DO $$"));
assert_eq!(r[1], "CREATE TABLE t ()");
}
#[test]
fn splitter_handles_tagged_dollar_quotes() {
let script = "SELECT $tag$body; with ; semicolons$tag$; SELECT 1";
let r = split_owned(script);
assert_eq!(r.len(), 2);
assert_eq!(r[1], "SELECT 1");
}
#[test]
fn splitter_drops_empty_statements_between_semicolons() {
let r = split_owned("SELECT 1;;; SELECT 2;");
assert_eq!(r, vec!["SELECT 1", "SELECT 2"]);
}
#[test]
fn splitter_handles_escaped_single_quote_in_string() {
let r = split_owned("SELECT 'it''s; ok'; SELECT 1");
assert_eq!(r, vec!["SELECT 'it''s; ok'", "SELECT 1"]);
}
#[test]
fn write_cell_timestamp_matches_engine_format() {
let cases: &[i64] = &[
0, 1_700_000_000_000_000, 1_700_000_000_123_456, 1_700_000_000_123_000, 1_700_000_000_100_000, -1_000_000_000_000, 253_402_300_799_000_000, 1_577_836_800_000_000, ];
for µs in cases {
let expected = spg_engine::eval::format_timestamp(micros);
let mut buf = Vec::new();
write_cell_timestamp(&mut buf, micros, false).unwrap();
let got = std::str::from_utf8(read_cell(&buf)).unwrap();
assert_eq!(got, expected, "timestamp mismatch @ micros={micros}");
let expected_tz = spg_engine::eval::format_timestamptz(micros);
let mut buf_tz = Vec::new();
write_cell_timestamp(&mut buf_tz, micros, true).unwrap();
let got_tz = std::str::from_utf8(read_cell(&buf_tz)).unwrap();
assert_eq!(
got_tz, expected_tz,
"timestamptz mismatch @ micros={micros}"
);
}
}
#[test]
fn write_cell_date_matches_pgwire_format() {
let cases: &[i32] = &[
0, 19_723, -10_957, 2_932_896, ];
for &days in cases {
let expected = format_date(days);
let mut buf = Vec::new();
write_cell_date(&mut buf, days).unwrap();
let got = std::str::from_utf8(read_cell(&buf)).unwrap();
assert_eq!(got, expected, "date mismatch @ days={days}");
}
}
#[test]
fn write_cell_bytes_matches_extend_form() {
fn extend_form(out: &mut Vec<u8>, bytes: &[u8]) {
let len = i32::try_from(bytes.len()).unwrap();
out.extend_from_slice(&len.to_be_bytes());
out.extend_from_slice(bytes);
}
let cases: &[&[u8]] = &[
b"",
b"a",
b"hello",
b"abcdefghij" as &[u8],
&[0u8; 256],
&[0xffu8; 1024],
];
for bytes in cases {
let mut got = Vec::new();
write_cell_bytes(&mut got, bytes).unwrap();
let mut want = Vec::new();
extend_form(&mut want, bytes);
assert_eq!(got, want, "len={}", bytes.len());
}
}
#[test]
fn parse_copy_with_column_list() {
let sql = "COPY posts (id, title, body) FROM stdin;";
match parse_copy_intent(sql) {
Some(CopyIntent::From(table, _, _)) => assert_eq!(table, "posts"),
other => panic!("expected From(posts), got {other:?}"),
}
}
#[test]
fn parse_copy_without_column_list() {
let sql = "COPY accounts FROM STDIN";
match parse_copy_intent(sql) {
Some(CopyIntent::From(table, _, _)) => assert_eq!(table, "accounts"),
other => panic!("expected From(accounts), got {other:?}"),
}
}
#[test]
fn parse_copy_to_stdout_with_column_list() {
let sql = "COPY t (a, b) TO STDOUT";
match parse_copy_intent(sql) {
Some(CopyIntent::To(table, _)) => assert_eq!(table, "t"),
other => panic!("expected To(t), got {other:?}"),
}
}
#[test]
fn parse_copy_query_to_stdout() {
let sql =
"COPY (SELECT a, b FROM t WHERE a > 1 ORDER BY a) TO STDOUT WITH (FORMAT csv, HEADER)";
match parse_copy_intent(sql) {
Some(CopyIntent::ToQuery(query, opts)) => {
assert_eq!(query, "SELECT a, b FROM t WHERE a > 1 ORDER BY a");
assert!(opts.format_csv);
assert!(opts.header);
}
other => panic!("expected ToQuery, got {other:?}"),
}
}
#[test]
fn parse_copy_query_with_cte_not_confused_by_inner_with() {
let sql = "COPY (WITH x AS (SELECT 1 AS n) SELECT n FROM x) TO STDOUT";
match parse_copy_intent(sql) {
Some(CopyIntent::ToQuery(query, opts)) => {
assert_eq!(query, "WITH x AS (SELECT 1 AS n) SELECT n FROM x");
assert!(!opts.format_csv);
assert!(!opts.header);
}
other => panic!("expected ToQuery, got {other:?}"),
}
}
#[test]
fn parse_copy_with_with_options() {
let sql = "COPY t FROM stdin WITH (format json)";
match parse_copy_intent(sql) {
Some(CopyIntent::From(table, _, opts)) => {
assert_eq!(table, "t");
assert!(opts.format_json);
}
other => panic!("expected From(t) with format_json, got {other:?}"),
}
}
#[test]
fn parse_to_file_intent() {
match parse_copy_intent("COPY ct TO '/tmp/spg-r252/x.txt'") {
Some(CopyIntent::ToFile(spec)) => {
assert_eq!(spec.table, "ct");
assert_eq!(spec.path, "/tmp/spg-r252/x.txt");
}
other => panic!("expected ToFile, got {other:?}"),
}
}
#[test]
fn parse_non_copy_returns_none() {
assert!(parse_copy_intent("SELECT 1").is_none());
match parse_copy_intent("COPY t FROM '/etc/passwd'") {
Some(CopyIntent::FromFile(spec)) => {
assert_eq!(spec.table, "t");
assert_eq!(spec.path, "/etc/passwd");
}
other => panic!("expected FromFile, got {other:?}"),
}
}
#[test]
fn parse_copy_from_csv_options() {
let sql = "COPY t FROM stdin WITH (FORMAT csv, HEADER true, DELIMITER ';', QUOTE '#')";
match parse_copy_intent(sql) {
Some(CopyIntent::From(table, _, opts)) => {
assert_eq!(table, "t");
assert!(opts.format_csv);
assert!(!opts.format_json);
assert_eq!(opts.skip, 1); assert_eq!(opts.csv_delimiter, Some(';'));
assert_eq!(opts.csv_quote, Some('#'));
}
other => panic!("expected From(t) csv opts, got {other:?}"),
}
}
#[test]
fn parse_copy_from_bare_csv_header() {
let sql = "COPY t FROM stdin WITH (FORMAT csv, HEADER)";
match parse_copy_intent(sql) {
Some(CopyIntent::From(_, _, opts)) => {
assert!(opts.format_csv);
assert_eq!(opts.skip, 1);
}
other => panic!("expected csv, got {other:?}"),
}
}
#[test]
fn decode_binary_param_uuid_16_bytes_round_trip() {
let bytes = [
0x55, 0x0e, 0x84, 0x00, 0xe2, 0x9b, 0x41, 0xd4, 0xa7, 0x16, 0x44, 0x66, 0x55, 0x44,
0x00, 0x00,
];
let v = decode_binary_param(2950, &bytes).expect("UUID binary BIND must succeed");
match v {
spg_storage::Value::Uuid(b) => assert_eq!(b, bytes),
other => panic!("expected Value::Uuid, got {other:?}"),
}
}
#[test]
fn decode_binary_param_uuid_rejects_wrong_length() {
for &len in &[0usize, 1, 8, 15, 17, 32] {
let bytes = vec![0u8; len];
let r = decode_binary_param(2950, &bytes);
assert!(r.is_err(), "len={len} must reject (UUID is 16 bytes)");
}
}
#[test]
fn decode_binary_param_interval_16_bytes_round_trip() {
let mut bytes = [0u8; 16];
bytes[0..8].copy_from_slice(&3_i64.to_be_bytes()); bytes[8..12].copy_from_slice(&2_i32.to_be_bytes()); bytes[12..16].copy_from_slice(&1_i32.to_be_bytes()); let v = decode_binary_param(1186, &bytes).expect("INTERVAL binary BIND must succeed");
assert_eq!(
v,
spg_storage::Value::Interval {
months: 1,
days: 2,
micros: 3,
}
);
}
#[test]
fn decode_binary_param_interval_signed_round_trip() {
let mut bytes = [0u8; 16];
bytes[0..8].copy_from_slice(&(-86_400_000_000_i64).to_be_bytes());
bytes[8..12].copy_from_slice(&(-1_i32).to_be_bytes());
bytes[12..16].copy_from_slice(&(-1_i32).to_be_bytes());
let v =
decode_binary_param(1186, &bytes).expect("signed INTERVAL binary BIND must succeed");
assert_eq!(
v,
spg_storage::Value::Interval {
months: -1,
days: -1,
micros: -86_400_000_000,
}
);
}
#[test]
fn decode_binary_param_interval_rejects_wrong_length() {
for &len in &[0usize, 1, 8, 12, 15, 17, 32] {
let bytes = vec![0u8; len];
let r = decode_binary_param(1186, &bytes);
assert!(
r.is_err(),
"len={len} must reject (INTERVAL binary is 16 bytes)"
);
}
}
}