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]