mikebridge commented on code in PR #41549: URL: https://github.com/apache/superset/pull/41549#discussion_r3574285338
########## superset/commands/deletion_retention/purge_cascade.py: ########## @@ -0,0 +1,526 @@ +# 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. +"""Shared hard-delete cascade for the purge task and force-purge command. + +A single code path keeps the two surfaces from drifting. Every dependent row +is removed by an explicit ``sa.delete``; +the DB ``ON DELETE CASCADE`` constraints are a backstop only — SQLite does +not enforce FKs unless ``PRAGMA foreign_keys=ON`` and Core bulk DML fires +only DB-level cascades, so relying on cascade silently leaks rows. + +Cascade tiers: + +* **M:N join rows** — hard-deleted for the purged entity, including join + rows owned by *surviving* entities (e.g. a live dashboard's + ``dashboard_slices`` row to a purged chart). The entity on the other side + is never touched except to lose that one relationship row. +* **Owned children** (``delete-orphan``, no independent existence) — a + dataset's columns and metrics, hard-deleted with it. +* **Independently-owned entities** (a dashboard's charts, a chart's dataset) + are **preserved**. A live chart's loose ``datasource_id`` to a purged + dataset is left dangling — legacy hard-delete semantics, no guard. +* **Version history** — the entity's own ``*_version`` shadows and + the ``version_changes`` scoped to them, plus an orphan-sweep of any + ``version_transaction`` left owning zero surviving shadows. Runs + behind a ``has_table`` check so it no-ops when versioning is absent. +""" + +from __future__ import annotations + +import logging +from collections.abc import Iterator +from contextlib import contextmanager +from dataclasses import dataclass, field +from datetime import datetime +from typing import Any + +import sqlalchemy as sa +from sqlalchemy.exc import IntegrityError +from sqlalchemy.orm import Session + +logger = logging.getLogger(__name__) + + +@contextmanager +def suspend_version_capture() -> Iterator[None]: + """Turn Continuum's write tracking off for the duration of the block. + + A purge *removes* version history; it must not *generate* it. Flipping + Continuum's master ``versioning`` option off also stops the association + tracker (``track_association_operations``) from firing on the cascade's + raw ``sa.delete`` of tracked join tables (``dashboard_slices``), which + otherwise raises ``KeyError`` because there is no Continuum unit-of-work + for a Core delete. No-op when Continuum is absent or capture is already + off; the prior value is always restored. + """ + try: + from sqlalchemy_continuum import versioning_manager + except ImportError: + yield + return + previous = versioning_manager.options.get("versioning", True) + versioning_manager.options["versioning"] = False + try: + yield + finally: + versioning_manager.options["versioning"] = previous + + +def entity_uuid(entity: Any) -> str | None: + """Return the entity's UUID as a string, or ``None`` if it has none.""" + value = getattr(entity, "uuid", None) + return str(value) if value is not None else None + + +def _version_tables_present(bind: Any) -> bool: + """Whether the version-history tables exist (a testable seam, so the + version cascade no-ops cleanly before versioning is installed).""" + return sa.inspect(bind).has_table("version_transaction") + + +@dataclass +class CascadeResult: + """Outcome of one entity's cascade. + + ``purged`` is False when the conditional entity-row delete matched no + row — the entity was restored between selection and delete or was already + gone. In that case + no dependents are touched. + """ + + purged: bool + entity_type: str + entity_uuid: str | None + dangling_chart_uuids: list[str] = field(default_factory=list) + removed_dashboard_slices: int = 0 + version_rows_removed: int = 0 + 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.""" + + +def cascade_hard_delete( + session: Session, + entity: Any, + *, + enforce_window: bool, + cutoff: datetime | None = None, +) -> CascadeResult: + """Remove *entity* and everything that depends on it in one transaction. + + The entity row is locked and its eligibility is re-checked before any + dependent state is touched. Ordinary deletion blockers remain + authoritative. All cleanup and the conditional parent delete run inside a + savepoint so a lost race or restrictive foreign key leaves no side effects. + """ + # 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") + + model = type(entity) + table = model.__table__ + entity_id = entity.id + uuid = entity_uuid(entity) + entity_type = _USER_FACING_TYPE.get(table.name, table.name) + + dangling_chart_uuids: list[str] = [] + removed_dashboard_slices = 0 + version_rows = 0 + permission_name = _dataset_permission_name(entity) if model is SqlaTable else None + + try: + with session.begin_nested(): + claim = sa.select(table.c.id).where(table.c.id == entity_id) + if enforce_window: + claim = claim.where(table.c.deleted_at.is_not(None)).where( + table.c.deleted_at < cutoff + ) + 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 + ) + 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) + version_rows = _delete_version_history(session, entity, entity_id) + + delete_entity = sa.delete(table).where(table.c.id == entity_id) + if enforce_window: + delete_entity = delete_entity.where( + table.c.deleted_at.is_not(None) + ).where(table.c.deleted_at < cutoff) + if session.execute(delete_entity).rowcount == 0: + raise PurgeRaceLostError + + if permission_name is not None: + _cleanup_dataset_permission(session, permission_name, entity_id) + except PurgeRaceLostError: + logger.info( + "deletion_retention: %s id=%s not purged (restored or already gone)", + entity_type, + entity_id, + ) + return CascadeResult(purged=False, entity_type=entity_type, entity_uuid=uuid) + except (PurgeBlockedError, IntegrityError) as ex: + logger.info( + "deletion_retention: %s id=%s blocked by existing deletion rules", + entity_type, + entity_id, + ) + return CascadeResult( + purged=False, + entity_type=entity_type, + entity_uuid=uuid, + blocked_reason=str(ex), + ) + + return CascadeResult( + purged=True, + entity_type=entity_type, + entity_uuid=uuid, + dangling_chart_uuids=dangling_chart_uuids, + removed_dashboard_slices=removed_dashboard_slices, + version_rows_removed=version_rows, + ) + + +_USER_FACING_TYPE = { + "slices": "chart", + "dashboards": "dashboard", + "tables": "dataset", +} Review Comment: Addressed in 5b4b2460fc: _USER_FACING_TYPE now has an explicit dict[str, str] annotation. ########## superset/commands/deletion_retention/purge_cascade.py: ########## @@ -0,0 +1,526 @@ +# 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. +"""Shared hard-delete cascade for the purge task and force-purge command. + +A single code path keeps the two surfaces from drifting. Every dependent row +is removed by an explicit ``sa.delete``; +the DB ``ON DELETE CASCADE`` constraints are a backstop only — SQLite does +not enforce FKs unless ``PRAGMA foreign_keys=ON`` and Core bulk DML fires +only DB-level cascades, so relying on cascade silently leaks rows. + +Cascade tiers: + +* **M:N join rows** — hard-deleted for the purged entity, including join + rows owned by *surviving* entities (e.g. a live dashboard's + ``dashboard_slices`` row to a purged chart). The entity on the other side + is never touched except to lose that one relationship row. +* **Owned children** (``delete-orphan``, no independent existence) — a + dataset's columns and metrics, hard-deleted with it. +* **Independently-owned entities** (a dashboard's charts, a chart's dataset) + are **preserved**. A live chart's loose ``datasource_id`` to a purged + dataset is left dangling — legacy hard-delete semantics, no guard. +* **Version history** — the entity's own ``*_version`` shadows and + the ``version_changes`` scoped to them, plus an orphan-sweep of any + ``version_transaction`` left owning zero surviving shadows. Runs + behind a ``has_table`` check so it no-ops when versioning is absent. +""" + +from __future__ import annotations + +import logging +from collections.abc import Iterator +from contextlib import contextmanager +from dataclasses import dataclass, field +from datetime import datetime +from typing import Any + +import sqlalchemy as sa +from sqlalchemy.exc import IntegrityError +from sqlalchemy.orm import Session + +logger = logging.getLogger(__name__) + + +@contextmanager +def suspend_version_capture() -> Iterator[None]: + """Turn Continuum's write tracking off for the duration of the block. + + A purge *removes* version history; it must not *generate* it. Flipping + Continuum's master ``versioning`` option off also stops the association + tracker (``track_association_operations``) from firing on the cascade's + raw ``sa.delete`` of tracked join tables (``dashboard_slices``), which + otherwise raises ``KeyError`` because there is no Continuum unit-of-work + for a Core delete. No-op when Continuum is absent or capture is already + off; the prior value is always restored. + """ + try: + from sqlalchemy_continuum import versioning_manager + except ImportError: + yield + return + previous = versioning_manager.options.get("versioning", True) + versioning_manager.options["versioning"] = False + try: + yield + finally: + versioning_manager.options["versioning"] = previous + + +def entity_uuid(entity: Any) -> str | None: + """Return the entity's UUID as a string, or ``None`` if it has none.""" + value = getattr(entity, "uuid", None) + return str(value) if value is not None else None + + +def _version_tables_present(bind: Any) -> bool: + """Whether the version-history tables exist (a testable seam, so the + version cascade no-ops cleanly before versioning is installed).""" + return sa.inspect(bind).has_table("version_transaction") + + +@dataclass +class CascadeResult: + """Outcome of one entity's cascade. + + ``purged`` is False when the conditional entity-row delete matched no + row — the entity was restored between selection and delete or was already + gone. In that case + no dependents are touched. + """ + + purged: bool + entity_type: str + entity_uuid: str | None + dangling_chart_uuids: list[str] = field(default_factory=list) + removed_dashboard_slices: int = 0 + version_rows_removed: int = 0 + 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.""" + + +def cascade_hard_delete( + session: Session, + entity: Any, + *, + enforce_window: bool, + cutoff: datetime | None = None, +) -> CascadeResult: + """Remove *entity* and everything that depends on it in one transaction. + + The entity row is locked and its eligibility is re-checked before any + dependent state is touched. Ordinary deletion blockers remain + authoritative. All cleanup and the conditional parent delete run inside a + savepoint so a lost race or restrictive foreign key leaves no side effects. + """ + # 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") + + model = type(entity) + table = model.__table__ + entity_id = entity.id + uuid = entity_uuid(entity) + entity_type = _USER_FACING_TYPE.get(table.name, table.name) + + dangling_chart_uuids: list[str] = [] + removed_dashboard_slices = 0 + version_rows = 0 + permission_name = _dataset_permission_name(entity) if model is SqlaTable else None + + try: + with session.begin_nested(): + claim = sa.select(table.c.id).where(table.c.id == entity_id) + if enforce_window: + claim = claim.where(table.c.deleted_at.is_not(None)).where( + table.c.deleted_at < cutoff + ) + 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 + ) + 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) + version_rows = _delete_version_history(session, entity, entity_id) + + delete_entity = sa.delete(table).where(table.c.id == entity_id) + if enforce_window: + delete_entity = delete_entity.where( + table.c.deleted_at.is_not(None) + ).where(table.c.deleted_at < cutoff) + if session.execute(delete_entity).rowcount == 0: + raise PurgeRaceLostError + + if permission_name is not None: + _cleanup_dataset_permission(session, permission_name, entity_id) + except PurgeRaceLostError: + logger.info( + "deletion_retention: %s id=%s not purged (restored or already gone)", + entity_type, + entity_id, + ) + return CascadeResult(purged=False, entity_type=entity_type, entity_uuid=uuid) + except (PurgeBlockedError, IntegrityError) as ex: + logger.info( + "deletion_retention: %s id=%s blocked by existing deletion rules", + entity_type, + entity_id, + ) + return CascadeResult( + purged=False, + entity_type=entity_type, + entity_uuid=uuid, + blocked_reason=str(ex), + ) + + return CascadeResult( + purged=True, + entity_type=entity_type, + entity_uuid=uuid, + dangling_chart_uuids=dangling_chart_uuids, + removed_dashboard_slices=removed_dashboard_slices, + version_rows_removed=version_rows, + ) + + +_USER_FACING_TYPE = { + "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.reports.models import ReportSchedule + + column = None Review Comment: Addressed in 5b4b2460fc: the deletion-guard column now has an explicit Any | None annotation. ########## superset/commands/deletion_retention/purge_cascade.py: ########## @@ -0,0 +1,526 @@ +# 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. +"""Shared hard-delete cascade for the purge task and force-purge command. + +A single code path keeps the two surfaces from drifting. Every dependent row +is removed by an explicit ``sa.delete``; +the DB ``ON DELETE CASCADE`` constraints are a backstop only — SQLite does +not enforce FKs unless ``PRAGMA foreign_keys=ON`` and Core bulk DML fires +only DB-level cascades, so relying on cascade silently leaks rows. + +Cascade tiers: + +* **M:N join rows** — hard-deleted for the purged entity, including join + rows owned by *surviving* entities (e.g. a live dashboard's + ``dashboard_slices`` row to a purged chart). The entity on the other side + is never touched except to lose that one relationship row. +* **Owned children** (``delete-orphan``, no independent existence) — a + dataset's columns and metrics, hard-deleted with it. +* **Independently-owned entities** (a dashboard's charts, a chart's dataset) + are **preserved**. A live chart's loose ``datasource_id`` to a purged + dataset is left dangling — legacy hard-delete semantics, no guard. +* **Version history** — the entity's own ``*_version`` shadows and + the ``version_changes`` scoped to them, plus an orphan-sweep of any + ``version_transaction`` left owning zero surviving shadows. Runs + behind a ``has_table`` check so it no-ops when versioning is absent. +""" + +from __future__ import annotations + +import logging +from collections.abc import Iterator +from contextlib import contextmanager +from dataclasses import dataclass, field +from datetime import datetime +from typing import Any + +import sqlalchemy as sa +from sqlalchemy.exc import IntegrityError +from sqlalchemy.orm import Session + +logger = logging.getLogger(__name__) + + +@contextmanager +def suspend_version_capture() -> Iterator[None]: + """Turn Continuum's write tracking off for the duration of the block. + + A purge *removes* version history; it must not *generate* it. Flipping + Continuum's master ``versioning`` option off also stops the association + tracker (``track_association_operations``) from firing on the cascade's + raw ``sa.delete`` of tracked join tables (``dashboard_slices``), which + otherwise raises ``KeyError`` because there is no Continuum unit-of-work + for a Core delete. No-op when Continuum is absent or capture is already + off; the prior value is always restored. + """ + try: + from sqlalchemy_continuum import versioning_manager + except ImportError: + yield + return + previous = versioning_manager.options.get("versioning", True) + versioning_manager.options["versioning"] = False + try: + yield + finally: + versioning_manager.options["versioning"] = previous + + +def entity_uuid(entity: Any) -> str | None: + """Return the entity's UUID as a string, or ``None`` if it has none.""" + value = getattr(entity, "uuid", None) + return str(value) if value is not None else None + + +def _version_tables_present(bind: Any) -> bool: + """Whether the version-history tables exist (a testable seam, so the + version cascade no-ops cleanly before versioning is installed).""" + return sa.inspect(bind).has_table("version_transaction") + + +@dataclass +class CascadeResult: + """Outcome of one entity's cascade. + + ``purged`` is False when the conditional entity-row delete matched no + row — the entity was restored between selection and delete or was already + gone. In that case + no dependents are touched. + """ + + purged: bool + entity_type: str + entity_uuid: str | None + dangling_chart_uuids: list[str] = field(default_factory=list) + removed_dashboard_slices: int = 0 + version_rows_removed: int = 0 + 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.""" + + +def cascade_hard_delete( + session: Session, + entity: Any, + *, + enforce_window: bool, + cutoff: datetime | None = None, +) -> CascadeResult: + """Remove *entity* and everything that depends on it in one transaction. + + The entity row is locked and its eligibility is re-checked before any + dependent state is touched. Ordinary deletion blockers remain + authoritative. All cleanup and the conditional parent delete run inside a + savepoint so a lost race or restrictive foreign key leaves no side effects. + """ + # 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") + + model = type(entity) + table = model.__table__ + entity_id = entity.id + uuid = entity_uuid(entity) + entity_type = _USER_FACING_TYPE.get(table.name, table.name) + + dangling_chart_uuids: list[str] = [] + removed_dashboard_slices = 0 + version_rows = 0 + permission_name = _dataset_permission_name(entity) if model is SqlaTable else None + + try: + with session.begin_nested(): + claim = sa.select(table.c.id).where(table.c.id == entity_id) + if enforce_window: + claim = claim.where(table.c.deleted_at.is_not(None)).where( + table.c.deleted_at < cutoff + ) + 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 + ) + 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) + version_rows = _delete_version_history(session, entity, entity_id) + + delete_entity = sa.delete(table).where(table.c.id == entity_id) + if enforce_window: + delete_entity = delete_entity.where( + table.c.deleted_at.is_not(None) + ).where(table.c.deleted_at < cutoff) + if session.execute(delete_entity).rowcount == 0: + raise PurgeRaceLostError + + if permission_name is not None: + _cleanup_dataset_permission(session, permission_name, entity_id) + except PurgeRaceLostError: + logger.info( + "deletion_retention: %s id=%s not purged (restored or already gone)", + entity_type, + entity_id, + ) + return CascadeResult(purged=False, entity_type=entity_type, entity_uuid=uuid) + except (PurgeBlockedError, IntegrityError) as ex: + logger.info( + "deletion_retention: %s id=%s blocked by existing deletion rules", + entity_type, + entity_id, + ) + return CascadeResult( + purged=False, + entity_type=entity_type, + entity_uuid=uuid, + blocked_reason=str(ex), + ) + + return CascadeResult( + purged=True, + entity_type=entity_type, + entity_uuid=uuid, + dangling_chart_uuids=dangling_chart_uuids, + removed_dashboard_slices=removed_dashboard_slices, + version_rows_removed=version_rows, + ) + + +_USER_FACING_TYPE = { + "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.reports.models import ReportSchedule + + column = 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") + + +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 = None + if model is Dashboard: Review Comment: Addressed in 5b4b2460fc: the dashboard-slice predicate now has an explicit Any | None annotation. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
