fix: cascade delete uses restrict() for proper SQL generation and OR convergence

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 now uses OR convergence (a row is deleted
if ANY FK reference points to a deleted row), while restrict/export uses AND.

Also adds dry_run support to delete() and drop(), fixes CLI test subprocess
invocation to use `sys.executable -m datajoint.cli`, and fixes test fixture
cleanup to respect part_integrity policy.

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

Author: dimitri-yatsenko

pixi.lock

@@ -97,6 +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 = getattr(source, "_part_integrity", "enforce")
             super().__init__(source)
             return
 
@@ -369,6 +370,7 @@ def cascade(self, table_expr, part_integrity="enforce"):
                 "cascade and restrict modes are mutually exclusive."
             )
         result = Diagram(self)
+        result._part_integrity = part_integrity
         node = table_expr.full_table_name
         if node not in result.nodes():
             raise DataJointError(f"Table {node} is not in the diagram.")
@@ -380,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.
@@ -445,11 +475,8 @@ def _propagate_restrictions(self, start_node, mode, part_integrity="enforce"):
                 # Build parent FreeTable with current restriction
                 parent_ft = FreeTable(self._connection, node)
                 restr = restrictions[node]
-                if mode == "cascade" and restr:
-                    parent_ft._restriction = restr  # plain list → OR
-                elif mode == "restrict":
-                    parent_ft._restriction = restr  # AndList → AND
-                # else: cascade with empty list → unrestricted
+                if restr:
+                    parent_ft = self._restrict_freetable(parent_ft, restr, mode=mode)
 
                 parent_attrs = self._restriction_attrs.get(node, set())
 
@@ -507,14 +534,14 @@ 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._restriction = 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
 
                                 master_restr = make_condition(
                                     master_ft,
                                     (master_ft.proj() & child_ft.proj()).to_arrays(),
-                                    master_ft._restriction_attributes,
+                                    master_ft.restriction_attributes,
                                 )
                                 restrictions[master_name] = [master_restr]
                                 self._restriction_attrs[master_name] = set()
@@ -579,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.
 
@@ -589,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:
@@ -606,14 +639,15 @@ def delete(self, transaction=True, prompt=None):
 
         # Pre-check part_integrity="enforce": ensure no part is deleted
         # before its master
-        for node in self._cascade_restrictions:
-            master = extract_master(node)
-            if master and master not in self._cascade_restrictions:
-                raise DataJointError(
-                    f"Attempt to delete part table {node} before "
-                    f"its master {master}. Delete from the master first, "
-                    f"or use part_integrity='ignore' or 'cascade'."
-                )
+        if getattr(self, "_part_integrity", "enforce") == "enforce":
+            for node in self._cascade_restrictions:
+                master = extract_master(node)
+                if master and master not in self._cascade_restrictions:
+                    raise DataJointError(
+                        f"Attempt to delete part table {node} before "
+                        f"its master {master}. Delete from the master first, "
+                        f"or use part_integrity='ignore' or 'cascade'."
+                    )
 
         # Get non-alias nodes with restrictions in topological order
         all_sorted = topo_sort(self)
@@ -623,9 +657,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._restriction = restr
+                ft = self._restrict_freetable(ft, self._cascade_restrictions[t])
                 logger.info("{table} ({count} tuples)".format(table=t, count=len(ft)))
 
         # Start transaction
@@ -647,9 +679,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._restriction = restr
+                ft = self._restrict_freetable(ft, self._cascade_restrictions[table_name])
                 count = ft.delete_quick(get_count=True)
                 logger.info("Deleting {count} rows from {table}".format(count=count, table=table_name))
                 if table_name == tables[0]:
@@ -692,7 +722,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.
 
@@ -702,6 +732,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
 
@@ -720,6 +757,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:
@@ -742,6 +787,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.")
 
@@ -750,9 +796,7 @@ def preview(self):
             if node.isdigit() or node not in restrictions:
                 continue
             ft = FreeTable(self._connection, node)
-            restr = restrictions[node]
-            if restr:
-                ft._restriction = restr
+            ft = self._restrict_freetable(ft, restrictions[node], mode=mode)
             result[node] = len(ft)
 
         for t, count in result.items():

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:

src/datajoint/user_tables.py

@@ -188,3 +188,86 @@ class Observation(dj.Manual):
     assert remaining_obs[0]["obs_id"] == 3
     assert remaining_obs[0]["subject_id"] == 2
     assert remaining_obs[0]["measurement"] == 15.3
+
+
+def test_delete_dry_run(schema_by_backend):
+    """dry_run=True returns affected row counts without deleting data."""
+
+    @schema_by_backend
+    class Parent(dj.Manual):
+        definition = """
+        parent_id : int
+        ---
+        name : varchar(255)
+        """
+
+    @schema_by_backend
+    class Child(dj.Manual):
+        definition = """
+        -> Parent
+        child_id : int
+        ---
+        data : varchar(255)
+        """
+
+    Parent.insert1((1, "P1"))
+    Parent.insert1((2, "P2"))
+    Child.insert1((1, 1, "C1-1"))
+    Child.insert1((1, 2, "C1-2"))
+    Child.insert1((2, 1, "C2-1"))
+
+    # dry_run on restricted delete
+    counts = (Parent & {"parent_id": 1}).delete(dry_run=True)
+
+    assert isinstance(counts, dict)
+    assert counts[Parent.full_table_name] == 1
+    assert counts[Child.full_table_name] == 2
+
+    # Data must still be intact
+    assert len(Parent()) == 2
+    assert len(Child()) == 3
+
+    # dry_run on unrestricted delete
+    counts_all = Parent.delete(dry_run=True)
+    assert counts_all[Parent.full_table_name] == 2
+    assert counts_all[Child.full_table_name] == 3
+
+    # Still intact
+    assert len(Parent()) == 2
+    assert len(Child()) == 3
+
+
+def test_drop_dry_run(schema_by_backend):
+    """dry_run=True returns row counts without dropping tables."""
+
+    @schema_by_backend
+    class Parent(dj.Manual):
+        definition = """
+        parent_id : int
+        ---
+        name : varchar(255)
+        """
+
+    @schema_by_backend
+    class Child(dj.Manual):
+        definition = """
+        -> Parent
+        child_id : int
+        ---
+        data : varchar(255)
+        """
+
+    Parent.insert1((1, "P1"))
+    Child.insert1((1, 1, "C1"))
+
+    counts = Parent.drop(dry_run=True)
+
+    assert isinstance(counts, dict)
+    assert counts[Parent.full_table_name] == 1
+    assert counts[Child.full_table_name] == 1
+
+    # Tables must still exist and have data
+    assert Parent.is_declared
+    assert Child.is_declared
+    assert len(Parent()) == 1
+    assert len(Child()) == 1