diff --git a/superset/commands/deletion_retention/purge_cascade.py b/superset/commands/deletion_retention/purge_cascade.py index e58bb6c35a9..554736c6092 100644 --- a/superset/commands/deletion_retention/purge_cascade.py +++ b/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 diff --git a/superset/commands/deletion_retention/purge_policy.py b/superset/commands/deletion_retention/purge_policy.py new file mode 100644 index 00000000000..9af54c361ef --- /dev/null +++ b/superset/commands/deletion_retention/purge_policy.py @@ -0,0 +1,1213 @@ +# 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. +"""Declarative policies for explicit deletion-retention purges.""" + +from __future__ import annotations + +import logging +from collections.abc import Callable, Hashable, Iterable, Mapping +from dataclasses import dataclass +from enum import Enum +from functools import lru_cache +from types import MappingProxyType +from typing import Any, cast + +import sqlalchemy as sa +from sqlalchemy.orm import Mapper, Session + +from superset.utils.sqlalchemy_events import ( + declared_delete_listeners, + DeleteListenerEffect, +) + +logger: logging.Logger = logging.getLogger(__name__) + + +class PurgeBlockedError(Exception): + """Raised when ordinary deletion policy forbids purging an entity.""" + + +class DependencyClassification(str, Enum): + """Describe how purge treats a persistence dependency.""" + + OWNED = "owned" + ASSOCIATION = "association" + PRESERVE = "preserve" + BLOCK = "block" + LISTENER_EFFECT = "listener_effect" + VERSION_OWNED = "version_owned" + + +class ExecutionPhase(str, Enum): + """Describe the execution phase associated with a dependency.""" + + VALIDATE = "validate" + ASSOCIATIONS = "associations" + OWNED = "owned" + VERSION = "version" + POST_DELETE = "post_delete" + + +class ListenerAction(str, Enum): + """Identify one typed listener-equivalent purge action.""" + + DELETE_TAGGED_OBJECTS = "delete_tagged_objects" + DELETE_DATASET_PERMISSION = "delete_dataset_permission" + + +PolicyAction = Callable[[Session, "PurgeEntityPolicy", int], None] +CountSnapshot = Callable[[Session, "PurgeEntityPolicy", int], int] +UuidSnapshot = Callable[[Session, "PurgeEntityPolicy", int], list[str]] +PermissionSnapshot = Callable[[Session, "PurgeEntityPolicy", int], "str | None"] +PermissionCleanup = Callable[[Session, "PurgeEntityPolicy", "str | None", int], None] + + +@dataclass(frozen=True, order=True) +class DependencyKey: + """Identify one physical or non-FK persistence dependency.""" + + kind: str + owner_table: str + related_table: str + local_columns: tuple[str, ...] = () + remote_columns: tuple[str, ...] = () + direction: str = "" + relationship: str = "" + + def describe(self) -> str: + """Return a deterministic human-readable dependency identity.""" + columns: str = ",".join(self.local_columns) + remote: str = ",".join(self.remote_columns) + relationship: str = f":{self.relationship}" if self.relationship else "" + return ( + f"{self.kind}:{self.owner_table}->{self.related_table}:" + f"{columns}->{remote}:{self.direction}{relationship}" + ) + + +@dataclass(frozen=True) +class DependencyPolicy: + """Classify one dependency for a purge root.""" + + key: DependencyKey + classification: DependencyClassification + phase: ExecutionPhase | None = None + blocked_reason: str | None = None + optional_listener: bool = False + listener_action: ListenerAction | None = None + version_column: str | None = None + + +@dataclass(frozen=True) +class PurgeEntityPolicy: + """Declare the complete purge behavior for one root model.""" + + model: type[Any] + entity_type: str + dependencies: tuple[DependencyPolicy, ...] + validate: PolicyAction + count_dashboard_slices: CountSnapshot + collect_dangling_chart_uuids: UuidSnapshot + delete_associations: PolicyAction + delete_owned_children: PolicyAction + capture_permission_name: PermissionSnapshot + cleanup_permission: PermissionCleanup + + @property + def listener_responsibilities(self) -> frozenset[str]: + """Derive listener obligations from synthetic dependency declarations.""" + return frozenset( + dependency.key.relationship + for dependency in self.dependencies + if dependency.classification is DependencyClassification.LISTENER_EFFECT + ) + + @property + def optional_listener_responsibilities(self) -> frozenset[str]: + """Return listener effects that are conditional on runtime configuration.""" + return frozenset( + dependency.key.relationship + for dependency in self.dependencies + if dependency.classification is DependencyClassification.LISTENER_EFFECT + and dependency.optional_listener + ) + + @property + def version_shadow_names(self) -> tuple[tuple[str, str], ...]: + """Derive executable shadow targets from version-owned declarations.""" + return tuple( + (dependency.key.related_table, dependency.version_column or "") + for dependency in self.dependencies + if dependency.classification is DependencyClassification.VERSION_OWNED + ) + + +@dataclass(frozen=True) +class PolicyCoverage: + """Report missing, duplicate, and stale dependency declarations.""" + + missing: tuple[DependencyKey, ...] + duplicates: tuple[DependencyKey, ...] + stale: tuple[DependencyKey, ...] + missing_listeners: tuple[str, ...] = () + stale_listeners: tuple[str, ...] = () + + @property + def complete(self) -> bool: + """Return whether the policy covers the discovered graph exactly.""" + return not ( + self.missing + or self.duplicates + or self.stale + or self.missing_listeners + or self.stale_listeners + ) + + +def _fk_key( + owner_table: sa.Table, + constraint: sa.ForeignKeyConstraint, + direction: str, +) -> DependencyKey: + """Normalize one physical foreign-key constraint as an atomic edge.""" + elements: tuple[sa.ForeignKey, ...] = tuple(constraint.elements) + if direction == "outbound": + related_table: sa.Table = elements[0].column.table + local: tuple[str, ...] = tuple(element.parent.name for element in elements) + remote: tuple[str, ...] = tuple(element.column.name for element in elements) + else: + related_table = elements[0].parent.table + local = tuple(element.column.name for element in elements) + remote = tuple(element.parent.name for element in elements) + return DependencyKey( + kind="foreign_key", + owner_table=owner_table.name, + related_table=related_table.name, + local_columns=local, + remote_columns=remote, + direction=direction, + ) + + +def _relationship_has_physical_edge(relationship: Any) -> bool: + """Return whether a mapper relationship is represented by a physical FK.""" + for local, remote in relationship.local_remote_pairs: + if local.foreign_keys or remote.foreign_keys: + return True + if relationship.secondary is None: + return False + owner_table: sa.Table = relationship.parent.local_table + return any( + foreign_key.column.table is owner_table + for foreign_key in relationship.secondary.foreign_keys + ) + + +def discover_dependencies( + mapper: Mapper[Any], + *, + recursive_tables: frozenset[str] = frozenset(), + visited: frozenset[str] = frozenset(), +) -> frozenset[DependencyKey]: + """Discover physical FKs and relationships, recursing through owned tables.""" + table: sa.Table = mapper.local_table + if table.name in visited: + return frozenset() + next_visited: frozenset[str] = visited | {table.name} + discovered: set[DependencyKey] = { + _fk_key(table, constraint, "outbound") + for constraint in table.foreign_key_constraints + } + for candidate in table.metadata.tables.values(): + for constraint in candidate.foreign_key_constraints: + if constraint.elements[0].column.table is table: + discovered.add(_fk_key(table, constraint, "inbound")) + for relationship in mapper.relationships: + related_mapper: Mapper[Any] = relationship.mapper + if related_mapper.local_table.name in recursive_tables: + discovered.update( + discover_dependencies( + related_mapper, + recursive_tables=recursive_tables, + visited=next_visited, + ) + ) + if not _relationship_has_physical_edge(relationship): + discovered.add( + DependencyKey( + kind="relationship", + owner_table=table.name, + related_table=relationship.mapper.local_table.name, + direction=relationship.direction.name.lower(), + relationship=relationship.key, + ) + ) + discovered.update( + _discover_recursive_table_dependencies( + table, + recursive_tables=recursive_tables, + visited=next_visited, + ) + ) + return frozenset(discovered) + + +def _discover_table_dependencies( + table: sa.Table, + *, + recursive_tables: frozenset[str], + visited: frozenset[str], +) -> frozenset[DependencyKey]: + """Discover physical edges for unmapped association and owned tables.""" + if table.name in visited: + return frozenset() + next_visited: frozenset[str] = visited | {table.name} + discovered: set[DependencyKey] = { + _fk_key(table, constraint, "outbound") + for constraint in table.foreign_key_constraints + } + for candidate in table.metadata.tables.values(): + for constraint in candidate.foreign_key_constraints: + if constraint.elements[0].column.table is table: + discovered.add(_fk_key(table, constraint, "inbound")) + discovered.update( + _discover_recursive_table_dependencies( + table, + recursive_tables=recursive_tables, + visited=next_visited, + ) + ) + return frozenset(discovered) + + +def _discover_recursive_table_dependencies( + table: sa.Table, + *, + recursive_tables: frozenset[str], + visited: frozenset[str], +) -> frozenset[DependencyKey]: + """Discover dependencies for named recursive tables not already visited.""" + discovered: set[DependencyKey] = set() + for recursive_table_name in recursive_tables - visited: + recursive_table: sa.Table | None = table.metadata.tables.get( + recursive_table_name + ) + if recursive_table is not None: + discovered.update( + _discover_table_dependencies( + recursive_table, + recursive_tables=recursive_tables, + visited=visited, + ) + ) + return frozenset(discovered) + + +def validate_unique_root_policies( + policies: Iterable[PurgeEntityPolicy], +) -> Mapping[type[Any], PurgeEntityPolicy]: + """Create an immutable root index and reject duplicate model policies.""" + registry: dict[type[Any], PurgeEntityPolicy] = {} + for policy in policies: + if policy.model in registry: + raise ValueError(f"Duplicate purge policy for {policy.model.__name__}") + registry[policy.model] = policy + return MappingProxyType(registry) + + +def compare_policy( + discovered: Iterable[DependencyKey], + declared: Iterable[DependencyPolicy], + *, + discovered_listeners: Iterable[str] = (), + declared_listeners: Iterable[str] = (), + optional_declared_listeners: Iterable[str] = (), +) -> PolicyCoverage: + """Compare discovered dependencies with one policy's declarations.""" + discovered_set: set[DependencyKey] = set(discovered) + declared_keys: list[DependencyKey] = [item.key for item in declared] + declared_set: set[DependencyKey] = set(declared_keys) + duplicates: set[DependencyKey] = { + key for key in declared_set if declared_keys.count(key) > 1 + } + return PolicyCoverage( + missing=tuple(sorted(discovered_set - declared_set)), + duplicates=tuple(sorted(duplicates)), + stale=tuple( + sorted( + key for key in declared_set - discovered_set if key.kind != "synthetic" + ) + ), + missing_listeners=tuple( + sorted(set(discovered_listeners) - set(declared_listeners)) + ), + stale_listeners=tuple( + sorted( + set(declared_listeners) + - set(discovered_listeners) + - set(optional_declared_listeners) + ) + ), + ) + + +@lru_cache(maxsize=1) +def purge_policy_registry() -> Mapping[type[Any], PurgeEntityPolicy]: + """Build the supported purge registry after model initialization.""" + # avoid circular import: model listener registration imports neutral event helpers + from superset.connectors.sqla.models import SqlaTable + from superset.models.dashboard import Dashboard + from superset.models.embedded_dashboard import EmbeddedDashboard # noqa: F401 + from superset.models.slice import Slice + from superset.models.user_attributes import UserAttribute # noqa: F401 + from superset.reports.models import ReportSchedule # noqa: F401 + + def fk( + owner: str, + related: str, + local: str, + remote: str, + direction: str, + ) -> DependencyKey: + return DependencyKey( + "foreign_key", + owner, + related, + (local,), + (remote,), + direction, + ) + + def relationship( + owner: str, related: str, direction: str, name: str + ) -> DependencyKey: + return DependencyKey( + "relationship", + owner, + related, + direction=direction, + relationship=name, + ) + + def version(owner: str, related: str, name: str = "versions") -> DependencyKey: + return DependencyKey( + "relationship", + owner, + related, + direction="onetomany", + relationship=name, + ) + + def policies( + keys: tuple[DependencyKey, ...], + classifications: tuple[DependencyClassification, ...], + synthetic: tuple[DependencyPolicy, ...], + blocked_reasons: Mapping[str, str] = MappingProxyType({}), + version_columns: Mapping[str, str] = MappingProxyType({}), + ) -> tuple[DependencyPolicy, ...]: + phases: dict[DependencyClassification, ExecutionPhase | None] = { + DependencyClassification.OWNED: ExecutionPhase.OWNED, + DependencyClassification.ASSOCIATION: ExecutionPhase.ASSOCIATIONS, + DependencyClassification.PRESERVE: None, + DependencyClassification.BLOCK: ExecutionPhase.VALIDATE, + DependencyClassification.VERSION_OWNED: ExecutionPhase.VERSION, + } + if len(keys) != len(classifications): + raise ValueError("Every dependency key requires one classification") + return ( + tuple( + DependencyPolicy( + key, + classification, + phases[classification], + blocked_reason=blocked_reasons.get(key.related_table), + version_column=version_columns.get(key.related_table), + ) + for key, classification in zip(keys, classifications, strict=True) + ) + + synthetic + ) + + tag_cleanup: DependencyPolicy = DependencyPolicy( + DependencyKey( + "synthetic", + "", + "tagged_object", + relationship="tagged_object_cleanup", + ), + DependencyClassification.LISTENER_EFFECT, + ExecutionPhase.ASSOCIATIONS, + optional_listener=True, + listener_action=ListenerAction.DELETE_TAGGED_OBJECTS, + ) + permission_cleanup: DependencyPolicy = DependencyPolicy( + DependencyKey( + "synthetic", + "tables", + "ab_permission_view", + relationship="datasource_permission_cleanup", + ), + DependencyClassification.LISTENER_EFFECT, + ExecutionPhase.POST_DELETE, + listener_action=ListenerAction.DELETE_DATASET_PERMISSION, + ) + chart_membership_versions: DependencyPolicy = DependencyPolicy( + DependencyKey( + "synthetic", + "slices", + "dashboard_slices_version", + relationship="association_versions", + ), + DependencyClassification.VERSION_OWNED, + ExecutionPhase.VERSION, + version_column="slice_id", + ) + dashboard_membership_versions: DependencyPolicy = DependencyPolicy( + DependencyKey( + "synthetic", + "dashboards", + "dashboard_slices_version", + relationship="association_versions", + ), + DependencyClassification.VERSION_OWNED, + ExecutionPhase.VERSION, + version_column="dashboard_id", + ) + registry: dict[type[Any], PurgeEntityPolicy] = { + Slice: PurgeEntityPolicy( + model=Slice, + entity_type="chart", + dependencies=policies( + ( + fk("slices", "ab_user", "changed_by_fk", "id", "outbound"), + fk("slices", "ab_user", "created_by_fk", "id", "outbound"), + fk("slices", "ab_user", "last_saved_by_fk", "id", "outbound"), + fk("slices", "chart_editors", "id", "chart_id", "inbound"), + fk("slices", "chart_viewers", "id", "chart_id", "inbound"), + fk("slices", "dashboard_slices", "id", "slice_id", "inbound"), + fk("slices", "report_schedule", "id", "chart_id", "inbound"), + version("slices", "slices_version"), + relationship("slices", "tables", "manytoone", "table"), + fk("chart_editors", "slices", "chart_id", "id", "outbound"), + fk("chart_editors", "subjects", "subject_id", "id", "outbound"), + fk("chart_viewers", "slices", "chart_id", "id", "outbound"), + fk("chart_viewers", "subjects", "subject_id", "id", "outbound"), + fk( + "dashboard_slices", + "dashboards", + "dashboard_id", + "id", + "outbound", + ), + fk( + "dashboard_slices", + "slices", + "slice_id", + "id", + "outbound", + ), + ), + ( + DependencyClassification.PRESERVE, + DependencyClassification.PRESERVE, + DependencyClassification.PRESERVE, + DependencyClassification.ASSOCIATION, + DependencyClassification.ASSOCIATION, + DependencyClassification.ASSOCIATION, + DependencyClassification.BLOCK, + DependencyClassification.VERSION_OWNED, + DependencyClassification.PRESERVE, + DependencyClassification.PRESERVE, + DependencyClassification.PRESERVE, + DependencyClassification.PRESERVE, + DependencyClassification.PRESERVE, + DependencyClassification.PRESERVE, + DependencyClassification.PRESERVE, + ), + (tag_cleanup, chart_membership_versions), + {"report_schedule": "associated alerts or reports exist"}, + {"slices_version": "id"}, + ), + validate=validate_deletion_allowed, + count_dashboard_slices=count_dashboard_slices, + collect_dangling_chart_uuids=dangling_chart_uuids, + delete_associations=delete_associations, + delete_owned_children=delete_owned_children, + capture_permission_name=dataset_permission_name, + cleanup_permission=cleanup_dataset_permission, + ), + Dashboard: PurgeEntityPolicy( + model=Dashboard, + entity_type="dashboard", + dependencies=policies( + ( + fk("dashboards", "ab_user", "changed_by_fk", "id", "outbound"), + fk("dashboards", "ab_user", "created_by_fk", "id", "outbound"), + fk( + "dashboards", + "dashboard_editors", + "id", + "dashboard_id", + "inbound", + ), + fk( + "dashboards", + "dashboard_slices", + "id", + "dashboard_id", + "inbound", + ), + fk( + "dashboards", + "dashboard_viewers", + "id", + "dashboard_id", + "inbound", + ), + fk( + "dashboards", + "embedded_dashboards", + "id", + "dashboard_id", + "inbound", + ), + fk( + "dashboards", + "report_schedule", + "id", + "dashboard_id", + "inbound", + ), + fk("dashboards", "themes", "theme_id", "id", "outbound"), + fk( + "dashboards", + "user_attribute", + "id", + "welcome_dashboard_id", + "inbound", + ), + version("dashboards", "dashboards_version"), + fk( + "embedded_dashboards", + "ab_user", + "changed_by_fk", + "id", + "outbound", + ), + fk( + "embedded_dashboards", + "ab_user", + "created_by_fk", + "id", + "outbound", + ), + fk( + "embedded_dashboards", + "dashboards", + "dashboard_id", + "id", + "outbound", + ), + fk( + "dashboard_editors", + "dashboards", + "dashboard_id", + "id", + "outbound", + ), + fk( + "dashboard_editors", + "subjects", + "subject_id", + "id", + "outbound", + ), + fk( + "dashboard_slices", + "dashboards", + "dashboard_id", + "id", + "outbound", + ), + fk( + "dashboard_slices", + "slices", + "slice_id", + "id", + "outbound", + ), + fk( + "dashboard_viewers", + "dashboards", + "dashboard_id", + "id", + "outbound", + ), + fk( + "dashboard_viewers", + "subjects", + "subject_id", + "id", + "outbound", + ), + ), + ( + DependencyClassification.PRESERVE, + DependencyClassification.PRESERVE, + DependencyClassification.ASSOCIATION, + DependencyClassification.ASSOCIATION, + DependencyClassification.ASSOCIATION, + DependencyClassification.OWNED, + DependencyClassification.BLOCK, + DependencyClassification.PRESERVE, + DependencyClassification.BLOCK, + DependencyClassification.VERSION_OWNED, + DependencyClassification.PRESERVE, + DependencyClassification.PRESERVE, + DependencyClassification.PRESERVE, + DependencyClassification.PRESERVE, + DependencyClassification.PRESERVE, + DependencyClassification.PRESERVE, + DependencyClassification.PRESERVE, + DependencyClassification.PRESERVE, + DependencyClassification.PRESERVE, + ), + (tag_cleanup, dashboard_membership_versions), + { + "report_schedule": "associated alerts or reports exist", + "user_attribute": ( + "a user has this dashboard set as their welcome page" + ), + }, + {"dashboards_version": "id"}, + ), + validate=validate_deletion_allowed, + count_dashboard_slices=count_dashboard_slices, + collect_dangling_chart_uuids=dangling_chart_uuids, + delete_associations=delete_associations, + delete_owned_children=delete_owned_children, + capture_permission_name=dataset_permission_name, + cleanup_permission=cleanup_dataset_permission, + ), + SqlaTable: PurgeEntityPolicy( + model=SqlaTable, + entity_type="dataset", + dependencies=policies( + ( + fk("tables", "ab_user", "changed_by_fk", "id", "outbound"), + fk("tables", "ab_user", "created_by_fk", "id", "outbound"), + fk("tables", "dbs", "database_id", "id", "outbound"), + fk("tables", "rls_filter_tables", "id", "table_id", "inbound"), + fk("tables", "sql_metrics", "id", "table_id", "inbound"), + fk("tables", "sqlatable_editors", "id", "table_id", "inbound"), + fk("tables", "table_columns", "id", "table_id", "inbound"), + relationship("tables", "slices", "onetomany", "slices"), + version("tables", "tables_version"), + fk("sql_metrics", "ab_user", "changed_by_fk", "id", "outbound"), + fk("sql_metrics", "ab_user", "created_by_fk", "id", "outbound"), + fk("sql_metrics", "tables", "table_id", "id", "outbound"), + version("sql_metrics", "sql_metrics_version"), + fk("table_columns", "ab_user", "changed_by_fk", "id", "outbound"), + fk("table_columns", "ab_user", "created_by_fk", "id", "outbound"), + fk("table_columns", "tables", "table_id", "id", "outbound"), + version("table_columns", "table_columns_version"), + fk( + "rls_filter_tables", + "row_level_security_filters", + "rls_filter_id", + "id", + "outbound", + ), + fk( + "rls_filter_tables", + "tables", + "table_id", + "id", + "outbound", + ), + fk( + "sqlatable_editors", + "subjects", + "subject_id", + "id", + "outbound", + ), + fk( + "sqlatable_editors", + "tables", + "table_id", + "id", + "outbound", + ), + ), + ( + DependencyClassification.PRESERVE, + DependencyClassification.PRESERVE, + DependencyClassification.PRESERVE, + DependencyClassification.ASSOCIATION, + DependencyClassification.OWNED, + DependencyClassification.ASSOCIATION, + DependencyClassification.OWNED, + DependencyClassification.PRESERVE, + DependencyClassification.VERSION_OWNED, + DependencyClassification.PRESERVE, + DependencyClassification.PRESERVE, + DependencyClassification.PRESERVE, + DependencyClassification.VERSION_OWNED, + DependencyClassification.PRESERVE, + DependencyClassification.PRESERVE, + DependencyClassification.PRESERVE, + DependencyClassification.VERSION_OWNED, + DependencyClassification.PRESERVE, + DependencyClassification.PRESERVE, + DependencyClassification.PRESERVE, + DependencyClassification.PRESERVE, + ), + (tag_cleanup, permission_cleanup), + {}, + { + "tables_version": "id", + "sql_metrics_version": "table_id", + "table_columns_version": "table_id", + }, + ), + validate=validate_deletion_allowed, + count_dashboard_slices=count_dashboard_slices, + collect_dangling_chart_uuids=dangling_chart_uuids, + delete_associations=delete_associations, + delete_owned_children=delete_owned_children, + capture_permission_name=dataset_permission_name, + cleanup_permission=cleanup_dataset_permission, + ), + } + return validate_unique_root_policies(registry.values()) + + +@lru_cache(maxsize=None) +def _validated_purge_policy(model: type[Any]) -> PurgeEntityPolicy: + """Validate and return one root policy without blocking unrelated roots.""" + try: + policy: PurgeEntityPolicy = purge_policy_registry()[model] + except KeyError as ex: + raise ValueError(f"Unsupported purge model: {model.__name__}") from ex + 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, + ) + _validate_executable_declarations(policy) + if not coverage.complete: + details: str = "; ".join( + f"{label}=[{', '.join(key.describe() for key in dependencies)}]" + for label, dependencies in ( + ("missing", coverage.missing), + ("duplicates", coverage.duplicates), + ("stale", coverage.stale), + ) + if dependencies + ) + if coverage.missing_listeners: + details = ( + f"{details}; " if details else "" + ) + f"missing_listeners=[{', '.join(coverage.missing_listeners)}]" + if coverage.stale_listeners: + details = ( + f"{details}; " if details else "" + ) + f"stale_listeners=[{', '.join(coverage.stale_listeners)}]" + raise RuntimeError(f"Incomplete purge policy for {model.__name__}: {details}") + return policy + + +def _validate_executable_declarations(policy: PurgeEntityPolicy) -> None: + """Reject executable classifications missing their required action metadata.""" + for dependency in policy.dependencies: + if ( + dependency.classification is DependencyClassification.LISTENER_EFFECT + and dependency.listener_action is None + ): + raise RuntimeError( + f"Missing listener action for {dependency.key.describe()}" + ) + if ( + dependency.classification is DependencyClassification.VERSION_OWNED + and dependency.version_column is None + ): + raise RuntimeError( + f"Missing version target column for {dependency.key.describe()}" + ) + + +def get_purge_policy(model: type[Any]) -> PurgeEntityPolicy: + """Resolve a complete policy or reject an unsupported purge model.""" + return _validated_purge_policy(cast(Hashable, model)) + + +def listener_responsibilities(model: type[Any]) -> frozenset[str]: + """Return persistent listener responsibilities declared for a model.""" + return frozenset( + declaration.responsibility + for declaration in declared_delete_listeners() + if declaration.target is model + and declaration.effect is not DeleteListenerEffect.OBSERVATIONAL + ) + + +def validate_deletion_allowed( + session: Session, policy: PurgeEntityPolicy, entity_id: int +) -> None: + """Apply every blocker declared for a purge root.""" + metadata: sa.MetaData = sa.inspect(policy.model).local_table.metadata + for dependency in policy.dependencies: + if dependency.classification is not DependencyClassification.BLOCK: + continue + key: DependencyKey = dependency.key + table: sa.Table = _dependency_table(metadata, key) + predicates: tuple[Any, ...] = _dependency_predicates( + policy, key, entity_id, table + ) + if session.execute( + sa.select(sa.literal(1)).select_from(table).where(*predicates).limit(1) + ).first(): + if dependency.blocked_reason is None: + raise RuntimeError(f"Missing blocker reason for {key.describe()}") + raise PurgeBlockedError(dependency.blocked_reason) + + +def count_dashboard_slices( + session: Session, policy: PurgeEntityPolicy, entity_id: int +) -> int: + """Snapshot dashboard membership before explicit cleanup.""" + # avoid circular import: dashboard imports the chart model + from superset.models.dashboard import dashboard_slices + + column: Any | None = { + "chart": dashboard_slices.c.slice_id, + "dashboard": dashboard_slices.c.dashboard_id, + }.get(policy.entity_type) + if column is None: + return 0 + return int( + session.execute( + sa.select(sa.func.count()) + .select_from(dashboard_slices) + .where(column == entity_id) + ).scalar_one() + ) + + +def dangling_chart_uuids( + session: Session, policy: PurgeEntityPolicy, entity_id: int +) -> list[str]: + """Return chart UUIDs left by the dataset preservation policy.""" + if policy.entity_type != "dataset": + return [] + # avoid circular import: the chart model participates in registry assembly + from superset.models.slice import Slice + + return [ + str(chart_uuid) + for (chart_uuid,) in session.execute( + sa.select(Slice.uuid) + .where(Slice.datasource_id == entity_id) + .where(Slice.datasource_type == "table") + ) + ] + + +def delete_associations( + session: Session, policy: PurgeEntityPolicy, entity_id: int +) -> None: + """Execute declared association and tag cleanup with Core DML.""" + _delete_declared_dependencies( + session, policy, entity_id, DependencyClassification.ASSOCIATION + ) + _execute_listener_effects( + session, + policy, + entity_id, + phase=ExecutionPhase.ASSOCIATIONS, + permission_name=None, + ) + + +def delete_owned_children( + session: Session, policy: PurgeEntityPolicy, entity_id: int +) -> None: + """Execute declared owned-child cleanup with Core DML.""" + _delete_declared_dependencies( + session, policy, entity_id, DependencyClassification.OWNED + ) + + +def _delete_declared_dependencies( + session: Session, + policy: PurgeEntityPolicy, + entity_id: int, + classification: DependencyClassification, +) -> None: + """Delete inbound FK dependencies declared for one execution class.""" + metadata: sa.MetaData = sa.inspect(policy.model).local_table.metadata + dependencies: list[DependencyPolicy] = sorted( + ( + dependency + for dependency in policy.dependencies + if dependency.classification is classification + ), + key=lambda dependency: _dependency_owner_depth(policy, dependency.key), + reverse=True, + ) + for dependency in dependencies: + key: DependencyKey = dependency.key + table: sa.Table = _dependency_table(metadata, key) + predicates: tuple[Any, ...] = _dependency_predicates( + policy, key, entity_id, table + ) + session.execute(sa.delete(table).where(*predicates)) + + +def _dependency_table(metadata: sa.MetaData, key: DependencyKey) -> sa.Table: + """Resolve a declared dependency table or fail with policy context.""" + try: + return metadata.tables[key.related_table] + except KeyError as ex: + raise RuntimeError(f"Purge execution cannot resolve {key.describe()}") from ex + + +def _dependency_predicates( + policy: PurgeEntityPolicy, + key: DependencyKey, + entity_id: int, + table: sa.Table, +) -> tuple[Any, ...]: + """Build an atomic predicate for a simple or composite inbound constraint.""" + if key.kind != "foreign_key" or key.direction != "inbound": + raise RuntimeError(f"Purge execution cannot use {key.describe()}") + if len(key.local_columns) != len(key.remote_columns): + raise RuntimeError(f"Mismatched dependency columns for {key.describe()}") + owner_values: Any = _owner_value_select( + policy, key.owner_table, key.local_columns, entity_id + ) + remote_columns: tuple[sa.Column[Any], ...] = tuple( + table.c[column_name] for column_name in key.remote_columns + ) + if len(remote_columns) == 1: + return (remote_columns[0].in_(owner_values),) + return (sa.tuple_(*remote_columns).in_(owner_values),) + + +def _owner_value_select( + policy: PurgeEntityPolicy, + owner_table_name: str, + column_names: tuple[str, ...], + entity_id: int, +) -> Any: + """Select owner-column values reachable from the purge root through ownership.""" + root_table: sa.Table = sa.inspect(policy.model).local_table + metadata: sa.MetaData = root_table.metadata + owner_table: sa.Table = metadata.tables[owner_table_name] + selected_columns: tuple[sa.Column[Any], ...] = tuple( + owner_table.c[column_name] for column_name in column_names + ) + if owner_table_name == root_table.name: + return sa.select(*selected_columns).where(root_table.c.id == entity_id) + reverse_path: list[DependencyKey] = [] + visited: set[str] = set() + path_table_name: str = owner_table_name + while path_table_name != root_table.name: + if path_table_name in visited: + raise RuntimeError(f"Cyclic ownership path to {owner_table_name}") + visited.add(path_table_name) + ownership_key: DependencyKey = _ownership_edge(policy, path_table_name) + reverse_path.append(ownership_key) + path_table_name = ownership_key.owner_table + + path: list[DependencyKey] = list(reversed(reverse_path)) + parent_table: sa.Table = root_table + parent_predicate: Any = root_table.c.id == entity_id + for index, ownership_key in enumerate(path): + parent_columns: tuple[sa.Column[Any], ...] = tuple( + parent_table.c[column_name] for column_name in ownership_key.local_columns + ) + parent_values: Any = sa.select(*parent_columns).where(parent_predicate) + child_table: sa.Table = metadata.tables[ownership_key.related_table] + child_link_columns: tuple[sa.Column[Any], ...] = tuple( + child_table.c[column_name] for column_name in ownership_key.remote_columns + ) + parent_predicate = ( + child_link_columns[0].in_(parent_values) + if len(child_link_columns) == 1 + else sa.tuple_(*child_link_columns).in_(parent_values) + ) + parent_table = child_table + if index == len(path) - 1: + return sa.select(*selected_columns).where(parent_predicate) + raise RuntimeError(f"Missing ownership path to {owner_table_name}") + + +def _ownership_edge(policy: PurgeEntityPolicy, owner_table_name: str) -> DependencyKey: + """Resolve the unique owned/association edge linking a table to the root.""" + ownership_edges: tuple[DependencyKey, ...] = tuple( + dependency.key + for dependency in policy.dependencies + if dependency.classification + in {DependencyClassification.OWNED, DependencyClassification.ASSOCIATION} + and dependency.key.related_table == owner_table_name + and dependency.key.direction == "inbound" + ) + if len(ownership_edges) != 1: + raise RuntimeError( + f"Expected one ownership path to {owner_table_name}, " + f"found {len(ownership_edges)}" + ) + return ownership_edges[0] + + +def _dependency_owner_depth( + policy: PurgeEntityPolicy, + key: DependencyKey, +) -> int: + """Return the number of ownership edges between a dependency and its root.""" + root_table: sa.Table = sa.inspect(policy.model).local_table + owner_table_name: str = key.owner_table + visited: set[str] = set() + depth: int = 0 + while owner_table_name != root_table.name: + if owner_table_name in visited: + raise RuntimeError( + f"Cyclic ownership path while resolving {key.describe()}" + ) + visited.add(owner_table_name) + ownership_key: DependencyKey = _ownership_edge(policy, owner_table_name) + owner_table_name = ownership_key.owner_table + depth += 1 + return depth + + +def _execute_listener_effects( + session: Session, + policy: PurgeEntityPolicy, + entity_id: int, + *, + phase: ExecutionPhase, + permission_name: str | None, +) -> None: + """Execute every listener-equivalent action declared for one phase.""" + for dependency in policy.dependencies: + if ( + dependency.classification is not DependencyClassification.LISTENER_EFFECT + or dependency.phase is not phase + ): + continue + if dependency.listener_action is ListenerAction.DELETE_TAGGED_OBJECTS: + _delete_tagged_objects(session, policy, entity_id) + elif dependency.listener_action is ListenerAction.DELETE_DATASET_PERMISSION: + _delete_dataset_permission(session, permission_name, entity_id) + else: + raise RuntimeError( + f"Unsupported listener action for {dependency.key.describe()}" + ) + + +def _delete_tagged_objects( + session: Session, policy: PurgeEntityPolicy, entity_id: int +) -> None: + """Delete tag rows using the declared root object type.""" + from superset.tags.models import ObjectType, TaggedObject + + try: + object_type: ObjectType = { + "chart": ObjectType.chart, + "dashboard": ObjectType.dashboard, + "dataset": ObjectType.dataset, + }[policy.entity_type] + except KeyError as ex: + raise ValueError(f"Unsupported purge entity type: {policy.entity_type}") from ex + session.execute( + sa.delete(TaggedObject.__table__).where( + TaggedObject.object_id == entity_id, + TaggedObject.object_type == object_type, + ) + ) + + +def dataset_permission_name( + session: Session, policy: PurgeEntityPolicy, entity_id: int +) -> str | None: + """Capture the dataset permission identifier under the purge row lock. + + Reads the identity columns from the database rather than the in-memory + entity, so a rename or database move committed before the purge claimed + the row cannot leave the cleanup targeting a stale permission name. + """ + if policy.entity_type != "dataset": + return None + from superset import security_manager + + metadata: sa.MetaData = sa.inspect(policy.model).local_table.metadata + tables: sa.Table = metadata.tables["tables"] + dbs: sa.Table = metadata.tables["dbs"] + row = session.execute( + sa.select(tables.c.table_name, dbs.c.database_name) + .select_from(tables.join(dbs, tables.c.database_id == dbs.c.id)) + .where(tables.c.id == entity_id) + ).one_or_none() + if row is None: + return None + return str( + security_manager.get_dataset_perm(entity_id, row.table_name, row.database_name) + ) + + +def cleanup_dataset_permission( + session: Session, + policy: PurgeEntityPolicy, + permission_name: str | None, + entity_id: int, +) -> None: + """Execute the declared dataset permission-listener equivalent.""" + _execute_listener_effects( + session, + policy, + entity_id, + phase=ExecutionPhase.POST_DELETE, + permission_name=permission_name, + ) + + +def _delete_dataset_permission( + session: Session, permission_name: str | None, entity_id: int +) -> None: + """Remove the permission artifact represented by a listener declaration.""" + if permission_name is None: + raise RuntimeError("Dataset permission cleanup requires a captured name") + 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) diff --git a/superset/connectors/sqla/models.py b/superset/connectors/sqla/models.py index 45a4fc418b5..819e3e9f364 100644 --- a/superset/connectors/sqla/models.py +++ b/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", diff --git a/superset/migrations/versions/2018-07-26_11-10_c82ee8a39623_add_implicit_tags.py b/superset/migrations/versions/2018-07-26_11-10_c82ee8a39623_add_implicit_tags.py index 0b072df8d10..9fbf0284eea 100644 --- a/superset/migrations/versions/2018-07-26_11-10_c82ee8a39623_add_implicit_tags.py +++ b/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) diff --git a/superset/tags/core.py b/superset/tags/core.py index 6c4f56a2e66..00647cea01c 100644 --- a/superset/tags/core.py +++ b/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) diff --git a/superset/utils/sqlalchemy_events.py b/superset/utils/sqlalchemy_events.py new file mode 100644 index 00000000000..6d7abc66998 --- /dev/null +++ b/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, + ), + ) + ) diff --git a/tests/integration_tests/deletion_retention/purge_performance_tests.py b/tests/integration_tests/deletion_retention/purge_performance_tests.py new file mode 100644 index 00000000000..fdf410a1664 --- /dev/null +++ b/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) diff --git a/tests/integration_tests/deletion_retention/purge_tests.py b/tests/integration_tests/deletion_retention/purge_tests.py index ad8c3b1dbe5..f2fd9d3526d 100644 --- a/tests/integration_tests/deletion_retention/purge_tests.py +++ b/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.""" diff --git a/tests/unit_tests/commands/deletion_retention/__init__.py b/tests/unit_tests/commands/deletion_retention/__init__.py new file mode 100644 index 00000000000..13a83393a91 --- /dev/null +++ b/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. diff --git a/tests/unit_tests/commands/deletion_retention/test_purge_policy.py b/tests/unit_tests/commands/deletion_retention/test_purge_policy.py new file mode 100644 index 00000000000..2b9582fcd7c --- /dev/null +++ b/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) diff --git a/tests/unit_tests/migrations/test_add_implicit_tags_mapper_isolation.py b/tests/unit_tests/migrations/test_add_implicit_tags_mapper_isolation.py new file mode 100644 index 00000000000..729290875f7 --- /dev/null +++ b/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()