sql: Allow dropping non-owned descendants (MaterializeInc#18873)

This commit reverts the changes made in
acc39ea. Specifically it allows
dropping non-owned objects with a CASCADE statement. This aligns with
the behavior in PostgreSQL. Roles are not allowed to directly drop
objects if they don't own the object, but they can drop those objects
indirectly through CASCADE. The author of
acc39ea, who's also the author of this
commit, misunderstood the intended behavior.

Works towards resolving MaterializeInc#18551
Part of MaterializeInc#11579

src/adapter/src/catalog.rs

@@ -252,33 +252,35 @@ impl CatalogState {
     /// objects.
     fn object_dependents(
         &self,
-        object_ids: Vec<ObjectId>,
+        object_ids: &Vec<ObjectId>,
         seen: &mut BTreeSet<ObjectId>,
     ) -> Vec<ObjectId> {
         let mut dependents = Vec::new();
         for object_id in object_ids {
             match object_id {
                 ObjectId::Cluster(id) => {
-                    dependents.extend_from_slice(&self.cluster_dependents(id, seen));
+                    dependents.extend_from_slice(&self.cluster_dependents(*id, seen));
                 }
                 ObjectId::ClusterReplica((cluster_id, replica_id)) => dependents.extend_from_slice(
-                    &self.cluster_replica_dependents(cluster_id, replica_id, seen),
+                    &self.cluster_replica_dependents(*cluster_id, *replica_id, seen),
                 ),
                 ObjectId::Database(id) => {
-                    dependents.extend_from_slice(&self.database_dependents(id, seen))
+                    dependents.extend_from_slice(&self.database_dependents(*id, seen))
                 }
                 ObjectId::Schema((database_spec, schema_id)) => {
                     dependents.extend_from_slice(&self.schema_dependents(
-                        database_spec,
-                        schema_id,
+                        *database_spec,
+                        *schema_id,
                         seen,
                     ));
                 }
                 id @ ObjectId::Role(_) => {
                     seen.insert(id.clone());
-                    dependents.push(id);
+                    dependents.push(id.clone());
+                }
+                ObjectId::Item(id) => {
+                    dependents.extend_from_slice(&self.item_dependents(*id, seen))
                 }
-                ObjectId::Item(id) => dependents.extend_from_slice(&self.item_dependents(id, seen)),
             }
         }
         dependents
@@ -4468,7 +4470,7 @@ impl Catalog {
             .cloned()
             .map(ObjectId::Item)
             .collect();
-        self.object_dependents(temp_ids)
+        self.object_dependents(&temp_ids)
             .into_iter()
             .map(Op::DropObject)
             .collect()
@@ -4481,7 +4483,7 @@ impl Catalog {
         Ok(())
     }
 
-    pub(crate) fn object_dependents(&self, object_ids: Vec<ObjectId>) -> Vec<ObjectId> {
+    pub(crate) fn object_dependents(&self, object_ids: &Vec<ObjectId>) -> Vec<ObjectId> {
         let mut seen = BTreeSet::new();
         self.state.object_dependents(object_ids, &mut seen)
     }
@@ -7420,7 +7422,7 @@ impl SessionCatalog for ConnCatalog<'_> {
         }
     }
 
-    fn object_dependents(&self, ids: Vec<ObjectId>) -> Vec<ObjectId> {
+    fn object_dependents(&self, ids: &Vec<ObjectId>) -> Vec<ObjectId> {
         let mut seen = BTreeSet::new();
         self.state.object_dependents(ids, &mut seen)
     }

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

@@ -1087,7 +1087,7 @@ impl Coordinator {
                 session,
                 plan.name.clone(),
                 plan.view.clone(),
-                plan.replace,
+                plan.drop_ids,
                 depends_on,
             )
             .await?;
@@ -1166,7 +1166,8 @@ impl Coordinator {
                     column_names,
                     cluster_id,
                 },
-            replace,
+            replace: _,
+            drop_ids,
             if_not_exists,
             ambiguous_columns,
         } = plan;
@@ -1221,7 +1222,7 @@ impl Coordinator {
 
         let mut ops = Vec::new();
         ops.extend(
-            replace
+            drop_ids
                 .into_iter()
                 .map(|id| catalog::Op::DropObject(ObjectId::Item(id))),
         );
@@ -1404,7 +1405,7 @@ impl Coordinator {
         let mut dropped_active_cluster = false;
 
         let mut dropped_roles: BTreeMap<_, _> = plan
-            .ids
+            .drop_ids
             .iter()
             .filter_map(|id| match id {
                 ObjectId::Role(role_id) => Some(role_id),
@@ -1437,7 +1438,7 @@ impl Coordinator {
             }
         }
 
-        for id in &plan.ids {
+        for id in &plan.drop_ids {
             match id {
                 ObjectId::Database(id) => {
                     let name = self.catalog().get_database(id).name();
@@ -1474,7 +1475,7 @@ impl Coordinator {
             }
         }
 
-        ops.extend(plan.ids.into_iter().map(catalog::Op::DropObject));
+        ops.extend(plan.drop_ids.into_iter().map(catalog::Op::DropObject));
         self.catalog_transact(Some(session), ops).await?;
 
         fail::fail_point!("after_sequencer_drop_replica");

src/adapter/src/rbac.rs

@@ -345,10 +345,11 @@ fn generate_required_ownership(plan: &Plan) -> Vec<ObjectId> {
         | Plan::GrantRole(_)
         | Plan::RevokeRole(_) => Vec::new(),
         Plan::CreateView(CreateViewPlan { replace, .. })
-        | Plan::CreateMaterializedView(CreateMaterializedViewPlan { replace, .. }) => {
-            replace.iter().map(|id| ObjectId::Item(*id)).collect()
-        }
-        Plan::DropObjects(plan) => plan.ids.clone(),
+        | Plan::CreateMaterializedView(CreateMaterializedViewPlan { replace, .. }) => replace
+            .map(|id| vec![ObjectId::Item(id)])
+            .unwrap_or_default(),
+        // Do not need ownership of descendant objects.
+        Plan::DropObjects(plan) => plan.referenced_ids.clone(),
         Plan::AlterIndexSetOptions(plan) => vec![ObjectId::Item(plan.id)],
         Plan::AlterIndexResetOptions(plan) => vec![ObjectId::Item(plan.id)],
         Plan::AlterSink(plan) => vec![ObjectId::Item(plan.id)],

src/sql/src/catalog.rs

@@ -260,7 +260,7 @@ pub trait SessionCatalog: fmt::Debug + ExprHumanizer + Send + Sync {
     /// The order is guaranteed to be in reverse dependency order, i.e. the leafs will appear
     /// earlier in the list than the roots. This is particularly userful for the order to drop
     /// objects.
-    fn object_dependents(&self, ids: Vec<ObjectId>) -> Vec<ObjectId>;
+    fn object_dependents(&self, ids: &Vec<ObjectId>) -> Vec<ObjectId>;
 
     /// Returns all the IDs of all objects that depend on `id`, including `id` themselves.
     ///

src/sql/src/plan.rs

@@ -524,8 +524,10 @@ pub struct CreateTablePlan {
 pub struct CreateViewPlan {
     pub name: QualifiedItemName,
     pub view: View,
-    /// The IDs of the objects that this view is replacing, if any.
-    pub replace: Vec<GlobalId>,
+    /// The ID of the object that this view is replacing, if any.
+    pub replace: Option<GlobalId>,
+    /// The IDs of all objects that need to be dropped. This includes `replace` and any dependents.
+    pub drop_ids: Vec<GlobalId>,
     pub if_not_exists: bool,
     /// True if the view contains an expression that can make the exact column list
     /// ambiguous. For example `NATURAL JOIN` or `SELECT *`.
@@ -536,8 +538,10 @@ pub struct CreateViewPlan {
 pub struct CreateMaterializedViewPlan {
     pub name: QualifiedItemName,
     pub materialized_view: MaterializedView,
-    /// The IDs of the objects that this view is replacing, if any.
-    pub replace: Vec<GlobalId>,
+    /// The ID of the object that this view is replacing, if any.
+    pub replace: Option<GlobalId>,
+    /// The IDs of all objects that need to be dropped. This includes `replace` and any dependents.
+    pub drop_ids: Vec<GlobalId>,
     pub if_not_exists: bool,
     /// True if the materialized view contains an expression that can make the exact column list
     /// ambiguous. For example `NATURAL JOIN` or `SELECT *`.
@@ -560,7 +564,10 @@ pub struct CreateTypePlan {
 
 #[derive(Debug)]
 pub struct DropObjectsPlan {
-    pub ids: Vec<ObjectId>,
+    /// The IDs of only the objects directly referenced in the `DROP` statement.
+    pub referenced_ids: Vec<ObjectId>,
+    /// All object IDs to drop. Includes `referenced_ids` and all descendants.
+    pub drop_ids: Vec<ObjectId>,
     /// The type of object that was dropped explicitly in the DROP statement. `ids` may contain
     /// objects of different types due to CASCADE.
     pub object_type: ObjectType,

src/sql/src/plan/statement/ddl.rs

@@ -1735,17 +1735,22 @@ pub fn plan_create_view(
                     scx.catalog.resolve_full_name(item.name())
                 );
             }
+            Some(id)
+        } else {
+            None
+        }
+    } else {
+        None
+    };
+    let drop_ids = replace
+        .map(|id| {
             scx.catalog
                 .item_dependents(id)
                 .into_iter()
                 .map(|id| id.unwrap_item_id())
                 .collect()
-        } else {
-            Vec::new()
-        }
-    } else {
-        Vec::new()
-    };
+        })
+        .unwrap_or_default();
 
     // Check for an object in the catalog with this same name
     let full_name = scx.catalog.resolve_full_name(&name);
@@ -1763,6 +1768,7 @@ pub fn plan_create_view(
         name,
         view,
         replace,
+        drop_ids,
         if_not_exists: *if_exists == IfExistsBehavior::Skip,
         ambiguous_columns: *scx.ambiguous_columns.borrow(),
     }))
@@ -1823,7 +1829,7 @@ pub fn plan_create_materialized_view(
         sql_bail!("column {} specified more than once", dup.as_str().quoted());
     }
 
-    let mut replace = Vec::new();
+    let mut replace = None;
     let mut if_not_exists = false;
     match stmt.if_exists {
         IfExistsBehavior::Replace => {
@@ -1844,17 +1850,21 @@ pub fn plan_create_materialized_view(
                         scx.catalog.resolve_full_name(item.name())
                     );
                 }
-                replace.extend(
-                    scx.catalog
-                        .item_dependents(id)
-                        .into_iter()
-                        .map(|id| id.unwrap_item_id()),
-                );
+                replace = Some(id);
             }
         }
         IfExistsBehavior::Skip => if_not_exists = true,
         IfExistsBehavior::Error => (),
     }
+    let drop_ids = replace
+        .map(|id| {
+            scx.catalog
+                .item_dependents(id)
+                .into_iter()
+                .map(|id| id.unwrap_item_id())
+                .collect()
+        })
+        .unwrap_or_default();
 
     // Check for an object in the catalog with this same name
     let full_name = scx.catalog.resolve_full_name(&name);
@@ -1877,6 +1887,7 @@ pub fn plan_create_materialized_view(
             cluster_id,
         },
         replace,
+        drop_ids,
         if_not_exists,
         ambiguous_columns: *scx.ambiguous_columns.borrow(),
     }))
@@ -3379,7 +3390,7 @@ pub fn plan_drop_objects(
 ) -> Result<Plan, PlanError> {
     assert_ne!(object_type, ObjectType::Func, "rejected in parser");
 
-    let mut ids = Vec::new();
+    let mut referenced_ids = Vec::new();
     for name in names {
         let id = match name {
             UnresolvedObjectName::Cluster(name) => {
@@ -3402,12 +3413,16 @@ pub fn plan_drop_objects(
             }
         };
         if let Some(id) = id {
-            ids.push(id);
+            referenced_ids.push(id);
         }
     }
-    let ids = scx.catalog.object_dependents(ids);
+    let drop_ids = scx.catalog.object_dependents(&referenced_ids);
 
-    Ok(Plan::DropObjects(DropObjectsPlan { ids, object_type }))
+    Ok(Plan::DropObjects(DropObjectsPlan {
+        referenced_ids,
+        drop_ids,
+        object_type,
+    }))
 }
 
 fn plan_drop_schema(

test/sqllogictest/object_ownership.slt

@@ -702,7 +702,7 @@ SELECT mz_roles.name
 ----
 joe
 
-statement error must be owner of SCHEMA jdb.public, DATABASE jdb
+statement error must be owner of DATABASE jdb
 DROP DATABASE jdb
 
 statement error must be owner of DATABASE jdb
@@ -835,7 +835,7 @@ SELECT mz_roles.name
 ----
 joe
 
-statement error must be owner of CLUSTER REPLICA jclus.jr1, CLUSTER jclus
+statement error must be owner of CLUSTER jclus
 DROP CLUSTER jclus
 
 statement error must be owner of CLUSTER jclus
@@ -1376,7 +1376,7 @@ DROP TABLE t;
 ----
 COMPLETE 0
 
-# Test that ownership is checked with cascading deletes
+# Test that ownership is not checked with cascading deletes
 
 statement ok
 CREATE TABLE t (a INT);
@@ -1387,12 +1387,7 @@ CREATE VIEW v AS SELECT * FROM t;
 COMPLETE 0
 
 statement error must be owner of VIEW materialize.public.v
-DROP TABLE t CASCADE;
-
-simple conn=mz_system,user=mz_system
-ALTER VIEW v OWNER TO materialize;
-----
-COMPLETE 0
+DROP VIEW v
 
 statement ok
 DROP TABLE t CASCADE;
@@ -1405,14 +1400,6 @@ 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;
 
@@ -1427,14 +1414,6 @@ CREATE INDEX i ON mv(a);
 ----
 COMPLETE 0
 
-statement error must be owner of INDEX materialize.public.i
-CREATE OR REPLACE MATERIALIZED VIEW mv 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 MATERIALIZED VIEW mv AS SELECT 2 AS a;