sql: Implement grant and revoke privilege

src/adapter/src/catalog/builtin_table_updates.rs

@@ -7,6 +7,7 @@
 // the Business Source License, use of this software will be governed
 // by the Apache License, Version 2.0.
 
+use std::collections::BTreeMap;
 use std::net::Ipv4Addr;
 
 use bytesize::ByteSize;
@@ -24,10 +25,11 @@ use mz_ore::cast::CastFrom;
 use mz_ore::collections::CollectionExt;
 use mz_repr::adt::array::ArrayDimension;
 use mz_repr::adt::jsonb::Jsonb;
+use mz_repr::adt::mz_acl_item::MzAclItem;
 use mz_repr::role_id::RoleId;
 use mz_repr::{Datum, Diff, GlobalId, Row};
 use mz_sql::ast::{CreateIndexStatement, Statement};
-use mz_sql::catalog::{CatalogDatabase, CatalogType, TypeCategory};
+use mz_sql::catalog::{CatalogCluster, CatalogDatabase, CatalogSchema, CatalogType, TypeCategory};
 use mz_sql::func::FuncImplCatalogDetails;
 use mz_sql::names::{ResolvedDatabaseSpecifier, SchemaId, SchemaSpecifier};
 use mz_sql_parser::ast::display::AstDisplay;
@@ -88,19 +90,7 @@ impl CatalogState {
         database: &Database,
         diff: Diff,
     ) -> BuiltinTableUpdate {
-        let mut row = Row::default();
-        row.packer()
-            .push_array(
-                &[ArrayDimension {
-                    lower_bound: 1,
-                    length: database.privileges.len(),
-                }],
-                database
-                    .privileges
-                    .iter()
-                    .map(|mz_acl_item| Datum::MzAclItem(mz_acl_item.clone())),
-            )
-            .expect("privileges is 1 dimensional, and its length is used for the array length");
+        let row = self.pack_privilege_array_row(database.privileges());
         let privileges = row.unpack_first();
         BuiltinTableUpdate {
             id: self.resolve_builtin_table(&MZ_DATABASES),
@@ -128,19 +118,7 @@ impl CatalogState {
                 &self.database_by_id[id].schemas_by_id[schema_id],
             ),
         };
-        let mut row = Row::default();
-        row.packer()
-            .push_array(
-                &[ArrayDimension {
-                    lower_bound: 1,
-                    length: schema.privileges.len(),
-                }],
-                schema
-                    .privileges
-                    .iter()
-                    .map(|mz_acl_item| Datum::MzAclItem(mz_acl_item.clone())),
-            )
-            .expect("privileges is 1 dimensional, and its length is used for the array length");
+        let row = self.pack_privilege_array_row(schema.privileges());
         let privileges = row.unpack_first();
         BuiltinTableUpdate {
             id: self.resolve_builtin_table(&MZ_SCHEMAS),
@@ -205,19 +183,7 @@ impl CatalogState {
     pub(super) fn pack_cluster_update(&self, name: &str, diff: Diff) -> BuiltinTableUpdate {
         let id = self.clusters_by_name[name];
         let cluster = &self.clusters_by_id[&id];
-        let mut row = Row::default();
-        row.packer()
-            .push_array(
-                &[ArrayDimension {
-                    lower_bound: 1,
-                    length: cluster.privileges.len(),
-                }],
-                cluster
-                    .privileges
-                    .iter()
-                    .map(|mz_acl_item| Datum::MzAclItem(mz_acl_item.clone())),
-            )
-            .expect("privileges is 1 dimensional, and its length is used for the array length");
+        let row = self.pack_privilege_array_row(cluster.privileges());
         let privileges = row.unpack_first();
         BuiltinTableUpdate {
             id: self.resolve_builtin_table(&MZ_CLUSTERS),
@@ -331,20 +297,7 @@ impl CatalogState {
             .id;
         let name = &entry.name().item;
         let owner_id = entry.owner_id();
-        let mut privileges_row = Row::default();
-        privileges_row
-            .packer()
-            .push_array(
-                &[ArrayDimension {
-                    lower_bound: 1,
-                    length: entry.privileges.len(),
-                }],
-                entry
-                    .privileges
-                    .iter()
-                    .map(|mz_acl_item| Datum::MzAclItem(mz_acl_item.clone())),
-            )
-            .expect("privileges is 1 dimensional, and its length is used for the array length");
+        let privileges_row = self.pack_privilege_array_row(entry.privileges());
         let privileges = privileges_row.unpack_first();
         let mut updates = match entry.item() {
             CatalogItem::Log(_) => self.pack_source_update(
@@ -1278,4 +1231,21 @@ impl CatalogState {
             diff,
         }
     }
+
+    fn pack_privilege_array_row(&self, privileges: &BTreeMap<RoleId, Vec<MzAclItem>>) -> Row {
+        let mut row = Row::default();
+        let flat_privileges = MzAclItem::flatten(privileges);
+        row.packer()
+            .push_array(
+                &[ArrayDimension {
+                    lower_bound: 1,
+                    length: flat_privileges.len(),
+                }],
+                flat_privileges
+                    .into_iter()
+                    .map(|mz_acl_item| Datum::MzAclItem(mz_acl_item.clone())),
+            )
+            .expect("privileges is 1 dimensional, and its length is used for the array length");
+        row
+    }
 }

src/adapter/src/catalog/error.rs

@@ -52,6 +52,8 @@ pub enum ErrorKind {
     ReservedReplicaName(String),
     #[error("system cluster '{0}' cannot be modified")]
     ReadOnlyCluster(String),
+    #[error("system database '{0}' cannot be modified")]
+    ReadOnlyDatabase(String),
     #[error("system schema '{0}' cannot be modified")]
     ReadOnlySystemSchema(String),
     #[error("system item '{0}' cannot be modified")]

src/adapter/src/catalog/storage.rs

@@ -2067,7 +2067,7 @@ impl<'a> Transaction<'a> {
                     name: cluster.name().to_string(),
                     linked_object_id: cluster.linked_object_id(),
                     owner_id: cluster.owner_id,
-                    privileges: Some(cluster.privileges.clone()),
+                    privileges: Some(MzAclItem::flatten(cluster.privileges())),
                 })
             } else {
                 None
@@ -2129,7 +2129,7 @@ impl<'a> Transaction<'a> {
                 Some(DatabaseValue {
                     name: database.name().to_string(),
                     owner_id: database.owner_id,
-                    privileges: Some(database.privileges.clone()),
+                    privileges: Some(MzAclItem::flatten(database.privileges())),
                 })
             } else {
                 None
@@ -2167,7 +2167,7 @@ impl<'a> Transaction<'a> {
                     database_ns: db_ns,
                     name: schema.name().schema.clone(),
                     owner_id: schema.owner_id,
-                    privileges: Some(schema.privileges.clone()),
+                    privileges: Some(MzAclItem::flatten(schema.privileges())),
                 })
             } else {
                 None

src/adapter/src/command.rs

@@ -317,6 +317,8 @@ pub enum ExecuteResponse {
         /// How long to wait for results to arrive.
         timeout: ExecuteTimeout,
     },
+    /// The requested privilege was granted.
+    GrantedPrivilege,
     /// The requested role was granted.
     GrantedRole,
     /// The specified number of rows were inserted into the requested table.
@@ -325,6 +327,8 @@ pub enum ExecuteResponse {
     Prepare,
     /// A user-requested warning was raised.
     Raised,
+    /// The requested privilege was revoked.
+    RevokedPrivilege,
     /// The requested role was revoked.
     RevokedRole,
     /// Rows will be delivered via the specified future.
@@ -389,6 +393,7 @@ impl ExecuteResponse {
             DroppedObject(o) => Some(format!("DROP {o}")),
             EmptyQuery => None,
             Fetch { .. } => None,
+            GrantedPrivilege => Some("GRANT".into()),
             GrantedRole => Some("GRANT ROLE".into()),
             Inserted(n) => {
                 // "On successful completion, an INSERT command returns a
@@ -402,6 +407,7 @@ impl ExecuteResponse {
             }
             Prepare => Some("PREPARE".into()),
             Raised => Some("RAISE".into()),
+            RevokedPrivilege => Some("REVOKE".into()),
             RevokedRole => Some("REVOKE ROLE".into()),
             SendingRows { .. } => None,
             SetVariable { reset: true, .. } => Some("RESET".into()),
@@ -461,11 +467,13 @@ impl ExecuteResponse {
             }
             Execute | ReadThenWrite => vec![Deleted, Inserted, SendingRows, Updated],
             PlanKind::Fetch => vec![ExecuteResponseKind::Fetch],
+            GrantPrivilege => vec![GrantedPrivilege],
             GrantRole => vec![GrantedRole],
             CopyRows => vec![Inserted],
             Insert => vec![Inserted, SendingRows],
             PlanKind::Prepare => vec![ExecuteResponseKind::Prepare],
             PlanKind::Raise => vec![ExecuteResponseKind::Raised],
+            RevokePrivilege => vec![RevokedPrivilege],
             RevokeRole => vec![RevokedRole],
             PlanKind::SetVariable | ResetVariable => vec![ExecuteResponseKind::SetVariable],
             PlanKind::Subscribe => vec![Subscribing, CopyTo],

src/adapter/src/coord/ddl.rs

@@ -960,9 +960,11 @@ impl Coordinator {
                 | Op::AlterSink { .. }
                 | Op::AlterSource { .. }
                 | Op::DropTimeline(_)
+                | Op::GrantPrivilege { .. }
                 | Op::GrantRole { .. }
                 | Op::RenameItem { .. }
                 | Op::UpdateOwner { .. }
+                | Op::RevokePrivilege { .. }
                 | Op::RevokeRole { .. }
                 | Op::UpdateClusterReplicaStatus { .. }
                 | Op::UpdateStorageUsage { .. }

src/adapter/src/coord/introspection.rs

@@ -209,6 +209,8 @@ pub fn user_privilege_hack(
         | Plan::RotateKeys(_)
         | Plan::GrantRole(_)
         | Plan::RevokeRole(_)
+        | Plan::GrantPrivilege(_)
+        | Plan::RevokePrivilege(_)
         | Plan::CopyRows(_) => {
             return Err(AdapterError::Unauthorized(
                 rbac::UnauthorizedError::Privilege {

src/adapter/src/coord/sequencer.rs

@@ -421,6 +421,18 @@ impl Coordinator {
             Plan::RotateKeys(RotateKeysPlan { id }) => {
                 tx.send(self.sequence_rotate_keys(&session, id).await, session);
             }
+            Plan::GrantPrivilege(plan) => {
+                tx.send(
+                    self.sequence_grant_privilege(&mut session, plan).await,
+                    session,
+                );
+            }
+            Plan::RevokePrivilege(plan) => {
+                tx.send(
+                    self.sequence_revoke_privilege(&mut session, plan).await,
+                    session,
+                );
+            }
             Plan::GrantRole(plan) => {
                 tx.send(self.sequence_grant_role(&mut session, plan).await, session);
             }