fix: restrict prune() to non-cascade Diagrams

prune() removes tables with zero matching rows from the diagram.
For cascade (delete), this is unsafe: between cascade computation
and the actual DELETE, concurrent inserts could add rows to a
pruned table, causing FK errors during delete. Zero-count tables
in the cascade are harmless — delete_quick() on an empty result
is a no-op.

prune() now raises DataJointError on cascade-produced Diagrams.
It remains valid for restrict() (export subsetting) and
unrestricted Diagrams (showing populated tables).

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: dimitri-yatsenko

src/datajoint/diagram.py

@@ -648,8 +648,10 @@ def prune(self):
         Remove tables with zero matching rows from the diagram.
 
         Without prior restrictions, removes physically empty tables.
-        With restrictions (``cascade()`` or ``restrict()``), removes
-        tables where the restricted query yields zero rows.
+        After ``restrict()``, removes tables where the restricted query
+        yields zero rows. Cannot be used on a cascade Diagram (cascade
+        is for delete, where zero-count tables must remain in the graph
+        to handle concurrent inserts safely).
 
         Returns
         -------
@@ -658,16 +660,20 @@ def prune(self):
         """
         from .table import FreeTable
 
+        if self._cascade_restrictions:
+            raise DataJointError(
+                "prune() cannot be used on a Diagram produced by Diagram.cascade(). "
+                "Cascade diagrams must retain all descendant tables for safe deletion."
+            )
+
         result = Diagram(self)
-        restrictions = result._cascade_restrictions or result._restrict_conditions
 
-        if restrictions:
-            # Restricted: check row counts under restriction
-            for node in list(restrictions):
+        if result._restrict_conditions:
+            for node in list(result._restrict_conditions):
                 if node.isdigit():
                     continue
                 if len(result._restricted_table(node)) == 0:
-                    restrictions.pop(node)
+                    result._restrict_conditions.pop(node)
                     result._restriction_attrs.pop(node, None)
                     result.nodes_to_show.discard(node)
         else:

tests/integration/test_erd.py

@@ -124,19 +124,11 @@ def test_prune_after_restrict(schema_simp_pop):
             assert table not in pruned._restrict_conditions, f"{table} had 0 rows but was not pruned"
 
 
-def test_prune_after_cascade(schema_simp_pop):
-    """Prune after cascade removes tables with zero matching rows."""
+def test_prune_raises_on_cascade(schema_simp_pop):
+    """prune() raises on a cascade Diagram — cascade must retain all tables for safe deletion."""
     cascaded = dj.Diagram.cascade(A & "id_a=0")
-    counts = cascaded.counts()
-
-    pruned = cascaded.prune()
-    pruned_counts = pruned.counts()
-
-    assert all(c > 0 for c in pruned_counts.values())
-
-    for table, count in counts.items():
-        if count == 0:
-            assert table not in pruned._cascade_restrictions, f"{table} had 0 rows but was not pruned"
+    with _pytest.raises(dj.DataJointError, match="prune.*cannot be used.*cascade"):
+        cascaded.prune()
 
 
 def test_prune_idempotent(schema_simp_pop):