fix: Backend-agnostic fixes for cascade delete and FreeTable

- Fix FreeTable.__init__ to strip both backticks and double quotes
- Fix heading.py error message to not add hardcoded backticks
- Fix Attribute.original_name to accept both quote types
- Fix delete_quick() to use cursor.rowcount instead of ROW_COUNT()
- Update PostgreSQL FK error parser with clearer naming
- Add cascade delete integration tests

All 4 PostgreSQL multi-backend tests passing.
Cascade delete logic working correctly.

Author: dimitri-yatsenko

src/datajoint/adapters/postgres.py

@@ -685,13 +685,19 @@ def get_constraint_info_sql(self, constraint_name: str, schema_name: str, table_
         )
 
     def parse_foreign_key_error(self, error_message: str) -> dict[str, str | list[str]] | None:
-        """Parse PostgreSQL foreign key violation error message."""
+        """
+        Parse PostgreSQL foreign key violation error message.
+
+        PostgreSQL FK error format:
+        'update or delete on table "X" violates foreign key constraint "Y" on table "Z"'
+        Where:
+        - "X" is the referenced table (being deleted/updated)
+        - "Z" is the referencing table (has the FK, needs cascade delete)
+        """
         import re
 
-        # PostgreSQL FK error pattern
-        # Example: 'update or delete on table "parent" violates foreign key constraint "child_parent_id_fkey" on table "child"'
         pattern = re.compile(
-            r'.*table "(?P<parent_table>[^"]+)" violates foreign key constraint "(?P<name>[^"]+)" on table "(?P<child_table>[^"]+)"'
+            r'.*table "(?P<referenced_table>[^"]+)" violates foreign key constraint "(?P<name>[^"]+)" on table "(?P<referencing_table>[^"]+)"'
         )
 
         match = pattern.match(error_message)
@@ -700,16 +706,17 @@ def parse_foreign_key_error(self, error_message: str) -> dict[str, str | list[st
 
         result = match.groupdict()
 
-        # Build child table name (assume same schema as parent for now)
+        # The child is the referencing table (the one with the FK that needs cascade delete)
+        # The parent is the referenced table (the one being deleted)
         # The error doesn't include schema, so we return unqualified names
-        # and let the caller add schema context
-        child = f'"{result["child_table"]}"'
+        child = f'"{result["referencing_table"]}"'
+        parent = f'"{result["referenced_table"]}"'
 
         return {
             "child": child,
             "name": f'"{result["name"]}"',
             "fk_attrs": None,  # Not in error message, will need constraint query
-            "parent": f'"{result["parent_table"]}"',
+            "parent": parent,
             "pk_attrs": None,  # Not in error message, will need constraint query
         }
 

src/datajoint/heading.py

@@ -164,8 +164,9 @@ def original_name(self) -> str:
         """
         if self.attribute_expression is None:
             return self.name
-        assert self.attribute_expression.startswith("`")
-        return self.attribute_expression.strip("`")
+        # Backend-agnostic quote stripping (MySQL uses `, PostgreSQL uses ")
+        assert self.attribute_expression.startswith(("`", '"'))
+        return self.attribute_expression.strip('`"')
 
 
 class Heading:
@@ -365,7 +366,7 @@ def _init_from_database(self) -> None:
         ).fetchone()
         if info is None:
             raise DataJointError(
-                "The table `{database}`.`{table_name}` is not defined.".format(table_name=table_name, database=database)
+                f"The table {database}.{table_name} is not defined."
             )
         # Normalize table_comment to comment for backward compatibility
         self._table_status = {k.lower(): v for k, v in info.items()}

src/datajoint/table.py

@@ -834,8 +834,9 @@ def delete_quick(self, get_count=False):
         If this table has populated dependent tables, this will fail.
         """
         query = "DELETE FROM " + self.full_table_name + self.where_clause()
-        self.connection.query(query)
-        count = self.connection.query("SELECT ROW_COUNT()").fetchone()[0] if get_count else None
+        cursor = self.connection.query(query)
+        # Use cursor.rowcount (DB-API 2.0 standard, works for both MySQL and PostgreSQL)
+        count = cursor.rowcount if get_count else None
         return count
 
     def delete(
@@ -876,9 +877,17 @@ def cascade(table):
             """service function to perform cascading deletes recursively."""
             max_attempts = 50
             for _ in range(max_attempts):
+                # Set savepoint before delete attempt (for PostgreSQL transaction handling)
+                savepoint_name = f"cascade_delete_{id(table)}"
+                if transaction:
+                    table.connection.query(f"SAVEPOINT {savepoint_name}")
+
                 try:
                     delete_count = table.delete_quick(get_count=True)
                 except IntegrityError as error:
+                    # Rollback to savepoint so we can continue querying (PostgreSQL requirement)
+                    if transaction:
+                        table.connection.query(f"ROLLBACK TO SAVEPOINT {savepoint_name}")
                     # Use adapter to parse FK error message
                     match = table.connection.adapter.parse_foreign_key_error(error.args[0])
                     if match is None:
@@ -895,43 +904,47 @@ def strip_quotes(s):
                             return s.strip('`"')
                         return s
 
-                    # Ensure child table has schema
-                    child_table = match["child"]
-                    if "." not in strip_quotes(child_table):
+                    # Extract schema and table name from child (work with unquoted names)
+                    child_table_raw = strip_quotes(match["child"])
+                    if "." in child_table_raw:
+                        child_parts = child_table_raw.split(".")
+                        child_schema = strip_quotes(child_parts[0])
+                        child_table_name = strip_quotes(child_parts[1])
+                    else:
                         # Add schema from current table
-                        schema = table.full_table_name.split(".")[0].strip('`"')
-                        child_unquoted = strip_quotes(child_table)
-                        child_table = f"{table.connection.adapter.quote_identifier(schema)}.{table.connection.adapter.quote_identifier(child_unquoted)}"
-                        match["child"] = child_table
+                        schema_parts = table.full_table_name.split(".")
+                        child_schema = strip_quotes(schema_parts[0])
+                        child_table_name = child_table_raw
 
                     # If FK/PK attributes not in error message, query information_schema
                     if match["fk_attrs"] is None or match["pk_attrs"] is None:
-                        # Extract schema and table name from child
-                        child_parts = [strip_quotes(p) for p in child_table.split(".")]
-                        if len(child_parts) == 2:
-                            child_schema, child_table_name = child_parts
-                        else:
-                            child_schema = table.full_table_name.split(".")[0].strip('`"')
-                            child_table_name = child_parts[0]
-
                         constraint_query = table.connection.adapter.get_constraint_info_sql(
                             strip_quotes(match["name"]),
                             child_schema,
                             child_table_name,
                         )
 
-                        results = table.connection.query(constraint_query).fetchall()
+                        results = table.connection.query(
+                            constraint_query,
+                            args=(strip_quotes(match["name"]), child_schema, child_table_name),
+                        ).fetchall()
                         if results:
                             match["fk_attrs"], match["parent"], match["pk_attrs"] = list(
                                 map(list, zip(*results))
                             )
                             match["parent"] = match["parent"][0]  # All rows have same parent
 
+                    # Build properly quoted full table name for FreeTable
+                    child_full_name = (
+                        f"{table.connection.adapter.quote_identifier(child_schema)}."
+                        f"{table.connection.adapter.quote_identifier(child_table_name)}"
+                    )
+
                     # Restrict child by table if
                     #   1. if table's restriction attributes are not in child's primary key
                     #   2. if child renames any attributes
                     # Otherwise restrict child by table's restriction.
-                    child = FreeTable(table.connection, match["child"])
+                    child = FreeTable(table.connection, child_full_name)
                     if set(table.restriction_attributes) <= set(child.primary_key) and match["fk_attrs"] == match["pk_attrs"]:
                         child._restriction = table._restriction
                         child._restriction_attributes = table.restriction_attributes
@@ -961,6 +974,9 @@ def strip_quotes(s):
                     else:
                         cascade(child)
                 else:
+                    # Successful delete - release savepoint
+                    if transaction:
+                        table.connection.query(f"RELEASE SAVEPOINT {savepoint_name}")
                     deleted.add(table.full_table_name)
                     logger.info("Deleting {count} rows from {table}".format(count=delete_count, table=table.full_table_name))
                     break
@@ -1381,7 +1397,8 @@ class FreeTable(Table):
     """
 
     def __init__(self, conn, full_table_name):
-        self.database, self._table_name = (s.strip("`") for s in full_table_name.split("."))
+        # Backend-agnostic quote stripping (MySQL uses `, PostgreSQL uses ")
+        self.database, self._table_name = (s.strip('`"') for s in full_table_name.split("."))
         self._connection = conn
         self._support = [full_table_name]
         self._heading = Heading(

tests/integration/test_cascade_delete.py

@@ -0,0 +1,170 @@
+"""
+Integration tests for cascade delete on multiple backends.
+"""
+
+import os
+
+import pytest
+
+import datajoint as dj
+
+
+@pytest.fixture(scope="function")
+def schema_by_backend(connection_by_backend, db_creds_by_backend, request):
+    """Create a schema for cascade delete tests."""
+    backend = db_creds_by_backend["backend"]
+    # Use unique schema name for each test
+    import time
+    test_id = str(int(time.time() * 1000))[-8:]  # Last 8 digits of timestamp
+    schema_name = f"djtest_cascade_{backend}_{test_id}"[:64]  # Limit length
+
+    # Drop schema if exists (cleanup from any previous failed runs)
+    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  # Ignore errors during cleanup
+
+    # Create fresh schema
+    schema = dj.Schema(schema_name, connection=connection_by_backend)
+
+    yield schema
+
+    # Cleanup after test
+    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  # Ignore errors during cleanup
+
+
+def test_simple_cascade_delete(schema_by_backend):
+    """Test basic cascade delete with foreign keys."""
+
+    @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)
+        """
+
+    # Insert test data
+    Parent.insert1((1, "Parent1"))
+    Parent.insert1((2, "Parent2"))
+    Child.insert1((1, 1, "Child1-1"))
+    Child.insert1((1, 2, "Child1-2"))
+    Child.insert1((2, 1, "Child2-1"))
+
+    assert len(Parent()) == 2
+    assert len(Child()) == 3
+
+    # Delete parent with cascade
+    (Parent & {"parent_id": 1}).delete()
+
+    # Check cascade worked
+    assert len(Parent()) == 1
+    assert len(Child()) == 1
+    assert (Child & {"parent_id": 2, "child_id": 1}).fetch1("data") == "Child2-1"
+
+
+def test_multi_level_cascade_delete(schema_by_backend):
+    """Test cascade delete through multiple levels of foreign keys."""
+
+    @schema_by_backend
+    class GrandParent(dj.Manual):
+        definition = """
+        gp_id : int
+        ---
+        name : varchar(255)
+        """
+
+    @schema_by_backend
+    class Parent(dj.Manual):
+        definition = """
+        -> GrandParent
+        parent_id : int
+        ---
+        name : varchar(255)
+        """
+
+    @schema_by_backend
+    class Child(dj.Manual):
+        definition = """
+        -> Parent
+        child_id : int
+        ---
+        data : varchar(255)
+        """
+
+    # Insert test data
+    GrandParent.insert1((1, "GP1"))
+    Parent.insert1((1, 1, "P1"))
+    Parent.insert1((1, 2, "P2"))
+    Child.insert1((1, 1, 1, "C1"))
+    Child.insert1((1, 1, 2, "C2"))
+    Child.insert1((1, 2, 1, "C3"))
+
+    assert len(GrandParent()) == 1
+    assert len(Parent()) == 2
+    assert len(Child()) == 3
+
+    # Delete grandparent - should cascade through parent to child
+    (GrandParent & {"gp_id": 1}).delete()
+
+    # Check everything is deleted
+    assert len(GrandParent()) == 0
+    assert len(Parent()) == 0
+    assert len(Child()) == 0
+
+
+def test_cascade_delete_with_renamed_attrs(schema_by_backend):
+    """Test cascade delete when foreign key renames attributes."""
+
+    @schema_by_backend
+    class Animal(dj.Manual):
+        definition = """
+        animal_id : int
+        ---
+        species : varchar(255)
+        """
+
+    @schema_by_backend
+    class Observation(dj.Manual):
+        definition = """
+        obs_id : int
+        ---
+        -> Animal.proj(subject_id='animal_id')
+        measurement : float
+        """
+
+    # Insert test data
+    Animal.insert1((1, "Mouse"))
+    Animal.insert1((2, "Rat"))
+    Observation.insert1((1, 1, 10.5))
+    Observation.insert1((2, 1, 11.2))
+    Observation.insert1((3, 2, 15.3))
+
+    assert len(Animal()) == 2
+    assert len(Observation()) == 3
+
+    # Delete animal - should cascade to observations
+    (Animal & {"animal_id": 1}).delete()
+
+    # Check cascade worked
+    assert len(Animal()) == 1
+    assert len(Observation()) == 1
+    assert (Observation & {"obs_id": 3}).fetch1("measurement") == 15.3