fix(cascade): restrict only the rows a cascade reaches (2.3)

src/datajoint/diagram.py

@@ -154,6 +154,20 @@ def stroke_rule(light_hex, dark_hex, opacity=None):
     return "<style>\n@media (prefers-color-scheme: dark) {\n" + body + "\n}\n</style>"
 
 
+def _seed_restriction(table_expr):
+    """
+    Return a traversal seed's restriction and the attributes it names.
+
+    A ``dj.Top`` restriction is an ORDER BY ... LIMIT, not part of the WHERE
+    clause, so a seed restricted by ``Top`` is replaced by the primary keys
+    it selects. Otherwise the traversal would start from every row that
+    matches the seed's other conditions.
+    """
+    if table_expr._top is not None:
+        return AndList([table_expr.keys()]), set(table_expr.primary_key)
+    return AndList(table_expr.restriction), set(table_expr.restriction_attributes)
+
+
 class Diagram(nx.MultiDiGraph):  # noqa: C901
     """
     Schema diagram as a directed acyclic graph (DAG).
@@ -210,9 +224,12 @@ def __init__(self, source, context=None) -> None:
             self._expanded_nodes = set(source._expanded_nodes)
             self.context = source.context
             self._connection = source._connection
-            self._cascade_restrictions = copy.deepcopy(source._cascade_restrictions)
-            self._restrict_conditions = copy.deepcopy(source._restrict_conditions)
-            self._restriction_attrs = copy.deepcopy(source._restriction_attrs)
+            # Copy each node's restriction list, not its items: projected
+            # restrictions are query expressions bound to a live connection,
+            # which cannot be deep-copied.
+            self._cascade_restrictions = {k: copy.copy(v) for k, v in source._cascade_restrictions.items()}
+            self._restrict_conditions = {k: copy.copy(v) for k, v in source._restrict_conditions.items()}
+            self._restriction_attrs = {k: set(v) for k, v in source._restriction_attrs.items()}
             super().__init__(source)
             return
 
@@ -471,9 +488,9 @@ def cascade(cls, table_expr, part_integrity="enforce"):
         result._expanded_nodes = set(descendants)
 
         # Seed restriction
-        restriction = AndList(table_expr.restriction)
+        restriction, restriction_attrs = _seed_restriction(table_expr)
         result._cascade_restrictions[node] = [restriction] if restriction else []
-        result._restriction_attrs[node] = set(table_expr.restriction_attributes)
+        result._restriction_attrs[node] = restriction_attrs
 
         # Propagate downstream
         result._propagate_restrictions(node, mode="cascade", part_integrity=part_integrity)
@@ -557,9 +574,9 @@ def trace(cls, table_expr):
         result._expanded_nodes = set(ancestors)
 
         # Seed restriction
-        restriction = AndList(table_expr.restriction)
+        restriction, restriction_attrs = _seed_restriction(table_expr)
         result._cascade_restrictions[node] = [restriction] if restriction else []
-        result._restriction_attrs[node] = set(table_expr.restriction_attributes)
+        result._restriction_attrs[node] = restriction_attrs
 
         # Propagate upstream
         result._propagate_restrictions_upstream(node)
@@ -883,16 +900,22 @@ def _apply_propagation_rule(
 
         Rules (from table.py restriction propagation):
 
-        1. Non-aliased AND parent restriction attrs ⊆ child PK:
+        1. Non-aliased AND parent restriction attrs ⊆ FK attrs ⊆ child PK:
            Copy parent restriction directly.
         2. Aliased FK (attr_map renames columns):
            ``parent.proj(**{fk: pk for fk, pk in attr_map.items()})``
-        3. Non-aliased AND parent restriction attrs ⊄ child PK:
-           ``parent.proj()``
+        3. Otherwise (non-aliased): ``parent.proj()``
+
+        Copying is exact only when the restriction names attributes that this
+        foreign key carries and the foreign key lies in the child's primary
+        key. A restriction on an attribute that the child inherits through
+        another foreign key would select child rows by that attribute instead
+        of by the parent rows they reference. A secondary foreign key may be
+        NULL, and then the child row references no parent row at all.
         """
         child_pk = self.nodes[child_node].get("primary_key", set())
 
-        if not aliased and parent_attrs and parent_attrs <= child_pk:
+        if not aliased and parent_attrs and parent_attrs <= set(attr_map) <= child_pk:
             # Rule 1: copy parent restriction directly
             parent_restr = restrictions.get(
                 parent_ft.full_table_name,
@@ -912,7 +935,7 @@ def _apply_propagation_rule(
                 restrictions.setdefault(child_node, AndList()).append(child_item)
             child_attrs = set(attr_map.keys())
         else:
-            # Rule 3: non-aliased, restriction attrs ⊄ child PK — project
+            # Rule 3: non-aliased, not a pure copy — project
             child_item = parent_ft.proj()
             if mode == "cascade":
                 restrictions.setdefault(child_node, []).append(child_item)
@@ -922,7 +945,9 @@ def _apply_propagation_rule(
 
         self._restriction_attrs.setdefault(child_node, set()).update(child_attrs)
 
-    def _apply_propagation_rule_upward(self, child_ft, child_attrs, parent_node, attr_map, aliased, mode, restrictions):
+    def _apply_propagation_rule_upward(
+        self, child_ft, child_attrs, parent_node, attr_map, aliased, mode, restrictions, copy_allowed=True
+    ):
         """
         Apply the symmetric (upward) propagation rule to a parent←child edge.
 
@@ -947,10 +972,15 @@ def _apply_propagation_rule_upward(self, child_ft, child_attrs, parent_node, att
            ``child.proj(*attr_map.keys())`` — project child onto its FK columns
            (which, being non-aliased, share names with parent's PK columns) so
            the subsequent restriction on the parent joins on the right columns.
+
+        ``copy_allowed=False`` disables Rule 1, so the parent is restricted to
+        the rows that the restricted child rows actually reference. Copying
+        the child's condition would also select parent rows that no child row
+        references.
         """
         parent_pk = self.nodes[parent_node].get("primary_key", set())
 
-        if not aliased and child_attrs and child_attrs <= parent_pk:
+        if copy_allowed and not aliased and child_attrs and child_attrs <= parent_pk:
             # Backward Rule 1: copy child restriction directly
             child_restr = restrictions.get(
                 child_ft.full_table_name,
@@ -1044,6 +1074,21 @@ def _propagate_part_to_master(self, part_node, master_name, mode, restrictions):
         if len(real_path) < 2 or real_path[-1] != part_node or real_path[0] != master_name:
             return False
 
+        # The walk restricts the intermediate parts only to carry the
+        # restriction up to the master. Those restrictions reference the
+        # descendant part, which is deleted first, so they are restored once
+        # the master is materialized. The master then cascades back down to
+        # all of its parts.
+        intermediates = real_path[1:-1]
+        # Copy each node's attribute set too: the walk below extends it in place.
+        saved = {
+            node: (
+                copy.copy(restrictions[node]) if node in restrictions else None,
+                set(self._restriction_attrs[node]) if node in self._restriction_attrs else None,
+            )
+            for node in intermediates
+        }
+
         # Walk real_path in reverse (child → parent direction). For each
         # adjacent (parent, child) pair, look up the FK edge props.
         any_propagated = False
@@ -1059,6 +1104,8 @@ def _propagate_part_to_master(self, part_node, master_name, mode, restrictions):
             child_ft = self._restricted_table(child)
             child_attrs = self._restriction_attrs.get(child, set())
 
+            # The lift is existential: a master joins the cascade only if one
+            # of its part rows did, so the part's condition is never copied.
             self._apply_propagation_rule_upward(
                 child_ft,
                 child_attrs,
@@ -1067,6 +1114,7 @@ def _propagate_part_to_master(self, part_node, master_name, mode, restrictions):
                 aliased,
                 mode,
                 restrictions,
+                copy_allowed=False,
             )
             any_propagated = True
 
@@ -1094,6 +1142,16 @@ def _propagate_part_to_master(self, part_node, master_name, mode, restrictions):
                     restrictions[master_name] = [False]
                 self._restriction_attrs.setdefault(master_name, set())
 
+        for node, (restriction, attrs) in saved.items():
+            if restriction is None:
+                restrictions.pop(node, None)
+            else:
+                restrictions[node] = restriction
+            if attrs is None:
+                self._restriction_attrs.pop(node, None)
+            else:
+                self._restriction_attrs[node] = attrs
+
         return any_propagated
 
     def _edge_props(self, parent, child):

src/datajoint/table.py

@@ -1,4 +1,5 @@
 import collections
+import copy
 import csv
 import inspect
 import itertools
@@ -1083,6 +1084,12 @@ def delete_quick(self, get_count=False):
         Deletes the table without cascading and without user prompt.
         If this table has populated dependent tables, this will fail.
         """
+        if self._top is not None:
+            # dj.Top is an ORDER BY ... LIMIT, not part of the WHERE clause:
+            # delete exactly the rows it selects.
+            target = copy.copy(self)
+            target._top = None
+            return (target & self.keys()).delete_quick(get_count=get_count)
         query = "DELETE FROM " + self.full_table_name + self.where_clause()
         cursor = self.connection.query(query)
         # Use cursor.rowcount (DB-API 2.0 standard, works for both MySQL and PostgreSQL)