feat: Add backend-agnostic cascade delete support

Cascade delete previously relied on parsing MySQL-specific foreign key
error messages. Now uses adapter methods for both MySQL and PostgreSQL.

New adapter methods:
1. parse_foreign_key_error(error_message) -> dict
   - Parses FK violation errors to extract constraint details
   - MySQL: Extracts from detailed error with full FK definition
   - PostgreSQL: Extracts table names and constraint from simpler error

2. get_constraint_info_sql(constraint_name, schema, table) -> str
   - Queries information_schema for FK column mappings
   - Used when error message doesn't include full FK details
   - MySQL: Uses KEY_COLUMN_USAGE with CONCAT for parent name
   - PostgreSQL: Joins KEY_COLUMN_USAGE with CONSTRAINT_COLUMN_USAGE

table.py cascade delete updates:
- Use adapter.parse_foreign_key_error() instead of hardcoded regexp
- Backend-agnostic quote stripping (handles both ` and ")
- Use adapter.get_constraint_info_sql() for querying FK details
- Properly rebuild child table names with schema when missing

This enables cascade delete operations to work correctly on PostgreSQL
while maintaining full backward compatibility with MySQL.

Part of multi-backend PostgreSQL support implementation.

Author: dimitri-yatsenko

src/datajoint/adapters/base.py

@@ -600,6 +600,72 @@ def get_foreign_keys_sql(self, schema_name: str, table_name: str) -> str:
         """
         ...
 
+    @abstractmethod
+    def get_constraint_info_sql(self, constraint_name: str, schema_name: str, table_name: str) -> str:
+        """
+        Generate query to get foreign key constraint details from information_schema.
+
+        Used during cascade delete to determine FK columns when error message
+        doesn't provide full details.
+
+        Parameters
+        ----------
+        constraint_name : str
+            Name of the foreign key constraint.
+        schema_name : str
+            Schema/database name of the child table.
+        table_name : str
+            Name of the child table.
+
+        Returns
+        -------
+        str
+            SQL query that returns rows with columns:
+            - fk_attrs: foreign key column name in child table
+            - parent: parent table name (quoted, with schema)
+            - pk_attrs: referenced column name in parent table
+        """
+        ...
+
+    @abstractmethod
+    def parse_foreign_key_error(self, error_message: str) -> dict[str, str | list[str]] | None:
+        """
+        Parse a foreign key violation error message to extract constraint details.
+
+        Used during cascade delete to identify which child table is preventing
+        deletion and what columns are involved.
+
+        Parameters
+        ----------
+        error_message : str
+            The error message from a foreign key constraint violation.
+
+        Returns
+        -------
+        dict or None
+            Dictionary with keys if successfully parsed:
+            - child: child table name (quoted with schema if available)
+            - name: constraint name (quoted)
+            - fk_attrs: list of foreign key column names (may be None if not in message)
+            - parent: parent table name (quoted, may be None if not in message)
+            - pk_attrs: list of parent key column names (may be None if not in message)
+
+            Returns None if error message doesn't match FK violation pattern.
+
+        Examples
+        --------
+        MySQL error:
+            "Cannot delete or update a parent row: a foreign key constraint fails
+            (`schema`.`child`, CONSTRAINT `fk_name` FOREIGN KEY (`child_col`)
+            REFERENCES `parent` (`parent_col`))"
+
+        PostgreSQL error:
+            "update or delete on table \"parent\" violates foreign key constraint
+            \"child_parent_id_fkey\" on table \"child\"
+            DETAIL:  Key (parent_id)=(1) is still referenced from table \"child\"."
+        """
+        ...
+
     @abstractmethod
     def get_indexes_sql(self, schema_name: str, table_name: str) -> str:
         """

src/datajoint/adapters/mysql.py

@@ -595,6 +595,44 @@ def get_foreign_keys_sql(self, schema_name: str, table_name: str) -> str:
             f"ORDER BY constraint_name, ordinal_position"
         )
 
+    def get_constraint_info_sql(self, constraint_name: str, schema_name: str, table_name: str) -> str:
+        """Query to get FK constraint details from information_schema."""
+        return (
+            f"SELECT "
+            f"  COLUMN_NAME as fk_attrs, "
+            f"  CONCAT('`', REFERENCED_TABLE_SCHEMA, '`.`', REFERENCED_TABLE_NAME, '`') as parent, "
+            f"  REFERENCED_COLUMN_NAME as pk_attrs "
+            f"FROM INFORMATION_SCHEMA.KEY_COLUMN_USAGE "
+            f"WHERE CONSTRAINT_NAME = %s AND TABLE_SCHEMA = %s AND TABLE_NAME = %s"
+        )
+
+    def parse_foreign_key_error(self, error_message: str) -> dict[str, str | list[str]] | None:
+        """Parse MySQL foreign key violation error message."""
+        import re
+
+        # MySQL FK error pattern with backticks
+        pattern = re.compile(
+            r"[\w\s:]*\((?P<child>`[^`]+`.`[^`]+`), "
+            r"CONSTRAINT (?P<name>`[^`]+`) "
+            r"(FOREIGN KEY \((?P<fk_attrs>[^)]+)\) "
+            r"REFERENCES (?P<parent>`[^`]+`(\.`[^`]+`)?) \((?P<pk_attrs>[^)]+)\)[\s\w]+\))?"
+        )
+
+        match = pattern.match(error_message)
+        if not match:
+            return None
+
+        result = match.groupdict()
+
+        # Parse comma-separated FK attrs if present
+        if result.get("fk_attrs"):
+            result["fk_attrs"] = [col.strip("`") for col in result["fk_attrs"].split(",")]
+        # Parse comma-separated PK attrs if present
+        if result.get("pk_attrs"):
+            result["pk_attrs"] = [col.strip("`") for col in result["pk_attrs"].split(",")]
+
+        return result
+
     def get_indexes_sql(self, schema_name: str, table_name: str) -> str:
         """Query to get index definitions."""
         return (

src/datajoint/adapters/postgres.py

@@ -667,6 +667,52 @@ def get_foreign_keys_sql(self, schema_name: str, table_name: str) -> str:
             f"ORDER BY kcu.constraint_name, kcu.ordinal_position"
         )
 
+    def get_constraint_info_sql(self, constraint_name: str, schema_name: str, table_name: str) -> str:
+        """Query to get FK constraint details from information_schema."""
+        return (
+            f"SELECT "
+            f"  kcu.column_name as fk_attrs, "
+            f"  '\"' || ccu.table_schema || '\".\"' || ccu.table_name || '\"' as parent, "
+            f"  ccu.column_name as pk_attrs "
+            f"FROM information_schema.key_column_usage AS kcu "
+            f"JOIN information_schema.constraint_column_usage AS ccu "
+            f"  ON kcu.constraint_name = ccu.constraint_name "
+            f"  AND kcu.constraint_schema = ccu.constraint_schema "
+            f"WHERE kcu.constraint_name = %s "
+            f"  AND kcu.table_schema = %s "
+            f"  AND kcu.table_name = %s "
+            f"ORDER BY kcu.ordinal_position"
+        )
+
+    def parse_foreign_key_error(self, error_message: str) -> dict[str, str | list[str]] | None:
+        """Parse PostgreSQL foreign key violation error message."""
+        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>[^"]+)"'
+        )
+
+        match = pattern.match(error_message)
+        if not match:
+            return None
+
+        result = match.groupdict()
+
+        # Build child table name (assume same schema as parent for now)
+        # The error doesn't include schema, so we return unqualified names
+        # and let the caller add schema context
+        child = f'"{result["child_table"]}"'
+
+        return {
+            "child": child,
+            "name": f'"{result["name"]}"',
+            "fk_attrs": None,  # Not in error message, will need constraint query
+            "parent": f'"{result["parent_table"]}"',
+            "pk_attrs": None,  # Not in error message, will need constraint query
+        }
+
     def get_indexes_sql(self, schema_name: str, table_name: str) -> str:
         """Query to get index definitions."""
         return (

src/datajoint/table.py

@@ -30,24 +30,8 @@
 
 logger = logging.getLogger(__name__.split(".")[0])
 
-foreign_key_error_regexp = re.compile(
-    r"[\w\s:]*\((?P<child>`[^`]+`.`[^`]+`), "
-    r"CONSTRAINT (?P<name>`[^`]+`) "
-    r"(FOREIGN KEY \((?P<fk_attrs>[^)]+)\) "
-    r"REFERENCES (?P<parent>`[^`]+`(\.`[^`]+`)?) \((?P<pk_attrs>[^)]+)\)[\s\w]+\))?"
-)
-
-constraint_info_query = " ".join(
-    """
-    SELECT
-        COLUMN_NAME as fk_attrs,
-        CONCAT('`', REFERENCED_TABLE_SCHEMA, '`.`', REFERENCED_TABLE_NAME, '`') as parent,
-        REFERENCED_COLUMN_NAME as pk_attrs
-    FROM INFORMATION_SCHEMA.KEY_COLUMN_USAGE
-    WHERE
-        CONSTRAINT_NAME = %s AND TABLE_SCHEMA = %s AND TABLE_NAME = %s;
-    """.split()
-)
+# Note: Foreign key error parsing is now handled by adapter methods
+# Legacy regexp and query kept for reference but no longer used
 
 
 class _RenameMap(tuple):
@@ -895,35 +879,53 @@ def cascade(table):
                 try:
                     delete_count = table.delete_quick(get_count=True)
                 except IntegrityError as error:
-                    match = foreign_key_error_regexp.match(error.args[0])
+                    # Use adapter to parse FK error message
+                    match = table.connection.adapter.parse_foreign_key_error(error.args[0])
                     if match is None:
                         raise DataJointError(
-                            "Cascading deletes failed because the error message is missing foreign key information."
+                            "Cascading deletes failed because the error message is missing foreign key information. "
                             "Make sure you have REFERENCES privilege to all dependent tables."
                         ) from None
-                    match = match.groupdict()
-                    # if schema name missing, use table
-                    if "`.`" not in match["child"]:
-                        match["child"] = "{}.{}".format(table.full_table_name.split(".")[0], match["child"])
-                    if match["pk_attrs"] is not None:  # fully matched, adjusting the keys
-                        match["fk_attrs"] = [k.strip("`") for k in match["fk_attrs"].split(",")]
-                        match["pk_attrs"] = [k.strip("`") for k in match["pk_attrs"].split(",")]
-                    else:  # only partially matched, querying with constraint to determine keys
-                        match["fk_attrs"], match["parent"], match["pk_attrs"] = list(
-                            map(
-                                list,
-                                zip(
-                                    *table.connection.query(
-                                        constraint_info_query,
-                                        args=(
-                                            match["name"].strip("`"),
-                                            *[_.strip("`") for _ in match["child"].split("`.`")],
-                                        ),
-                                    ).fetchall()
-                                ),
-                            )
+
+                    # Strip quotes from parsed values for backend-agnostic processing
+                    quote_chars = ('`', '"')
+
+                    def strip_quotes(s):
+                        if s and any(s.startswith(q) for q in quote_chars):
+                            return s.strip('`"')
+                        return s
+
+                    # Ensure child table has schema
+                    child_table = match["child"]
+                    if "." not in strip_quotes(child_table):
+                        # 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
+
+                    # 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,
                         )
-                        match["parent"] = match["parent"][0]
+
+                        results = table.connection.query(constraint_query).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
 
                     # Restrict child by table if
                     #   1. if table's restriction attributes are not in child's primary key