diff --git a/CHANGELOG.md b/CHANGELOG.md index a431be35..1c7881e1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,7 @@ All changes that a user can see are in this file. The version is 0.M.patch, wher ### Added - `EXPLAIN` in the text, JSON, XML and YAML forms, with a planner that makes the plan of a query with the costs of PostgreSQL (#439). +- `ALTER DATABASE ... SET` and `RESET`, with the settings applied at the start of each session of the database and shown in `pg_db_role_setting` (#440). ## 0.1.51 (2026-10-20) diff --git a/crates/rupg-server/src/lib.rs b/crates/rupg-server/src/lib.rs index 9ba4f885..46da9098 100644 --- a/crates/rupg-server/src/lib.rs +++ b/crates/rupg-server/src/lib.rs @@ -140,6 +140,8 @@ struct Shared { tasks: Arc, /// The catalog that all the sessions share. store: Arc, + /// The file system, which `COPY ... FROM` a file reads. + io: Arc, /// The cancel key and the cancel flag of each session, by process ID. keys: Mutex>, stopping: AtomicBool, @@ -353,6 +355,7 @@ impl Server { ), entropy, tasks: tasks.clone(), + io: io.clone(), keys: Mutex::new(BTreeMap::new()), stopping: AtomicBool::new(false), }); @@ -978,6 +981,7 @@ fn session(shared: &Shared, wire: &mut Wire, start: &Start, protocol: u32) { connection.set_clock(Arc::clone(clock)); } connection.set_store(Arc::clone(&shared.store)); + connection.set_io(Arc::clone(&shared.io)); let client = address(wire, &wire.stream.peer_addr()); connection.set_addresses(client, address(wire, &wire.stream.local_addr())); if let Some(secured) = &wire.secured { diff --git a/crates/rupg-session/src/connection.rs b/crates/rupg-session/src/connection.rs index 5dba4995..aa82aa9b 100644 --- a/crates/rupg-session/src/connection.rs +++ b/crates/rupg-session/src/connection.rs @@ -13,8 +13,8 @@ use rupg_exec::Plan; use rupg_func::{BackendState, Randomness, Sequences}; pub use rupg_func::{Client, Ssl}; use rupg_pgcatalog::builtin; -use rupg_platform::Clock; use rupg_platform::os::OsClock; +use rupg_platform::{Clock, Io}; use rupg_sql::Severity; use rupg_sql::nodes::{LockStmt, Node, ObjectType}; use rupg_types::Value; @@ -48,6 +48,7 @@ mod upsert; mod writes; pub(crate) use activity::Activity; +use copy_from::Begin; use extended::Failed; pub(crate) use extended::{Portal, Prepared}; use writes::Writes; @@ -365,6 +366,8 @@ pub struct Connection { role: u32, /// The catalog that the sessions of the server share. store: Arc, + /// The file system of the server, which `COPY ... FROM` a file reads. A session with no server has none. + io: Option>, /// The lock of the store and the copy of the catalog, while the transaction changes the catalog. lease: Option, /// The row of the session in the table of the sessions of the store. @@ -437,6 +440,7 @@ impl Connection { statement_start: 0, role: role_oid(start), store: Arc::new(Store::new()), + io: None, lease: None, activity: Activity::default(), sequences: RefCell::new(Sequences::default()), @@ -458,6 +462,11 @@ impl Connection { self.clock = clock; } + /// Sets the file system of the server, which `COPY ... FROM` a file reads. + pub fn set_io(&mut self, io: Arc) { + self.io = Some(io); + } + /// Sets the catalog and the table of the sessions that the sessions of the server share, before [`Connection::greet`]. A new session has a store of its own. pub fn set_store(&mut self, store: Arc) { self.database_oid = @@ -1260,11 +1269,17 @@ impl Connection { } else if let Node::ConstraintsSetStmt(stmt) = node { drop(self.portals.close(b"")); self.set_constraints(stmt, &mut notices) + } else if let Node::CopyStmt(stmt) = node + && stmt.is_from + && stmt.filename.is_some() + { + drop(self.portals.close(b"")); + self.copy_from_file(stmt, &mut notices) } else if let Node::CopyStmt(stmt) = node && stmt.is_from { drop(self.portals.close(b"")); - match self.copy_from(stmt, out) { + match self.copy_from(stmt, Begin::Client(out)) { Ok(mut copy) => { copy.text = text.to_string(); copy.index = i; @@ -4135,14 +4150,49 @@ mod tests { big_stack(|| script(include_str!("connection/copy_to.test"))); } - /// `COPY FROM` a file or a program, the JSON format, an encoding other than `UTF8` and `COPY TO` a file or a program give `0A000`. + /// `COPY ... FROM` a file of the server: the rows of the file in the simple and the extended protocol, an error of a line with the context of the line, and a file that does not exist. A session with no server cannot read files. + #[test] + fn copy_from_file() { + use rupg_platform::{OpenMode, sim::SimIo}; + big_stack(|| { + let (mut c, _) = connect(); + assert_eq!(script_lines(&mut c, "> CREATE TABLE t (a int, b text)"), ["CREATE TABLE"]); + assert_eq!( + script_lines(&mut c, "> COPY t FROM '/data/t.data'"), + ["ERROR 0A000 COPY FROM a file in a session with no server is not supported yet"] + ); + let io = SimIo::new(1); + let file = io.open(std::path::Path::new("/data/t.data"), OpenMode::Create).unwrap(); + file.write_at(0, b"1\tone\n2\ttwo\n").unwrap(); + let file = io.open(std::path::Path::new("/data/bad.data"), OpenMode::Create).unwrap(); + file.write_at(0, b"3\tthree\nx\tfour\n").unwrap(); + c.set_io(Arc::new(io)); + assert_eq!(script_lines(&mut c, "> COPY t FROM '/data/t.data'"), ["COPY 2"]); + assert!(script_lines(&mut c, "P s|COPY t (a, b) FROM '/data/t.data'").is_empty()); + assert_eq!(script_lines(&mut c, "X s"), ["COPY 2"]); + assert_eq!( + script_lines(&mut c, "> COPY t FROM '/data/bad.data'"), + [ + "ERROR 22P02 invalid input syntax for type integer: \"x\" (COPY t, line 2, column a: \"x\")" + ] + ); + assert_eq!( + script_lines(&mut c, "> COPY t FROM '/data/none.data'"), + ["ERROR 58P01 could not open file \"/data/none.data\" for reading: no such file"] + ); + let rows = + script_lines(&mut c, "> SELECT a, b, count(*) FROM t GROUP BY a, b ORDER BY a"); + assert_eq!(rows[1..], ["row 1,one,2", "row 2,two,2", "SELECT 2"]); + }); + } + + /// `COPY FROM` a program, the JSON format, an encoding other than `UTF8` and `COPY TO` a file or a program give `0A000`. #[test] fn copy_not_supported() { big_stack(|| { let (mut c, _) = connect(); assert_eq!(script_lines(&mut c, "> CREATE TABLE t (a int)"), ["CREATE TABLE"]); for (sql, error) in [ - ("COPY t FROM '/tmp/rupg-copy.txt'", "COPY FROM a file is not supported yet"), ("COPY t FROM PROGRAM 'cat'", "COPY FROM a program is not supported yet"), ("COPY t TO stdout (FORMAT json)", "COPY FORMAT JSON is not supported yet"), ( diff --git a/crates/rupg-session/src/connection/copy_from.rs b/crates/rupg-session/src/connection/copy_from.rs index 29153d8d..791f829b 100644 --- a/crates/rupg-session/src/connection/copy_from.rs +++ b/crates/rupg-session/src/connection/copy_from.rs @@ -1,5 +1,6 @@ -//! `COPY ... FROM STDIN`, as `DoCopy`, `BeginCopyFrom`, `CopyFrom` and `copyfromparse.c` run it. The copy reads the fields of each line, and `NextCopyFrom` makes the values of the columns with the input functions or the receive functions of their types. The other columns get their defaults from a plan of `INSERT`, which also checks the constraints. The copy keeps up to `MAX_BUFFERED_TUPLES` rows and then writes them in the heap, as `CopyMultiInsertInfoFlush` does. The triggers of the foreign keys fire at the end of the copy. +//! `COPY ... FROM STDIN` and `COPY ... FROM` a file of the server, as `DoCopy`, `BeginCopyFrom`, `CopyFrom` and `copyfromparse.c` run it. The copy reads the fields of each line, and `NextCopyFrom` makes the values of the columns with the input functions or the receive functions of their types. The other columns get their defaults from a plan of `INSERT`, which also checks the constraints. The copy keeps up to `MAX_BUFFERED_TUPLES` rows and then writes them in the heap, as `CopyMultiInsertInfoFlush` does. The triggers of the foreign keys fire at the end of the copy. +use std::path::Path; use std::rc::Rc; use rupg_analyze::{CopyTarget, Params}; @@ -9,11 +10,11 @@ use rupg_pgcatalog::builtin; use rupg_sql::Severity; use rupg_sql::nodes::{CopyStmt, RangeVar}; use rupg_types::Value; -use rupg_wire::OutBuf; +use rupg_wire::{CommandTag, OutBuf}; -use super::Connection; use super::copy::{BINARY_SIGNATURE, ForceQuote, Format, OnError, Verbosity, options}; use super::foreign::Batch; +use super::{Connection, Done}; use crate::heap::{Changed, View}; use crate::query::{self, Reader}; use crate::utility::Notice; @@ -24,6 +25,14 @@ const MAX_DISPLAY: usize = 100; /// `MAX_BUFFERED_TUPLES`: the number of rows that the copy keeps before it writes them in the heap. const MAX_BUFFERED: usize = 1000; +/// Where the data of a copy comes from: the client, which gets `CopyInResponse`, or a file of the server, which the copy reads at the same point. +#[derive(Debug)] +pub(super) enum Begin<'a> { + Client(&'a mut OutBuf), + /// The name of the file, and the bytes of the file after the copy reads it. + File(&'a str, &'a mut Vec), +} + /// A line of data that the copy read, before `NextCopyFrom` makes its values. #[derive(Debug)] struct Row { @@ -157,9 +166,68 @@ pub(super) struct CopyIn { } impl Connection { - /// The checks of `DoCopy` and `BeginCopyFrom` for `COPY ... FROM STDIN`, then `CopyInResponse` and the checks of `CopyFrom`. The caller starts the copy with the state that this gives. - pub(super) fn copy_from(&self, stmt: &CopyStmt, out: &mut OutBuf) -> Result { - if stmt.filename.is_some() { + /// `COPY ... FROM` a file: the checks of `DoCopy`, then the copy reads the whole file where `BeginCopyFrom` opens it, and runs as a copy that got all the bytes at once. A program is not supported yet. + pub(super) fn copy_from_file( + &self, + stmt: &CopyStmt, + notices: &mut Vec, + ) -> Result { + if stmt.is_program { + return Err(Error::new( + SqlState::FEATURE_NOT_SUPPORTED, + "COPY FROM a program is not supported yet", + )); + } + if !self.superuser() { + return Err(Error::new( + SqlState::INSUFFICIENT_PRIVILEGE, + "permission denied to COPY from a file", + ) + .with_detail( + "Only roles with privileges of the \"pg_read_server_files\" role may COPY from a file.", + ) + .with_hint( + "Anyone can COPY to stdout or from stdin. psql's \\copy command also works for anyone.", + )); + } + let name = stmt.filename.as_deref().unwrap_or_default(); + let mut bytes = Vec::new(); + let mut copy = self.copy_from(stmt, Begin::File(name, &mut bytes))?; + let read = self.copy_data(&mut copy, &bytes); + let count = read.and_then(|()| self.copy_done(&mut copy)); + notices.append(&mut copy.notices); + Ok(Done::Count(CommandTag::Copy, count?)) + } + + /// `AllocateFile` and the check of `BeginCopyFrom` for a file of the server: the bytes of the file. A relative name is in the data directory, which is the working directory of the server. + fn read_server_file(&self, name: &str) -> Result, Error> { + let Some(io) = &self.io else { + return Err(Error::new( + SqlState::FEATURE_NOT_SUPPORTED, + "COPY FROM a file in a session with no server is not supported yet", + )); + }; + let path = Path::new(name); + let reading = |error: Error| { + let prefix = format!("could not open file \"{name}\": "); + let text = error.message().strip_prefix(&prefix).unwrap_or(error.message()).to_owned(); + Error::new(error.state(), format!("could not open file \"{name}\" for reading: {text}")) + }; + if let Ok(mode) = io.mode(path) + && !mode.regular + && io.read_dir(path).is_ok() + { + return Err(Error::new( + SqlState::WRONG_OBJECT_TYPE, + format!("\"{name}\" is a directory"), + )); + } + io.read_file(path).map_err(reading) + } + + /// The checks of `DoCopy` and `BeginCopyFrom` for `COPY ... FROM STDIN` or a file, then `CopyInResponse` or the read of the file, and the checks of `CopyFrom`. The caller starts the copy with the state that this gives. + pub(super) fn copy_from(&self, stmt: &CopyStmt, begin: Begin<'_>) -> Result { + if stmt.filename.is_some() && matches!(begin, Begin::Client(_)) { let what = if stmt.is_program { "a program" } else { "a file" }; return Err(Error::new( SqlState::FEATURE_NOT_SUPPORTED, @@ -262,7 +330,13 @@ impl Connection { let force_not_null = flags(&options.force_not_null)?; let force_null = flags(&options.force_null)?; let binary = options.format == Format::Binary; - out.copy_in_response(i8::from(binary), &vec![i16::from(binary); target.selected.len()]); + match begin { + Begin::Client(out) => out.copy_in_response( + i8::from(binary), + &vec![i16::from(binary); target.selected.len()], + ), + Begin::File(name, bytes) => *bytes = self.read_server_file(name)?, + } let name = &target.name; let wrong = |message: String| Error::new(SqlState::WRONG_OBJECT_TYPE, message); match target.kind { diff --git a/crates/rupg-session/src/connection/databases.rs b/crates/rupg-session/src/connection/databases.rs index ae2a4298..f001e697 100644 --- a/crates/rupg-session/src/connection/databases.rs +++ b/crates/rupg-session/src/connection/databases.rs @@ -265,7 +265,7 @@ impl Connection { } /// `superuser()` for the role of the session. - fn superuser(&self) -> bool { + pub(super) fn superuser(&self) -> bool { builtin::roles().iter().any(|role| role.oid == self.role && role.superuser) } diff --git a/crates/rupg-session/src/connection/extended.rs b/crates/rupg-session/src/connection/extended.rs index 8494f644..ed14ef31 100644 --- a/crates/rupg-session/src/connection/extended.rs +++ b/crates/rupg-session/src/connection/extended.rs @@ -552,6 +552,14 @@ impl Connection { let result = self.set_constraints(stmt, &mut notices); self.notices(&mut notices, &statement.text, &[], out); result? + } else if let Node::CopyStmt(stmt) = node + && stmt.is_from + && stmt.filename.is_some() + { + let mut notices = Vec::new(); + let result = self.copy_from_file(stmt, &mut notices); + self.notices(&mut notices, &statement.text, &[], out); + result? } else if let Node::CopyStmt(stmt) = node { let mut started = None; let result = self.copy(stmt, &mut started);