refactor(deletion-retention): add declarative purge policies (#42888)

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>

superset/commands/deletion_retention/purge_cascade.py

@@ -52,6 +52,12 @@ import sqlalchemy as sa
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.orm import Session
 
+from superset.commands.deletion_retention.purge_policy import (
+    get_purge_policy,
+    PurgeBlockedError,
+    PurgeEntityPolicy,
+)
+
 logger: logging.Logger = logging.getLogger(__name__)
 
 
@@ -141,10 +147,6 @@ class CascadeResult:
     blocked_reason: str | None = None
 
 
-class PurgeBlockedError(Exception):
-    """Raised when ordinary deletion policy forbids purging an entity."""
-
-
 class PurgeRaceLostError(Exception):
     """Raised to roll back dependent cleanup when the entity delete loses."""
 
@@ -196,10 +198,6 @@ def cascade_hard_delete(
     window) requires only that it is still soft-deleted, so a restore
     committed after the caller resolved the entity cannot be destroyed.
     """
-    # pylint: disable=import-outside-toplevel
-    from superset.connectors.sqla.models import SqlaTable
-    from superset.models.slice import Slice
-
     if enforce_window and cutoff is None:
         raise ValueError("cutoff is required when enforce_window=True")
 
@@ -207,12 +205,13 @@ def cascade_hard_delete(
     table = model.__table__
     entity_id = entity.id
     uuid = entity_uuid(entity)
-    entity_type = _USER_FACING_TYPE.get(table.name, table.name)
+    policy: PurgeEntityPolicy = get_purge_policy(model)
+    entity_type: str = policy.entity_type
 
     dangling_chart_uuids: list[str] = []
     removed_dashboard_slices = 0
     version_rows = 0
-    permission_name = _dataset_permission_name(entity) if model is SqlaTable else None
+    permission_name: str | None = None
 
     try:
         with session.begin_nested():
@@ -230,22 +229,19 @@ def cascade_hard_delete(
             if session.execute(claim.with_for_update()).scalar_one_or_none() is None:
                 raise PurgeRaceLostError
 
-            _validate_deletion_allowed(session, model, entity_id)
-            removed_dashboard_slices = _count_dashboard_slices(
-                session, model, entity_id
+            policy.validate(session, policy, entity_id)
+            # Captured under the lock: the row is claimed, so the identity
+            # the permission name is built from can no longer change.
+            permission_name = policy.capture_permission_name(session, policy, entity_id)
+            removed_dashboard_slices = policy.count_dashboard_slices(
+                session, policy, entity_id
+            )
+            dangling_chart_uuids = policy.collect_dangling_chart_uuids(
+                session, policy, entity_id
             )
-            if model is SqlaTable:
-                dangling_chart_uuids = [
-                    str(chart_uuid)
-                    for (chart_uuid,) in session.execute(
-                        sa.select(Slice.uuid)
-                        .where(Slice.datasource_id == entity_id)
-                        .where(Slice.datasource_type == "table")
-                    )
-                ]
 
-            _delete_m2m_joins(session, model, entity_id)
-            _delete_owned_children(session, model, entity_id)
+            policy.delete_associations(session, policy, entity_id)
+            policy.delete_owned_children(session, policy, entity_id)
             version_rows = _delete_version_history(session, entity, entity_id)
 
             delete_entity = sa.delete(table).where(*identity).where(*eligibility)
@@ -253,7 +249,7 @@ def cascade_hard_delete(
                 raise PurgeRaceLostError
 
             if permission_name is not None:
-                _cleanup_dataset_permission(session, permission_name, entity_id)
+                policy.cleanup_permission(session, policy, permission_name, entity_id)
     except PurgeRaceLostError:
         logger.info(
             "deletion_retention: %s id=%s not purged (restored or already gone)",
@@ -306,207 +302,10 @@ def cascade_hard_delete(
     )
 
 
-_USER_FACING_TYPE: dict[str, str] = {
-    "slices": "chart",
-    "dashboards": "dashboard",
-    "tables": "dataset",
-}
-
-
-def _validate_deletion_allowed(
-    session: Session, model: type[Any], entity_id: int
-) -> None:
-    """Apply the dependency guards used by ordinary delete commands."""
-    # pylint: disable=import-outside-toplevel
-    from superset.models.dashboard import Dashboard
-    from superset.models.slice import Slice
-    from superset.models.user_attributes import UserAttribute
-    from superset.reports.models import ReportSchedule
-
-    column: Any | None = None
-    if model is Slice:
-        column = ReportSchedule.chart_id
-    elif model is Dashboard:
-        column = ReportSchedule.dashboard_id
-    if (
-        column is not None
-        and session.execute(
-            sa.select(ReportSchedule.id).where(column == entity_id).limit(1)
-        ).first()
-    ):
-        raise PurgeBlockedError("associated alerts or reports exist")
-
-    # The welcome-dashboard reference must be an explicit guard, not a hope
-    # that the database enforces it: user_attributes.welcome_dashboard_id has
-    # no ondelete, so on FK-enforcing backends the delete fails with an
-    # IntegrityError misreported as a policy block -- while on SQLite with
-    # FKs off the dashboard purges "successfully", strands a dangling pointer
-    # (a broken user homepage), and the audit row says confirmed. One check,
-    # both dialect families, and a reason the blocked entity can be named by.
-    if (
-        model is Dashboard
-        and session.execute(
-            sa.select(UserAttribute.id)
-            .where(UserAttribute.welcome_dashboard_id == entity_id)
-            .limit(1)
-        ).first()
-    ):
-        raise PurgeBlockedError("a user has this dashboard set as their welcome page")
-
-
-def _count_dashboard_slices(session: Session, model: type[Any], entity_id: int) -> int:
-    """Snapshot relationship counts before DB cascades can remove rows."""
-    # pylint: disable=import-outside-toplevel
-    from superset.models.dashboard import Dashboard, dashboard_slices
-    from superset.models.slice import Slice
-
-    predicate: Any | None = None
-    if model is Dashboard:
-        predicate = dashboard_slices.c.dashboard_id == entity_id
-    elif model is Slice:
-        predicate = dashboard_slices.c.slice_id == entity_id
-    if predicate is None:
-        return 0
-    return int(
-        session.execute(
-            sa.select(sa.func.count()).select_from(dashboard_slices).where(predicate)
-        ).scalar_one()
-    )
-
-
 def dashboard_slice_count(session: Session, entity: Any) -> int:
     """Return the current dashboard relationship count for audit write-ahead."""
-    return _count_dashboard_slices(session, type(entity), entity.id)
-
-
-def _delete_m2m_joins(session: Session, model: type[Any], entity_id: int) -> None:
-    """Hard-delete every M:N join / association row the entity owns.
-
-    Relationship counts are captured before this function runs so database
-    cascades cannot make the reported values dialect-dependent.
-    """
-    # pylint: disable=import-outside-toplevel
-    from superset.connectors.sqla.models import SqlaTable
-    from superset.models.dashboard import Dashboard, dashboard_slices
-    from superset.models.slice import Slice
-    from superset.subjects.models import (
-        chart_editors,
-        chart_viewers,
-        dashboard_editors,
-        dashboard_viewers,
-        sqlatable_editors,
-    )
-    from superset.tags.models import ObjectType, TaggedObject
-
-    if model is Dashboard:
-        session.execute(
-            sa.delete(dashboard_slices).where(
-                dashboard_slices.c.dashboard_id == entity_id
-            )
-        )
-        for association in (dashboard_editors, dashboard_viewers):
-            session.execute(
-                sa.delete(association).where(association.c.dashboard_id == entity_id)
-            )
-        _delete_tags(session, TaggedObject, ObjectType.dashboard, entity_id)
-    elif model is Slice:
-        # Every dashboard_slices row pointing at this chart, including those
-        # owned by live dashboards (the live dashboard survives, minus this
-        # chart from its layout).
-        session.execute(
-            sa.delete(dashboard_slices).where(dashboard_slices.c.slice_id == entity_id)
-        )
-        for association in (chart_editors, chart_viewers):
-            session.execute(
-                sa.delete(association).where(association.c.chart_id == entity_id)
-            )
-        _delete_tags(session, TaggedObject, ObjectType.chart, entity_id)
-    elif model is SqlaTable:
-        from superset.connectors.sqla.models import RLSFilterTables
-
-        session.execute(
-            sa.delete(sqlatable_editors).where(
-                sqlatable_editors.c.table_id == entity_id
-            )
-        )
-        session.execute(
-            sa.delete(RLSFilterTables).where(RLSFilterTables.c.table_id == entity_id)
-        )
-        _delete_tags(session, TaggedObject, ObjectType.dataset, entity_id)
-
-
-def _delete_tags(
-    session: Session, tagged_object: type[Any], object_type: Any, entity_id: int
-) -> None:
-    """Remove ``tagged_object`` rows skipped by the Core bulk delete."""
-    session.execute(
-        sa.delete(tagged_object.__table__).where(
-            tagged_object.object_id == entity_id,
-            tagged_object.object_type == object_type,
-        )
-    )
-
-
-def _delete_owned_children(session: Session, model: type[Any], entity_id: int) -> None:
-    """Hard-delete the entity's owned children — rows with no independent
-    existence: a dataset's columns and metrics, a dashboard's embedded
-    configs. Charts have no such owned child tables today.
-    """
-    # pylint: disable=import-outside-toplevel
-    from superset.connectors.sqla.models import SqlaTable, SqlMetric, TableColumn
-    from superset.models.dashboard import Dashboard
-    from superset.models.embedded_dashboard import EmbeddedDashboard
-
-    if model is SqlaTable:
-        session.execute(
-            sa.delete(TableColumn.__table__).where(
-                TableColumn.__table__.c.table_id == entity_id
-            )
-        )
-        session.execute(
-            sa.delete(SqlMetric.__table__).where(
-                SqlMetric.__table__.c.table_id == entity_id
-            )
-        )
-    elif model is Dashboard:
-        # Embedded configs (delete-orphan children carrying the public
-        # embed UUID and allowed_domains) — the ORM cascade does not fire
-        # for Core deletes and the DB cascade is a backstop only.
-        session.execute(
-            sa.delete(EmbeddedDashboard.__table__).where(
-                EmbeddedDashboard.__table__.c.dashboard_id == entity_id
-            )
-        )
-
-
-def _dataset_permission_name(entity: Any) -> str:
-    """Capture the permission identifier while dataset attributes are readable."""
-    # pylint: disable=import-outside-toplevel
-    from superset import security_manager
-
-    return str(
-        security_manager.get_dataset_perm(
-            entity.id, entity.table_name, entity.database.database_name
-        )
-    )
-
-
-def _cleanup_dataset_permission(
-    session: Session, permission_name: str, entity_id: int
-) -> None:
-    """Replicate ``SqlaTable.after_delete`` permission cleanup.
-
-    Core ``sa.delete`` does not fire the ORM ``after_delete`` listener that
-    normally removes the dataset's ``datasource access`` view-menu /
-    permission-view, so it is done explicitly here or the PVM is orphaned.
-    """
-    # pylint: disable=import-outside-toplevel
-    from superset import security_manager
-
-    security_manager._delete_pvm_on_sqla_event(  # pylint: disable=protected-access
-        None, session.connection(), "datasource_access", permission_name
-    )
-    logger.debug("deletion_retention: removed dataset permission for id=%s", entity_id)
+    policy: PurgeEntityPolicy = get_purge_policy(type(entity))
+    return policy.count_dashboard_slices(session, policy, entity.id)
 
 
 def _entity_version_targets(
@@ -520,24 +319,19 @@ def _entity_version_targets(
     the dashboard/chart M2M shadow (``dashboard_slices_version``) and a
     dataset's child shadows (``table_columns_version`` / ``sql_metrics_version``
     keyed by ``table_id``). It never touches another entity's rows."""
-    # pylint: disable=import-outside-toplevel
-    from superset.connectors.sqla.models import SqlaTable
-    from superset.models.dashboard import Dashboard
-    from superset.models.slice import Slice
-
-    targets: list[tuple[sa.Table, Any]] = [
-        (parent_shadow, parent_shadow.c.id == entity_id)
-    ]
-    m2m = metadata.tables.get("dashboard_slices_version")
-    if m2m is not None and model is Dashboard:
-        targets.append((m2m, m2m.c.dashboard_id == entity_id))
-    elif m2m is not None and model is Slice:
-        targets.append((m2m, m2m.c.slice_id == entity_id))
-    elif model is SqlaTable:
-        for child_name in ("table_columns_version", "sql_metrics_version"):
-            child = metadata.tables.get(child_name)
-            if child is not None and "table_id" in child.c:
-                targets.append((child, child.c.table_id == entity_id))
+    targets: list[tuple[sa.Table, Any]] = []
+    for table_name, column_name in get_purge_policy(model).version_shadow_names:
+        shadow: sa.Table | None = (
+            parent_shadow
+            if table_name == parent_shadow.name
+            else metadata.tables.get(table_name)
+        )
+        if shadow is None or column_name not in shadow.c:
+            raise RuntimeError(
+                f"Invalid version shadow declaration for {model.__name__}: "
+                f"{table_name}.{column_name}"
+            )
+        targets.append((shadow, shadow.c[column_name] == entity_id))
     return targets
 
 

superset/connectors/sqla/models.py

@@ -120,6 +120,11 @@ from superset.superset_typing import (
 )
 from superset.utils import core as utils, json
 from superset.utils.backports import StrEnum
+from superset.utils.sqlalchemy_events import (
+    DeleteListenerDeclaration,
+    DeleteListenerEffect,
+    register_delete_listener,
+)
 
 config = current_app.config  # Backward compatibility for tests
 metadata = Model.metadata  # pylint: disable=no-member
@@ -2392,7 +2397,14 @@ class SqlaTable(
 
 sa.event.listen(SqlaTable, "before_update", SqlaTable.before_update)
 sa.event.listen(SqlaTable, "after_insert", SqlaTable.after_insert)
-sa.event.listen(SqlaTable, "after_delete", SqlaTable.after_delete)
+register_delete_listener(
+    DeleteListenerDeclaration(
+        SqlaTable,
+        "datasource_permission_cleanup",
+        DeleteListenerEffect.PERMISSION_ARTIFACT,
+        SqlaTable.after_delete,
+    )
+)
 
 RLSFilterSubjects = DBTable(
     "rls_filter_subjects",

superset/migrations/versions/2018-07-26_11-10_c82ee8a39623_add_implicit_tags.py

@@ -29,7 +29,6 @@ down_revision = "c617da68de7d"
 from datetime import datetime  # noqa: E402
 
 from alembic import op  # noqa: E402
-from flask_appbuilder.models.mixins import AuditMixin  # noqa: E402
 from sqlalchemy import Column, DateTime, Enum, ForeignKey, Integer, String  # noqa: E402
 from sqlalchemy.orm import declarative_base, declared_attr  # noqa: E402
 
@@ -39,10 +38,15 @@ from superset.utils.core import get_user_id  # noqa: E402
 Base = declarative_base()
 
 
-class AuditMixinNullable(AuditMixin):
-    """Altering the AuditMixin to use nullable fields
+class AuditMixinNullable:
+    """Nullable audit columns, without FAB's ``AuditMixin`` relationships.
 
-    Allows creating objects programmatically outside of CRUD
+    This module only needs the audit *columns* for ``__table__.create``.
+    Inheriting FAB's ``AuditMixin`` would also declare ``created_by`` /
+    ``changed_by`` relationships on these throwaway mapped classes, and once
+    alembic imports this script the resulting mapper cannot be configured —
+    breaking ``sqlalchemy.orm.configure_mappers()`` process-wide for any
+    later caller.
     """
 
     created_on = Column(DateTime, default=datetime.now, nullable=True)

superset/tags/core.py

@@ -16,6 +16,42 @@
 # under the License.
 # pylint: disable=import-outside-toplevel
 
+from superset.utils.sqlalchemy_events import (
+    DeleteListenerDeclaration,
+    DeleteListenerEffect,
+    register_delete_listener,
+    remove_delete_listener,
+)
+
+
+def _tag_delete_listener_declarations() -> tuple[DeleteListenerDeclaration, ...]:
+    """Build tag cleanup declarations without introducing model import cycles."""
+    from superset.connectors.sqla.models import SqlaTable
+    from superset.models.dashboard import Dashboard
+    from superset.models.slice import Slice
+    from superset.tags.models import ChartUpdater, DashboardUpdater, DatasetUpdater
+
+    return (
+        DeleteListenerDeclaration(
+            SqlaTable,
+            "tagged_object_cleanup",
+            DeleteListenerEffect.PERSISTENT_RECORD,
+            DatasetUpdater.after_delete,
+        ),
+        DeleteListenerDeclaration(
+            Slice,
+            "tagged_object_cleanup",
+            DeleteListenerEffect.PERSISTENT_RECORD,
+            ChartUpdater.after_delete,
+        ),
+        DeleteListenerDeclaration(
+            Dashboard,
+            "tagged_object_cleanup",
+            DeleteListenerEffect.PERSISTENT_RECORD,
+            DashboardUpdater.after_delete,
+        ),
+    )
+
 
 def register_sqla_event_listeners() -> None:
     import sqlalchemy as sqla
@@ -33,17 +69,21 @@ def register_sqla_event_listeners() -> None:
         QueryUpdater,
     )
 
+    declarations: tuple[DeleteListenerDeclaration, ...] = (
+        _tag_delete_listener_declarations()
+    )
+
     sqla.event.listen(SqlaTable, "after_insert", DatasetUpdater.after_insert)
     sqla.event.listen(SqlaTable, "after_update", DatasetUpdater.after_update)
-    sqla.event.listen(SqlaTable, "after_delete", DatasetUpdater.after_delete)
+    register_delete_listener(declarations[0])
 
     sqla.event.listen(Slice, "after_insert", ChartUpdater.after_insert)
     sqla.event.listen(Slice, "after_update", ChartUpdater.after_update)
-    sqla.event.listen(Slice, "after_delete", ChartUpdater.after_delete)
+    register_delete_listener(declarations[1])
 
     sqla.event.listen(Dashboard, "after_insert", DashboardUpdater.after_insert)
     sqla.event.listen(Dashboard, "after_update", DashboardUpdater.after_update)
-    sqla.event.listen(Dashboard, "after_delete", DashboardUpdater.after_delete)
+    register_delete_listener(declarations[2])
 
     sqla.event.listen(FavStar, "after_insert", FavStarUpdater.after_insert)
     sqla.event.listen(FavStar, "after_delete", FavStarUpdater.after_delete)
@@ -69,17 +109,21 @@ def clear_sqla_event_listeners() -> None:
         QueryUpdater,
     )
 
+    declarations: tuple[DeleteListenerDeclaration, ...] = (
+        _tag_delete_listener_declarations()
+    )
+
     sqla.event.remove(SqlaTable, "after_insert", DatasetUpdater.after_insert)
     sqla.event.remove(SqlaTable, "after_update", DatasetUpdater.after_update)
-    sqla.event.remove(SqlaTable, "after_delete", DatasetUpdater.after_delete)
+    remove_delete_listener(declarations[0])
 
     sqla.event.remove(Slice, "after_insert", ChartUpdater.after_insert)
     sqla.event.remove(Slice, "after_update", ChartUpdater.after_update)
-    sqla.event.remove(Slice, "after_delete", ChartUpdater.after_delete)
+    remove_delete_listener(declarations[1])
 
     sqla.event.remove(Dashboard, "after_insert", DashboardUpdater.after_insert)
     sqla.event.remove(Dashboard, "after_update", DashboardUpdater.after_update)
-    sqla.event.remove(Dashboard, "after_delete", DashboardUpdater.after_delete)
+    remove_delete_listener(declarations[2])
 
     sqla.event.remove(FavStar, "after_insert", FavStarUpdater.after_insert)
     sqla.event.remove(FavStar, "after_delete", FavStarUpdater.after_delete)

superset/utils/sqlalchemy_events.py

@@ -0,0 +1,103 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""Typed declarations for SQLAlchemy deletion-listener effects."""
+
+from __future__ import annotations
+
+from collections.abc import Callable
+from dataclasses import dataclass
+from enum import Enum
+from threading import RLock
+from typing import Any
+
+import sqlalchemy as sa
+
+
+class DeleteListenerEffect(str, Enum):
+    """Classify the durable effect of an ORM deletion listener."""
+
+    PERSISTENT_RECORD = "persistent_record"
+    PERMISSION_ARTIFACT = "permission_artifact"
+    OBSERVATIONAL = "observational"
+
+
+@dataclass(frozen=True)
+class DeleteListenerDeclaration:
+    """Describe one supported-root ``after_delete`` listener."""
+
+    target: type[Any]
+    responsibility: str
+    effect: DeleteListenerEffect
+    listener: Callable[..., None]
+
+    @property
+    def key(self) -> tuple[type[Any], str]:
+        """Return the stable catalog key for this declaration."""
+        return self.target, self.responsibility
+
+
+_DELETE_LISTENERS: dict[tuple[type[Any], str], DeleteListenerDeclaration] = {}
+_DELETE_LISTENER_LOCK: RLock = RLock()
+
+
+def register_delete_listener(declaration: DeleteListenerDeclaration) -> None:
+    """Register a declared listener idempotently."""
+    with _DELETE_LISTENER_LOCK:
+        existing: DeleteListenerDeclaration | None = _DELETE_LISTENERS.get(
+            declaration.key
+        )
+        if existing is not None and existing != declaration:
+            raise ValueError(
+                "Conflicting delete-listener declaration: "
+                f"{declaration.target.__name__}.{declaration.responsibility}"
+            )
+        _DELETE_LISTENERS[declaration.key] = declaration
+        if not sa.event.contains(
+            declaration.target, "after_delete", declaration.listener
+        ):
+            sa.event.listen(declaration.target, "after_delete", declaration.listener)
+
+
+def remove_delete_listener(declaration: DeleteListenerDeclaration) -> None:
+    """Remove a declared listener while preserving catalog identity."""
+    with _DELETE_LISTENER_LOCK:
+        existing: DeleteListenerDeclaration | None = _DELETE_LISTENERS.get(
+            declaration.key
+        )
+        if existing is not None and existing != declaration:
+            raise ValueError(
+                "Conflicting delete-listener declaration: "
+                f"{declaration.target.__name__}.{declaration.responsibility}"
+            )
+        if sa.event.contains(declaration.target, "after_delete", declaration.listener):
+            sa.event.remove(declaration.target, "after_delete", declaration.listener)
+        _DELETE_LISTENERS.pop(declaration.key, None)
+
+
+def declared_delete_listeners() -> tuple[DeleteListenerDeclaration, ...]:
+    """Return declared listeners in deterministic order."""
+    with _DELETE_LISTENER_LOCK:
+        return tuple(
+            sorted(
+                _DELETE_LISTENERS.values(),
+                key=lambda declaration: (
+                    declaration.target.__module__,
+                    declaration.target.__qualname__,
+                    declaration.responsibility,
+                ),
+            )
+        )

tests/integration_tests/deletion_retention/purge_performance_tests.py

@@ -0,0 +1,259 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""Deterministic query-count guard for representative purge graphs."""
+
+from __future__ import annotations
+
+import os
+from datetime import datetime, timedelta
+from statistics import median
+from time import perf_counter
+from typing import Any
+
+import pytest
+import sqlalchemy as sa
+from sqlalchemy.engine import Connection
+
+from superset import db
+from superset.commands.deletion_retention.purge_cascade import (
+    cascade_hard_delete,
+    CascadeResult,
+    suppress_purge_association_versions,
+)
+from superset.connectors.sqla.models import (
+    RowLevelSecurityFilter,
+    SqlaTable,
+    SqlMetric,
+    TableColumn,
+)
+from superset.models.core import Database
+from superset.models.dashboard import Dashboard
+from superset.models.slice import Slice
+
+from ._base import DeletionRetentionTestBase
+
+CHART_PURGE_BASELINE_STATEMENTS: int = 35
+DASHBOARD_PURGE_BASELINE_STATEMENTS: int = 36
+DATASET_PURGE_BASELINE_STATEMENTS: int = 43
+MAX_STATEMENT_REGRESSION: float = 0.10
+WARMUP_RUNS: int = 5
+MEASURED_RUNS: int = 20
+
+
+def statement_budget(baseline: int) -> int:
+    """Return the inclusive integer budget for a measured SQL baseline."""
+    return int(baseline * (1 + MAX_STATEMENT_REGRESSION) + 0.9999)
+
+
+class TestPurgeQueryCount(DeletionRetentionTestBase):
+    """Guard against accidental graph loading or query-count explosions."""
+
+    def test_representative_chart_purge_query_count(self) -> None:
+        """A chart with five dashboard memberships stays within its SQL budget."""
+        chart: Slice = self.make_chart("perf_chart")
+        for index in range(5):
+            self.make_dashboard(f"perf_dashboard_{index}", slices=[chart])
+        statement_count: int = self._purge_statement_count(chart)
+        if os.environ.get("SUPERSET_PURGE_BENCHMARK") == "1":
+            print(f"chart purge statements: {statement_count}")
+        assert statement_count <= statement_budget(CHART_PURGE_BASELINE_STATEMENTS)
+
+    def test_representative_dashboard_purge_query_count(self) -> None:
+        """A dashboard with five chart memberships stays within its SQL budget."""
+        charts: list[Slice] = [
+            self.make_chart(f"perf_dashboard_chart_{index}") for index in range(5)
+        ]
+        dashboard: Dashboard = self.make_dashboard("perf_dashboard", slices=charts)
+
+        statement_count: int = self._purge_statement_count(dashboard)
+
+        if os.environ.get("SUPERSET_PURGE_BENCHMARK") == "1":
+            print(f"dashboard purge statements: {statement_count}")
+        assert statement_count <= statement_budget(DASHBOARD_PURGE_BASELINE_STATEMENTS)
+
+    def test_representative_dataset_purge_query_count(self) -> None:
+        """A dataset's fixed owned and association graph stays bounded."""
+        dataset: SqlaTable = self.make_dataset("perf_dataset")
+        for index in range(10):
+            db.session.add(
+                TableColumn(column_name=f"retention_it_column_{index}", table=dataset)
+            )
+        for index in range(5):
+            db.session.add(
+                SqlMetric(
+                    metric_name=f"retention_it_metric_{index}",
+                    expression="count(*)",
+                    table=dataset,
+                )
+            )
+            db.session.add(
+                RowLevelSecurityFilter(
+                    name=f"retention_it_rls_{index}",
+                    clause="1=1",
+                    filter_type="Regular",
+                    tables=[dataset],
+                )
+            )
+        db.session.commit()
+
+        statement_count: int = self._purge_statement_count(dataset)
+
+        if os.environ.get("SUPERSET_PURGE_BENCHMARK") == "1":
+            print(f"dataset purge statements: {statement_count}")
+        assert statement_count <= statement_budget(DATASET_PURGE_BASELINE_STATEMENTS)
+
+    @pytest.mark.skipif(
+        os.environ.get("SUPERSET_PURGE_BENCHMARK") != "1",
+        reason="manual fixed-cardinality timing protocol",
+    )
+    @pytest.mark.parametrize(
+        ("entity_type", "baseline_environment_variable"),
+        [
+            ("chart", "SUPERSET_PURGE_BASELINE_CHART_SECONDS"),
+            ("dashboard", "SUPERSET_PURGE_BASELINE_DASHBOARD_SECONDS"),
+            ("dataset", "SUPERSET_PURGE_BASELINE_DATASET_SECONDS"),
+        ],
+    )
+    def test_representative_purge_median(
+        self,
+        entity_type: str,
+        baseline_environment_variable: str,
+    ) -> None:
+        """Report fixed-cardinality medians and enforce supplied baselines."""
+        durations: list[float] = []
+        for iteration in range(WARMUP_RUNS + MEASURED_RUNS):
+            if iteration:
+                self._reset_benchmark_fixture()
+            entity: Slice | Dashboard | SqlaTable = self._make_benchmark_entity(
+                entity_type, iteration
+            )
+            self.soft_delete(entity, days_ago=90)
+
+            started_at: float = perf_counter()
+            with suppress_purge_association_versions(db.session):
+                result: CascadeResult = cascade_hard_delete(
+                    db.session,
+                    entity,
+                    enforce_window=True,
+                    cutoff=datetime.now() - timedelta(days=30),
+                )
+            db.session.commit()
+            elapsed: float = perf_counter() - started_at
+
+            assert result.purged
+            if iteration >= WARMUP_RUNS:
+                durations.append(elapsed)
+
+        measured_median: float = median(durations)
+        baseline_value: str | None = os.environ.get(baseline_environment_variable)
+        print(f"{entity_type} purge median: {measured_median:.6f}s")
+        if baseline_value is not None:
+            baseline_median: float = float(baseline_value)
+            regression: float = (measured_median - baseline_median) / baseline_median
+            print(f"{entity_type} purge elapsed-time delta: {regression:+.2%}")
+            assert regression <= MAX_STATEMENT_REGRESSION
+        assert len(durations) == MEASURED_RUNS
+
+    def _make_benchmark_entity(
+        self, entity_type: str, iteration: int
+    ) -> Slice | Dashboard | SqlaTable:
+        """Create one fixed-cardinality root for the manual timing protocol."""
+        if entity_type == "chart":
+            chart: Slice = self.make_chart(f"benchmark_chart_{iteration}")
+            for dashboard_index in range(5):
+                self.make_dashboard(
+                    f"benchmark_dashboard_{iteration}_{dashboard_index}",
+                    slices=[chart],
+                )
+            return chart
+        if entity_type == "dashboard":
+            charts: list[Slice] = [
+                self.make_chart(f"benchmark_chart_{iteration}_{index}")
+                for index in range(5)
+            ]
+            return self.make_dashboard(
+                f"benchmark_dashboard_{iteration}", slices=charts
+            )
+        if entity_type == "dataset":
+            dataset: SqlaTable = self.make_dataset(f"benchmark_dataset_{iteration}")
+            for index in range(10):
+                db.session.add(
+                    TableColumn(
+                        column_name=f"benchmark_column_{iteration}_{index}",
+                        table=dataset,
+                    )
+                )
+            for index in range(5):
+                db.session.add(
+                    SqlMetric(
+                        metric_name=f"benchmark_metric_{iteration}_{index}",
+                        expression="count(*)",
+                        table=dataset,
+                    )
+                )
+                db.session.add(
+                    RowLevelSecurityFilter(
+                        name=f"benchmark_rls_{iteration}_{index}",
+                        clause="1=1",
+                        filter_type="Regular",
+                        tables=[dataset],
+                    )
+                )
+            db.session.commit()
+            return dataset
+        raise ValueError(f"Unsupported benchmark entity type: {entity_type}")
+
+    def _reset_benchmark_fixture(self) -> None:
+        """Rebuild the fixed fixture between manual timing samples."""
+        self._cleanup()
+        self.database: Database = Database(
+            database_name="retention_it_db", sqlalchemy_uri="sqlite://"
+        )
+        db.session.add(self.database)
+        db.session.commit()
+        self.dataset: SqlaTable = self.make_dataset("ds")
+
+    def _purge_statement_count(self, entity: Any) -> int:
+        """Purge one root and return SQL statements within the measured region."""
+        self.soft_delete(entity, days_ago=90)
+        statements: list[str] = []
+
+        def count_statement(
+            _connection: Connection,
+            _cursor: object,
+            statement: str,
+            _parameters: object,
+            _context: object,
+            _executemany: bool,
+        ) -> None:
+            statements.append(statement)
+
+        sa.event.listen(db.engine, "before_cursor_execute", count_statement)
+        try:
+            with suppress_purge_association_versions(db.session):
+                result: CascadeResult = cascade_hard_delete(
+                    db.session,
+                    entity,
+                    enforce_window=True,
+                    cutoff=datetime.now() - timedelta(days=30),
+                )
+            db.session.commit()
+        finally:
+            sa.event.remove(db.engine, "before_cursor_execute", count_statement)
+
+        assert result.purged
+        return len(statements)

tests/integration_tests/deletion_retention/purge_tests.py

@@ -24,25 +24,35 @@ guarantee under FK enforcement OFF, and the version-tables-absent no-op.
 
 from __future__ import annotations
 
+from dataclasses import replace
 from datetime import datetime, timedelta
 from typing import Any
 from unittest.mock import MagicMock, patch
 
+import pytest
 import sqlalchemy as sa
 from sqlalchemy.exc import IntegrityError
+from sqlalchemy.orm import Session
 from sqlalchemy.orm.attributes import set_committed_value
 from sqlalchemy.sql.dml import Delete
 
 from superset import db, security_manager
 from superset.commands.deletion_retention import audit
-from superset.commands.deletion_retention.purge_cascade import cascade_hard_delete
+from superset.commands.deletion_retention.purge_cascade import (
+    cascade_hard_delete,
+    suppress_purge_association_versions,
+)
+from superset.commands.deletion_retention.purge_policy import (
+    get_purge_policy,
+    PurgeEntityPolicy,
+)
 from superset.connectors.sqla.models import (
     RLSFilterTables,
     RowLevelSecurityFilter,
     SqlaTable,
 )
 from superset.constants import SKIP_VISIBILITY_FILTER_CLASSES
-from superset.models.dashboard import Dashboard
+from superset.models.dashboard import Dashboard, dashboard_slices
 from superset.models.slice import Slice
 from superset.models.user_attributes import UserAttribute
 from superset.reports.models import ReportSchedule
@@ -779,9 +789,19 @@ class TestExplicitBlockerGuards(DeletionRetentionTestBase):
             "(sqlite3.IntegrityError) FOREIGN KEY constraint failed "
             "[SQL: DELETE FROM slices WHERE slices.id = ?] [parameters: (1,)]"
         )
+
+        def fail_association_cleanup(
+            _session: Session, _policy: PurgeEntityPolicy, _entity_id: int
+        ) -> None:
+            raise IntegrityError(driver_text, None, Exception("fk"))
+
+        policy: PurgeEntityPolicy = replace(
+            get_purge_policy(Slice),
+            delete_associations=fail_association_cleanup,
+        )
         with patch(
-            "superset.commands.deletion_retention.purge_cascade._delete_m2m_joins",
-            side_effect=IntegrityError(driver_text, None, Exception("fk")),
+            "superset.commands.deletion_retention.purge_cascade.get_purge_policy",
+            return_value=policy,
         ):
             result = cascade_hard_delete(
                 db.session,
@@ -796,6 +816,88 @@ class TestExplicitBlockerGuards(DeletionRetentionTestBase):
         assert "SQL:" not in result.blocked_reason
         assert self.exists(Slice, chart_id)
 
+    def test_policy_action_failure_rolls_back_prior_phases(self) -> None:
+        """A later policy-action failure restores earlier association cleanup."""
+        chart: Slice = self.make_chart("action_rollback")
+        chart_id: int = chart.id
+        dashboard: Dashboard = self.make_dashboard("action_rollback", slices=[chart])
+        dashboard_id: int = dashboard.id
+        self.soft_delete(chart, days_ago=90)
+
+        def fail_owned_cleanup(
+            _session: Session, _policy: PurgeEntityPolicy, _entity_id: int
+        ) -> None:
+            raise RuntimeError("injected owned cleanup failure")
+
+        policy: PurgeEntityPolicy = replace(
+            get_purge_policy(Slice),
+            delete_owned_children=fail_owned_cleanup,
+        )
+        with (
+            patch(
+                "superset.commands.deletion_retention.purge_cascade.get_purge_policy",
+                return_value=policy,
+            ),
+            pytest.raises(RuntimeError, match="injected owned cleanup failure"),
+        ):
+            with suppress_purge_association_versions(db.session):
+                cascade_hard_delete(
+                    db.session,
+                    chart,
+                    enforce_window=True,
+                    cutoff=datetime.now() - timedelta(days=30),
+                )
+
+        membership_count: int = int(
+            db.session.execute(
+                sa.select(sa.func.count())
+                .select_from(dashboard_slices)
+                .where(
+                    dashboard_slices.c.dashboard_id == dashboard_id,
+                    dashboard_slices.c.slice_id == chart_id,
+                )
+            ).scalar_one()
+        )
+        assert membership_count == 1
+        assert self.exists(Slice, chart_id)
+
+    def test_history_cleanup_failure_rolls_back_prior_phases(self) -> None:
+        """A history-phase failure restores association cleanup and the root."""
+        chart: Slice = self.make_chart("history_rollback")
+        chart_id: int = chart.id
+        dashboard: Dashboard = self.make_dashboard("history_rollback", slices=[chart])
+        dashboard_id: int = dashboard.id
+        self.soft_delete(chart, days_ago=90)
+
+        with (
+            patch(
+                "superset.commands.deletion_retention.purge_cascade."
+                "_delete_version_history",
+                side_effect=RuntimeError("injected history cleanup failure"),
+            ),
+            pytest.raises(RuntimeError, match="injected history cleanup failure"),
+        ):
+            with suppress_purge_association_versions(db.session):
+                cascade_hard_delete(
+                    db.session,
+                    chart,
+                    enforce_window=True,
+                    cutoff=datetime.now() - timedelta(days=30),
+                )
+
+        membership_count: int = int(
+            db.session.execute(
+                sa.select(sa.func.count())
+                .select_from(dashboard_slices)
+                .where(
+                    dashboard_slices.c.dashboard_id == dashboard_id,
+                    dashboard_slices.c.slice_id == chart_id,
+                )
+            ).scalar_one()
+        )
+        assert membership_count == 1
+        assert self.exists(Slice, chart_id)
+
 
 class TestFailClosedAudienceDefault(DeletionRetentionTestBase):
     """A soft-delete model without editors must not enumerate to everyone."""

tests/unit_tests/commands/deletion_retention/__init__.py

@@ -0,0 +1,16 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.

tests/unit_tests/commands/deletion_retention/test_purge_policy.py

@@ -0,0 +1,570 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""Contract tests for declarative hard-purge policies."""
+
+from __future__ import annotations
+
+from dataclasses import replace
+from typing import Any
+from unittest.mock import MagicMock
+
+import pytest
+import sqlalchemy as sa
+from sqlalchemy.engine import Dialect
+from sqlalchemy.orm import configure_mappers, registry
+from sqlalchemy.sql import Select
+
+from superset.commands.deletion_retention.purge_policy import (
+    _dependency_owner_depth,
+    _dependency_predicates,
+    _fk_key,
+    compare_policy,
+    delete_associations,
+    delete_owned_children,
+    DependencyClassification,
+    DependencyKey,
+    DependencyPolicy,
+    discover_dependencies,
+    get_purge_policy,
+    listener_responsibilities,
+    PolicyCoverage,
+    purge_policy_registry,
+    PurgeEntityPolicy,
+    validate_deletion_allowed,
+    validate_unique_root_policies,
+)
+from superset.connectors.sqla.models import SqlaTable
+from superset.models.dashboard import Dashboard
+from superset.models.slice import Slice
+from superset.tasks.deletion_retention import _soft_delete_models
+from superset.utils.sqlalchemy_events import (
+    declared_delete_listeners,
+    DeleteListenerDeclaration,
+    DeleteListenerEffect,
+    register_delete_listener,
+    remove_delete_listener,
+)
+
+
+def test_compare_policy_reports_missing_duplicate_and_stale_dependencies() -> None:
+    """Coverage diagnostics identify every kind of registry drift."""
+    discovered: set[DependencyKey] = {
+        DependencyKey("foreign_key", "root", "owned"),
+        DependencyKey("relationship", "root", "preserved", relationship="item"),
+    }
+    duplicate: DependencyPolicy = DependencyPolicy(
+        DependencyKey("foreign_key", "root", "owned"),
+        DependencyClassification.OWNED,
+    )
+    stale: DependencyPolicy = DependencyPolicy(
+        DependencyKey("foreign_key", "root", "stale"),
+        DependencyClassification.PRESERVE,
+    )
+
+    coverage: PolicyCoverage = compare_policy(discovered, (duplicate, duplicate, stale))
+
+    assert len(coverage.missing) == 1
+    assert coverage.duplicates == (duplicate.key,)
+    assert coverage.stale == (stale.key,)
+
+
+def test_adding_a_complete_policy_restores_coverage() -> None:
+    """A new dependency passes after its policy is supplied."""
+    key: DependencyKey = DependencyKey("foreign_key", "root", "child")
+
+    assert not compare_policy({key}, ()).complete
+    assert compare_policy(
+        {key},
+        (DependencyPolicy(key, DependencyClassification.OWNED),),
+    ).complete
+
+
+def test_omitted_inbound_fk_and_listener_are_reported() -> None:
+    """Inbound metadata edges and persistent listeners are obligations."""
+    inbound: DependencyKey = DependencyKey(
+        "foreign_key",
+        "root",
+        "referrer",
+        ("id",),
+        ("root_id",),
+        "inbound",
+    )
+    coverage: PolicyCoverage = compare_policy(
+        {inbound},
+        (),
+        discovered_listeners={"persistent_cleanup"},
+    )
+
+    assert coverage.missing == (inbound,)
+    assert coverage.missing_listeners == ("persistent_cleanup",)
+
+
+@pytest.mark.parametrize(
+    "classification",
+    [DependencyClassification.PRESERVE, DependencyClassification.BLOCK],
+)
+def test_terminal_dependency_classifications_are_complete(
+    classification: DependencyClassification,
+) -> None:
+    """Preserve and block are explicit terminal treatments, not omissions."""
+    key: DependencyKey = DependencyKey("foreign_key", "root", "terminal")
+
+    assert compare_policy({key}, (DependencyPolicy(key, classification),)).complete
+
+
+def test_composite_foreign_key_is_one_atomic_dependency() -> None:
+    """Composite constraints retain ordered local and remote column tuples."""
+    metadata: sa.MetaData = sa.MetaData()
+    root: sa.Table = sa.Table(
+        "root",
+        metadata,
+        sa.Column("tenant_id", sa.Integer, primary_key=True),
+        sa.Column("id", sa.Integer, primary_key=True),
+    )
+    child: sa.Table = sa.Table(
+        "child",
+        metadata,
+        sa.Column("tenant_id", sa.Integer),
+        sa.Column("root_id", sa.Integer),
+        sa.ForeignKeyConstraint(
+            ("tenant_id", "root_id"), ("root.tenant_id", "root.id")
+        ),
+    )
+    constraint: sa.ForeignKeyConstraint = next(iter(child.foreign_key_constraints))
+
+    key: DependencyKey = _fk_key(root, constraint, "inbound")
+
+    assert key.local_columns == ("tenant_id", "id")
+    assert key.remote_columns == ("tenant_id", "root_id")
+
+
+def test_duplicate_root_policies_are_rejected() -> None:
+    """Registry construction cannot silently replace a root declaration."""
+    policy: PurgeEntityPolicy = get_purge_policy(Slice)
+
+    with pytest.raises(ValueError, match="Duplicate purge policy for Slice"):
+        validate_unique_root_policies((policy, policy))
+
+
+def test_every_soft_delete_root_has_a_purge_policy() -> None:
+    """Every built-in retention root has exactly one purge policy."""
+    production_roots: set[type[Any]] = {
+        model
+        for model in _soft_delete_models()
+        if model.__module__.startswith("superset.")
+    }
+    assert set(purge_policy_registry()) == production_roots
+
+
+def test_listener_coverage_reports_stale_and_optional_declarations() -> None:
+    """Required stale listeners fail while disabled optional listeners pass."""
+    stale: PolicyCoverage = compare_policy(
+        (), (), declared_listeners={"removed_cleanup"}
+    )
+    optional: PolicyCoverage = compare_policy(
+        (),
+        (),
+        declared_listeners={"optional_cleanup"},
+        optional_declared_listeners={"optional_cleanup"},
+    )
+
+    assert stale.stale_listeners == ("removed_cleanup",)
+    assert optional.complete
+
+
+@pytest.mark.parametrize("model", [Slice, Dashboard, SqlaTable])
+def test_real_mapper_graph_has_complete_policy(model: type[Any]) -> None:
+    """Every supported root mapper dependency has one policy."""
+    configure_mappers()
+    metadata_tables: set[str] = set(sa.inspect(Slice).local_table.metadata.tables)
+    assert {
+        "embedded_dashboards",
+        "report_schedule",
+        "rls_filter_tables",
+        "tagged_object",
+        "user_attribute",
+    } <= metadata_tables
+    policy: PurgeEntityPolicy = get_purge_policy(model)
+    recursive_tables: frozenset[str] = frozenset(
+        dependency.key.related_table
+        for dependency in policy.dependencies
+        if dependency.classification
+        in {
+            DependencyClassification.OWNED,
+            DependencyClassification.ASSOCIATION,
+        }
+    )
+    coverage: PolicyCoverage = compare_policy(
+        discover_dependencies(sa.inspect(model), recursive_tables=recursive_tables),
+        policy.dependencies,
+        discovered_listeners=listener_responsibilities(model),
+        declared_listeners=policy.listener_responsibilities,
+        optional_declared_listeners=policy.optional_listener_responsibilities,
+    )
+
+    assert coverage.complete, coverage
+
+
+@pytest.mark.parametrize("model", [Slice, Dashboard, SqlaTable])
+def test_non_preserve_dependencies_carry_phases(model: type[Any]) -> None:
+    """Every executable classification declares its execution phase."""
+    policy: PurgeEntityPolicy = get_purge_policy(model)
+
+    assert all(
+        dependency.phase is not None
+        for dependency in policy.dependencies
+        if dependency.classification is not DependencyClassification.PRESERVE
+    )
+
+
+def test_recursive_discovery_stops_at_owned_cycles() -> None:
+    """Owned-child backrefs terminate instead of walking the graph forever."""
+    dependencies: frozenset[DependencyKey] = discover_dependencies(
+        sa.inspect(Dashboard),
+        recursive_tables=frozenset({"dashboards", "embedded_dashboards"}),
+    )
+
+    assert dependencies
+    assert len(dependencies) == len(set(dependencies))
+    assert any(
+        dependency.owner_table == "embedded_dashboards"
+        and dependency.related_table == "dashboards"
+        for dependency in dependencies
+    )
+
+
+def test_dependency_owner_depth_handles_deep_paths_without_recursion() -> None:
+    """Ownership ordering and predicates are independent of recursion limits."""
+    path_length: int = 1_100
+    metadata: sa.MetaData = sa.MetaData()
+    root_table: sa.Table = sa.Table(
+        "synthetic_root",
+        metadata,
+        sa.Column("id", sa.Integer, primary_key=True),
+    )
+    mapper_registry: registry = registry()
+
+    class SyntheticRoot:
+        """Temporary mapped root for deep ownership-path construction."""
+
+    mapper_registry.map_imperatively(SyntheticRoot, root_table)
+    for index in range(path_length):
+        sa.Table(
+            f"owned_{index}",
+            metadata,
+            sa.Column("id", sa.Integer, primary_key=True),
+            sa.Column("owner_id", sa.Integer),
+        )
+    sa.Table(
+        "leaf",
+        metadata,
+        sa.Column("id", sa.Integer, primary_key=True),
+        sa.Column("owner_id", sa.Integer),
+    )
+    dependencies: tuple[DependencyPolicy, ...] = tuple(
+        DependencyPolicy(
+            DependencyKey(
+                "foreign_key",
+                "synthetic_root" if index == 0 else f"owned_{index - 1}",
+                f"owned_{index}",
+                ("id",),
+                ("owner_id",),
+                "inbound",
+            ),
+            DependencyClassification.OWNED,
+        )
+        for index in range(path_length)
+    )
+    policy: PurgeEntityPolicy = replace(
+        get_purge_policy(Slice), model=SyntheticRoot, dependencies=dependencies
+    )
+    leaf: DependencyKey = DependencyKey(
+        "foreign_key",
+        f"owned_{path_length - 1}",
+        "leaf",
+        ("id",),
+        ("owner_id",),
+        "inbound",
+    )
+
+    assert _dependency_owner_depth(policy, leaf) == path_length
+    predicates: tuple[Any, ...] = _dependency_predicates(
+        policy, leaf, 1, metadata.tables["leaf"]
+    )
+    assert len(predicates) == 1
+    mapper_registry.dispose()
+
+
+def test_dependency_owner_depth_rejects_cycles() -> None:
+    """Malformed ownership declarations fail clearly instead of looping."""
+    first: DependencyPolicy = DependencyPolicy(
+        DependencyKey("foreign_key", "owned_b", "owned_a", direction="inbound"),
+        DependencyClassification.OWNED,
+    )
+    second: DependencyPolicy = DependencyPolicy(
+        DependencyKey("foreign_key", "owned_a", "owned_b", direction="inbound"),
+        DependencyClassification.OWNED,
+    )
+    policy: PurgeEntityPolicy = replace(
+        get_purge_policy(Slice), dependencies=(first, second)
+    )
+    leaf: DependencyKey = DependencyKey("foreign_key", "owned_a", "leaf")
+
+    with pytest.raises(RuntimeError, match="Cyclic ownership path"):
+        _dependency_owner_depth(policy, leaf)
+
+
+@pytest.mark.parametrize(
+    ("model", "expected_targets"),
+    [
+        (
+            Slice,
+            (
+                ("slices_version", "id"),
+                ("dashboard_slices_version", "slice_id"),
+            ),
+        ),
+        (
+            Dashboard,
+            (
+                ("dashboards_version", "id"),
+                ("dashboard_slices_version", "dashboard_id"),
+            ),
+        ),
+        (
+            SqlaTable,
+            (
+                ("tables_version", "id"),
+                ("sql_metrics_version", "table_id"),
+                ("table_columns_version", "table_id"),
+            ),
+        ),
+    ],
+)
+def test_version_targets_are_policy_owned(
+    model: type[Any], expected_targets: tuple[tuple[str, str], ...]
+) -> None:
+    """Root, association, and owned-child shadows come from the policy."""
+    policy: PurgeEntityPolicy = get_purge_policy(model)
+
+    assert policy.version_shadow_names == expected_targets
+
+
+def test_version_target_resolution_rejects_invalid_declarations(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """A shadow-table typo cannot silently leave version history behind."""
+    from superset.commands.deletion_retention import purge_cascade
+
+    metadata: sa.MetaData = sa.MetaData()
+    parent_shadow: sa.Table = sa.Table(
+        "slices_version", metadata, sa.Column("id", sa.Integer)
+    )
+    slice_policy: PurgeEntityPolicy = purge_policy_registry()[Slice]
+    dependencies: tuple[DependencyPolicy, ...] = tuple(
+        replace(
+            dependency,
+            key=replace(dependency.key, related_table="missing_version"),
+        )
+        if dependency.key.related_table == "slices_version"
+        else dependency
+        for dependency in slice_policy.dependencies
+    )
+    policy: PurgeEntityPolicy = replace(slice_policy, dependencies=dependencies)
+
+    def fake_policy(_model: type[Any]) -> PurgeEntityPolicy:
+        return policy
+
+    monkeypatch.setattr(purge_cascade, "get_purge_policy", fake_policy)
+
+    with pytest.raises(RuntimeError, match="missing_version.id"):
+        purge_cascade._entity_version_targets(
+            Slice, metadata, parent_shadow, entity_id=1
+        )
+
+
+def test_listener_registration_is_idempotent_and_symmetric() -> None:
+    """Declared listener registration can safely repeat and clear."""
+
+    def observe(*_args: Any) -> None:
+        return None
+
+    declaration: DeleteListenerDeclaration = DeleteListenerDeclaration(
+        Slice,
+        "test_observer",
+        DeleteListenerEffect.OBSERVATIONAL,
+        observe,
+    )
+    try:
+        register_delete_listener(declaration)
+        register_delete_listener(declaration)
+        assert declaration in declared_delete_listeners()
+    finally:
+        remove_delete_listener(declaration)
+    assert declaration not in declared_delete_listeners()
+
+
+def test_listener_removal_rejects_a_conflicting_declaration() -> None:
+    """Removal cannot erase a different declaration with the same key."""
+
+    def first(*_args: Any) -> None:
+        return None
+
+    def conflicting(*_args: Any) -> None:
+        return None
+
+    declaration: DeleteListenerDeclaration = DeleteListenerDeclaration(
+        Slice,
+        "test_conflict",
+        DeleteListenerEffect.OBSERVATIONAL,
+        first,
+    )
+    conflicting_declaration: DeleteListenerDeclaration = DeleteListenerDeclaration(
+        Slice,
+        "test_conflict",
+        DeleteListenerEffect.OBSERVATIONAL,
+        conflicting,
+    )
+    try:
+        register_delete_listener(declaration)
+        with pytest.raises(ValueError, match="Conflicting delete-listener"):
+            remove_delete_listener(conflicting_declaration)
+        assert declaration in declared_delete_listeners()
+    finally:
+        remove_delete_listener(declaration)
+
+
+@pytest.mark.parametrize("model", [Slice, Dashboard, SqlaTable])
+def test_supported_root_after_delete_listeners_are_declared_and_installed(
+    model: type[Any],
+) -> None:
+    """Every runtime root listener is represented in the listener catalog."""
+    declarations: tuple[DeleteListenerDeclaration, ...] = tuple(
+        declaration
+        for declaration in declared_delete_listeners()
+        if declaration.target is model
+        and declaration.effect is not DeleteListenerEffect.OBSERVATIONAL
+    )
+    policy: PurgeEntityPolicy = get_purge_policy(model)
+
+    assert {declaration.responsibility for declaration in declarations} <= set(
+        policy.listener_responsibilities
+    )
+    assert all(
+        sa.event.contains(model, "after_delete", declaration.listener)
+        for declaration in declarations
+    )
+    installed_listeners: set[Any] = {
+        cell.cell_contents
+        for wrapper in model.__mapper__.dispatch.after_delete
+        for cell in (wrapper.__closure__ or ())
+        if callable(cell.cell_contents)
+        and not getattr(cell.cell_contents, "__module__", "").startswith(
+            "sqlalchemy_continuum"
+        )
+    }
+    assert installed_listeners == {declaration.listener for declaration in declarations}
+
+
+def test_delete_associations_rejects_unknown_entity_type() -> None:
+    """An unsupported root cannot fall through to dataset cleanup."""
+    from superset.commands.deletion_retention.purge_policy import delete_associations
+
+    slice_policy: PurgeEntityPolicy = get_purge_policy(Slice)
+    listener_dependencies: tuple[DependencyPolicy, ...] = tuple(
+        dependency
+        for dependency in slice_policy.dependencies
+        if dependency.classification is DependencyClassification.LISTENER_EFFECT
+    )
+    policy: PurgeEntityPolicy = replace(
+        slice_policy,
+        entity_type="unsupported",
+        dependencies=listener_dependencies,
+    )
+
+    with pytest.raises(ValueError, match="Unsupported purge entity type"):
+        delete_associations(MagicMock(), policy, 1)
+
+
+def test_association_owned_children_are_deleted_before_their_owner() -> None:
+    """Association traversal deletes nested rows before their owning rows."""
+    association: DependencyPolicy = DependencyPolicy(
+        DependencyKey(
+            "foreign_key",
+            "slices",
+            "dashboard_slices",
+            ("id",),
+            ("slice_id",),
+            "inbound",
+        ),
+        DependencyClassification.ASSOCIATION,
+    )
+    association_child: DependencyPolicy = DependencyPolicy(
+        DependencyKey(
+            "foreign_key",
+            "dashboard_slices",
+            "dashboard_slices_version",
+            ("dashboard_id", "slice_id"),
+            ("dashboard_id", "slice_id"),
+            "inbound",
+        ),
+        DependencyClassification.ASSOCIATION,
+    )
+    policy: PurgeEntityPolicy = replace(
+        get_purge_policy(Slice),
+        dependencies=(association, association_child),
+    )
+    session: MagicMock = MagicMock()
+
+    delete_associations(session, policy, 7)
+
+    statements: list[Any] = [call.args[0] for call in session.execute.call_args_list]
+    assert [statement.table.name for statement in statements] == [
+        "dashboard_slices_version",
+        "dashboard_slices",
+    ]
+    assert "dashboard_slices.slice_id" in str(statements[0])
+    assert "slices.id" in str(statements[0])
+
+
+@pytest.mark.parametrize("dialect", ["sqlite", "postgresql", "mysql"])
+def test_core_delete_actions_compile_for_supported_dialects(dialect: str) -> None:
+    """Statements emitted by policy callbacks compile for supported dialects."""
+    from sqlalchemy.dialects import mysql, postgresql, sqlite
+
+    dialects: dict[str, Dialect] = {
+        "sqlite": sqlite.dialect(),
+        "postgresql": postgresql.dialect(),
+        "mysql": mysql.dialect(),
+    }
+    compiled: list[str] = []
+    for model in (Slice, Dashboard, SqlaTable):
+        policy: PurgeEntityPolicy = get_purge_policy(model)
+        session: MagicMock = MagicMock()
+        session.execute.return_value.first.return_value = None
+        validate_deletion_allowed(session, policy, 1)
+        delete_associations(session, policy, 1)
+        delete_owned_children(session, policy, 1)
+        calls: list[Any] = list(session.execute.call_args_list)
+        for call in calls:
+            statement: Any = call.args[0]
+            compiled.append(str(statement.compile(dialect=dialects[dialect])))
+    claim: Select = sa.select(Slice.id).where(Slice.id == 1).with_for_update()
+    compiled.append(str(claim.compile(dialect=dialects[dialect])))
+
+    assert compiled
+    assert all(statement.startswith(("SELECT", "DELETE")) for statement in compiled)

tests/unit_tests/migrations/test_add_implicit_tags_mapper_isolation.py

@@ -0,0 +1,35 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+import importlib
+
+import sqlalchemy.orm as orm
+
+
+def test_add_implicit_tags_script_leaves_mappers_configurable() -> None:
+    """Importing the migration script must not poison the mapper registry.
+
+    Alembic imports every version script when it walks revision history (the
+    unit-test app fixture triggers this via its pending-migration check), so a
+    script whose throwaway declarative models cannot be configured breaks
+    ``configure_mappers()`` process-wide for whichever test happens to trigger
+    mapper configuration next.
+    """
+    importlib.import_module(
+        "superset.migrations.versions.2018-07-26_11-10_c82ee8a39623_add_implicit_tags"
+    )
+
+    orm.configure_mappers()