diff --git a/Cargo.lock b/Cargo.lock index 6a90905..93ffb59 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -471,6 +471,7 @@ name = "rupg-session" version = "0.1.1" dependencies = [ "rupg-analyze", + "rupg-catalog", "rupg-common", "rupg-exec", "rupg-func", diff --git a/crates/rupg-server/src/lib.rs b/crates/rupg-server/src/lib.rs index 50da68d..81bcc18 100644 --- a/crates/rupg-server/src/lib.rs +++ b/crates/rupg-server/src/lib.rs @@ -32,6 +32,7 @@ use rupg_common::{Error, Result, SqlState}; use rupg_platform::{Clock, Entropy, Io, Listener, Net, Stream, TaskHandle, Tasks}; use rupg_session::connection::{self, Connection, Next, Start}; use rupg_session::guc::Settings; +use rupg_session::store::Store; use rupg_wire::{ CANCEL_KEY_LEN, CancelKey, Handshake, Hashes, Level, MOCK_NONCE_LEN, OutBuf, Replication, SCRAM_ITERATIONS, SCRAM_SALT_LEN, ScramSecret, Step, cancel_target, md5_encrypt, @@ -127,6 +128,8 @@ struct Shared { log: Option>, clock: Option>, entropy: Arc, + /// The catalog that all the sessions share. + store: Arc, /// The cancel key and the cancel flag of each session, by process ID. keys: Mutex>, stopping: AtomicBool, @@ -279,6 +282,7 @@ impl Server { iterations, log: config.log.clone(), clock: config.clock.clone(), + store: Arc::new(Store::with_tasks(tasks.clone())), entropy, keys: Mutex::new(BTreeMap::new()), stopping: AtomicBool::new(false), @@ -749,6 +753,7 @@ fn session(shared: &Shared, wire: &mut Wire, start: &Start, protocol: u32) { if let Some(clock) = &shared.clock { connection.set_clock(Arc::clone(clock)); } + connection.set_store(Arc::clone(&shared.store)); let (_registered, key) = register(shared, protocol, connection.cancel_flag()); connection.greet(&key, &mut wire.out); loop { diff --git a/crates/rupg-session/Cargo.toml b/crates/rupg-session/Cargo.toml index 5c205b8..378892f 100644 --- a/crates/rupg-session/Cargo.toml +++ b/crates/rupg-session/Cargo.toml @@ -15,6 +15,7 @@ publish = false [dependencies] rupg-analyze.workspace = true +rupg-catalog.workspace = true rupg-common.workspace = true rupg-exec.workspace = true rupg-func.workspace = true diff --git a/crates/rupg-session/src/connection.rs b/crates/rupg-session/src/connection.rs index c39ff3d..5890308 100644 --- a/crates/rupg-session/src/connection.rs +++ b/crates/rupg-session/src/connection.rs @@ -6,8 +6,10 @@ use std::cell::{Ref, RefCell}; use std::sync::Arc; use std::sync::atomic::{AtomicBool, Ordering}; -use rupg_analyze::Params; +use rupg_analyze::{Message, Params}; +use rupg_catalog::Catalog; use rupg_common::{Error, SqlState}; +use rupg_pgcatalog::builtin; use rupg_platform::Clock; use rupg_platform::os::OsClock; use rupg_sql::Severity; @@ -20,6 +22,7 @@ use rupg_wire::{ use crate::block::{Block, Ending, Transaction}; use crate::guc::{Action, Origin, Settings}; use crate::query::{self, Column, Reader, Row}; +use crate::store::{Lease, Store}; use crate::utility::{self, Context, Notice, Outcome}; mod extended; @@ -278,8 +281,17 @@ pub struct Connection { /// `xactStartTimestamp` and `stmtStartTimestamp`, in microseconds since 2000-01-01 UTC. transaction_start: i64, statement_start: i64, + /// The OID of the role of the session. + role: u32, + /// The catalog that the sessions of the server share. + store: Arc, + /// The lock of the store and the copy of the catalog, while the transaction changes the catalog. + lease: Option, } +/// The OID of the bootstrap superuser, which owns the objects of initdb. +const BOOTSTRAP_SUPERUSER: u32 = 10; + impl Connection { /// A session with the settings that [`session_settings`] gave. `protocol` is the version of the protocol that the client uses. pub fn new(start: &Start, settings: Settings, protocol: u32) -> Connection { @@ -299,6 +311,9 @@ impl Connection { pid: 0, transaction_start: 0, statement_start: 0, + role: role_oid(start), + store: Arc::new(Store::new()), + lease: None, }; connection.session.set_utf8(connection.utf8()); connection @@ -309,6 +324,11 @@ impl Connection { self.clock = clock; } + /// Sets the catalog that the sessions of the server share. A new session has a catalog of its own. + pub fn set_store(&mut self, store: Arc) { + self.store = store; + } + /// `SetCurrentStatementStartTimestamp`: the start of a message that runs a statement. fn start_statement(&mut self) { self.statement_start = query::now(&*self.clock); @@ -323,9 +343,48 @@ impl Connection { (self.transaction_start, self.statement_start), &*self.clock, self.pid, + self.catalog(), ) } + /// The catalog that a statement sees: the copy of the transaction when it changed the catalog, or the catalog of the last commit. + fn catalog(&self) -> Arc { + match &self.lease { + Some(lease) => Arc::clone(&lease.catalog), + None => self.store.committed(), + } + } + + /// Runs a statement that defines an object. The statement takes the lock of the store, which the transaction holds until it ends. After an error the changes of the statement go, but its OIDs stay used. + fn define(&mut self, node: &Node, notices: &mut Vec) -> Result { + if self.lease.is_none() { + self.lease = Some(self.store.lease(&self.cancel)?); + } + let mut work = Catalog::clone(&self.catalog()); + let defined = rupg_analyze::define(node, &self.reader(), &mut work, self.role); + notices.extend(defined.messages.into_iter().map(|message| match message { + Message::Warning(error) => Notice::warning(error), + Message::Notice(error) => Notice { severity: Severity::Notice, error }, + })); + let Some(lease) = self.lease.as_mut() else { + return Err(Error::internal("the lock of the catalog is gone")); + }; + match defined.result { + Ok(()) => { + lease.catalog = Arc::new(work); + Ok(Done::Tag(match node { + Node::IndexStmt(_) => CommandTag::CreateIndex, + Node::CreateSchemaStmt(_) => CommandTag::CreateSchema, + _ => CommandTag::CreateTable, + })) + } + Err(error) => { + Arc::make_mut(&mut lease.catalog).advance_oid(work.next_oid()); + Err(error) + } + } + } + /// The flag that a `CancelRequest` sets. The server keeps a clone of it with the cancel key of the session. pub fn cancel_flag(&self) -> Arc { Arc::clone(&self.cancel) @@ -557,6 +616,9 @@ impl Connection { } else if query::is_query(node) { drop(self.portals.close(b"")); self.select(node, &mut notices) + } else if rupg_analyze::is_definition(node) { + drop(self.portals.close(b"")); + self.define(node, &mut notices) } else { drop(self.portals.close(b"")); let mut cx = Context { @@ -639,6 +701,11 @@ impl Connection { let kept = chained.then(|| self.settings.get_mut().characteristics()); if ending != Ending::None { self.settings.get_mut().end(ending == Ending::Commit); + if let Some(lease) = self.lease.take() + && ending == Ending::Commit + { + lease.commit(); + } } if kept.is_some() { self.settings.get_mut().start_transaction(kept); @@ -661,6 +728,7 @@ impl Connection { self.xact_started = false; if self.transaction.abort_current() == Ending::Rollback { self.settings.get_mut().end(false); + self.lease = None; } if self.transaction.failed() { self.portals.retain(|_, portal| { @@ -673,6 +741,14 @@ impl Connection { } } +/// The OID of the role of a session. The superuser of the server is the bootstrap superuser. +fn role_oid(start: &Start) -> u32 { + if start.superuser { + return BOOTSTRAP_SUPERUSER; + } + builtin::roles().iter().find(|role| role.name == start.user).map_or(0, |role| role.oid) +} + /// The position of a byte offset in characters, from 1, as the field `P` gives it. fn character_position(text: &str, at: usize) -> usize { let at = at.min(text.len()); @@ -1267,6 +1343,88 @@ mod tests { ); } + /// `CREATE SCHEMA`, `CREATE TABLE` and `CREATE INDEX`, with the messages of PostgreSQL 19 for the same statements. + fn definitions_cases() { + let (mut c, _) = connect(); + assert_eq!(send(&mut c, &query("CREATE SCHEMA s")), ["CREATE SCHEMA", "ready I"]); + assert_eq!( + send(&mut c, &query("SET search_path = s, public; SELECT current_schemas(false)")), + [ + "SET", + "columns current_schemas", + "row {s,public}", + "SELECT 1", + "search_path=s, public", + "ready I" + ] + ); + assert_eq!( + send(&mut c, &query("CREATE TABLE t (a int PRIMARY KEY)")), + ["CREATE TABLE", "ready I"] + ); + assert_eq!( + send(&mut c, &query("CREATE TABLE t (a int)")), + ["ERROR 42P07 relation \"t\" already exists", "ready I"] + ); + assert_eq!( + send(&mut c, &query("CREATE TABLE IF NOT EXISTS t (a int)")), + ["NOTICE 42P07 relation \"t\" already exists, skipping", "CREATE TABLE", "ready I"] + ); + assert_eq!( + send(&mut c, &query("BEGIN; CREATE TABLE u (a int); ROLLBACK")), + ["BEGIN", "CREATE TABLE", "ROLLBACK", "ready I"] + ); + assert_eq!( + send(&mut c, &query("CREATE TABLE u (b nosuch)")), + ["ERROR 42704 type \"nosuch\" does not exist at 19", "ready I"] + ); + assert_eq!( + send(&mut c, &query("CREATE TABLE u (a int); CREATE INDEX ON u (a)")), + ["CREATE TABLE", "CREATE INDEX", "ready I"] + ); + let catalog = c.store.committed(); + let names: Vec<&str> = catalog.relations().map(|r| r.name.as_str()).collect(); + assert_eq!(names, ["t", "t_pkey", "u", "u_a_idx"]); + assert_eq!( + send(&mut c, &query("BEGIN; CREATE TABLE w (a int)")), + ["BEGIN", "CREATE TABLE", "ready T"] + ); + assert_eq!( + send(&mut c, &query("CREATE TABLE w (a int)")), + ["ERROR 42P07 relation \"w\" already exists", "ready E"] + ); + assert_eq!( + send(&mut c, &query("CREATE TABLE x (a int)")), + [ + "ERROR 25P02 current transaction is aborted, commands ignored until end of transaction block", + "ready E" + ] + ); + assert_eq!(send(&mut c, &query("ROLLBACK")), ["ROLLBACK", "ready I"]); + assert_eq!(send(&mut c, &query("CREATE TABLE w (a int)")), ["CREATE TABLE", "ready I"]); + assert_eq!( + send(&mut c, &run("CREATE TABLE pg_catalog.z (a int)")), + [ + "ParseComplete", + "BindComplete", + "ERROR 42501 permission denied to create \"pg_catalog.z\"" + ] + ); + assert_eq!(send(&mut c, &sync()), ["ready I"]); + // A second session of the same server sees the catalog of the last commit. + let (mut d, _) = connect(); + d.set_store(Arc::clone(&c.store)); + assert_eq!( + send(&mut d, &query("CREATE TABLE s.t (a int)")), + ["ERROR 42P07 relation \"t\" already exists", "ready I"] + ); + } + + #[test] + fn definitions() { + big_stack(definitions_cases); + } + #[test] fn extended_selects() { big_stack(extended_selects_cases); diff --git a/crates/rupg-session/src/connection/extended.rs b/crates/rupg-session/src/connection/extended.rs index f1add21..6236c4f 100644 --- a/crates/rupg-session/src/connection/extended.rs +++ b/crates/rupg-session/src/connection/extended.rs @@ -161,7 +161,7 @@ impl Connection { columns = Some(query::columns(&made)); params = made.params().to_vec(); plan = Some(made); - } else { + } else if !rupg_analyze::is_definition(node) { utility::check(node, &text)?; columns = utility::columns(node, &self.settings.borrow())? .map(|names| names.into_iter().map(Column::text).collect()); @@ -327,6 +327,14 @@ impl Connection { self.settings.borrow_mut().take_snapshot(); let rows = query::run(plan, ¶ms, &self.reader(), &formats)?; Done::Rows { columns: Vec::new(), rows, tag: CommandTag::Select } + } else if rupg_analyze::is_definition(node) { + let mut notices = Vec::new(); + let result = self.define(node, &mut notices); + self.notices(&mut notices, &statement.text, &[], out); + result.map_err(|error| Failed { + position: error.position().map(|at| character_position(&statement.text, at)), + error, + })? } else { let mut notices = Vec::new(); let mut cx = Context { diff --git a/crates/rupg-session/src/lib.rs b/crates/rupg-session/src/lib.rs index cdcbd62..dc44869 100644 --- a/crates/rupg-session/src/lib.rs +++ b/crates/rupg-session/src/lib.rs @@ -12,4 +12,5 @@ mod generated; pub mod guc; mod param; mod query; +pub mod store; pub mod utility; diff --git a/crates/rupg-session/src/query.rs b/crates/rupg-session/src/query.rs index b42ccc5..1e49982 100644 --- a/crates/rupg-session/src/query.rs +++ b/crates/rupg-session/src/query.rs @@ -4,8 +4,10 @@ use std::cell::{Ref, RefCell}; use std::rc::Rc; +use std::sync::Arc; use rupg_analyze::{Env, Params}; +use rupg_catalog::Catalog; use rupg_common::{Error, Result, SqlState}; use rupg_exec::Plan; use rupg_platform::Clock; @@ -81,6 +83,8 @@ pub(crate) struct Reader<'a> { pub(crate) statement_start: i64, pub(crate) clock: &'a dyn Clock, pub(crate) pid: i32, + /// The catalog that the statement sees: the catalog of the last commit, or the copy of the transaction that changes it. + pub(crate) catalog: Arc, /// The text of `TimeZone` and its zone, or the name of a zone whose rules the engine does not have yet. A call of `set_config` can change the setting, so the zone is made again when the text changes. zone: RefCell<(String, ZoneOf)>, } @@ -96,6 +100,7 @@ impl<'a> Reader<'a> { times: (i64, i64), clock: &'a dyn Clock, pid: i32, + catalog: Arc, ) -> Reader<'a> { let name = settings.borrow().get("TimeZone").unwrap_or_else(|| "UTC".to_owned()); let zone = fixed_zone(&name).map(Rc::new); @@ -107,6 +112,7 @@ impl<'a> Reader<'a> { statement_start: times.1, clock, pid, + catalog, zone: RefCell::new((name, zone)), } } @@ -117,19 +123,28 @@ impl<'a> Reader<'a> { } /// The schemas of `search_path` that exist, in order, without the implicit schemas, as `recomputeNamespacePath` finds them. `$user` names the schema of the user, and `pg_temp` the temporary schema, which do not exist yet. - fn path(&self) -> Vec<(&'static str, u32)> { + fn path(&self) -> Vec<(String, u32)> { let text = self.settings().get("search_path").unwrap_or_default(); - let mut path: Vec<(&'static str, u32)> = Vec::new(); + let mut path: Vec<(String, u32)> = Vec::new(); for name in guc::split_identifiers(&text, ',').unwrap_or_default() { let name = if name == "$user" { self.user.to_owned() } else { name }; - if let Some(&found) = SCHEMAS.iter().find(|(n, _)| *n == name) - && !path.contains(&found) + if let Some(oid) = self.schema(&name) + && !path.iter().any(|&(_, o)| o == oid) { - path.push(found); + path.push((name, oid)); } } path } + + /// The OID of a built-in schema or a schema of the catalog. + fn schema(&self, name: &str) -> Option { + SCHEMAS + .iter() + .find(|(n, _)| *n == name) + .map(|(_, oid)| *oid) + .or_else(|| self.catalog.schema_by_name(name).map(|s| s.oid)) + } } /// The zone of a `TimeZone` value that the setting accepted, or the name of a zone that the engine does not have yet. @@ -245,7 +260,7 @@ impl rupg_func::Session for Reader<'_> { } fn schemas(&self) -> Vec { - self.path().into_iter().map(|(name, _)| name.to_owned()).collect() + self.path().into_iter().map(|(name, _)| name).collect() } fn backend_pid(&self) -> i32 { @@ -277,7 +292,7 @@ impl Env for Reader<'_> { } fn namespace(&self, name: &str) -> Option { - SCHEMAS.iter().find(|(n, _)| *n == name).map(|(_, oid)| *oid) + self.schema(name) } fn database(&self) -> String { diff --git a/crates/rupg-session/src/store.rs b/crates/rupg-session/src/store.rs new file mode 100644 index 0000000..b2c2481 --- /dev/null +++ b/crates/rupg-session/src/store.rs @@ -0,0 +1,164 @@ +//! The catalog that the sessions of a server share, and the lock of the statements that change it. + +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::{Arc, Condvar, Mutex, MutexGuard, PoisonError}; +use std::time::Duration; + +use rupg_catalog::Catalog; +use rupg_common::{Error, Result, SqlState}; +use rupg_platform::Tasks; + +/// How long a session waits for the lock before it looks at its cancel flag again. +const WAIT_SLICE: Duration = Duration::from_millis(50); + +/// The catalog that the sessions of a server share. A statement reads the catalog of the last commit. A transaction that defines an object holds the lock of the store until it ends, so only one transaction at a time changes the catalog. +#[derive(Debug, Default)] +pub struct Store { + state: Mutex, + free: Condvar, + /// The tasks of the server. A session that waits for the lock lets the other tasks run, because the simulated scheduler runs all the tasks on one thread. + tasks: Option>, +} + +#[derive(Debug, Default)] +struct State { + committed: Arc, + locked: bool, +} + +impl Store { + /// A store with an empty catalog. + pub fn new() -> Store { + Store::default() + } + + /// A store with an empty catalog for the sessions that run as `tasks`. + pub fn with_tasks(tasks: Arc) -> Store { + Store { tasks: Some(tasks), ..Store::default() } + } + + /// The catalog of the last commit. + pub fn committed(&self) -> Arc { + Arc::clone(&self.state().committed) + } + + fn state(&self) -> MutexGuard<'_, State> { + self.state.lock().unwrap_or_else(PoisonError::into_inner) + } + + /// Takes the lock for a transaction, with a copy of the catalog of the last commit. The session waits while another transaction holds the lock. A cancel of the session stops the wait with `57014`. + pub(crate) fn lease(self: &Arc, cancel: &AtomicBool) -> Result { + let mut state = self.state(); + while state.locked { + if cancel.swap(false, Ordering::Relaxed) { + return Err(Error::new( + SqlState::QUERY_CANCELED, + "canceling statement due to user request", + )); + } + if let Some(tasks) = &self.tasks { + drop(state); + tasks.yield_now(); + state = self.state(); + if !state.locked { + break; + } + } + state = + self.free.wait_timeout(state, WAIT_SLICE).unwrap_or_else(PoisonError::into_inner).0; + } + state.locked = true; + Ok(Lease { store: Arc::clone(self), catalog: Arc::clone(&state.committed) }) + } +} + +/// The lock of a transaction and its copy of the catalog. The drop of a lease releases the lock and discards the changes, but the OIDs that the transaction used stay used, as in PostgreSQL. +#[derive(Debug)] +pub(crate) struct Lease { + store: Arc, + /// The catalog with the changes of the transaction. + pub(crate) catalog: Arc, +} + +impl Lease { + /// Makes the changes of the transaction the catalog of the last commit, and releases the lock. + pub(crate) fn commit(mut self) { + let catalog = std::mem::take(&mut self.catalog); + self.store.state().committed = catalog; + } +} + +impl Drop for Lease { + fn drop(&mut self) { + let mut state = self.store.state(); + let next = self.catalog.next_oid(); + if state.committed.next_oid() < next { + Arc::make_mut(&mut state.committed).advance_oid(next); + } + state.locked = false; + drop(state); + self.store.free.notify_one(); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn rollback_keeps_the_oids() { + let store = Arc::new(Store::new()); + let cancel = AtomicBool::new(false); + let mut lease = store.lease(&cancel).unwrap(); + Arc::make_mut(&mut lease.catalog).create_schema("s", 10).unwrap(); + let next = lease.catalog.next_oid(); + drop(lease); + assert!(store.committed().schema_by_name("s").is_none()); + assert_eq!(store.committed().next_oid(), next); + let mut lease = store.lease(&cancel).unwrap(); + let oid = Arc::make_mut(&mut lease.catalog).create_schema("s", 10).unwrap(); + assert_eq!(oid, next); + lease.commit(); + assert_eq!(store.committed().schema_by_name("s").map(|s| s.oid), Some(oid)); + } + + #[test] + fn cancel_stops_the_wait() { + let store = Arc::new(Store::new()); + let held = store.lease(&AtomicBool::new(false)).unwrap(); + let error = store.lease(&AtomicBool::new(true)).unwrap_err(); + assert_eq!(error.state(), SqlState::QUERY_CANCELED); + drop(held); + assert!(store.lease(&AtomicBool::new(false)).is_ok()); + } + + /// Tasks whose `yield_now` runs the other task: the end of the transaction that holds the lock. + #[derive(Debug, Default)] + struct Holder(Mutex>); + + impl Tasks for Holder { + fn spawn(&self, _: &str, _: rupg_platform::Task) -> Result { + Err(Error::internal("no tasks in this test")) + } + + fn sleep(&self, _: Duration) {} + + fn yield_now(&self) { + drop(self.0.lock().unwrap().take()); + } + + fn parallelism(&self) -> usize { + 1 + } + } + + #[test] + fn a_waiter_lets_the_holder_run() { + let holder = Arc::new(Holder::default()); + let store = Arc::new(Store::with_tasks(holder.clone())); + let held = store.lease(&AtomicBool::new(false)).unwrap(); + *holder.0.lock().unwrap() = Some(held); + assert!(store.lease(&AtomicBool::new(false)).is_ok()); + assert!(holder.0.lock().unwrap().is_none()); + } +}