Auto Delete Empty Anchors and Sources in SQL Registry

registry/sql-registry/main.py

@@ -97,10 +97,12 @@ def delete_entity(entity: str):
     entity_id = registry.get_entity_id(entity)
     downstream_entities = registry.get_dependent_entities(entity_id)
     if len(downstream_entities) > 0:
-        raise HTTPException(
-            status_code=412, detail=f"""Entity cannot be deleted as it has downstream/dependent entities.
-            Entities: {list([e.qualified_name for e in downstream_entities])}"""
-        )
+        registry.delete_empty_entities(downstream_entities)
+        if len(registry.get_dependent_entities(entity_id)) > 0:
+            raise HTTPException(
+                status_code=412, detail=f"""Entity cannot be deleted as it has downstream/dependent entities.
+                Entities: {list([e.qualified_name for e in downstream_entities])}"""
+            )
     registry.delete_entity(entity_id)
 
 @router.get("/projects/{project}/datasources")

registry/sql-registry/registry/db_registry.py

@@ -204,6 +204,28 @@ def get_dependent_entities(self, entity_id: Union[str, UUID]) -> List[Entity]:
             downstream_entities, _ = self._bfs(entity_id, RelationshipType.Produces)
         return [e for e in downstream_entities if str(e.id) != str(entity_id)]
 
+    def delete_empty_entities(self, entities: List[Entity]):
+        """
+        Given entity list, deleting all anchors that have no features and all sources that have no anchors.
+        """
+        if len(entities) == 0:
+            return
+
+        # clean up empty anchors
+        for e in entities:
+            if e.entity_type == EntityType.Anchor:
+                downstream_entities, _ = self._bfs(e.id, RelationshipType.Contains) 
+                if len(downstream_entities) == 1: # only anchor itself
+                    self.delete_entity(e.id)
+        # clean up empty sources
+        for e in entities:
+            if e.entity_type == EntityType.Source:
+                downstream_entities, _ = self._bfs(e.id, RelationshipType.Produces) 
+                if len(downstream_entities) == 1: # only source itself
+                    self.delete_entity(e.id)
+
+        return
+
     def delete_entity(self, entity_id: Union[str, UUID]):
         """
         Deletes given entity