fix: cascade delete with proper SQL generation, OR convergence, and post-check part integrity

Replace direct `_restriction` assignment with `restrict()` calls in Diagram
so that AndList and QueryExpression objects are converted to valid SQL via
`make_condition()`. Cascade delete uses OR convergence (a row is deleted if
ANY FK reference points to a deleted row), while restrict/export uses AND.

Part integrity enforcement uses a data-driven post-check: only raises when
rows were actually deleted from a Part without its master also being deleted.
This avoids false positives when a Part table appears in the cascade graph
but has zero affected rows.

Also adds dry_run support to delete()/drop(), prune() method, fixes CLI test
subprocess invocation, and updates test fixtures.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: dimitri-yatsenko

pixi.lock

@@ -97,7 +97,7 @@ def __init__(self, source, context=None) -> None:
             self._cascade_restrictions = copy_module.deepcopy(source._cascade_restrictions)
             self._restrict_conditions = copy_module.deepcopy(source._restrict_conditions)
             self._restriction_attrs = copy_module.deepcopy(source._restriction_attrs)
-            self._part_integrity = source._part_integrity
+            self._part_integrity = getattr(source, "_part_integrity", "enforce")
             super().__init__(source)
             return
 
@@ -125,7 +125,6 @@ def __init__(self, source, context=None) -> None:
         self._cascade_restrictions = {}
         self._restrict_conditions = {}
         self._restriction_attrs = {}
-        self._part_integrity = "enforce"
 
         # Enumerate nodes from all the items in the list
         self.nodes_to_show = set()
@@ -193,7 +192,6 @@ def _from_table(cls, table_expr) -> "Diagram":
         result._cascade_restrictions = {}
         result._restrict_conditions = {}
         result._restriction_attrs = {}
-        result._part_integrity = "enforce"
         return result
 
     def add_parts(self) -> "Diagram":
@@ -384,6 +382,34 @@ def cascade(self, table_expr, part_integrity="enforce"):
         result._propagate_restrictions(node, mode="cascade", part_integrity=part_integrity)
         return result
 
+    @staticmethod
+    def _restrict_freetable(ft, restrictions, mode="cascade"):
+        """
+        Apply cascade/restrict restrictions to a FreeTable.
+
+        Uses ``restrict()`` to properly convert each restriction (AndList,
+        QueryExpression, etc.) into SQL via ``make_condition``, rather than
+        assigning raw objects to ``_restriction`` which would produce
+        invalid SQL in ``where_clause``.
+
+        For cascade mode (delete), restrictions from different parent edges
+        are OR-ed: a row is deleted if ANY of its FK references point to a
+        deleted row.
+
+        For restrict mode (export), restrictions are AND-ed: a row is
+        included only if ALL ancestor conditions are satisfied.
+        """
+        if not restrictions:
+            return ft
+        if mode == "cascade":
+            # OR semantics — passing a list to restrict() creates an OrList
+            return ft.restrict(restrictions)
+        else:
+            # AND semantics — each restriction narrows further
+            for r in restrictions:
+                ft = ft.restrict(r)
+            return ft
+
     def restrict(self, table_expr):
         """
         Apply restrict condition and propagate downstream.
@@ -450,10 +476,7 @@ def _propagate_restrictions(self, start_node, mode, part_integrity="enforce"):
                 parent_ft = FreeTable(self._connection, node)
                 restr = restrictions[node]
                 if restr:
-                    if mode == "cascade":
-                        parent_ft.restrict_in_place(restr)  # list → OR
-                    else:
-                        parent_ft._restriction = restr  # AndList → AND
+                    parent_ft = self._restrict_freetable(parent_ft, restr, mode=mode)
 
                 parent_attrs = self._restriction_attrs.get(node, set())
 
@@ -511,7 +534,7 @@ def _propagate_restrictions(self, start_node, mode, part_integrity="enforce"):
                                 child_ft = FreeTable(self._connection, target)
                                 child_restr = restrictions.get(target, [])
                                 if child_restr:
-                                    child_ft.restrict_in_place(child_restr)
+                                    child_ft = self._restrict_freetable(child_ft, child_restr, mode=mode)
                                 master_ft = FreeTable(self._connection, master_name)
                                 from .condition import make_condition
 
@@ -583,7 +606,7 @@ def _apply_propagation_rule(
 
         self._restriction_attrs.setdefault(child_node, set()).update(child_attrs)
 
-    def delete(self, transaction=True, prompt=None):
+    def delete(self, transaction=True, prompt=None, dry_run=False):
         """
         Execute cascading delete using cascade restrictions.
 
@@ -593,14 +616,20 @@ def delete(self, transaction=True, prompt=None):
             Wrap in a transaction. Default True.
         prompt : bool or None, optional
             Show preview and ask confirmation. Default ``dj.config['safemode']``.
+        dry_run : bool, optional
+            If True, return affected row counts without deleting. Default False.
 
         Returns
         -------
-        int
-            Number of rows deleted from the root table.
+        int or dict[str, int]
+            Number of rows deleted from the root table, or (if ``dry_run``)
+            a mapping of full table name to affected row count.
         """
         from .table import FreeTable
 
+        if dry_run:
+            return self.preview()
+
         prompt = self._connection._config["safemode"] if prompt is None else prompt
 
         if not self._cascade_restrictions:
@@ -616,9 +645,7 @@ def delete(self, transaction=True, prompt=None):
         if prompt:
             for t in tables:
                 ft = FreeTable(conn, t)
-                restr = self._cascade_restrictions[t]
-                if restr:
-                    ft.restrict_in_place(restr)
+                ft = self._restrict_freetable(ft, self._cascade_restrictions[t])
                 logger.info("{table} ({count} tuples)".format(table=t, count=len(ft)))
 
         # Start transaction
@@ -641,9 +668,7 @@ def delete(self, transaction=True, prompt=None):
         try:
             for table_name in reversed(tables):
                 ft = FreeTable(conn, table_name)
-                restr = self._cascade_restrictions[table_name]
-                if restr:
-                    ft.restrict_in_place(restr)
+                ft = self._restrict_freetable(ft, self._cascade_restrictions[table_name])
                 count = ft.delete_quick(get_count=True)
                 if count > 0:
                     deleted_tables.add(table_name)
@@ -668,7 +693,7 @@ def delete(self, transaction=True, prompt=None):
 
         # Post-check part_integrity="enforce": roll back if a part table
         # had rows deleted without its master also having rows deleted.
-        if self._part_integrity == "enforce" and deleted_tables:
+        if getattr(self, "_part_integrity", "enforce") == "enforce" and deleted_tables:
             for table_name in deleted_tables:
                 master = extract_master(table_name)
                 if master and master not in deleted_tables:
@@ -702,7 +727,7 @@ def delete(self, transaction=True, prompt=None):
                 root_count = 0
         return root_count
 
-    def drop(self, prompt=None, part_integrity="enforce"):
+    def drop(self, prompt=None, part_integrity="enforce", dry_run=False):
         """
         Drop all tables in the diagram in reverse topological order.
 
@@ -712,6 +737,13 @@ def drop(self, prompt=None, part_integrity="enforce"):
             Show preview and ask confirmation. Default ``dj.config['safemode']``.
         part_integrity : str, optional
             ``"enforce"`` (default) or ``"ignore"``.
+        dry_run : bool, optional
+            If True, return row counts without dropping. Default False.
+
+        Returns
+        -------
+        dict[str, int] or None
+            If ``dry_run``, mapping of full table name to row count.
         """
         from .table import FreeTable
 
@@ -730,6 +762,14 @@ def drop(self, prompt=None, part_integrity="enforce"):
                         )
                     )
 
+        if dry_run:
+            result = {}
+            for t in tables:
+                count = len(FreeTable(conn, t))
+                result[t] = count
+                logger.info("{table} ({count} tuples)".format(table=t, count=count))
+            return result
+
         do_drop = True
         if prompt:
             for t in tables:
@@ -752,6 +792,7 @@ def preview(self):
         from .table import FreeTable
 
         restrictions = self._cascade_restrictions or self._restrict_conditions
+        mode = "cascade" if self._cascade_restrictions else "restrict"
         if not restrictions:
             raise DataJointError("No restrictions applied. " "Call cascade() or restrict() first.")
 
@@ -760,12 +801,7 @@ def preview(self):
             if node.isdigit() or node not in restrictions:
                 continue
             ft = FreeTable(self._connection, node)
-            restr = restrictions[node]
-            if restr:
-                if isinstance(restr, list) and not isinstance(restr, AndList):
-                    ft.restrict_in_place(restr)  # cascade: list → OR
-                else:
-                    ft._restriction = restr  # restrict: AndList → AND
+            ft = self._restrict_freetable(ft, restrictions[node], mode=mode)
             result[node] = len(ft)
 
         for t, count in result.items():
@@ -789,19 +825,15 @@ def prune(self):
 
         result = Diagram(self)
         restrictions = result._cascade_restrictions or result._restrict_conditions
+        mode = "cascade" if result._cascade_restrictions else "restrict"
 
         if restrictions:
             # Restricted: check row counts under restriction
             for node in list(restrictions):
                 if node.isdigit():
                     continue
                 ft = FreeTable(self._connection, node)
-                restr = restrictions[node]
-                if restr:
-                    if isinstance(restr, list) and not isinstance(restr, AndList):
-                        ft.restrict_in_place(restr)
-                    else:
-                        ft._restriction = restr
+                ft = self._restrict_freetable(ft, restrictions[node], mode=mode)
                 if len(ft) == 0:
                     restrictions.pop(node)
                     result._restriction_attrs.pop(node, None)

src/datajoint/diagram.py

@@ -973,7 +973,8 @@ def delete(
         transaction: bool = True,
         prompt: bool | None = None,
         part_integrity: str = "enforce",
-    ) -> int:
+        dry_run: bool = False,
+    ) -> int | dict[str, int]:
         """
         Deletes the contents of the table and its dependent tables, recursively.
 
@@ -991,9 +992,12 @@ def delete(
                 - ``"enforce"`` (default): Error if parts would be deleted without masters.
                 - ``"ignore"``: Allow deleting parts without masters (breaks integrity).
                 - ``"cascade"``: Also delete masters when parts are deleted (maintains integrity).
+            dry_run: If `True`, return a dict mapping full table names to affected
+                row counts without deleting any data. Default False.
 
         Returns:
-            Number of deleted rows (excluding those from dependent tables).
+            Number of deleted rows (excluding those from dependent tables), or
+            (if ``dry_run``) a dict mapping full table name to affected row count.
 
         Raises:
             DataJointError: When deleting within an existing transaction.
@@ -1006,7 +1010,7 @@ def delete(
 
         diagram = Diagram._from_table(self)
         diagram = diagram.cascade(self, part_integrity=part_integrity)
-        return diagram.delete(transaction=transaction, prompt=prompt)
+        return diagram.delete(transaction=transaction, prompt=prompt, dry_run=dry_run)
 
     def drop_quick(self):
         """
@@ -1046,7 +1050,7 @@ def drop_quick(self):
         else:
             logger.info("Nothing to drop: table %s is not declared" % self.full_table_name)
 
-    def drop(self, prompt: bool | None = None, part_integrity: str = "enforce"):
+    def drop(self, prompt: bool | None = None, part_integrity: str = "enforce", dry_run: bool = False):
         """
         Drop the table and all tables that reference it, recursively.
 
@@ -1059,6 +1063,12 @@ def drop(self, prompt: bool | None = None, part_integrity: str = "enforce"):
             part_integrity: Policy for master-part integrity. One of:
                 - ``"enforce"`` (default): Error if parts would be dropped without masters.
                 - ``"ignore"``: Allow dropping parts without masters.
+            dry_run: If `True`, return a dict mapping full table names to row
+                counts without dropping any tables. Default False.
+
+        Returns:
+            dict[str, int] or None: If ``dry_run``, mapping of full table name
+            to row count. Otherwise None.
         """
         if self.restriction:
             raise DataJointError(
@@ -1067,7 +1077,7 @@ def drop(self, prompt: bool | None = None, part_integrity: str = "enforce"):
         from .diagram import Diagram
 
         diagram = Diagram._from_table(self)
-        diagram.drop(prompt=prompt, part_integrity=part_integrity)
+        return diagram.drop(prompt=prompt, part_integrity=part_integrity, dry_run=dry_run)
 
     def describe(self, context=None, printout=False):
         """

src/datajoint/table.py

@@ -239,7 +239,7 @@ def delete(self, part_integrity: str = "enforce", **kwargs):
             )
         super().delete(part_integrity=part_integrity, **kwargs)
 
-    def drop(self, part_integrity: str = "enforce"):
+    def drop(self, part_integrity: str = "enforce", dry_run: bool = False):
         """
         Drop a Part table.
 
@@ -248,12 +248,13 @@ def drop(self, part_integrity: str = "enforce"):
                 - ``"enforce"`` (default): Error - drop master instead.
                 - ``"ignore"``: Allow direct drop (breaks master-part structure).
                 Note: ``"cascade"`` is not supported for drop (too destructive).
+            dry_run: If `True`, return row counts without dropping. Default False.
 
         Raises:
             DataJointError: If part_integrity="enforce" (direct Part drops prohibited)
         """
         if part_integrity == "ignore":
-            super().drop(part_integrity="ignore")
+            return super().drop(part_integrity="ignore", dry_run=dry_run)
         elif part_integrity == "enforce":
             raise DataJointError("Cannot drop a Part directly. Drop master instead, or use part_integrity='ignore' to force.")
         else: