warren.set_owner: ownership transfer; re-check ownership after the lock (M10 step 5)

CHANGELOG.md

@@ -2,6 +2,28 @@
 
 ## 0.6.0 — in development
 
+### Added
+
+- **`warren.set_owner(name, owner regrole)`** gives a graph a new owner, as
+  `ALTER MATERIALIZED VIEW … OWNER TO` does (DESIGN.md §5.7, §14 item 43).
+  The caller must be a member of the current owner and able to `SET ROLE`
+  to the new one (on PostgreSQL 15, a member of it). The source queries
+  are opened as the new owner first, so one that cannot read them is
+  refused (42501), as are queries whose columns no longer fit (42804).
+  Versions and vectors stay as built until the next `refresh`; readers are
+  kept, except the new owner; the old owner keeps no access unless granted
+  (docs/semantics.md, "Access"). `ALTER EXTENSION pg_warren UPDATE`
+  creates it.
+
+### Fixed
+
+- `refresh`, `compact`, `promote`, `drop_version`, `grant`, `revoke`,
+  `drop_graph` and `contract` into an existing quotient check ownership
+  again once they hold the graph's row lock (`grant`, `revoke` and
+  `drop_graph` now take it). They checked only before it, so a call queued behind an ownership
+  transfer ran with an ownership its caller no longer had (harness test
+  `set_owner_race`).
+
 ## 0.5.0 — the shared cache (2026-10-23)
 
 Milestone M9 of DESIGN.md §13, the first on the road to 1.0. A loaded

README.md

@@ -290,8 +290,8 @@ See [CHANGELOG.md](CHANGELOG.md). This is pre-1.0 software: the SQL surface
 may still change between minor versions, with an upgrade script each time.
 
 Not there yet: the rest of the road to 1.0 (DESIGN.md §13) is text ids,
-ownership transfer, cleanup when an owner role is dropped, quotas and
-observability (M10);
+cleanup when an owner role is dropped, quotas and observability (M10;
+ownership transfer, `warren.set_owner`, is built for 0.6.0);
 benchmarks and robustness on real graphs and deployments, and packages
 (M11); and a reviewed, frozen SQL surface (M12, 1.0.0). Incremental
 maintenance, general sparse matrix algebra (contraction's `PᵀAP` is the

crates/pg_warren/src/access/error.rs

@@ -60,6 +60,18 @@ pub(crate) enum AccessError {
         /// The owner role's name.
         owner: String,
     },
+    /// The caller may not `SET ROLE` to the role it would make a graph's
+    /// owner (DESIGN.md §5.7).
+    #[error("must be able to SET ROLE \"{role}\" to make it the owner of graph \"{graph}\"")]
+    CannotBecome {
+        /// The would-be owner's name.
+        role: String,
+        /// The graph.
+        graph: String,
+    },
+    /// No role has this oid.
+    #[error("role with OID {0} does not exist")]
+    RoleNotFound(u32),
     /// The graph's owner role was dropped (DESIGN.md §7).
     #[error("the owner of graph \"{0}\" no longer exists")]
     OwnerDropped(String),
@@ -77,8 +89,9 @@ impl SqlError for AccessError {
             Self::NotOwner { .. }
             | Self::NotReader { .. }
             | Self::NotCreator { .. }
-            | Self::CreatorDropped { .. } => PgSqlErrorCode::ERRCODE_INSUFFICIENT_PRIVILEGE,
-            Self::OwnerDropped(_) | Self::ExtensionMissing => {
+            | Self::CreatorDropped { .. }
+            | Self::CannotBecome { .. } => PgSqlErrorCode::ERRCODE_INSUFFICIENT_PRIVILEGE,
+            Self::OwnerDropped(_) | Self::RoleNotFound(_) | Self::ExtensionMissing => {
                 PgSqlErrorCode::ERRCODE_UNDEFINED_OBJECT
             }
             Self::Spi(_) => PgSqlErrorCode::ERRCODE_INTERNAL_ERROR,

crates/pg_warren/src/access/guard.rs

@@ -294,6 +294,7 @@ mod tests {
         ("refresh", "warren.refresh('victim')"),
         ("revoke", "warren.revoke('victim', 'eve')"),
         ("s", "warren.s(NULL)"),
+        ("set_owner", "warren.set_owner('victim', 'eve')"),
         ("shared_cache", "warren.shared_cache()"),
         ("shortest_path", "warren.shortest_path('victim', 1::bigint, 2::bigint)"),
         ("similar_pairs", "warren.similar_pairs('victim', NULL::bigint)"),

crates/pg_warren/src/access/mod.rs

@@ -15,5 +15,6 @@ pub(crate) use error::AccessError;
 pub(crate) use guard::{entry, with_read_only, with_search_path, Caller, SqlError};
 pub(crate) use identity::{as_catalog_owner, as_role, current_user};
 pub(crate) use privilege::{
-    is_owner, is_reader, is_superuser, require_creator_or_owner, require_owner, require_reader,
+    is_owner, is_reader, is_superuser, require_can_become, require_creator_or_owner, require_owner,
+    require_reader,
 };

crates/pg_warren/src/access/privilege.rs

@@ -76,6 +76,34 @@ pub(crate) fn require_creator_or_owner(
     }
 }
 
+/// Checks that the caller may make `role` the owner of graph `graph`, as
+/// `ALTER … OWNER TO` checks a new owner (DESIGN.md §5.7): on PG 16+, the
+/// caller can `SET ROLE` to it (`member_can_set_role`, the test of
+/// `check_can_set_role`); on PG 15, which has no `SET` option, it is a
+/// member of it. A superuser may always.
+pub(crate) fn require_can_become(role: pg_sys::Oid, graph: &str) -> Result<(), AccessError> {
+    let name = role_name(role).ok_or(AccessError::RoleNotFound(role.to_u32()))?;
+    match can_set_role(role) {
+        true => Ok(()),
+        false => Err(AccessError::CannotBecome { role: name, graph: graph.to_owned() }),
+    }
+}
+
+/// Whether the current user may `SET ROLE` to `role`.
+#[cfg(not(feature = "pg15"))]
+fn can_set_role(role: pg_sys::Oid) -> bool {
+    // SAFETY: reads the current user and its cached role memberships, on
+    // the main thread.
+    unsafe { pg_sys::member_can_set_role(pg_sys::GetUserId(), role) }
+}
+
+/// Whether the current user may `SET ROLE` to `role`: on PG 15, whether
+/// it is a member of it.
+#[cfg(feature = "pg15")]
+fn can_set_role(role: pg_sys::Oid) -> bool {
+    is_member(role)
+}
+
 /// Whether the caller is a superuser.
 pub(crate) fn is_superuser() -> Result<bool, AccessError> {
     Ok(read_one::<bool>(

crates/pg_warren/src/build/refresh.rs

@@ -30,7 +30,7 @@ fn refresh_graph(name: &str, promote: bool, label: Option<&str>) -> Result<Versi
     let mut graph = GraphDefinition::find(name)?;
     require_owner(graph.owner, &graph.name, "refresh")?;
     graph.require_query_source("refresh")?;
-    graph.lock()?;
+    graph.lock("refresh")?;
     graph.require_sound_current_version()?;
     if let Some(base) = graph.current_version {
         require_current_id_space(&graph, base)?;

crates/pg_warren/src/build/spi_source.rs

@@ -10,8 +10,8 @@ use super::source::{EdgeRow, EdgeSink, SourceReader};
 use crate::access::{as_role, with_read_only, with_search_path};
 use crate::catalog::GraphDefinition;
 use crate::source::{
-    check_edge_columns, check_node_columns, for_each_row, for_each_tuple, open_read_only,
-    text_in_place, IdDatum, SourceError,
+    check_edge_columns, for_each_row, for_each_tuple, open_read_only, require_node_id_type,
+    text_in_place, IdDatum,
 };
 
 /// A graph's sources, read through SPI.
@@ -35,16 +35,7 @@ impl<E: IdDatum> SourceReader<E> for SpiSource<'_> {
                     Spi::connect(|client| {
                         let mut opened =
                             open_read_only::<BuildError>(client, "node_query", &graph.node_query)?;
-                        if check_node_columns(&opened.columns)? != graph.id_type {
-                            return Err(SourceError::ColumnType {
-                                query: "node_query",
-                                column: 1,
-                                role: "node id",
-                                expected: graph.id_type.sql_name(),
-                                found: opened.columns[0].name.clone(),
-                            }
-                            .into());
-                        }
+                        require_node_id_type(&opened.columns, graph.id_type)?;
                         let mut ids = BuildBuffer::new(memory);
                         for_each_row::<BuildError>(&mut opened.cursor, |row| {
                             if let Some(id) = row.get::<E::Sql>(1)?.map(E::from_sql) {

crates/pg_warren/src/catalog/create_graph.rs

@@ -9,7 +9,8 @@ use crate::read_only::read_one;
 use crate::source::{check_edge_columns, check_node_columns, source_columns};
 
 /// Of the schemas in `$1`, those, in order and quoted, in which PUBLIC or
-/// a role other than the caller that is not a superuser and not predefined
+/// a role other than `$2`, the graph's owner, that is not a superuser and
+/// not predefined
 /// (`pg_*`; `pg_database_owner` counts through the database's owner) may
 /// create objects. (Membership is not excluded: a superuser is a member of
 /// every role, and a group role's other members could create there too.)
@@ -24,7 +25,7 @@ const WRITABLE_SCHEMAS: &str = "SELECT pg_catalog.string_agg(pg_catalog.quote_id
        AND (pg_catalog.has_schema_privilege('public', n.oid, 'CREATE')
             OR EXISTS (SELECT FROM pg_catalog.pg_roles r
                        WHERE NOT r.rolsuper AND NOT pg_catalog.starts_with(r.rolname, 'pg_')
-                         AND r.rolname <> pg_catalog.current_user()
+                         AND r.oid <> $2
                          AND pg_catalog.has_schema_privilege(r.oid, n.oid, 'CREATE')))";
 
 /// `warren.create_graph`: records graph `name` over `node_query` and
@@ -86,27 +87,30 @@ fn record(graph: &NewGraph<'_>) -> Result<(), CatalogError> {
     let id_type = check_node_columns(&node_columns)?;
     check_edge_columns(&edge_columns, id_type, graph.has_relations, graph.weighted)?;
     NewGraph { id_type: id_type.oid(), ..*graph }.create()?;
-    if let Some(schemas) = writable_schemas(graph.search_path)? {
+    if let Some(schemas) = writable_schemas(graph.search_path, graph.owner)? {
         notice!("{}", writable_notice(graph.name, graph.search_path, &schemas));
     }
     Ok(())
 }
 
-/// The schemas of `search_path` that roles other than the caller can create
+/// The schemas of `search_path` that roles other than `owner` can create
 /// objects in (`WRITABLE_SCHEMAS`), comma-separated, or `None`. A function
 /// or operator planted there can capture an unqualified name in the source
 /// queries, which every refresh runs as the graph owner (DESIGN.md §7).
-fn writable_schemas(search_path: &str) -> Result<Option<String>, CatalogError> {
+pub(super) fn writable_schemas(
+    search_path: &str,
+    owner: pg_sys::Oid,
+) -> Result<Option<String>, CatalogError> {
     // Only the effective schemas are read under that search_path; the
     // check runs under the pinned one.
     let schemas = with_search_path(search_path, || {
         read_one::<Vec<String>>("SELECT pg_catalog.current_schemas(false)::pg_catalog.text[]", &[])
     })?
     .unwrap_or_default();
-    Ok(read_one::<String>(WRITABLE_SCHEMAS, &[schemas.into()])?)
+    Ok(read_one::<String>(WRITABLE_SCHEMAS, &[schemas.into(), owner.into()])?)
 }
 
-fn writable_notice(graph: &str, search_path: &str, schemas: &str) -> String {
+pub(super) fn writable_notice(graph: &str, search_path: &str, schemas: &str) -> String {
     format!(
         "graph \"{graph}\" runs its source queries under search_path \"{search_path}\", in which \
          other roles can create objects (in {schemas}): a function or operator created there \
@@ -140,7 +144,10 @@ mod tests {
              CREATE SCHEMA cg_shared; GRANT CREATE ON SCHEMA cg_shared TO cg_other;
              CREATE SCHEMA cg_private; GRANT USAGE ON SCHEMA cg_private TO cg_other",
         )?;
-        let found = |path: &str| super::writable_schemas(path).map_err(|_| Error::InvalidPosition);
+        let found = |path: &str| {
+            super::writable_schemas(path, crate::access::current_user())
+                .map_err(|_| Error::InvalidPosition)
+        };
         assert_eq!(found("cg_private, cg_shared, public")?.as_deref(), Some("cg_shared"));
         assert_eq!(found("cg_private")?, None, "only the owner (a superuser) can create there");
         assert_eq!(found("public")?, None, "PG 15+: only the database owner, here a superuser");
@@ -151,7 +158,8 @@ mod tests {
     #[pg_test]
     fn a_schema_public_can_create_in_is_found_without_other_roles() -> Result<(), Error> {
         Spi::run("CREATE SCHEMA cg_open; GRANT CREATE ON SCHEMA cg_open TO PUBLIC")?;
-        let found = super::writable_schemas("cg_open").map_err(|_| Error::InvalidPosition)?;
+        let found = super::writable_schemas("cg_open", crate::access::current_user())
+            .map_err(|_| Error::InvalidPosition)?;
         assert_eq!(found.as_deref(), Some("cg_open"));
         Ok(())
     }
@@ -161,7 +169,8 @@ mod tests {
     #[pg_test]
     fn the_sessions_temporary_schema_is_not_found() -> Result<(), Error> {
         Spi::run("CREATE ROLE cg_temp_other; CREATE TEMP TABLE cg_temp (x int)")?;
-        let found = super::writable_schemas("pg_temp").map_err(|_| Error::InvalidPosition)?;
+        let found = super::writable_schemas("pg_temp", crate::access::current_user())
+            .map_err(|_| Error::InvalidPosition)?;
         assert_eq!(found, None);
         Ok(())
     }
@@ -171,7 +180,8 @@ mod tests {
         Spi::run(
             r#"CREATE ROLE cg_quoted; CREATE SCHEMA "cg Odd"; GRANT CREATE ON SCHEMA "cg Odd" TO cg_quoted"#,
         )?;
-        let found = super::writable_schemas(r#""cg Odd""#).map_err(|_| Error::InvalidPosition)?;
+        let found = super::writable_schemas(r#""cg Odd""#, crate::access::current_user())
+            .map_err(|_| Error::InvalidPosition)?;
         assert_eq!(found.as_deref(), Some(r#""cg Odd""#));
         Ok(())
     }

crates/pg_warren/src/catalog/drop.rs

@@ -18,7 +18,7 @@ fn drop_version(name: &str, version: i64) {
 fn drop_version_by_name(name: &str, version: VersionId) -> Result<(), CatalogError> {
     let mut graph = GraphDefinition::find(name)?;
     require_owner(graph.owner, &graph.name, "drop versions of")?;
-    graph.lock()?;
+    graph.lock("drop versions of")?;
     delete_version(&graph, version)
 }
 
@@ -31,11 +31,17 @@ fn drop_graph(name: &str) {
 }
 
 fn drop_graph_by_name(name: &str) -> Result<(), CatalogError> {
-    let graph = GraphDefinition::find(name)?;
+    let mut graph = GraphDefinition::find(name)?;
     match require_owner(graph.owner, &graph.name, "drop") {
         Err(AccessError::OwnerDropped(_)) if is_superuser()? => {}
         checked => checked?,
     }
+    // Checks ownership again once it holds the row: a set_owner may have
+    // held it (DESIGN.md §5.7). The lock is held whatever the check says.
+    match graph.lock("drop") {
+        Err(CatalogError::Access(AccessError::OwnerDropped(_))) if is_superuser()? => {}
+        locked => locked?,
+    }
     graph.drop_graph()
 }
 

crates/pg_warren/src/catalog/graph.rs

@@ -8,7 +8,7 @@ use pgrx::PgOid;
 
 use super::version::VersionId;
 use super::CatalogError;
-use crate::access::as_catalog_owner;
+use crate::access::{as_catalog_owner, require_owner};
 use crate::read_only::read_one;
 use crate::source::IdType;
 use warren_core::graph::NodeOrder;
@@ -37,9 +37,11 @@ const STILL_THERE: &str =
     "SELECT EXISTS (SELECT FROM warren.versions WHERE version_id = $1 AND graph_id = $2)";
 
 /// [`GraphDefinition::lock`]'s re-read once it holds the lock:
-/// `current_version` as in [`FIND`], and `node_order`, which a compaction
-/// that held the lock may have changed.
-const REREAD: &str = "SELECT g.current_version, v.version_id, g.node_order
+/// `current_version` as in [`FIND`], `node_order`, which a compaction that
+/// held the lock may have changed, and `owner` and `readers`, which a
+/// `set_owner` that held it may have changed.
+const REREAD: &str = "SELECT g.current_version, v.version_id, g.node_order, g.owner::oid,
+                             g.readers::oid[]
                       FROM warren.graphs g
                       LEFT JOIN warren.versions v
                              ON v.version_id = g.current_version AND v.graph_id = g.graph_id
@@ -209,30 +211,42 @@ impl GraphDefinition {
     /// Locks the graph's row until the end of the transaction, so that
     /// refreshes of one graph run one at a time, and re-reads its current
     /// version and node order, which a refresh or compaction that held the
-    /// lock may have changed. A graph dropped since [`find`](Self::find) is
-    /// not found.
-    pub(crate) fn lock(&mut self) -> Result<(), CatalogError> {
+    /// lock may have changed, and its owner and readers, which a
+    /// `set_owner` that held it may have changed. Then checks again that
+    /// the caller is a member of the owner role, to do `action`: a check
+    /// made before the lock may have been of the owner a transfer replaced
+    /// (DESIGN.md §5.7). A graph dropped since [`find`](Self::find) is not
+    /// found.
+    pub(crate) fn lock(&mut self, action: &'static str) -> Result<(), CatalogError> {
         // A row lock is a write (it assigns a transaction id): only the
         // functions that build or drop versions take it. The lock and the
         // re-read are two statements: when the lock waits for another
         // transaction's change to the row, READ COMMITTED re-checks only
         // the locked row (EvalPlanQual), so a version joined in the same
         // statement would be the one from before the wait. The re-read,
         // as a statement of its own, takes a snapshot after the lock.
-        let (stored, current, node_order) = as_catalog_owner(|| {
+        let (stored, current, node_order, owner, readers) = as_catalog_owner(|| {
             Spi::connect_mut(|client| {
                 let locked = client.update(LOCK, Some(1), &[self.graph_id.into()])?.len();
                 if locked == 0 {
                     return Err(CatalogError::GraphNotFound(self.name.clone()));
                 }
                 let row = client.update(REREAD, Some(1), &[self.graph_id.into()])?.first();
-                Ok((row.get::<i64>(1)?, row.get::<i64>(2)?, row.get::<String>(3)?))
+                Ok((
+                    row.get::<i64>(1)?,
+                    row.get::<i64>(2)?,
+                    row.get::<String>(3)?,
+                    row.get::<pg_sys::Oid>(4)?,
+                    row.get::<Vec<pg_sys::Oid>>(5)?,
+                ))
             })
         })?;
         self.node_order = parse_node_order(&node_order.unwrap_or_default())?;
         self.current_version = current;
         self.stray_current_version = stray(stored, current);
-        Ok(())
+        self.owner = owner.unwrap_or(pg_sys::Oid::INVALID);
+        self.readers = readers.unwrap_or_default();
+        Ok(require_owner(self.owner, &self.name, action)?)
     }
 
     /// Locks the graph's row `FOR SHARE` until the end of the transaction,
@@ -374,6 +388,29 @@ impl GraphDefinition {
             })
         })
     }
+
+    /// Makes `owner` the graph's owner, and removes it from the readers,
+    /// whose order is kept (DESIGN.md §5.7). Its row must be locked
+    /// ([`lock`](Self::lock)).
+    pub(crate) fn set_owner(&mut self, owner: pg_sys::Oid) -> Result<(), CatalogError> {
+        as_catalog_owner(|| {
+            Spi::connect_mut(|client| {
+                client.update(
+                    "UPDATE warren.graphs SET owner = $2::oid::regrole, readers = ARRAY(
+                         SELECT r.reader FROM pg_catalog.unnest(readers)
+                                              WITH ORDINALITY AS r (reader, position)
+                         WHERE r.reader::oid <> $2 ORDER BY r.position)::regrole[]
+                     WHERE graph_id = $1",
+                    None,
+                    &[self.graph_id.into(), owner.into()],
+                )?;
+                Ok::<_, CatalogError>(())
+            })
+        })?;
+        self.owner = owner;
+        self.readers.retain(|&reader| reader != owner);
+        Ok(())
+    }
 }
 
 /// The node order named `name` (DESIGN.md §14 item 22), or an

crates/pg_warren/src/catalog/mod.rs

@@ -12,6 +12,7 @@ mod drop;
 mod error;
 mod graph;
 mod info;
+mod owner;
 mod promote;
 mod readers;
 mod retention;