Run COPY FROM a file of the server

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)
 

crates/rupg-server/src/lib.rs

@@ -140,6 +140,8 @@ struct Shared {
     tasks: Arc<dyn Tasks>,
     /// The catalog that all the sessions share.
     store: Arc<Store>,
+    /// The file system, which `COPY ... FROM` a file reads.
+    io: Arc<dyn Io>,
     /// The cancel key and the cancel flag of each session, by process ID.
     keys: Mutex<BTreeMap<i32, Live>>,
     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 {

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<Store>,
+    /// The file system of the server, which `COPY ... FROM` a file reads. A session with no server has none.
+    io: Option<Arc<dyn Io>>,
     /// The lock of the store and the copy of the catalog, while the transaction changes the catalog.
     lease: Option<Lease>,
     /// 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<dyn Io>) {
+        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<Store>) {
         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"),
                 (

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<u8>),
+}
+
 /// 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<CopyIn, Error> {
-        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<Notice>,
+    ) -> Result<Done, Error> {
+        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<Vec<u8>, 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<CopyIn, Error> {
+        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 {

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)
     }
 

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);