From 18e5649228dec6aed7a5a8b34b43cb91c4f95340 Mon Sep 17 00:00:00 2001 From: Dimitri Yatsenko Date: Sat, 17 Oct 2026 02:04:50 -0500 Subject: [PATCH] fix(cascade): restrict only the rows a cascade reaches Three cases deleted rows outside the cascade: - `(T & dj.Top(n)).delete()` and `delete_quick()` ignored the Top and deleted every row matching the other conditions. A Top seed is now replaced by the primary keys it selects. - A parent restriction was copied to a child whenever the restricted attributes were in the child's primary key. Through a nullable foreign key that shares a primary-key column, the copy also deleted child rows whose foreign key is NULL; in a diamond it missed rows. The copy now applies only when the restricted attributes are among the foreign key's attributes and the foreign key lies in the child's primary key. Otherwise the child is restricted by the projected parent. - With part_integrity="cascade", the part-to-master lift copied the part's condition to the master, which also deleted masters with no matching part rows. The master is now restricted by the restricted part, so a master joins the cascade only if one of its part rows did. The walk's restrictions on intermediate parts are restored once the master is materialized; the master cascades back to all of its parts. The Diagram copy constructor now copies each node's restriction list instead of deep-copying projected query expressions, which hold a live connection. --- src/datajoint/diagram.py | 82 ++++++-- src/datajoint/table.py | 7 + tests/integration/test_cascade_overdelete.py | 194 +++++++++++++++++++ 3 files changed, 269 insertions(+), 14 deletions(-) create mode 100644 tests/integration/test_cascade_overdelete.py diff --git a/src/datajoint/diagram.py b/src/datajoint/diagram.py index 6ff60d061..6975850a2 100644 --- a/src/datajoint/diagram.py +++ b/src/datajoint/diagram.py @@ -154,6 +154,20 @@ def stroke_rule(light_hex, dark_hex, opacity=None): return "" +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,17 @@ 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] + saved = { + node: (copy.copy(restrictions[node]) if node in restrictions else None, self._restriction_attrs.get(node)) + 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 +1100,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 +1110,7 @@ def _propagate_part_to_master(self, part_node, master_name, mode, restrictions): aliased, mode, restrictions, + copy_allowed=False, ) any_propagated = True @@ -1094,6 +1138,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): diff --git a/src/datajoint/table.py b/src/datajoint/table.py index 9f6784fb3..fa4233a11 100644 --- a/src/datajoint/table.py +++ b/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) diff --git a/tests/integration/test_cascade_overdelete.py b/tests/integration/test_cascade_overdelete.py new file mode 100644 index 000000000..e8015c2c3 --- /dev/null +++ b/tests/integration/test_cascade_overdelete.py @@ -0,0 +1,194 @@ +""" +Regression tests for cascade restrictions that reached rows they should not. + +A restriction may be copied verbatim across a foreign key only when it names +attributes the foreign key carries and the foreign key lies in the child's +primary key. The part-to-master lift never copies: a master joins the cascade +only if one of its part rows did. A ``Top`` seed restricts the seed to the keys +it selects. +""" + +import pytest + +import datajoint as dj + + +@pytest.fixture(scope="function") +def schema_by_backend(connection_by_backend, db_creds_by_backend, request): + """Create a fresh schema per test (mirrors test_cascade_integrity.py).""" + backend = db_creds_by_backend["backend"] + import time + + test_id = str(int(time.time() * 1000))[-8:] + schema_name = f"djtest_overdelete_{backend}_{test_id}"[:64] + + if connection_by_backend.is_connected: + try: + connection_by_backend.query( + f"DROP DATABASE IF EXISTS {connection_by_backend.adapter.quote_identifier(schema_name)}" + ) + except Exception: + pass + + schema = dj.Schema(schema_name, connection=connection_by_backend) + yield schema + + if connection_by_backend.is_connected: + try: + connection_by_backend.query( + f"DROP DATABASE IF EXISTS {connection_by_backend.adapter.quote_identifier(schema_name)}" + ) + except Exception: + pass + + +def test_top_seed_deletes_only_the_top_rows(schema_by_backend): + """``(T & dj.Top(n)).delete()`` and ``delete_quick()`` delete n rows, not + every row of T.""" + + @schema_by_backend + class Item(dj.Manual): + definition = """ + item_id : int32 + """ + + @schema_by_backend + class Child(dj.Manual): + definition = """ + -> Item + child_id : int32 + """ + + Item.insert([(i,) for i in range(5)]) + Child.insert([(i, 0) for i in range(5)]) + + seed = Item & dj.Top(2, order_by="item_id") + counts = dj.Diagram.cascade(seed).counts() + assert counts[Item.full_table_name] == 2 + assert counts[Child.full_table_name] == 2 + + seed.delete(prompt=False) + assert sorted(Item.to_arrays("item_id")) == [2, 3, 4] + assert sorted(Child.to_arrays("item_id")) == [2, 3, 4] + + (Child & dj.Top(1, order_by="item_id")).delete_quick() + assert sorted(Child.to_arrays("item_id")) == [3, 4] + + +def test_nullable_fk_sharing_a_primary_key_column(schema_by_backend): + """A child whose nullable foreign key shares a primary-key column with the + parent keeps the rows whose foreign key is NULL: they reference nothing.""" + + @schema_by_backend + class Subject(dj.Manual): + definition = """ + subject_id : int32 + """ + + @schema_by_backend + class Surgery(dj.Manual): + definition = """ + -> Subject + surgery_id : int32 + """ + + @schema_by_backend + class Session(dj.Manual): + definition = """ + -> Subject + session_id : int32 + --- + -> [nullable] Surgery + """ + + Subject.insert([(1,)]) + Surgery.insert([(1, 1)]) + Session.insert( + [ + {"subject_id": 1, "session_id": 1, "surgery_id": 1}, + {"subject_id": 1, "session_id": 2, "surgery_id": None}, + ] + ) + + seed = Surgery & {"subject_id": 1} + assert dj.Diagram.cascade(seed).counts()[Session.full_table_name] == 1 + + seed.delete(prompt=False) + assert len(Surgery()) == 0 + assert Session.to_dicts() == [{"subject_id": 1, "session_id": 2, "surgery_id": None}] + + +def test_diamond_child_is_reached_through_its_secondary_fk(schema_by_backend): + """D references B directly (primary) and C (secondary). Deleting B=1 must + delete the D rows that reference the deleted C rows, whatever D's own b_id.""" + + @schema_by_backend + class B(dj.Manual): + definition = """ + b_id : int32 + """ + + @schema_by_backend + class C(dj.Manual): + definition = """ + c_id : int32 + --- + -> B + """ + + @schema_by_backend + class D(dj.Manual): + definition = """ + -> B + d_id : int32 + --- + -> C + """ + + B.insert([(1,), (2,)]) + C.insert([{"c_id": 10, "b_id": 1}]) + D.insert([{"b_id": 2, "d_id": 1, "c_id": 10}]) + + seed = B & {"b_id": 1} + assert dj.Diagram.cascade(seed).counts()[D.full_table_name] == 1 + + seed.delete(prompt=False) + assert B.to_arrays("b_id").tolist() == [2] + assert len(C()) == 0 + assert len(D()) == 0 + + +def test_part_cascade_lifts_only_masters_that_have_a_matching_part(schema_by_backend): + """With part_integrity="cascade", a part seed lifts to the masters of the + matching part rows. A master that has no matching part row is untouched.""" + + @schema_by_backend + class Master(dj.Manual): + definition = """ + master_id : int32 + """ + + class P(dj.Part): + definition = """ + -> master + p_id : int32 + """ + + @schema_by_backend + class Downstream(dj.Manual): + definition = """ + -> Master + d_id : int32 + """ + + Master.insert([(1,), (2,)]) + Master.P.insert([(1, 1)]) # master 2 has no parts + Downstream.insert([(1, 1), (2, 1)]) + + seed = Master.P & [{"master_id": 1}, {"master_id": 2}] + assert dj.Diagram.cascade(seed, part_integrity="cascade").counts()[Master.full_table_name] == 1 + + seed.delete(prompt=False, part_integrity="cascade") + assert Master.to_arrays("master_id").tolist() == [2] + assert len(Master.P()) == 0 + assert Downstream.to_arrays("master_id").tolist() == [2]