Recursively delete components (#30748)

GitOrigin-RevId: ef31f2926c553ce1284809adb75f6337c5638ae4

Author: Sujay Jayakar

crates/database/src/bootstrap_model/components/mod.rs

@@ -110,6 +110,15 @@ impl<'a, RT: Runtime> BootstrapComponentsModel<'a, RT> {
             .component_in_parent(parent_and_name, &mut self.tx.reads)
     }
 
+    pub fn component_children(
+        &mut self,
+        parent_id: DeveloperDocumentId,
+    ) -> anyhow::Result<Vec<ParsedDocument<ComponentMetadata>>> {
+        self.tx
+            .component_registry
+            .component_children(parent_id, &mut self.tx.reads)
+    }
+
     pub fn root_component(&mut self) -> anyhow::Result<Option<ParsedDocument<ComponentMetadata>>> {
         self.component_in_parent(None)
     }

crates/database/src/component_registry.rs

@@ -228,6 +228,36 @@ impl ComponentRegistry {
         self.component_in_parent(None, reads)
     }
 
+    pub fn component_children(
+        &self,
+        parent_id: DeveloperDocumentId,
+        reads: &mut TransactionReadSet,
+    ) -> anyhow::Result<Vec<ParsedDocument<ComponentMetadata>>> {
+        let interval =
+            Interval::prefix(values_to_bytes(&[Some(val!(parent_id.to_string()))]).into());
+        reads.record_indexed_derived(
+            TabletIndexName::new(
+                self.components_tablet,
+                COMPONENTS_BY_PARENT_INDEX.descriptor().clone(),
+            )?,
+            vec![PARENT_FIELD.clone(), NAME_FIELD.clone()].try_into()?,
+            interval,
+        );
+        let child_ids = self
+            .components
+            .iter()
+            .filter_map(|(_, doc)| {
+                let (component_parent_id, _) = doc.parent_and_name()?;
+                if component_parent_id == parent_id {
+                    Some(doc.clone())
+                } else {
+                    None
+                }
+            })
+            .collect();
+        Ok(child_ids)
+    }
+
     pub fn component_in_parent(
         &self,
         parent_and_name: Option<(DeveloperDocumentId, ComponentName)>,

crates/model/src/components/config.rs

@@ -706,51 +706,57 @@ impl<'a, RT: Runtime> ComponentConfigModel<'a, RT> {
 
     #[minitrace::trace]
     pub async fn delete_component(&mut self, component_id: ComponentId) -> anyhow::Result<()> {
-        let ComponentId::Child(id) = component_id else {
+        if component_id.is_root() {
             anyhow::bail!("Cannot delete root component");
-        };
+        }
 
+        // First, walk the component tree and validate that it's okay to delete each
+        // component.
         let component = BootstrapComponentsModel::new(self.tx)
             .load_component(component_id)
             .await?;
-
-        match component {
-            Some(component) => {
-                if component.state != ComponentState::Unmounted {
-                    anyhow::bail!(ErrorMetadata::bad_request(
-                        "ComponentMustBeUnmounted",
-                        "Component must be unmounted before deletion"
-                    ));
-                }
-            },
-            None => {
-                anyhow::bail!(ErrorMetadata::transient_not_found(
-                    "ComponentNotFound",
-                    format!("Component with ID {:?} not found", component_id)
+        let Some(component) = component else {
+            anyhow::bail!(ErrorMetadata::transient_not_found(
+                "ComponentNotFound",
+                format!("Component with ID {:?} not found", component_id)
+            ));
+        };
+        let mut stack = vec![component];
+        let mut all_ids = vec![];
+        while let Some(component) = stack.pop() {
+            if component.state != ComponentState::Unmounted {
+                anyhow::bail!(ErrorMetadata::bad_request(
+                    "ComponentMustBeUnmounted",
+                    "Component must be unmounted before deletion"
                 ));
-            },
+            }
+            let children =
+                BootstrapComponentsModel::new(self.tx).component_children(component.id().into())?;
+            stack.extend(children);
+            all_ids.push(component.id());
         }
 
-        let resolved_document_id =
-            BootstrapComponentsModel::new(self.tx).resolve_component_id(id)?;
-        SystemMetadataModel::new_global(self.tx)
-            .delete(resolved_document_id)
-            .await?;
-
-        let namespace = TableNamespace::from(component_id);
-        // delete the schema table first
-        // tables defined in the schema cannot be deleted, so we delete the _schemas
-        // table first to remove that restriction
-        TableModel::new(self.tx)
-            .delete_table(namespace, SCHEMAS_TABLE.clone())
-            .await?;
+        // Delete the components we found.
+        for component_id in all_ids {
+            SystemMetadataModel::new_global(self.tx)
+                .delete(component_id)
+                .await?;
 
-        // then delete all tables, including system tables
-        let namespaced_table_mapping = self.tx.table_mapping().namespace(namespace);
-        for (tablet_id, ..) in namespaced_table_mapping.iter() {
+            let namespace = TableNamespace::from(ComponentId::Child(component_id.into()));
+            // delete the schema table first
+            // tables defined in the schema cannot be deleted, so we delete the _schemas
+            // table first to remove that restriction
             TableModel::new(self.tx)
-                .delete_table_by_id(tablet_id)
+                .delete_table(namespace, SCHEMAS_TABLE.clone())
                 .await?;
+
+            // then delete all tables, including system tables
+            let namespaced_table_mapping = self.tx.table_mapping().namespace(namespace);
+            for (tablet_id, ..) in namespaced_table_mapping.iter() {
+                TableModel::new(self.tx)
+                    .delete_table_by_id(tablet_id)
+                    .await?;
+            }
         }
 
         Ok(())