sql: Correct ownership semantics of indexes

src/adapter/src/catalog.rs

@@ -2451,11 +2451,16 @@ impl CatalogEntry {
         matches!(self.item(), CatalogItem::Secret(_))
     }
 
-    /// Reports whether this catalog entry is a introspection source.
+    /// Reports whether this catalog entry is an introspection source.
     pub fn is_introspection_source(&self) -> bool {
         matches!(self.item(), CatalogItem::Log(_))
     }
 
+    /// Reports whether this catalog entry is an index.
+    pub fn is_index(&self) -> bool {
+        matches!(self.item(), CatalogItem::Index(_))
+    }
+
     /// Reports whether this catalog entry can be treated as a relation, it can produce rows.
     pub fn is_relation(&self) -> bool {
         match self.item {

src/adapter/src/coord/sequencer/inner.rs

@@ -1330,12 +1330,15 @@ impl Coordinator {
             custom_logical_compaction_window: None,
         };
         let oid = self.catalog_mut().allocate_oid()?;
+        let on = self.catalog().get_entry(&index.on);
+        // Indexes have the same owner as their parent relation.
+        let owner_id = *on.owner_id();
         let op = catalog::Op::CreateItem {
             id,
             oid,
             name: name.clone(),
             item: CatalogItem::Index(index),
-            owner_id: *session.role_id(),
+            owner_id,
         };
         match self
             .catalog_transact_with(Some(session), vec![op], |txn| {
@@ -3866,7 +3869,39 @@ impl Coordinator {
             new_owner,
         }: AlterOwnerPlan,
     ) -> Result<ExecuteResponse, AdapterError> {
-        self.catalog_transact(Some(session), vec![Op::UpdateOwner { id, new_owner }])
+        let entry = if let ObjectId::Item(global_id) = &id {
+            Some(self.catalog().get_entry(global_id))
+        } else {
+            None
+        };
+
+        // Cannot directly change the owner of an index.
+        if let Some(entry) = &entry {
+            if entry.is_index() {
+                let name = self
+                    .catalog()
+                    .resolve_full_name(entry.name(), Some(session.conn_id()))
+                    .to_string();
+                session.add_notice(AdapterNotice::AlterIndexOwner { name });
+                return Ok(ExecuteResponse::AlteredObject(object_type));
+            }
+        }
+
+        let mut ops = vec![Op::UpdateOwner { id, new_owner }];
+        // Alter owner cascades down to dependent indexes.
+        if let Some(entry) = entry {
+            let dependent_index_ops = entry
+                .used_by()
+                .into_iter()
+                .filter(|id| self.catalog().get_entry(id).is_index())
+                .map(|id| Op::UpdateOwner {
+                    id: ObjectId::Item(*id),
+                    new_owner,
+                });
+            ops.extend(dependent_index_ops);
+        }
+
+        self.catalog_transact(Some(session), ops)
             .await
             .map(|_| ExecuteResponse::AlteredObject(object_type))
     }

src/adapter/src/notice.rs

@@ -90,6 +90,9 @@ pub enum AdapterNotice {
         member_name: String,
     },
     AutoRunOnIntrospectionCluster,
+    AlterIndexOwner {
+        name: String,
+    },
 }
 
 impl AdapterNotice {
@@ -114,6 +117,7 @@ impl AdapterNotice {
             },
             AdapterNotice::RbacSystemDisabled => Some("To enable RBAC please reach out to support with a request to turn RBAC on.".into()),
             AdapterNotice::RbacUserDisabled => Some("To enable RBAC globally run `ALTER SYSTEM SET enable_rbac_checks TO TRUE` as a superuser. TO enable RBAC for just this session run `SET enable_session_rbac_checks TO TRUE`.".into()),
+            AdapterNotice::AlterIndexOwner {name: _} => Some("Change the ownership of the index's relation, instead.".into()),
             _ => None
         }
     }
@@ -230,6 +234,9 @@ impl fmt::Display for AdapterNotice {
                 f,
                 "query was automatically run on the \"mz_introspection\" cluster"
             ),
+            AdapterNotice::AlterIndexOwner { name } => {
+                write!(f, "cannot change owner of {}", name.quoted())
+            }
         }
     }
 }

src/adapter/src/rbac.rs

@@ -309,7 +309,6 @@ fn generate_required_ownership(plan: &Plan) -> Vec<ObjectId> {
         | Plan::CreateSecret(_)
         | Plan::CreateSink(_)
         | Plan::CreateTable(_)
-        | Plan::CreateIndex(_)
         | Plan::CreateType(_)
         | Plan::DiscardTemp
         | Plan::DiscardAll
@@ -344,6 +343,7 @@ fn generate_required_ownership(plan: &Plan) -> Vec<ObjectId> {
         | Plan::Raise(_)
         | Plan::GrantRole(_)
         | Plan::RevokeRole(_) => Vec::new(),
+        Plan::CreateIndex(plan) => vec![ObjectId::Item(plan.index.on)],
         Plan::CreateView(CreateViewPlan { replace, .. })
         | Plan::CreateMaterializedView(CreateMaterializedViewPlan { replace, .. }) => {
             replace.iter().map(|id| ObjectId::Item(*id)).collect()

src/pgwire/src/message.rs

@@ -434,6 +434,7 @@ impl ErrorResponse {
             AdapterNotice::RoleMembershipAlreadyExists { .. } => SqlState::WARNING,
             AdapterNotice::RoleMembershipDoesNotExists { .. } => SqlState::WARNING,
             AdapterNotice::AutoRunOnIntrospectionCluster => SqlState::WARNING,
+            AdapterNotice::AlterIndexOwner { .. } => SqlState::WARNING,
         };
         ErrorResponse {
             severity: Severity::for_adapter_notice(&notice),
@@ -596,6 +597,7 @@ impl Severity {
             AdapterNotice::RoleMembershipAlreadyExists { .. } => Severity::Notice,
             AdapterNotice::RoleMembershipDoesNotExists { .. } => Severity::Warning,
             AdapterNotice::AutoRunOnIntrospectionCluster => Severity::Debug,
+            AdapterNotice::AlterIndexOwner { .. } => Severity::Warning,
         }
     }
 }

test/sqllogictest/object_ownership.slt

@@ -219,13 +219,14 @@ ALTER INDEX mt_ind OWNER TO group_no_one
 statement ok
 ALTER INDEX mt_ind OWNER TO group_materialize
 
+# Altering the owner of an index is a no-op
 query T
 SELECT mz_roles.name
   FROM mz_indexes
   LEFT JOIN mz_roles ON mz_indexes.owner_id = mz_roles.id
   WHERE mz_indexes.name = 'mt_ind'
 ----
-group_materialize
+materialize
 
 statement ok
 ALTER INDEX mt_ind SET (LOGICAL COMPACTION WINDOW = 0)
@@ -961,8 +962,10 @@ DROP ROLE owner
 statement ok
 CREATE ROLE owner CREATEDB CREATECLUSTER
 
-statement ok
+simple conn=owner2,user=owner
 create TABLE t (a INT);
+----
+COMPLETE 0
 
 simple conn=owner2,user=owner
 CREATE INDEX ind ON t(a);
@@ -973,7 +976,10 @@ simple conn=materialize,user=materialize
 DROP ROLE owner
 ----
 db error: ERROR: role "owner" cannot be dropped because some objects depend on it
-DETAIL: owner: owner of index ind
+DETAIL: owner: owner of table t
+owner: privileges on table t granted by owner
+owner: privileges granted on table t to owner
+owner: owner of index ind
 owner: privileges on index ind granted by owner
 owner: privileges granted on index ind to owner
 
@@ -982,11 +988,13 @@ DROP INDEX ind;
 ----
 COMPLETE 0
 
-statement ok
-DROP ROLE owner
+simple conn=owner2,user=owner
+DROP TABLE t;
+----
+COMPLETE 0
 
 statement ok
-DROP TABLE t
+DROP ROLE owner
 
 ## Sources
 
@@ -1400,18 +1408,8 @@ DROP TABLE t CASCADE;
 statement ok
 CREATE VIEW v AS SELECT 1 AS a;
 
-simple conn=joe,user=joe
+statement ok
 CREATE INDEX i ON v(a);
-----
-COMPLETE 0
-
-statement error must be owner of INDEX materialize.public.i
-CREATE OR REPLACE VIEW v AS SELECT 2 AS a;
-
-simple conn=mz_system,user=mz_system
-ALTER INDEX i OWNER TO materialize;
-----
-COMPLETE 0
 
 statement ok
 CREATE OR REPLACE VIEW v AS SELECT 2 AS a;
@@ -1422,24 +1420,55 @@ DROP VIEW v;
 statement ok
 CREATE MATERIALIZED VIEW mv AS SELECT 1 AS a;
 
-simple conn=joe,user=joe
+statement ok
 CREATE INDEX i ON mv(a);
-----
-COMPLETE 0
 
-statement error must be owner of INDEX materialize.public.i
+statement ok
 CREATE OR REPLACE MATERIALIZED VIEW mv AS SELECT 2 AS a;
 
+statement ok
+DROP MATERIALIZED VIEW mv;
+
+# Test that index owners are consistent with their underlying relation
+
+statement ok
+CREATE TABLE t (a INT)
+
+simple conn=joe,user=joe
+CREATE INDEX i1 ON t (a);
+----
+db error: ERROR: must be owner of TABLE materialize.public.t
+
 simple conn=mz_system,user=mz_system
-ALTER INDEX i OWNER TO materialize;
+CREATE INDEX i1 IN CLUSTER default ON t (a);
 ----
 COMPLETE 0
 
 statement ok
-CREATE OR REPLACE MATERIALIZED VIEW mv AS SELECT 2 AS a;
+CREATE INDEX i2 ON t (a);
 
-statement ok
-DROP MATERIALIZED VIEW mv;
+query TT
+SELECT mz_indexes.name, mz_roles.name
+  FROM mz_indexes
+  LEFT JOIN mz_roles ON mz_indexes.owner_id = mz_roles.id
+  WHERE mz_indexes.name = 'i1' OR mz_indexes.name = 'i2'
+----
+i1  materialize
+i2  materialize
+
+simple conn=mz_system,user=mz_system
+ALTER TABLE t OWNER TO joe
+----
+COMPLETE 0
+
+query TT
+SELECT mz_indexes.name, mz_roles.name
+  FROM mz_indexes
+  LEFT JOIN mz_roles ON mz_indexes.owner_id = mz_roles.id
+  WHERE mz_indexes.name = 'i1' OR mz_indexes.name = 'i2'
+----
+i1  joe
+i2  joe
 
 # Disable rbac checks.
 simple conn=mz_system,user=mz_system