codeant-ai-for-open-source[bot] commented on code in PR #41549: URL: https://github.com/apache/superset/pull/41549#discussion_r3656645749
########## superset/commands/deletion_retention/purge_cascade.py: ########## @@ -0,0 +1,534 @@ +# 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.Logger = logging.getLogger(__name__) + + +@contextmanager +def suppress_purge_association_versions(session: Session) -> Iterator[None]: + """Discard only association versions generated by this purge session. + + ``dashboard_slices`` is a Continuum-tracked association table. Its + engine-level listener expects a unit-of-work even for Core deletes and + queues a shadow-table insert for each removed relationship. A purge + removes history rather than creating it, so preserve any pending work + that predates this block and discard only statements queued by the purge. + + This is deliberately session-scoped. Mutating Continuum's process-global + ``options["versioning"]`` would let concurrent requests silently lose + unrelated history while a purge is running. + """ + try: + from sqlalchemy_continuum import versioning_manager + except ImportError: + yield + return + + options = versioning_manager.options + if not (options.get("versioning") or options.get("native_versioning")): + yield + return + + unit_of_work = versioning_manager.unit_of_work(session) + pending_before = len(unit_of_work.pending_statements) + try: + yield + finally: + del unit_of_work.pending_statements[pending_before:] + + +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: 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.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") + + +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 + ) + ) Review Comment: **Suggestion:** The purge path deletes `dashboard_slices` without entering `suppress_purge_association_versions()`. Because this association is Continuum-tracked, the delete can enqueue association-version inserts during the purge, creating unwanted history or causing the transaction to fail when the corresponding entity/version rows are removed. Wrap the cascade's association deletes in the session-scoped suppression context. [logic error] <details> <summary><b>Severity Level:</b> Critical 🚨</summary> ```mdx - ⚠️ Dashboard purge can create unwanted association history. - ❌ Continuum-enabled purge transactions can fail during flush. - ⚠️ Version-history cleanup becomes inconsistent with relationship deletion. ``` </details> <details> <summary><b>Steps of Reproduction ✅ </b></summary> ```mdx 1. Start a deployment with SQLAlchemy-Continuum versioning enabled; the purge module explicitly detects Continuum at `superset/commands/deletion_retention/purge_cascade.py:72-83` and obtains the session unit of work. 2. Purge a soft-deleted dashboard through `cascade_hard_delete()` at `superset/commands/deletion_retention/purge_cascade.py:130`, which reaches `_delete_m2m_joins()` at line 186 and executes the `dashboard_slices` deletion at lines 304-309. 3. Observe that `suppress_purge_association_versions()` is defined at lines 58-89 but is not entered around `_delete_m2m_joins()` or the dashboard association delete. 4. Because `dashboard_slices` is documented as Continuum-tracked at lines 62-66, the Core delete can leave purge-generated association-version work queued in the session; the purge may create unwanted history or fail during flush when related entity/version rows have already been removed. ``` </details> [](https://app.codeant.ai/fix-in-ide?tool=cursor&prompt_id=8ef50d138e9c4b028eded1f4a0e21a9c&service=github&base_url=https%3A%2F%2Fgithub.com&org=apache&repo=apache%2Fsuperset) [](https://app.codeant.ai/fix-in-ide?tool=vscode-claude&prompt_id=8ef50d138e9c4b028eded1f4a0e21a9c&service=github&base_url=https%3A%2F%2Fgithub.com&org=apache&repo=apache%2Fsuperset) *(Use Cmd/Ctrl + Click for best experience)* <details> <summary><b>Prompt for AI Agent 🤖 </b></summary> ```mdx This is a comment left during a code review. **Path:** superset/commands/deletion_retention/purge_cascade.py **Line:** 305:309 **Comment:** *Logic Error: The purge path deletes `dashboard_slices` without entering `suppress_purge_association_versions()`. Because this association is Continuum-tracked, the delete can enqueue association-version inserts during the purge, creating unwanted history or causing the transaction to fail when the corresponding entity/version rows are removed. Wrap the cascade's association deletes in the session-scoped suppression context. Validate the correctness of the flagged issue. If correct, How can I resolve this? If you propose a fix, implement it and please make it concise. Once fix is implemented, also check other comments on the same PR, and ask user if the user wants to fix the rest of the comments as well. if said yes, then fetch all the comments validate the correctness and implement a minimal fix ``` </details> <a href='https://app.codeant.ai/feedback?pr_url=https%3A%2F%2Fgithub.com%2Fapache%2Fsuperset%2Fpull%2F41549&comment_hash=948e248def82c7b4846b7241ad88789b47f6059c0104e6347d4b8ec842b03d1d&reaction=like'>👍</a> | <a href='https://app.codeant.ai/feedback?pr_url=https%3A%2F%2Fgithub.com%2Fapache%2Fsuperset%2Fpull%2F41549&comment_hash=948e248def82c7b4846b7241ad88789b47f6059c0104e6347d4b8ec842b03d1d&reaction=dislike'>👎</a> ########## superset/commands/deletion_retention/purge_cascade.py: ########## @@ -0,0 +1,534 @@ +# 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.Logger = logging.getLogger(__name__) + + +@contextmanager +def suppress_purge_association_versions(session: Session) -> Iterator[None]: + """Discard only association versions generated by this purge session. + + ``dashboard_slices`` is a Continuum-tracked association table. Its + engine-level listener expects a unit-of-work even for Core deletes and + queues a shadow-table insert for each removed relationship. A purge + removes history rather than creating it, so preserve any pending work + that predates this block and discard only statements queued by the purge. + + This is deliberately session-scoped. Mutating Continuum's process-global + ``options["versioning"]`` would let concurrent requests silently lose + unrelated history while a purge is running. + """ + try: + from sqlalchemy_continuum import versioning_manager + except ImportError: + yield + return + + options = versioning_manager.options + if not (options.get("versioning") or options.get("native_versioning")): + yield + return + + unit_of_work = versioning_manager.unit_of_work(session) + pending_before = len(unit_of_work.pending_statements) + try: + yield + finally: + del unit_of_work.pending_statements[pending_before:] + + +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), + ) Review Comment: **Suggestion:** Catching every `IntegrityError` and converting it into a normal blocked result hides unrelated database failures, such as schema problems, connection-level transaction errors, or unexpected constraint violations. The caller will report that ordinary deletion rules blocked the entity even when the purge failed for an operational reason; only known dependency violations should be handled as `blocked`, while other integrity errors must propagate or be surfaced as failures. [error handling] <details> <summary><b>Severity Level:</b> Major ⚠️</summary> ```mdx - ❌ Unexpected database failures are reported as dependency blocks. - ⚠️ Operators receive misleading purge status and reasons. - ⚠️ Failed entities may remain pending without failure visibility. ``` </details> <details> <summary><b>Steps of Reproduction ✅ </b></summary> ```mdx 1. Invoke `cascade_hard_delete()` at `superset/commands/deletion_retention/purge_cascade.py:130` for an eligible entity; the function performs join cleanup, owned-child cleanup, version cleanup, and the parent delete at lines 186-195. 2. Cause any database integrity constraint to fail in one of those statements, including a constraint not represented by `_validate_deletion_allowed()` at lines 237-257; that validator only checks report or alert references for charts and dashboards. 3. The database raises `IntegrityError`, which is caught together with the intentional `PurgeBlockedError` at lines 207-208. 4. The function returns `CascadeResult(purged=False, blocked_reason=str(ex))` at lines 213-218, making an operational or unexpected constraint failure appear to callers as an ordinary dependency block instead of surfacing a failed purge for retry or operator investigation. ``` </details> [](https://app.codeant.ai/fix-in-ide?tool=cursor&prompt_id=7c14656ccbe44885819e5201c8494b79&service=github&base_url=https%3A%2F%2Fgithub.com&org=apache&repo=apache%2Fsuperset) [](https://app.codeant.ai/fix-in-ide?tool=vscode-claude&prompt_id=7c14656ccbe44885819e5201c8494b79&service=github&base_url=https%3A%2F%2Fgithub.com&org=apache&repo=apache%2Fsuperset) *(Use Cmd/Ctrl + Click for best experience)* <details> <summary><b>Prompt for AI Agent 🤖 </b></summary> ```mdx This is a comment left during a code review. **Path:** superset/commands/deletion_retention/purge_cascade.py **Line:** 207:218 **Comment:** *Error Handling: Catching every `IntegrityError` and converting it into a normal blocked result hides unrelated database failures, such as schema problems, connection-level transaction errors, or unexpected constraint violations. The caller will report that ordinary deletion rules blocked the entity even when the purge failed for an operational reason; only known dependency violations should be handled as `blocked`, while other integrity errors must propagate or be surfaced as failures. Validate the correctness of the flagged issue. If correct, How can I resolve this? If you propose a fix, implement it and please make it concise. Once fix is implemented, also check other comments on the same PR, and ask user if the user wants to fix the rest of the comments as well. if said yes, then fetch all the comments validate the correctness and implement a minimal fix ``` </details> <a href='https://app.codeant.ai/feedback?pr_url=https%3A%2F%2Fgithub.com%2Fapache%2Fsuperset%2Fpull%2F41549&comment_hash=8d993147c52be0605ae3c4f2b85c7f4fac04590880f85dfa5e1ef4b231a4d2b0&reaction=like'>👍</a> | <a href='https://app.codeant.ai/feedback?pr_url=https%3A%2F%2Fgithub.com%2Fapache%2Fsuperset%2Fpull%2F41549&comment_hash=8d993147c52be0605ae3c4f2b85c7f4fac04590880f85dfa5e1ef4b231a4d2b0&reaction=dislike'>👎</a> -- 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]
