codeant-ai-for-open-source[bot] commented on code in PR #41075:
URL: https://github.com/apache/superset/pull/41075#discussion_r3572405510
##########
superset/initialization/__init__.py:
##########
@@ -849,10 +857,61 @@ def init_versioning(self) -> None:
register_baseline_listener()
register_change_record_listener()
- # Retention pruning runs out-of-band as a scheduled Celery beat
- # task, shipped as a separate stacked PR. The previous
- # synchronous after_commit listener was retired so retention work
- # doesn't add latency to user saves.
+ # Retention is time-based and runs out-of-band as a Celery beat
+ # task — see ``superset/tasks/version_history_retention.py``
+ # and the ``version_history.prune_old_versions`` entry in
+ # ``CeleryConfig.beat_schedule`` (``superset/config.py``). The
+ # previous synchronous after_commit listener was retired so
+ # retention work doesn't add latency to user saves.
+
+ _RETENTION_TASK_NAME: str = "version_history.prune_old_versions"
+
+ def _warn_if_retention_beat_missing(self) -> None:
+ """WARN at startup when the resolved Celery beat schedule has no
+ ``version_history.prune_old_versions`` entry.
+
+ Operators who redefine ``CeleryConfig`` in ``superset_config.py``
+ — instead of subclassing or merging the default — silently lose
+ the retention task. Capture continues writing rows; the prune
+ never runs; disk grows until paged. The default config carries
+ the entry; this check makes the misconfiguration visible in the
+ deploy log before disk pressure makes it visible at 03:00.
+
+ Handles three shapes of ``CELERY_CONFIG``:
+ * ``None`` — Celery deliberately disabled, no retention either
+ way; return without warning.
+ * a class or module with a ``beat_schedule`` attribute — the
+ default ``CeleryConfig`` shape.
+ * a dict — Celery's documented "config as dict" shape, supported
+ by ``celery_app.config_from_object``.
+ """
+ celery_config = self.config.get("CELERY_CONFIG")
+ if celery_config is None:
+ return # Celery disabled entirely; no retention task to warn
about.
+ beat_schedule = (
+ celery_config.get("beat_schedule")
+ if isinstance(celery_config, dict)
+ else getattr(celery_config, "beat_schedule", None)
+ )
+ # Match on the ``task`` each entry runs, not the schedule entry key:
+ # an operator may register the retention task under any key (e.g.
+ # ``{"prune_versions": {"task":
"version_history.prune_old_versions"}}``),
+ # which is still correctly scheduled and must not warn. The default
+ # config happens to use the task name as the key, but that's
incidental.
+ registered_tasks = {
+ entry.get("task")
+ for entry in (beat_schedule or {}).values()
+ if isinstance(entry, dict)
+ }
Review Comment:
**Suggestion:** Add an explicit type annotation for `registered_tasks` to
satisfy the rule requiring type hints on annotatable variables. [custom_rule]
**Severity Level:** Minor 🧹
<details>
<summary><b>Why it matters? ⭐ </b></summary>
The new local variable `registered_tasks` is a set-like collection inferred
from the comprehension, and it is not annotated with a type hint. This matches
the rule requiring type hints for annotatable Python variables introduced or
modified in the PR.
</details>
<details>
<summary><b>Rule source 📖 </b></summary>
.cursor/rules/dev-standard.mdc (line 28)
</details>
[](https://app.codeant.ai/fix-in-ide?tool=cursor&prompt_id=a073748091af45daa6e318c42cca9a8c&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=a073748091af45daa6e318c42cca9a8c&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/initialization/__init__.py
**Line:** 901:905
**Comment:**
*Custom Rule: Add an explicit type annotation for `registered_tasks` to
satisfy the rule requiring type hints on annotatable variables.
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%2F41075&comment_hash=ca80cde11c19cd7aeb3ac54a4b696cc9d7a793eee5d6a0783049afb98f4a5cc1&reaction=like'>👍</a>
| <a
href='https://app.codeant.ai/feedback?pr_url=https%3A%2F%2Fgithub.com%2Fapache%2Fsuperset%2Fpull%2F41075&comment_hash=ca80cde11c19cd7aeb3ac54a4b696cc9d7a793eee5d6a0783049afb98f4a5cc1&reaction=dislike'>👎</a>
##########
tests/integration_tests/versioning/retention_prune_tests.py:
##########
@@ -0,0 +1,488 @@
+# 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.
+"""Integration coverage for version-history retention pruning.
+
+Exercises ``superset.tasks.version_history_retention`` end-to-end against
+a real database: that an aged-out transaction's shadow rows are pruned
+while the live row is always preserved, and that the SERIALIZABLE pass
+retries on transient serialization failures and gives up after the cap.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+import pytest
+from sqlalchemy_continuum import version_class
+
+from superset.extensions import db
+from superset.models.dashboard import Dashboard
+from superset.models.slice import Slice
+from superset.versioning.changes.table import version_changes_table
+from tests.integration_tests.base_tests import SupersetTestCase
+from tests.integration_tests.fixtures.birth_names_dashboard import ( # noqa:
F401
+ load_birth_names_dashboard_with_slices,
+ load_birth_names_data,
+)
+
+
+def _get_version_rows(dashboard: Dashboard) -> list[Any]:
+ ver_cls = version_class(Dashboard)
+ return (
+ db.session.query(ver_cls)
+ .filter(ver_cls.id == dashboard.id)
+ .order_by(ver_cls.transaction_id.asc())
+ .all()
+ )
+
+
+def _persist_fixture_state() -> None:
+ """Force the fixture's pending INSERTs to commit in their own transaction.
+
+ The birth_names fixture stages charts and the dashboard via session.add()
+ but does not commit. Without this, the test's first commit batches the
+ INSERTs and UPDATEs into the same Continuum transaction, causing the
+ existing version row to be updated in place instead of a new one being
+ created.
+ """
+ db.session.commit()
+
+
+class TestDashboardVersionRetention(SupersetTestCase):
+ """Retention pruning drops shadow rows older than
+ ``SUPERSET_VERSION_HISTORY_RETENTION_DAYS`` while preserving live rows."""
+
+ @pytest.fixture(autouse=True)
+ def _load_data(
+ self,
+ load_birth_names_dashboard_with_slices: None, # noqa: F811
+ ) -> None: # noqa: PT004
+ """Establish the capture state this integration class requires.
+
+ Continuum stores its option and SQLAlchemy event listeners globally.
+ Initialization unit tests intentionally exercise the kill-switch that
+ detaches those listeners, so a mixed or reordered pytest invocation can
+ otherwise enter these tests with capture disabled. Reasserting the
+ integration suite's prerequisite here makes each test
order-independent.
+ """
+ from sqlalchemy_continuum import versioning_manager
+
+ from superset.initialization import SupersetAppInitializer
+
+ versioning_manager.options["versioning"] = True
+ SupersetAppInitializer._add_continuum_write_listeners()
+
+ def test_retention_prunes_old_rows(self) -> None:
+ """``prune_old_versions`` removes shadow rows whose owning
+ ``version_transaction.issued_at`` is older than the retention
+ window, while preserving every transaction that anchors a live row."""
+ from datetime import datetime, timedelta
+
+ import sqlalchemy as sa
+
+ from superset.extensions import db as _db
+ from superset.tasks.version_history_retention import (
+ _prune_old_versions_impl,
+ )
+
+ _persist_fixture_state()
+ dashboard: Dashboard = (
+ db.session.query(Dashboard)
+ .filter(Dashboard.dashboard_title == "USA Births Names")
+ .first()
+ )
+ assert dashboard is not None
+
+ original_title = dashboard.dashboard_title
+
+ try:
+ # Force a few saves so we have ≥ 2 closed shadow rows plus
+ # a baseline plus the live row.
+ for i in range(3):
+ dashboard.dashboard_title = f"USA Births Names retention test
{i}"
+ db.session.commit()
+
+ rows_before = _get_version_rows(dashboard)
+ assert len(rows_before) >= 3, "Expected at least 3 version rows"
+
+ def _version_changes_count() -> int:
+ return (
+ _db.session.execute(
+ sa.text("SELECT COUNT(*) FROM version_changes")
+ ).scalar()
+ or 0
+ )
+
+ changes_before = _version_changes_count()
+ assert changes_before >= 1, (
+ "expected version_changes rows from the edits above"
+ )
+
+ # Backdate every version_transaction row by 100 days so the
+ # prune sees them as old. Skip baseline+live rows; the prune
+ # itself preserves them.
+ from sqlalchemy_continuum import versioning_manager
+
+ tx_table = versioning_manager.transaction_cls.__table__
+ with _db.engine.begin() as conn:
+ conn.execute(
+ sa.update(tx_table).values(
+ issued_at=datetime.utcnow() - timedelta(days=100)
+ )
+ )
+
+ stats = _prune_old_versions_impl(retention_days=30)
+ assert stats.get("pruned_transactions", 0) >= 1, stats
+
+ # version_changes rows for pruned transactions must be gone too.
+ # The prune deletes version_transaction rows and relies on the
+ # ON DELETE CASCADE FK from version_changes.transaction_id; assert
+ # the cascade actually fired rather than orphaning change records.
+ db.session.expire_all()
+ assert _version_changes_count() < changes_before, (
+ "version_changes rows for pruned transactions were not removed
"
+ f"(before={changes_before}, after={_version_changes_count()})"
+ )
+
+ rows_after = _get_version_rows(dashboard)
+ # Live row must still exist (this is the only preservation rule)
+ live_rows = [r for r in rows_after if r.end_transaction_id is None]
+ assert len(live_rows) >= 1, "Live row must never be pruned"
+ # Some rows should have been pruned. Closed historical rows —
+ # including the synthetic baseline (operation_type=0) — are
+ # subject to retention like everything else.
+ assert len(rows_after) < len(rows_before), (
+ f"Expected fewer rows after prune; before={len(rows_before)} "
+ f"after={len(rows_after)}"
+ )
+
+ finally:
+ dashboard.dashboard_title = original_title
+ db.session.commit()
+
+ def test_retention_preserves_live_child_and_m2m_rows(self) -> None:
+ """Regression for the live-row preservation BLOCKER.
+
+ A parent-only edit leaves the dashboard's chart associations (the
+ M2M ``dashboard_slices_version`` rows) live but anchored at an
+ *older* transaction than the dashboard's current live row. The
+ prune must preserve that older transaction too — if it scans only
+ parent shadows for live rows, it deletes the still-live
+ association and the surviving version silently loses its slices.
+
+ The parent-only ``test_retention_prunes_old_rows`` cannot catch
+ this: it asserts on the dashboard parent shadow alone, which is
+ exactly the blind spot the bug lived in.
+ """
+ from datetime import datetime, timedelta
+
+ import sqlalchemy as sa
+ from sqlalchemy_continuum import version_class, versioning_manager
+
+ from superset.extensions import db as _db
+ from superset.tasks.version_history_retention import
_prune_old_versions_impl
+
+ _persist_fixture_state()
+ dashboard: Dashboard = (
+ db.session.query(Dashboard)
+ .filter(Dashboard.dashboard_title == "USA Births Names")
+ .first()
+ )
+ assert dashboard is not None
+ assert dashboard.slices, "fixture dashboard must have slices for this
test"
+ dashboard_id = dashboard.id
+ original_title = dashboard.dashboard_title
+
+ m2m = version_class(Dashboard).__table__.metadata.tables[
+ "dashboard_slices_version"
+ ]
+
+ def _live_m2m_count() -> int:
+ with _db.engine.begin() as conn:
+ return conn.execute(
+ sa.select(sa.func.count())
+ .select_from(m2m)
+ .where(m2m.c.dashboard_id == dashboard_id)
+ .where(m2m.c.end_transaction_id.is_(None))
+ ).scalar_one()
+
+ try:
+ # Baseline capture synthesizes the live M2M association rows at
+ # the baseline tx. A parent-only edit then opens a newer parent
+ # live row while the association rows stay live at the older tx.
+ dashboard.dashboard_title = "USA Births Names (parent-only edit)"
+ db.session.commit()
+
+ live_before = _live_m2m_count()
+ assert live_before >= 1, (
+ "expected live dashboard_slices_version rows before prune"
+ )
+
+ # Backdate every transaction so the whole chain is older than
+ # the window — including the tx anchoring the live association.
+ tx_table = versioning_manager.transaction_cls.__table__
+ with _db.engine.begin() as conn:
+ conn.execute(
+ sa.update(tx_table).values(
+ issued_at=datetime.utcnow() - timedelta(days=100)
+ )
+ )
+
+ _prune_old_versions_impl(retention_days=30)
+
+ # The live association rows must survive: their anchoring tx is
+ # preserved because they are live, even though it predates the
+ # dashboard's current live parent row.
+ assert _live_m2m_count() == live_before, (
+ "prune deleted live dashboard_slices_version rows — the "
+ "surviving version lost its slices (preservation BLOCKER)"
+ )
+ finally:
+ dashboard.dashboard_title = original_title
+ db.session.commit()
+
+ def test_retention_preserves_multi_flush_transaction(self) -> None:
+ """Pruning preserves #41940's final multi-flush semantic projection."""
+ from datetime import datetime, timedelta, timezone
+
+ import sqlalchemy as sa
+ from sqlalchemy_continuum import versioning_manager
+
+ from superset.tasks.version_history_retention import
_prune_old_versions_impl
+
+ _persist_fixture_state()
+ dashboard = (
+ db.session.query(Dashboard)
+ .filter(Dashboard.dashboard_title == "USA Births Names")
+ .one()
+ )
+ chart = dashboard.slices[0]
Review Comment:
**Suggestion:** Add a concrete type annotation for this derived local
variable so the test keeps explicit typing for relevant values. [custom_rule]
**Severity Level:** Minor 🧹
<details>
<summary><b>Why it matters? ⭐ </b></summary>
`chart` is a new local variable assigned from a model relationship and can
be explicitly annotated. Since the file is new Python code, omitting the
annotation violates the type-hint rule.
</details>
<details>
<summary><b>Rule source 📖 </b></summary>
.cursor/rules/dev-standard.mdc (line 28)
</details>
[](https://app.codeant.ai/fix-in-ide?tool=cursor&prompt_id=168509fbfc674e39ba975b78b3f4cdbb&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=168509fbfc674e39ba975b78b3f4cdbb&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:** tests/integration_tests/versioning/retention_prune_tests.py
**Line:** 273:273
**Comment:**
*Custom Rule: Add a concrete type annotation for this derived local
variable so the test keeps explicit typing for relevant values.
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%2F41075&comment_hash=688dfb9aba43e65c017cbb64bfaec0cf1c91e2aabb7b5d2d142ab51633cafa2b&reaction=like'>👍</a>
| <a
href='https://app.codeant.ai/feedback?pr_url=https%3A%2F%2Fgithub.com%2Fapache%2Fsuperset%2Fpull%2F41075&comment_hash=688dfb9aba43e65c017cbb64bfaec0cf1c91e2aabb7b5d2d142ab51633cafa2b&reaction=dislike'>👎</a>
##########
tests/integration_tests/versioning/retention_prune_tests.py:
##########
@@ -0,0 +1,488 @@
+# 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.
+"""Integration coverage for version-history retention pruning.
+
+Exercises ``superset.tasks.version_history_retention`` end-to-end against
+a real database: that an aged-out transaction's shadow rows are pruned
+while the live row is always preserved, and that the SERIALIZABLE pass
+retries on transient serialization failures and gives up after the cap.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+import pytest
+from sqlalchemy_continuum import version_class
+
+from superset.extensions import db
+from superset.models.dashboard import Dashboard
+from superset.models.slice import Slice
+from superset.versioning.changes.table import version_changes_table
+from tests.integration_tests.base_tests import SupersetTestCase
+from tests.integration_tests.fixtures.birth_names_dashboard import ( # noqa:
F401
+ load_birth_names_dashboard_with_slices,
+ load_birth_names_data,
+)
+
+
+def _get_version_rows(dashboard: Dashboard) -> list[Any]:
+ ver_cls = version_class(Dashboard)
+ return (
+ db.session.query(ver_cls)
+ .filter(ver_cls.id == dashboard.id)
+ .order_by(ver_cls.transaction_id.asc())
+ .all()
+ )
+
+
+def _persist_fixture_state() -> None:
+ """Force the fixture's pending INSERTs to commit in their own transaction.
+
+ The birth_names fixture stages charts and the dashboard via session.add()
+ but does not commit. Without this, the test's first commit batches the
+ INSERTs and UPDATEs into the same Continuum transaction, causing the
+ existing version row to be updated in place instead of a new one being
+ created.
+ """
+ db.session.commit()
+
+
+class TestDashboardVersionRetention(SupersetTestCase):
+ """Retention pruning drops shadow rows older than
+ ``SUPERSET_VERSION_HISTORY_RETENTION_DAYS`` while preserving live rows."""
+
+ @pytest.fixture(autouse=True)
+ def _load_data(
+ self,
+ load_birth_names_dashboard_with_slices: None, # noqa: F811
+ ) -> None: # noqa: PT004
+ """Establish the capture state this integration class requires.
+
+ Continuum stores its option and SQLAlchemy event listeners globally.
+ Initialization unit tests intentionally exercise the kill-switch that
+ detaches those listeners, so a mixed or reordered pytest invocation can
+ otherwise enter these tests with capture disabled. Reasserting the
+ integration suite's prerequisite here makes each test
order-independent.
+ """
+ from sqlalchemy_continuum import versioning_manager
+
+ from superset.initialization import SupersetAppInitializer
+
+ versioning_manager.options["versioning"] = True
+ SupersetAppInitializer._add_continuum_write_listeners()
+
+ def test_retention_prunes_old_rows(self) -> None:
+ """``prune_old_versions`` removes shadow rows whose owning
+ ``version_transaction.issued_at`` is older than the retention
+ window, while preserving every transaction that anchors a live row."""
+ from datetime import datetime, timedelta
+
+ import sqlalchemy as sa
+
+ from superset.extensions import db as _db
+ from superset.tasks.version_history_retention import (
+ _prune_old_versions_impl,
+ )
+
+ _persist_fixture_state()
+ dashboard: Dashboard = (
+ db.session.query(Dashboard)
+ .filter(Dashboard.dashboard_title == "USA Births Names")
+ .first()
+ )
+ assert dashboard is not None
+
+ original_title = dashboard.dashboard_title
+
+ try:
+ # Force a few saves so we have ≥ 2 closed shadow rows plus
+ # a baseline plus the live row.
+ for i in range(3):
+ dashboard.dashboard_title = f"USA Births Names retention test
{i}"
+ db.session.commit()
+
+ rows_before = _get_version_rows(dashboard)
+ assert len(rows_before) >= 3, "Expected at least 3 version rows"
+
+ def _version_changes_count() -> int:
+ return (
+ _db.session.execute(
+ sa.text("SELECT COUNT(*) FROM version_changes")
+ ).scalar()
+ or 0
+ )
+
+ changes_before = _version_changes_count()
+ assert changes_before >= 1, (
+ "expected version_changes rows from the edits above"
+ )
+
+ # Backdate every version_transaction row by 100 days so the
+ # prune sees them as old. Skip baseline+live rows; the prune
+ # itself preserves them.
+ from sqlalchemy_continuum import versioning_manager
+
+ tx_table = versioning_manager.transaction_cls.__table__
+ with _db.engine.begin() as conn:
+ conn.execute(
+ sa.update(tx_table).values(
+ issued_at=datetime.utcnow() - timedelta(days=100)
+ )
+ )
+
+ stats = _prune_old_versions_impl(retention_days=30)
+ assert stats.get("pruned_transactions", 0) >= 1, stats
+
+ # version_changes rows for pruned transactions must be gone too.
+ # The prune deletes version_transaction rows and relies on the
+ # ON DELETE CASCADE FK from version_changes.transaction_id; assert
+ # the cascade actually fired rather than orphaning change records.
+ db.session.expire_all()
+ assert _version_changes_count() < changes_before, (
+ "version_changes rows for pruned transactions were not removed
"
+ f"(before={changes_before}, after={_version_changes_count()})"
+ )
+
+ rows_after = _get_version_rows(dashboard)
+ # Live row must still exist (this is the only preservation rule)
+ live_rows = [r for r in rows_after if r.end_transaction_id is None]
+ assert len(live_rows) >= 1, "Live row must never be pruned"
+ # Some rows should have been pruned. Closed historical rows —
+ # including the synthetic baseline (operation_type=0) — are
+ # subject to retention like everything else.
+ assert len(rows_after) < len(rows_before), (
+ f"Expected fewer rows after prune; before={len(rows_before)} "
+ f"after={len(rows_after)}"
+ )
+
+ finally:
+ dashboard.dashboard_title = original_title
+ db.session.commit()
+
+ def test_retention_preserves_live_child_and_m2m_rows(self) -> None:
+ """Regression for the live-row preservation BLOCKER.
+
+ A parent-only edit leaves the dashboard's chart associations (the
+ M2M ``dashboard_slices_version`` rows) live but anchored at an
+ *older* transaction than the dashboard's current live row. The
+ prune must preserve that older transaction too — if it scans only
+ parent shadows for live rows, it deletes the still-live
+ association and the surviving version silently loses its slices.
+
+ The parent-only ``test_retention_prunes_old_rows`` cannot catch
+ this: it asserts on the dashboard parent shadow alone, which is
+ exactly the blind spot the bug lived in.
+ """
+ from datetime import datetime, timedelta
+
+ import sqlalchemy as sa
+ from sqlalchemy_continuum import version_class, versioning_manager
+
+ from superset.extensions import db as _db
+ from superset.tasks.version_history_retention import
_prune_old_versions_impl
+
+ _persist_fixture_state()
+ dashboard: Dashboard = (
+ db.session.query(Dashboard)
+ .filter(Dashboard.dashboard_title == "USA Births Names")
+ .first()
+ )
+ assert dashboard is not None
+ assert dashboard.slices, "fixture dashboard must have slices for this
test"
+ dashboard_id = dashboard.id
+ original_title = dashboard.dashboard_title
+
+ m2m = version_class(Dashboard).__table__.metadata.tables[
+ "dashboard_slices_version"
+ ]
+
+ def _live_m2m_count() -> int:
+ with _db.engine.begin() as conn:
+ return conn.execute(
+ sa.select(sa.func.count())
+ .select_from(m2m)
+ .where(m2m.c.dashboard_id == dashboard_id)
+ .where(m2m.c.end_transaction_id.is_(None))
+ ).scalar_one()
+
+ try:
+ # Baseline capture synthesizes the live M2M association rows at
+ # the baseline tx. A parent-only edit then opens a newer parent
+ # live row while the association rows stay live at the older tx.
+ dashboard.dashboard_title = "USA Births Names (parent-only edit)"
+ db.session.commit()
+
+ live_before = _live_m2m_count()
+ assert live_before >= 1, (
+ "expected live dashboard_slices_version rows before prune"
+ )
+
+ # Backdate every transaction so the whole chain is older than
+ # the window — including the tx anchoring the live association.
+ tx_table = versioning_manager.transaction_cls.__table__
+ with _db.engine.begin() as conn:
+ conn.execute(
+ sa.update(tx_table).values(
+ issued_at=datetime.utcnow() - timedelta(days=100)
+ )
+ )
+
+ _prune_old_versions_impl(retention_days=30)
+
+ # The live association rows must survive: their anchoring tx is
+ # preserved because they are live, even though it predates the
+ # dashboard's current live parent row.
+ assert _live_m2m_count() == live_before, (
+ "prune deleted live dashboard_slices_version rows — the "
+ "surviving version lost its slices (preservation BLOCKER)"
+ )
+ finally:
+ dashboard.dashboard_title = original_title
+ db.session.commit()
+
+ def test_retention_preserves_multi_flush_transaction(self) -> None:
+ """Pruning preserves #41940's final multi-flush semantic projection."""
+ from datetime import datetime, timedelta, timezone
+
+ import sqlalchemy as sa
+ from sqlalchemy_continuum import versioning_manager
+
+ from superset.tasks.version_history_retention import
_prune_old_versions_impl
+
+ _persist_fixture_state()
+ dashboard = (
+ db.session.query(Dashboard)
+ .filter(Dashboard.dashboard_title == "USA Births Names")
+ .one()
+ )
Review Comment:
**Suggestion:** Add an explicit type annotation for this local variable to
satisfy the typing rule for relevant variables in new Python code. [custom_rule]
**Severity Level:** Minor 🧹
<details>
<summary><b>Why it matters? ⭐ </b></summary>
This is new Python code and `dashboard` is a clearly type-annotatable local
variable holding a `Dashboard` instance. The rule requires type hints for
relevant variables, so the omission is a real violation.
</details>
<details>
<summary><b>Rule source 📖 </b></summary>
.cursor/rules/dev-standard.mdc (line 28)
</details>
[](https://app.codeant.ai/fix-in-ide?tool=cursor&prompt_id=b4db4b0f12be4637b3a5080d3889307f&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=b4db4b0f12be4637b3a5080d3889307f&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:** tests/integration_tests/versioning/retention_prune_tests.py
**Line:** 268:272
**Comment:**
*Custom Rule: Add an explicit type annotation for this local variable
to satisfy the typing rule for relevant variables in new Python code.
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%2F41075&comment_hash=69bce5715be0660c6b339bd6d9ca1f564a09933d02733383c66c6a5854aca220&reaction=like'>👍</a>
| <a
href='https://app.codeant.ai/feedback?pr_url=https%3A%2F%2Fgithub.com%2Fapache%2Fsuperset%2Fpull%2F41075&comment_hash=69bce5715be0660c6b339bd6d9ca1f564a09933d02733383c66c6a5854aca220&reaction=dislike'>👎</a>
##########
tests/integration_tests/versioning/retention_prune_tests.py:
##########
@@ -0,0 +1,488 @@
+# 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.
+"""Integration coverage for version-history retention pruning.
+
+Exercises ``superset.tasks.version_history_retention`` end-to-end against
+a real database: that an aged-out transaction's shadow rows are pruned
+while the live row is always preserved, and that the SERIALIZABLE pass
+retries on transient serialization failures and gives up after the cap.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+import pytest
+from sqlalchemy_continuum import version_class
+
+from superset.extensions import db
+from superset.models.dashboard import Dashboard
+from superset.models.slice import Slice
+from superset.versioning.changes.table import version_changes_table
+from tests.integration_tests.base_tests import SupersetTestCase
+from tests.integration_tests.fixtures.birth_names_dashboard import ( # noqa:
F401
+ load_birth_names_dashboard_with_slices,
+ load_birth_names_data,
+)
+
+
+def _get_version_rows(dashboard: Dashboard) -> list[Any]:
+ ver_cls = version_class(Dashboard)
+ return (
+ db.session.query(ver_cls)
+ .filter(ver_cls.id == dashboard.id)
+ .order_by(ver_cls.transaction_id.asc())
+ .all()
+ )
+
+
+def _persist_fixture_state() -> None:
+ """Force the fixture's pending INSERTs to commit in their own transaction.
+
+ The birth_names fixture stages charts and the dashboard via session.add()
+ but does not commit. Without this, the test's first commit batches the
+ INSERTs and UPDATEs into the same Continuum transaction, causing the
+ existing version row to be updated in place instead of a new one being
+ created.
+ """
+ db.session.commit()
+
+
+class TestDashboardVersionRetention(SupersetTestCase):
+ """Retention pruning drops shadow rows older than
+ ``SUPERSET_VERSION_HISTORY_RETENTION_DAYS`` while preserving live rows."""
+
+ @pytest.fixture(autouse=True)
+ def _load_data(
+ self,
+ load_birth_names_dashboard_with_slices: None, # noqa: F811
+ ) -> None: # noqa: PT004
+ """Establish the capture state this integration class requires.
+
+ Continuum stores its option and SQLAlchemy event listeners globally.
+ Initialization unit tests intentionally exercise the kill-switch that
+ detaches those listeners, so a mixed or reordered pytest invocation can
+ otherwise enter these tests with capture disabled. Reasserting the
+ integration suite's prerequisite here makes each test
order-independent.
+ """
+ from sqlalchemy_continuum import versioning_manager
+
+ from superset.initialization import SupersetAppInitializer
+
+ versioning_manager.options["versioning"] = True
+ SupersetAppInitializer._add_continuum_write_listeners()
+
+ def test_retention_prunes_old_rows(self) -> None:
+ """``prune_old_versions`` removes shadow rows whose owning
+ ``version_transaction.issued_at`` is older than the retention
+ window, while preserving every transaction that anchors a live row."""
+ from datetime import datetime, timedelta
+
+ import sqlalchemy as sa
+
+ from superset.extensions import db as _db
+ from superset.tasks.version_history_retention import (
+ _prune_old_versions_impl,
+ )
+
+ _persist_fixture_state()
+ dashboard: Dashboard = (
+ db.session.query(Dashboard)
+ .filter(Dashboard.dashboard_title == "USA Births Names")
+ .first()
+ )
+ assert dashboard is not None
+
+ original_title = dashboard.dashboard_title
+
+ try:
+ # Force a few saves so we have ≥ 2 closed shadow rows plus
+ # a baseline plus the live row.
+ for i in range(3):
+ dashboard.dashboard_title = f"USA Births Names retention test
{i}"
+ db.session.commit()
+
+ rows_before = _get_version_rows(dashboard)
+ assert len(rows_before) >= 3, "Expected at least 3 version rows"
+
+ def _version_changes_count() -> int:
+ return (
+ _db.session.execute(
+ sa.text("SELECT COUNT(*) FROM version_changes")
+ ).scalar()
+ or 0
+ )
+
+ changes_before = _version_changes_count()
+ assert changes_before >= 1, (
+ "expected version_changes rows from the edits above"
+ )
+
+ # Backdate every version_transaction row by 100 days so the
+ # prune sees them as old. Skip baseline+live rows; the prune
+ # itself preserves them.
+ from sqlalchemy_continuum import versioning_manager
+
+ tx_table = versioning_manager.transaction_cls.__table__
+ with _db.engine.begin() as conn:
+ conn.execute(
+ sa.update(tx_table).values(
+ issued_at=datetime.utcnow() - timedelta(days=100)
+ )
+ )
+
+ stats = _prune_old_versions_impl(retention_days=30)
+ assert stats.get("pruned_transactions", 0) >= 1, stats
+
+ # version_changes rows for pruned transactions must be gone too.
+ # The prune deletes version_transaction rows and relies on the
+ # ON DELETE CASCADE FK from version_changes.transaction_id; assert
+ # the cascade actually fired rather than orphaning change records.
+ db.session.expire_all()
+ assert _version_changes_count() < changes_before, (
+ "version_changes rows for pruned transactions were not removed
"
+ f"(before={changes_before}, after={_version_changes_count()})"
+ )
+
+ rows_after = _get_version_rows(dashboard)
+ # Live row must still exist (this is the only preservation rule)
+ live_rows = [r for r in rows_after if r.end_transaction_id is None]
+ assert len(live_rows) >= 1, "Live row must never be pruned"
+ # Some rows should have been pruned. Closed historical rows —
+ # including the synthetic baseline (operation_type=0) — are
+ # subject to retention like everything else.
+ assert len(rows_after) < len(rows_before), (
+ f"Expected fewer rows after prune; before={len(rows_before)} "
+ f"after={len(rows_after)}"
+ )
+
+ finally:
+ dashboard.dashboard_title = original_title
+ db.session.commit()
+
+ def test_retention_preserves_live_child_and_m2m_rows(self) -> None:
+ """Regression for the live-row preservation BLOCKER.
+
+ A parent-only edit leaves the dashboard's chart associations (the
+ M2M ``dashboard_slices_version`` rows) live but anchored at an
+ *older* transaction than the dashboard's current live row. The
+ prune must preserve that older transaction too — if it scans only
+ parent shadows for live rows, it deletes the still-live
+ association and the surviving version silently loses its slices.
+
+ The parent-only ``test_retention_prunes_old_rows`` cannot catch
+ this: it asserts on the dashboard parent shadow alone, which is
+ exactly the blind spot the bug lived in.
+ """
+ from datetime import datetime, timedelta
+
+ import sqlalchemy as sa
+ from sqlalchemy_continuum import version_class, versioning_manager
+
+ from superset.extensions import db as _db
+ from superset.tasks.version_history_retention import
_prune_old_versions_impl
+
+ _persist_fixture_state()
+ dashboard: Dashboard = (
+ db.session.query(Dashboard)
+ .filter(Dashboard.dashboard_title == "USA Births Names")
+ .first()
+ )
+ assert dashboard is not None
+ assert dashboard.slices, "fixture dashboard must have slices for this
test"
+ dashboard_id = dashboard.id
+ original_title = dashboard.dashboard_title
+
+ m2m = version_class(Dashboard).__table__.metadata.tables[
+ "dashboard_slices_version"
+ ]
+
+ def _live_m2m_count() -> int:
+ with _db.engine.begin() as conn:
+ return conn.execute(
+ sa.select(sa.func.count())
+ .select_from(m2m)
+ .where(m2m.c.dashboard_id == dashboard_id)
+ .where(m2m.c.end_transaction_id.is_(None))
+ ).scalar_one()
+
+ try:
+ # Baseline capture synthesizes the live M2M association rows at
+ # the baseline tx. A parent-only edit then opens a newer parent
+ # live row while the association rows stay live at the older tx.
+ dashboard.dashboard_title = "USA Births Names (parent-only edit)"
+ db.session.commit()
+
+ live_before = _live_m2m_count()
+ assert live_before >= 1, (
+ "expected live dashboard_slices_version rows before prune"
+ )
+
+ # Backdate every transaction so the whole chain is older than
+ # the window — including the tx anchoring the live association.
+ tx_table = versioning_manager.transaction_cls.__table__
+ with _db.engine.begin() as conn:
+ conn.execute(
+ sa.update(tx_table).values(
+ issued_at=datetime.utcnow() - timedelta(days=100)
+ )
+ )
+
+ _prune_old_versions_impl(retention_days=30)
+
+ # The live association rows must survive: their anchoring tx is
+ # preserved because they are live, even though it predates the
+ # dashboard's current live parent row.
+ assert _live_m2m_count() == live_before, (
+ "prune deleted live dashboard_slices_version rows — the "
+ "surviving version lost its slices (preservation BLOCKER)"
+ )
+ finally:
+ dashboard.dashboard_title = original_title
+ db.session.commit()
+
+ def test_retention_preserves_multi_flush_transaction(self) -> None:
+ """Pruning preserves #41940's final multi-flush semantic projection."""
+ from datetime import datetime, timedelta, timezone
+
+ import sqlalchemy as sa
+ from sqlalchemy_continuum import versioning_manager
+
+ from superset.tasks.version_history_retention import
_prune_old_versions_impl
+
+ _persist_fixture_state()
+ dashboard = (
+ db.session.query(Dashboard)
+ .filter(Dashboard.dashboard_title == "USA Births Names")
+ .one()
+ )
+ chart = dashboard.slices[0]
+ dashboard_title = dashboard.dashboard_title
+ chart_name = chart.slice_name
+ tx_table = versioning_manager.transaction_cls.__table__
+ dashboard_version_table = version_class(Dashboard).__table__
+ chart_version_table = version_class(Slice).__table__
+ boundary = db.session.scalar(sa.select(sa.func.max(tx_table.c.id))) or 0
+
+ try:
+ dashboard.dashboard_title = "USA Births Names multi-flush
dashboard"
+ db.session.flush()
+ chart.slice_name = "USA Births Names multi-flush chart"
+ db.session.commit()
+
+ change_rows = db.session.execute(
+ sa.select(
+ version_changes_table.c.transaction_id,
+ version_changes_table.c.entity_kind,
+ )
+ .where(version_changes_table.c.transaction_id > boundary)
+ .where(
+ sa.or_(
+ sa.and_(
+ version_changes_table.c.entity_kind == "dashboard",
+ version_changes_table.c.entity_id == dashboard.id,
+ ),
+ sa.and_(
+ version_changes_table.c.entity_kind == "chart",
+ version_changes_table.c.entity_id == chart.id,
+ ),
+ )
+ )
+ ).all()
+ assert {row.entity_kind for row in change_rows} == {"dashboard",
"chart"}
+ transaction_ids = {row.transaction_id for row in change_rows}
+ assert len(transaction_ids) == 1
+ transaction_id = transaction_ids.pop()
+ assert (
+ db.session.scalar(
+ sa.select(dashboard_version_table.c.transaction_id)
+ .where(dashboard_version_table.c.id == dashboard.id)
+
.where(dashboard_version_table.c.end_transaction_id.is_(None))
+ )
+ == transaction_id
+ )
+ assert (
+ db.session.scalar(
+ sa.select(chart_version_table.c.transaction_id)
+ .where(chart_version_table.c.id == chart.id)
+ .where(chart_version_table.c.end_transaction_id.is_(None))
+ )
+ == transaction_id
+ )
+
+ with db.engine.begin() as conn:
+ conn.execute(
+ sa.update(tx_table).values(
+
issued_at=datetime.now(timezone.utc).replace(tzinfo=None)
+ - timedelta(days=100)
+ )
+ )
+
+ _prune_old_versions_impl(retention_days=30)
+ db.session.rollback()
+
+ assert (
+ db.session.scalar(
+ sa.select(sa.func.count())
+ .select_from(tx_table)
+ .where(tx_table.c.id == transaction_id)
+ )
+ == 1
+ )
+ surviving_kinds = set(
+ db.session.scalars(
+ sa.select(version_changes_table.c.entity_kind).where(
+ version_changes_table.c.transaction_id ==
transaction_id
+ )
+ )
+ )
+ assert {"dashboard", "chart"}.issubset(surviving_kinds)
+ assert (
+ db.session.scalar(
+ sa.select(sa.func.count())
+ .select_from(dashboard_version_table)
+ .where(dashboard_version_table.c.id == dashboard.id)
+
.where(dashboard_version_table.c.end_transaction_id.is_(None))
+ )
+ == 1
+ )
+ assert (
+ db.session.scalar(
+ sa.select(sa.func.count())
+ .select_from(chart_version_table)
+ .where(chart_version_table.c.id == chart.id)
+ .where(chart_version_table.c.end_transaction_id.is_(None))
+ )
+ == 1
+ )
+ finally:
+ dashboard.dashboard_title = dashboard_title
+ chart.slice_name = chart_name
+ db.session.commit()
+
+ def test_retention_retries_on_serialization_failure(self) -> None:
+ """A transient ``OperationalError`` from the SERIALIZABLE pass
+ triggers an inline retry; the prune completes on the second
+ attempt and the stats dict records the retry count."""
+ from datetime import datetime, timedelta
+ from unittest.mock import patch
+
+ import sqlalchemy as sa
+ from sqlalchemy.exc import OperationalError
+
+ from superset.tasks import version_history_retention
+ from superset.tasks.version_history_retention import (
+ _prune_old_versions_impl,
+ )
+
+ # Backdate transactions so the prune has work to do.
+ _persist_fixture_state()
+ dashboard: Dashboard = (
+ db.session.query(Dashboard)
+ .filter(Dashboard.dashboard_title == "USA Births Names")
+ .first()
+ )
+ original_title = dashboard.dashboard_title
+ try:
+ for i in range(3):
+ dashboard.dashboard_title = f"USA Births Names retry test {i}"
+ db.session.commit()
+
+ from sqlalchemy_continuum import versioning_manager
+
+ tx_table = versioning_manager.transaction_cls.__table__
+ from superset.extensions import db as _db
+
+ with _db.engine.begin() as conn:
+ conn.execute(
+ sa.update(tx_table).values(
+ issued_at=datetime.utcnow() - timedelta(days=100)
+ )
+ )
+
+ original_run = version_history_retention._run_prune_pass
+ calls: list[int] = []
+
+ def flaky_run(*args: Any, **kwargs: Any) -> dict[str, Any]:
+ calls.append(1)
+ if len(calls) == 1:
+ raise OperationalError(
+ "SELECT 1", {}, Exception("could not serialize access")
+ )
+ return original_run(*args, **kwargs)
+
+ with patch.object(
+ version_history_retention, "_run_prune_pass",
side_effect=flaky_run
+ ):
+ # Patch sleep so the test doesn't actually wait through
+ # the backoff.
+ with patch.object(version_history_retention.time, "sleep"):
+ stats = _prune_old_versions_impl(retention_days=30)
+
+ # At least 2 passes: the first conflicts (call 1) and is
+ # retried (call 2). There may be more when the backlog spans
+ # multiple id-ordered windows (the prune batches by
+ # _MAX_PRUNE_BATCH), but exactly one conflict was injected, so
+ # the retry counter must read 1 regardless of batch count.
+ assert len(calls) >= 2, (
+ f"Expected >= 2 _run_prune_pass calls (>= 1 failure + retry), "
+ f"got {len(calls)}"
+ )
+ assert stats.get("retried") == 1, stats
+ assert stats.get("pruned_transactions", 0) >= 1, stats
+ finally:
+ dashboard.dashboard_title = original_title
+ db.session.commit()
+
+ def test_retention_gives_up_after_max_attempts(self) -> None:
+ """When every attempt hits ``OperationalError``, the function
+ re-raises after the retry cap so the outer Celery wrapper logs
+ + returns ``{"error": 1}``."""
+ from unittest.mock import patch
+
+ from sqlalchemy.exc import OperationalError
+
+ from superset.tasks import version_history_retention
+ from superset.tasks.version_history_retention import (
+ _MAX_RETRY_ATTEMPTS,
+ _prune_old_versions_impl,
+ )
+
+ def always_fail(*args: Any, **kwargs: Any) -> dict[str, Any]:
+ raise OperationalError(
+ "SELECT 1", {}, Exception("could not serialize access")
+ )
+
+ call_count = 0
Review Comment:
**Suggestion:** Add an explicit integer type annotation to this counter
variable used across nested functions. [custom_rule]
**Severity Level:** Minor 🧹
<details>
<summary><b>Why it matters? ⭐ </b></summary>
`call_count` is a mutable counter used across nested functions and is
straightforward to annotate as `int`. Its omission fits the custom rule
requiring type hints for relevant variables in new Python code.
</details>
<details>
<summary><b>Rule source 📖 </b></summary>
.cursor/rules/dev-standard.mdc (line 28)
</details>
[](https://app.codeant.ai/fix-in-ide?tool=cursor&prompt_id=d0dc003a43a34ded829b0a312971cebd&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=d0dc003a43a34ded829b0a312971cebd&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:** tests/integration_tests/versioning/retention_prune_tests.py
**Line:** 470:470
**Comment:**
*Custom Rule: Add an explicit integer type annotation to this counter
variable used across nested functions.
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%2F41075&comment_hash=ab1deae71ff9849c235da3b14256b0f23c180caf0aba34b4317b66c9b13e508d&reaction=like'>👍</a>
| <a
href='https://app.codeant.ai/feedback?pr_url=https%3A%2F%2Fgithub.com%2Fapache%2Fsuperset%2Fpull%2F41075&comment_hash=ab1deae71ff9849c235da3b14256b0f23c180caf0aba34b4317b66c9b13e508d&reaction=dislike'>👎</a>
##########
superset/migrations/versions/2026-06-15_16-30_d3b9a1f6c204_version_transaction_issued_at_index.py:
##########
@@ -0,0 +1,48 @@
+# 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.
+"""Add an index on version_transaction.issued_at.
+
+The version-history retention prune selects candidates using
+``issued_at < cutoff`` and returns them in primary-key order. PostgreSQL 17
+and MySQL 8 planner checks with 500,000 transactions showed that the primary
+key remains optimal when expired rows exist near the low-id end. When no rows
+meet the cutoff, however, the primary-key plan scans the entire table. Both
+engines choose this index for that case, reducing it to an empty range scan,
+while retaining the primary-key plan for a populated backlog.
+
+Revision ID: d3b9a1f6c204
+Revises: 8f3a1b2c4d5e
+Create Date: 2026-06-15 16:30:00.000000
+"""
+
+from superset.migrations.shared.utils import create_index, drop_index
+
+revision: str = "d3b9a1f6c204"
+down_revision: str = "8f3a1b2c4d5e"
+
+INDEX_NAME = "ix_version_transaction_issued_at"
+TABLE_NAME = "version_transaction"
Review Comment:
**Suggestion:** Add explicit string type annotations to the new module-level
constants to keep new migration variables fully typed. [custom_rule]
**Severity Level:** Minor 🧹
<details>
<summary><b>Why it matters? ⭐ </b></summary>
The module introduces new constant variables that can be annotated, but they
are written without type hints. This matches the rule requiring type hints on
relevant Python variables in new or modified code.
</details>
<details>
<summary><b>Rule source 📖 </b></summary>
.cursor/rules/dev-standard.mdc (line 28)
</details>
[](https://app.codeant.ai/fix-in-ide?tool=cursor&prompt_id=5a201f51450d45a7a98f0de2d865d005&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=5a201f51450d45a7a98f0de2d865d005&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/migrations/versions/2026-06-15_16-30_d3b9a1f6c204_version_transaction_issued_at_index.py
**Line:** 37:38
**Comment:**
*Custom Rule: Add explicit string type annotations to the new
module-level constants to keep new migration variables fully typed.
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%2F41075&comment_hash=0b7c9637b81e1b9c2eadb3e4b0a7a18279b16bbbbefc9dcd5a84594fb5a9fe47&reaction=like'>👍</a>
| <a
href='https://app.codeant.ai/feedback?pr_url=https%3A%2F%2Fgithub.com%2Fapache%2Fsuperset%2Fpull%2F41075&comment_hash=0b7c9637b81e1b9c2eadb3e4b0a7a18279b16bbbbefc9dcd5a84594fb5a9fe47&reaction=dislike'>👎</a>
##########
superset/tasks/version_history_retention.py:
##########
@@ -0,0 +1,548 @@
+# 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.
+"""Celery task: prune old entity-version history.
+
+Retention is time-based. The task deletes parent + child shadow rows
+owned by ``version_transaction`` rows whose ``issued_at`` is older
+than ``SUPERSET_VERSION_HISTORY_RETENTION_DAYS`` (default 30, env
+overridable, non-positive to disable).
+
+One preservation rule, applied across every shadow table (parent,
+child, and the M2M association):
+
+* **Live** (``end_transaction_id IS NULL``) — never pruned.
+
+Baseline rows (``operation_type = 0``) and any closed historical row
+are subject to the same retention window as everything else. An
+entity that hasn't been edited within the window has only its live
+row remaining; the historical chain (including the synthetic
+baseline) ages out.
+
+If any shadow row anchored at a transaction is live (in a parent,
+child, or M2M shadow), the ``version_transaction`` and its
+``version_changes`` rows are preserved. Closed shadow rows whose lifecycle
+touches a pruned transaction are removed so their foreign keys cannot retain
+otherwise-expired history. Every other prunable transaction is dropped, and
+its ``version_changes`` rows cascade via the FK.
+
+Registered via ``CeleryConfig.beat_schedule`` in ``superset/config.py``.
+Idempotent: a second run prunes nothing.
+"""
+
+from __future__ import annotations
+
+import logging
+import time
+from collections.abc import Iterator
+from dataclasses import dataclass
+from datetime import datetime, timedelta, timezone
+from typing import Any
+
+import sqlalchemy as sa
+from flask import current_app
+from sqlalchemy.exc import OperationalError
+
+from superset.extensions import celery_app, db, stats_logger_manager
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass(frozen=True)
+class ShadowTables:
+ """The four Continuum-managed Table objects the prune walks.
+
+ Bundled here so the prune helper's signature stays at two arguments
+ instead of five. The shape is set once at task entry by
+ ``_resolve_shadow_tables`` and threaded through the retry loop.
+ """
+
+ parent: list[sa.Table]
+ child: list[sa.Table]
+ m2m: sa.Table | None
+ transaction: sa.Table
+
+
+def _resolve_shadow_tables(tx_table: sa.Table) -> ShadowTables:
+ """Resolve the parent / child / m2m shadow Tables from Continuum's
+ mapper registry and bundle them with the transaction Table.
+
+ ``dashboard_slices_version`` is M2M-tracked by Continuum and lives
+ in metadata under that name (Continuum auto-creates the Table; it
+ isn't registered as a versioned class). Carried separately on the
+ ``ShadowTables`` dataclass because it doesn't follow the parent /
+ child class shape.
+ """
+ # pylint: disable=import-outside-toplevel
+ from sqlalchemy_continuum import version_class
+ from sqlalchemy_continuum.exc import ClassNotVersioned
+
+ from superset.connectors.sqla.models import SqlaTable, SqlMetric,
TableColumn
+ from superset.models.dashboard import Dashboard
+ from superset.models.slice import Slice
+
+ # ``ClassNotVersioned`` is the only expected failure here — versioning
+ # init runs at startup; if it didn't, every class lookup raises this.
+ # Narrowing the catch keeps a real underlying failure (e.g. a metadata
+ # inconsistency after ``make_versioned``) from being silently swallowed
+ # into a no-op retention pass.
+ missing_tables: list[str] = []
+ parent_tables: list[sa.Table] = []
+ for cls in (Dashboard, Slice, SqlaTable):
+ try:
+ parent_tables.append(version_class(cls).__table__)
+ except ClassNotVersioned:
+ missing_tables.append(cls.__name__)
+
+ child_tables: list[sa.Table] = []
+ for cls in (TableColumn, SqlMetric):
+ try:
+ child_tables.append(version_class(cls).__table__)
+ except ClassNotVersioned:
+ missing_tables.append(cls.__name__)
+
+ metadata = parent_tables[0].metadata if parent_tables else None
+ m2m_table = (
+ metadata.tables.get("dashboard_slices_version")
+ if metadata is not None
+ else None
+ )
+ if m2m_table is None:
+ missing_tables.append("dashboard_slices_version")
+
+ if missing_tables:
+ raise RuntimeError(
+ "version-history retention requires every shadow table; missing: "
+ + ", ".join(missing_tables)
+ )
+
+ return ShadowTables(
+ parent=parent_tables,
+ child=child_tables,
+ m2m=m2m_table,
+ transaction=tx_table,
+ )
+
+
+@dataclass(frozen=True)
+class _PruneWindow:
+ """One id-ordered window of prune candidates.
+
+ ``prunable`` is the subset of the window's candidate transactions
+ that are safe to delete. ``candidate_count`` and ``max_candidate_id``
+ drive the batch loop in :func:`_prune_old_versions_impl`: the loop
+ stops when a window returns fewer than ``_MAX_PRUNE_BATCH``
+ candidates, and otherwise advances ``after_id`` to
+ ``max_candidate_id`` to fetch the next window.
+ """
+
+ prunable: list[int]
+ candidate_count: int
+ max_candidate_id: int
+
+
+def _resolve_prune_window(
+ conn: sa.engine.Connection,
+ cutoff: datetime,
+ shadow_tables: list[sa.Table],
+ after_id: int,
+ batch_size: int,
+) -> _PruneWindow:
+ """Resolve one id-ordered window of ``version_transaction`` rows
+ eligible to prune: ``issued_at < cutoff`` AND ``id > after_id``,
+ ordered by ``id`` and capped at *batch_size*. From that window,
+ ``prunable`` excludes any transaction still serving as the live row
+ (``end_transaction_id IS NULL``) of any versioned entity in *any*
+ shadow table — parent, child, or M2M.
+
+ A child (``table_columns_version`` / ``sql_metrics_version``) or the
+ M2M association (``dashboard_slices_version``) is versioned on a
+ validity lifecycle independent of its parent: after a parent-only
+ edit, an unchanged child stays live anchored at an *older*
+ transaction than the parent's current live row. That older
+ transaction must be preserved too — otherwise
+ :func:`_delete_for_transactions` would drop the still-live
+ child/association row and silently strip the surviving version of its
+ columns / metrics / slices. Scanning every shadow for live rows (not
+ just parents) is what prevents that corruption.
+
+ Windowing by ``id`` (rather than materializing the whole backlog)
+ keeps per-pass memory and lock/transaction-hold time bounded. Live
+ rows older than the cutoff are never deleted but are skipped via the
+ ``after_id`` watermark, so the batch loop still terminates.
+ """
+ # pylint: disable=import-outside-toplevel
+ from sqlalchemy_continuum import versioning_manager
+
+ tx_table = versioning_manager.transaction_cls.__table__
+ # ``ORDER BY id LIMIT`` lets PostgreSQL and MySQL use the primary key for
+ # a populated backlog, where expired transactions cluster at the low-id
+ # end. The separate ``issued_at`` index added by migration d3b9a1f6c204
+ # covers the complementary case: when no rows meet the cutoff, both
+ # engines choose it for an empty range scan instead of walking the full
+ # primary key. Planner checks with 500,000 rows confirmed that adding the
+ # cutoff index does not displace the efficient primary-key backlog plan.
+ candidate_ids: list[int] = [
+ row[0]
+ for row in conn.execute(
+ sa.select(tx_table.c.id)
+ .where(tx_table.c.issued_at < cutoff)
+ .where(tx_table.c.id > after_id)
+ .order_by(tx_table.c.id)
+ .limit(batch_size)
+ )
+ ]
+ if not candidate_ids:
+ return _PruneWindow(prunable=[], candidate_count=0,
max_candidate_id=after_id)
+
+ # The select is ordered by ``id`` ascending, so the last element is
+ # the watermark for the next window.
+ max_candidate_id = candidate_ids[-1]
+
+ # Build the set of transaction ids that still anchor a live row
+ # (``end_transaction_id IS NULL``) in some shadow table. Those
+ # transactions represent the current state of an entity (or one of
+ # its children / associations) and must be preserved regardless of
+ # age. Chunked over the candidates to keep the bind-parameter count
+ # inside SQLite's ``SQLITE_MAX_VARIABLE_NUMBER`` floor (see
+ # ``_TX_ID_CHUNK_SIZE`` below). Ids already confirmed live by an
+ # earlier table are skipped before probing the next one.
+ preserved_ids: set[int] = set()
+ for stbl in shadow_tables:
+ remaining = [tx_id for tx_id in candidate_ids if tx_id not in
preserved_ids]
+ if not remaining:
+ break
+ for chunk in _chunked(remaining, _TX_ID_CHUNK_SIZE):
+ for row in conn.execute(
+ sa.select(stbl.c.transaction_id)
+ .where(stbl.c.transaction_id.in_(chunk))
+ .where(stbl.c.end_transaction_id.is_(None))
+ .distinct()
+ ):
+ preserved_ids.add(row[0])
+
+ prunable = [tx_id for tx_id in candidate_ids if tx_id not in preserved_ids]
+ return _PruneWindow(
+ prunable=prunable,
+ candidate_count=len(candidate_ids),
+ max_candidate_id=max_candidate_id,
+ )
+
+
+# SQLite's ``SQLITE_MAX_VARIABLE_NUMBER`` defaults to 999 (lifted to
+# 32766 in 3.32+ but the older limit can still apply in shipped
+# builds). Postgres + MySQL handle tens of thousands of bind params
+# without complaint, so the chunk size is dictated by the SQLite floor.
+# 500 keeps each single-column DELETE / SELECT (see
``_delete_for_transactions``,
+# which splits the create/close predicates into two statements) well inside
+# that floor, with margin for any other bound params in the surrounding
+# statement.
+_TX_ID_CHUNK_SIZE = 500
+
+# Maximum ``version_transaction`` rows resolved + pruned per SERIALIZABLE
+# pass. The prune loops over id-ordered windows of this size (see
+# ``_prune_old_versions_impl``) so memory and lock/transaction-hold time
+# stay bounded per pass instead of scaling with the full backlog on the
+# first run after deploy.
+_MAX_PRUNE_BATCH = 1000
Review Comment:
**Suggestion:** Add an explicit integer type annotation to this batch-size
constant. [custom_rule]
**Severity Level:** Minor 🧹
<details>
<summary><b>Why it matters? ⭐ </b></summary>
The rule covers relevant variables that can be annotated. This module-level
integer constant is untyped, so the suggestion identifies a real violation.
</details>
<details>
<summary><b>Rule source 📖 </b></summary>
.cursor/rules/dev-standard.mdc (line 28)
</details>
[](https://app.codeant.ai/fix-in-ide?tool=cursor&prompt_id=570769fd85fc464e976c78ce03b97900&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=570769fd85fc464e976c78ce03b97900&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/tasks/version_history_retention.py
**Line:** 260:260
**Comment:**
*Custom Rule: Add an explicit integer type annotation to this
batch-size constant.
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%2F41075&comment_hash=2a73a5d0c387eec494ee58b1bcd7e534c255bfc1357ca114ae304e104c7d7bc3&reaction=like'>👍</a>
| <a
href='https://app.codeant.ai/feedback?pr_url=https%3A%2F%2Fgithub.com%2Fapache%2Fsuperset%2Fpull%2F41075&comment_hash=2a73a5d0c387eec494ee58b1bcd7e534c255bfc1357ca114ae304e104c7d7bc3&reaction=dislike'>👎</a>
##########
superset/tasks/version_history_retention.py:
##########
@@ -0,0 +1,548 @@
+# 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.
+"""Celery task: prune old entity-version history.
+
+Retention is time-based. The task deletes parent + child shadow rows
+owned by ``version_transaction`` rows whose ``issued_at`` is older
+than ``SUPERSET_VERSION_HISTORY_RETENTION_DAYS`` (default 30, env
+overridable, non-positive to disable).
+
+One preservation rule, applied across every shadow table (parent,
+child, and the M2M association):
+
+* **Live** (``end_transaction_id IS NULL``) — never pruned.
+
+Baseline rows (``operation_type = 0``) and any closed historical row
+are subject to the same retention window as everything else. An
+entity that hasn't been edited within the window has only its live
+row remaining; the historical chain (including the synthetic
+baseline) ages out.
+
+If any shadow row anchored at a transaction is live (in a parent,
+child, or M2M shadow), the ``version_transaction`` and its
+``version_changes`` rows are preserved. Closed shadow rows whose lifecycle
+touches a pruned transaction are removed so their foreign keys cannot retain
+otherwise-expired history. Every other prunable transaction is dropped, and
+its ``version_changes`` rows cascade via the FK.
+
+Registered via ``CeleryConfig.beat_schedule`` in ``superset/config.py``.
+Idempotent: a second run prunes nothing.
+"""
+
+from __future__ import annotations
+
+import logging
+import time
+from collections.abc import Iterator
+from dataclasses import dataclass
+from datetime import datetime, timedelta, timezone
+from typing import Any
+
+import sqlalchemy as sa
+from flask import current_app
+from sqlalchemy.exc import OperationalError
+
+from superset.extensions import celery_app, db, stats_logger_manager
+
+logger = logging.getLogger(__name__)
Review Comment:
**Suggestion:** Add an explicit type annotation to this module-level logger
variable. [custom_rule]
**Severity Level:** Minor 🧹
<details>
<summary><b>Why it matters? ⭐ </b></summary>
The custom rule requires type hints for relevant variables that can be
annotated. This module-level logger is a clearly annotatable variable and is
currently untyped, so the suggestion matches a real violation.
</details>
<details>
<summary><b>Rule source 📖 </b></summary>
.cursor/rules/dev-standard.mdc (line 28)
</details>
[](https://app.codeant.ai/fix-in-ide?tool=cursor&prompt_id=fe603025093c43bb87814bfc4ea66986&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=fe603025093c43bb87814bfc4ea66986&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/tasks/version_history_retention.py
**Line:** 61:61
**Comment:**
*Custom Rule: Add an explicit type annotation to this module-level
logger variable.
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%2F41075&comment_hash=3163236aa169daccaca9d9cc4a0dfc6abe08332b7108c9a581ba32c11a7f1f24&reaction=like'>👍</a>
| <a
href='https://app.codeant.ai/feedback?pr_url=https%3A%2F%2Fgithub.com%2Fapache%2Fsuperset%2Fpull%2F41075&comment_hash=3163236aa169daccaca9d9cc4a0dfc6abe08332b7108c9a581ba32c11a7f1f24&reaction=dislike'>👎</a>
##########
superset/tasks/version_history_retention.py:
##########
@@ -0,0 +1,548 @@
+# 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.
+"""Celery task: prune old entity-version history.
+
+Retention is time-based. The task deletes parent + child shadow rows
+owned by ``version_transaction`` rows whose ``issued_at`` is older
+than ``SUPERSET_VERSION_HISTORY_RETENTION_DAYS`` (default 30, env
+overridable, non-positive to disable).
+
+One preservation rule, applied across every shadow table (parent,
+child, and the M2M association):
+
+* **Live** (``end_transaction_id IS NULL``) — never pruned.
+
+Baseline rows (``operation_type = 0``) and any closed historical row
+are subject to the same retention window as everything else. An
+entity that hasn't been edited within the window has only its live
+row remaining; the historical chain (including the synthetic
+baseline) ages out.
+
+If any shadow row anchored at a transaction is live (in a parent,
+child, or M2M shadow), the ``version_transaction`` and its
+``version_changes`` rows are preserved. Closed shadow rows whose lifecycle
+touches a pruned transaction are removed so their foreign keys cannot retain
+otherwise-expired history. Every other prunable transaction is dropped, and
+its ``version_changes`` rows cascade via the FK.
+
+Registered via ``CeleryConfig.beat_schedule`` in ``superset/config.py``.
+Idempotent: a second run prunes nothing.
+"""
+
+from __future__ import annotations
+
+import logging
+import time
+from collections.abc import Iterator
+from dataclasses import dataclass
+from datetime import datetime, timedelta, timezone
+from typing import Any
+
+import sqlalchemy as sa
+from flask import current_app
+from sqlalchemy.exc import OperationalError
+
+from superset.extensions import celery_app, db, stats_logger_manager
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass(frozen=True)
+class ShadowTables:
+ """The four Continuum-managed Table objects the prune walks.
+
+ Bundled here so the prune helper's signature stays at two arguments
+ instead of five. The shape is set once at task entry by
+ ``_resolve_shadow_tables`` and threaded through the retry loop.
+ """
+
+ parent: list[sa.Table]
+ child: list[sa.Table]
+ m2m: sa.Table | None
+ transaction: sa.Table
+
+
+def _resolve_shadow_tables(tx_table: sa.Table) -> ShadowTables:
+ """Resolve the parent / child / m2m shadow Tables from Continuum's
+ mapper registry and bundle them with the transaction Table.
+
+ ``dashboard_slices_version`` is M2M-tracked by Continuum and lives
+ in metadata under that name (Continuum auto-creates the Table; it
+ isn't registered as a versioned class). Carried separately on the
+ ``ShadowTables`` dataclass because it doesn't follow the parent /
+ child class shape.
+ """
+ # pylint: disable=import-outside-toplevel
+ from sqlalchemy_continuum import version_class
+ from sqlalchemy_continuum.exc import ClassNotVersioned
+
+ from superset.connectors.sqla.models import SqlaTable, SqlMetric,
TableColumn
+ from superset.models.dashboard import Dashboard
+ from superset.models.slice import Slice
+
+ # ``ClassNotVersioned`` is the only expected failure here — versioning
+ # init runs at startup; if it didn't, every class lookup raises this.
+ # Narrowing the catch keeps a real underlying failure (e.g. a metadata
+ # inconsistency after ``make_versioned``) from being silently swallowed
+ # into a no-op retention pass.
+ missing_tables: list[str] = []
+ parent_tables: list[sa.Table] = []
+ for cls in (Dashboard, Slice, SqlaTable):
+ try:
+ parent_tables.append(version_class(cls).__table__)
+ except ClassNotVersioned:
+ missing_tables.append(cls.__name__)
+
+ child_tables: list[sa.Table] = []
+ for cls in (TableColumn, SqlMetric):
+ try:
+ child_tables.append(version_class(cls).__table__)
+ except ClassNotVersioned:
+ missing_tables.append(cls.__name__)
+
+ metadata = parent_tables[0].metadata if parent_tables else None
+ m2m_table = (
+ metadata.tables.get("dashboard_slices_version")
+ if metadata is not None
+ else None
+ )
+ if m2m_table is None:
+ missing_tables.append("dashboard_slices_version")
+
+ if missing_tables:
+ raise RuntimeError(
+ "version-history retention requires every shadow table; missing: "
+ + ", ".join(missing_tables)
+ )
+
+ return ShadowTables(
+ parent=parent_tables,
+ child=child_tables,
+ m2m=m2m_table,
+ transaction=tx_table,
+ )
+
+
+@dataclass(frozen=True)
+class _PruneWindow:
+ """One id-ordered window of prune candidates.
+
+ ``prunable`` is the subset of the window's candidate transactions
+ that are safe to delete. ``candidate_count`` and ``max_candidate_id``
+ drive the batch loop in :func:`_prune_old_versions_impl`: the loop
+ stops when a window returns fewer than ``_MAX_PRUNE_BATCH``
+ candidates, and otherwise advances ``after_id`` to
+ ``max_candidate_id`` to fetch the next window.
+ """
+
+ prunable: list[int]
+ candidate_count: int
+ max_candidate_id: int
+
+
+def _resolve_prune_window(
+ conn: sa.engine.Connection,
+ cutoff: datetime,
+ shadow_tables: list[sa.Table],
+ after_id: int,
+ batch_size: int,
+) -> _PruneWindow:
+ """Resolve one id-ordered window of ``version_transaction`` rows
+ eligible to prune: ``issued_at < cutoff`` AND ``id > after_id``,
+ ordered by ``id`` and capped at *batch_size*. From that window,
+ ``prunable`` excludes any transaction still serving as the live row
+ (``end_transaction_id IS NULL``) of any versioned entity in *any*
+ shadow table — parent, child, or M2M.
+
+ A child (``table_columns_version`` / ``sql_metrics_version``) or the
+ M2M association (``dashboard_slices_version``) is versioned on a
+ validity lifecycle independent of its parent: after a parent-only
+ edit, an unchanged child stays live anchored at an *older*
+ transaction than the parent's current live row. That older
+ transaction must be preserved too — otherwise
+ :func:`_delete_for_transactions` would drop the still-live
+ child/association row and silently strip the surviving version of its
+ columns / metrics / slices. Scanning every shadow for live rows (not
+ just parents) is what prevents that corruption.
+
+ Windowing by ``id`` (rather than materializing the whole backlog)
+ keeps per-pass memory and lock/transaction-hold time bounded. Live
+ rows older than the cutoff are never deleted but are skipped via the
+ ``after_id`` watermark, so the batch loop still terminates.
+ """
+ # pylint: disable=import-outside-toplevel
+ from sqlalchemy_continuum import versioning_manager
+
+ tx_table = versioning_manager.transaction_cls.__table__
+ # ``ORDER BY id LIMIT`` lets PostgreSQL and MySQL use the primary key for
+ # a populated backlog, where expired transactions cluster at the low-id
+ # end. The separate ``issued_at`` index added by migration d3b9a1f6c204
+ # covers the complementary case: when no rows meet the cutoff, both
+ # engines choose it for an empty range scan instead of walking the full
+ # primary key. Planner checks with 500,000 rows confirmed that adding the
+ # cutoff index does not displace the efficient primary-key backlog plan.
+ candidate_ids: list[int] = [
+ row[0]
+ for row in conn.execute(
+ sa.select(tx_table.c.id)
+ .where(tx_table.c.issued_at < cutoff)
+ .where(tx_table.c.id > after_id)
+ .order_by(tx_table.c.id)
+ .limit(batch_size)
+ )
+ ]
+ if not candidate_ids:
+ return _PruneWindow(prunable=[], candidate_count=0,
max_candidate_id=after_id)
+
+ # The select is ordered by ``id`` ascending, so the last element is
+ # the watermark for the next window.
+ max_candidate_id = candidate_ids[-1]
+
+ # Build the set of transaction ids that still anchor a live row
+ # (``end_transaction_id IS NULL``) in some shadow table. Those
+ # transactions represent the current state of an entity (or one of
+ # its children / associations) and must be preserved regardless of
+ # age. Chunked over the candidates to keep the bind-parameter count
+ # inside SQLite's ``SQLITE_MAX_VARIABLE_NUMBER`` floor (see
+ # ``_TX_ID_CHUNK_SIZE`` below). Ids already confirmed live by an
+ # earlier table are skipped before probing the next one.
+ preserved_ids: set[int] = set()
+ for stbl in shadow_tables:
+ remaining = [tx_id for tx_id in candidate_ids if tx_id not in
preserved_ids]
+ if not remaining:
+ break
+ for chunk in _chunked(remaining, _TX_ID_CHUNK_SIZE):
+ for row in conn.execute(
+ sa.select(stbl.c.transaction_id)
+ .where(stbl.c.transaction_id.in_(chunk))
+ .where(stbl.c.end_transaction_id.is_(None))
+ .distinct()
+ ):
+ preserved_ids.add(row[0])
+
+ prunable = [tx_id for tx_id in candidate_ids if tx_id not in preserved_ids]
+ return _PruneWindow(
+ prunable=prunable,
+ candidate_count=len(candidate_ids),
+ max_candidate_id=max_candidate_id,
+ )
+
+
+# SQLite's ``SQLITE_MAX_VARIABLE_NUMBER`` defaults to 999 (lifted to
+# 32766 in 3.32+ but the older limit can still apply in shipped
+# builds). Postgres + MySQL handle tens of thousands of bind params
+# without complaint, so the chunk size is dictated by the SQLite floor.
+# 500 keeps each single-column DELETE / SELECT (see
``_delete_for_transactions``,
+# which splits the create/close predicates into two statements) well inside
+# that floor, with margin for any other bound params in the surrounding
+# statement.
+_TX_ID_CHUNK_SIZE = 500
Review Comment:
**Suggestion:** Add an explicit integer type annotation to this module
constant. [custom_rule]
**Severity Level:** Minor 🧹
<details>
<summary><b>Why it matters? ⭐ </b></summary>
This module constant is an annotatable variable and lacks an explicit type
hint. Under the type-hint rule, this is a valid violation.
</details>
<details>
<summary><b>Rule source 📖 </b></summary>
.cursor/rules/dev-standard.mdc (line 28)
</details>
[](https://app.codeant.ai/fix-in-ide?tool=cursor&prompt_id=0778cab8ac024ea5b3e5a31b41bf5d2d&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=0778cab8ac024ea5b3e5a31b41bf5d2d&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/tasks/version_history_retention.py
**Line:** 253:253
**Comment:**
*Custom Rule: Add an explicit integer type annotation to this module
constant.
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%2F41075&comment_hash=ba2f631b4770d08cbe984b003fc548e1dde7d3787b4328bda7a0210a58da7165&reaction=like'>👍</a>
| <a
href='https://app.codeant.ai/feedback?pr_url=https%3A%2F%2Fgithub.com%2Fapache%2Fsuperset%2Fpull%2F41075&comment_hash=ba2f631b4770d08cbe984b003fc548e1dde7d3787b4328bda7a0210a58da7165&reaction=dislike'>👎</a>
##########
superset/tasks/version_history_retention.py:
##########
@@ -0,0 +1,548 @@
+# 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.
+"""Celery task: prune old entity-version history.
+
+Retention is time-based. The task deletes parent + child shadow rows
+owned by ``version_transaction`` rows whose ``issued_at`` is older
+than ``SUPERSET_VERSION_HISTORY_RETENTION_DAYS`` (default 30, env
+overridable, non-positive to disable).
+
+One preservation rule, applied across every shadow table (parent,
+child, and the M2M association):
+
+* **Live** (``end_transaction_id IS NULL``) — never pruned.
+
+Baseline rows (``operation_type = 0``) and any closed historical row
+are subject to the same retention window as everything else. An
+entity that hasn't been edited within the window has only its live
+row remaining; the historical chain (including the synthetic
+baseline) ages out.
+
+If any shadow row anchored at a transaction is live (in a parent,
+child, or M2M shadow), the ``version_transaction`` and its
+``version_changes`` rows are preserved. Closed shadow rows whose lifecycle
+touches a pruned transaction are removed so their foreign keys cannot retain
+otherwise-expired history. Every other prunable transaction is dropped, and
+its ``version_changes`` rows cascade via the FK.
+
+Registered via ``CeleryConfig.beat_schedule`` in ``superset/config.py``.
+Idempotent: a second run prunes nothing.
+"""
+
+from __future__ import annotations
+
+import logging
+import time
+from collections.abc import Iterator
+from dataclasses import dataclass
+from datetime import datetime, timedelta, timezone
+from typing import Any
+
+import sqlalchemy as sa
+from flask import current_app
+from sqlalchemy.exc import OperationalError
+
+from superset.extensions import celery_app, db, stats_logger_manager
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass(frozen=True)
+class ShadowTables:
+ """The four Continuum-managed Table objects the prune walks.
+
+ Bundled here so the prune helper's signature stays at two arguments
+ instead of five. The shape is set once at task entry by
+ ``_resolve_shadow_tables`` and threaded through the retry loop.
+ """
+
+ parent: list[sa.Table]
+ child: list[sa.Table]
+ m2m: sa.Table | None
+ transaction: sa.Table
+
+
+def _resolve_shadow_tables(tx_table: sa.Table) -> ShadowTables:
+ """Resolve the parent / child / m2m shadow Tables from Continuum's
+ mapper registry and bundle them with the transaction Table.
+
+ ``dashboard_slices_version`` is M2M-tracked by Continuum and lives
+ in metadata under that name (Continuum auto-creates the Table; it
+ isn't registered as a versioned class). Carried separately on the
+ ``ShadowTables`` dataclass because it doesn't follow the parent /
+ child class shape.
+ """
+ # pylint: disable=import-outside-toplevel
+ from sqlalchemy_continuum import version_class
+ from sqlalchemy_continuum.exc import ClassNotVersioned
+
+ from superset.connectors.sqla.models import SqlaTable, SqlMetric,
TableColumn
+ from superset.models.dashboard import Dashboard
+ from superset.models.slice import Slice
+
+ # ``ClassNotVersioned`` is the only expected failure here — versioning
+ # init runs at startup; if it didn't, every class lookup raises this.
+ # Narrowing the catch keeps a real underlying failure (e.g. a metadata
+ # inconsistency after ``make_versioned``) from being silently swallowed
+ # into a no-op retention pass.
+ missing_tables: list[str] = []
+ parent_tables: list[sa.Table] = []
+ for cls in (Dashboard, Slice, SqlaTable):
+ try:
+ parent_tables.append(version_class(cls).__table__)
+ except ClassNotVersioned:
+ missing_tables.append(cls.__name__)
+
+ child_tables: list[sa.Table] = []
+ for cls in (TableColumn, SqlMetric):
+ try:
+ child_tables.append(version_class(cls).__table__)
+ except ClassNotVersioned:
+ missing_tables.append(cls.__name__)
+
+ metadata = parent_tables[0].metadata if parent_tables else None
+ m2m_table = (
+ metadata.tables.get("dashboard_slices_version")
+ if metadata is not None
+ else None
+ )
+ if m2m_table is None:
+ missing_tables.append("dashboard_slices_version")
+
+ if missing_tables:
+ raise RuntimeError(
+ "version-history retention requires every shadow table; missing: "
+ + ", ".join(missing_tables)
+ )
+
+ return ShadowTables(
+ parent=parent_tables,
+ child=child_tables,
+ m2m=m2m_table,
+ transaction=tx_table,
+ )
+
+
+@dataclass(frozen=True)
+class _PruneWindow:
+ """One id-ordered window of prune candidates.
+
+ ``prunable`` is the subset of the window's candidate transactions
+ that are safe to delete. ``candidate_count`` and ``max_candidate_id``
+ drive the batch loop in :func:`_prune_old_versions_impl`: the loop
+ stops when a window returns fewer than ``_MAX_PRUNE_BATCH``
+ candidates, and otherwise advances ``after_id`` to
+ ``max_candidate_id`` to fetch the next window.
+ """
+
+ prunable: list[int]
+ candidate_count: int
+ max_candidate_id: int
+
+
+def _resolve_prune_window(
+ conn: sa.engine.Connection,
+ cutoff: datetime,
+ shadow_tables: list[sa.Table],
+ after_id: int,
+ batch_size: int,
+) -> _PruneWindow:
+ """Resolve one id-ordered window of ``version_transaction`` rows
+ eligible to prune: ``issued_at < cutoff`` AND ``id > after_id``,
+ ordered by ``id`` and capped at *batch_size*. From that window,
+ ``prunable`` excludes any transaction still serving as the live row
+ (``end_transaction_id IS NULL``) of any versioned entity in *any*
+ shadow table — parent, child, or M2M.
+
+ A child (``table_columns_version`` / ``sql_metrics_version``) or the
+ M2M association (``dashboard_slices_version``) is versioned on a
+ validity lifecycle independent of its parent: after a parent-only
+ edit, an unchanged child stays live anchored at an *older*
+ transaction than the parent's current live row. That older
+ transaction must be preserved too — otherwise
+ :func:`_delete_for_transactions` would drop the still-live
+ child/association row and silently strip the surviving version of its
+ columns / metrics / slices. Scanning every shadow for live rows (not
+ just parents) is what prevents that corruption.
+
+ Windowing by ``id`` (rather than materializing the whole backlog)
+ keeps per-pass memory and lock/transaction-hold time bounded. Live
+ rows older than the cutoff are never deleted but are skipped via the
+ ``after_id`` watermark, so the batch loop still terminates.
+ """
+ # pylint: disable=import-outside-toplevel
+ from sqlalchemy_continuum import versioning_manager
+
+ tx_table = versioning_manager.transaction_cls.__table__
+ # ``ORDER BY id LIMIT`` lets PostgreSQL and MySQL use the primary key for
+ # a populated backlog, where expired transactions cluster at the low-id
+ # end. The separate ``issued_at`` index added by migration d3b9a1f6c204
+ # covers the complementary case: when no rows meet the cutoff, both
+ # engines choose it for an empty range scan instead of walking the full
+ # primary key. Planner checks with 500,000 rows confirmed that adding the
+ # cutoff index does not displace the efficient primary-key backlog plan.
+ candidate_ids: list[int] = [
+ row[0]
+ for row in conn.execute(
+ sa.select(tx_table.c.id)
+ .where(tx_table.c.issued_at < cutoff)
+ .where(tx_table.c.id > after_id)
+ .order_by(tx_table.c.id)
+ .limit(batch_size)
+ )
+ ]
+ if not candidate_ids:
+ return _PruneWindow(prunable=[], candidate_count=0,
max_candidate_id=after_id)
+
+ # The select is ordered by ``id`` ascending, so the last element is
+ # the watermark for the next window.
+ max_candidate_id = candidate_ids[-1]
+
+ # Build the set of transaction ids that still anchor a live row
+ # (``end_transaction_id IS NULL``) in some shadow table. Those
+ # transactions represent the current state of an entity (or one of
+ # its children / associations) and must be preserved regardless of
+ # age. Chunked over the candidates to keep the bind-parameter count
+ # inside SQLite's ``SQLITE_MAX_VARIABLE_NUMBER`` floor (see
+ # ``_TX_ID_CHUNK_SIZE`` below). Ids already confirmed live by an
+ # earlier table are skipped before probing the next one.
+ preserved_ids: set[int] = set()
+ for stbl in shadow_tables:
+ remaining = [tx_id for tx_id in candidate_ids if tx_id not in
preserved_ids]
+ if not remaining:
+ break
+ for chunk in _chunked(remaining, _TX_ID_CHUNK_SIZE):
+ for row in conn.execute(
+ sa.select(stbl.c.transaction_id)
+ .where(stbl.c.transaction_id.in_(chunk))
+ .where(stbl.c.end_transaction_id.is_(None))
+ .distinct()
+ ):
+ preserved_ids.add(row[0])
+
+ prunable = [tx_id for tx_id in candidate_ids if tx_id not in preserved_ids]
+ return _PruneWindow(
+ prunable=prunable,
+ candidate_count=len(candidate_ids),
+ max_candidate_id=max_candidate_id,
+ )
+
+
+# SQLite's ``SQLITE_MAX_VARIABLE_NUMBER`` defaults to 999 (lifted to
+# 32766 in 3.32+ but the older limit can still apply in shipped
+# builds). Postgres + MySQL handle tens of thousands of bind params
+# without complaint, so the chunk size is dictated by the SQLite floor.
+# 500 keeps each single-column DELETE / SELECT (see
``_delete_for_transactions``,
+# which splits the create/close predicates into two statements) well inside
+# that floor, with margin for any other bound params in the surrounding
+# statement.
+_TX_ID_CHUNK_SIZE = 500
+
+# Maximum ``version_transaction`` rows resolved + pruned per SERIALIZABLE
+# pass. The prune loops over id-ordered windows of this size (see
+# ``_prune_old_versions_impl``) so memory and lock/transaction-hold time
+# stay bounded per pass instead of scaling with the full backlog on the
+# first run after deploy.
+_MAX_PRUNE_BATCH = 1000
+
+
+def _delete_for_transactions(
+ conn: sa.engine.Connection,
+ tables: list[sa.Table],
+ tx_ids: list[int],
+) -> int:
+ """Delete shadow rows in *tables* whose lifespan touches a pruned
+ transaction — either ``transaction_id`` (created at) or
+ ``end_transaction_id`` (closed at) is in *tx_ids*. Returns total
+ rowcount across all tables.
+
+ The ``end_transaction_id`` predicate is required to keep referential
+ integrity when transactions span multiple entities. A flush that
+ saves dashboard + slice + dataset at the same ``tx=X`` produces
+ three shadow rows sharing that tx. If only the dashboard is later
+ edited at ``tx=Y``, the dashboard row at ``tx=X`` is closed
+ (``end_tx=Y``) while the slice/dataset rows stay live at
+ ``tx=X``. Retention preserves ``tx=X`` (slice/dataset are live
+ there) and prunes ``tx=Y``. Without the ``end_tx`` predicate, the
+ dashboard's closed row at ``tx=X`` survives step 1 — its
+ ``end_transaction_id=Y`` then violates the FK when step 2 deletes
+ ``version_transaction`` row ``Y``.
+
+ Live rows are never matched by either predicate
+ (``end_transaction_id IS NULL`` is not ``IN`` anything; live rows'
+ ``transaction_id`` is preserved by construction in
+ :func:`_resolve_prune_window`).
+
+ ``tx_ids`` is chunked into batches of ``_TX_ID_CHUNK_SIZE`` so the
+ bind-parameter count stays inside SQLite's ``SQLITE_MAX_VARIABLE_
+ NUMBER`` limit. Postgres and MySQL would happily accept the full
+ list, but the floor is dialect-agnostic since the retention task is
+ the only path that accumulates open-ended id batches.
+ """
+ if not tx_ids:
+ return 0
+ total = 0
+ for tbl in tables:
+ for chunk in _chunked(tx_ids, _TX_ID_CHUNK_SIZE):
+ # Two single-column DELETEs rather than one ``OR`` of both
+ # columns. The OR form binds every id twice (once for
+ # ``transaction_id``, once for ``end_transaction_id``), so a
+ # full chunk would bind ``2 * _TX_ID_CHUNK_SIZE`` params and
+ # overflow SQLite's ``SQLITE_MAX_VARIABLE_NUMBER`` floor. Each
+ # single-column statement binds at most ``_TX_ID_CHUNK_SIZE``.
+ # A row whose create- and close-tx are both pruned is removed
+ # by the first matching DELETE; the second finds nothing, so
+ # there is no double count.
+ for col in (tbl.c.transaction_id, tbl.c.end_transaction_id):
+ result = conn.execute(sa.delete(tbl).where(col.in_(chunk)))
+ total += result.rowcount or 0
+ return total
+
+
+def _chunked(items: list[int], size: int) -> Iterator[list[int]]:
+ """Yield *items* in fixed-size lists. Final chunk may be smaller."""
+ for i in range(0, len(items), size):
+ yield items[i : i + size]
+
+
+#: Maximum number of attempts the prune will make before giving up.
+#: A daily Celery beat schedule means the next chance is 24h out, so
+#: a small inline retry materially improves the recovery time for the
+#: serialization-conflict path.
+_MAX_RETRY_ATTEMPTS = 3
+
+#: Base for exponential backoff between retries (seconds). Worst-case
+#: extra latency with the 3-attempt cap above and the factor below is
+#: ``BASE + BASE * FACTOR`` = ~0.5s — well inside the prune's own
+#: typical runtime.
+_RETRY_BACKOFF_BASE_SECONDS = 0.1
Review Comment:
**Suggestion:** Add an explicit float type annotation to this backoff
constant. [custom_rule]
**Severity Level:** Minor 🧹
<details>
<summary><b>Why it matters? ⭐ </b></summary>
This is a module-level constant that can be annotated, and it currently has
no explicit type hint. That fits the stated type-hint rule.
</details>
<details>
<summary><b>Rule source 📖 </b></summary>
.cursor/rules/dev-standard.mdc (line 28)
</details>
[](https://app.codeant.ai/fix-in-ide?tool=cursor&prompt_id=a686c8e293e6403cb54e30b57947bea9&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=a686c8e293e6403cb54e30b57947bea9&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/tasks/version_history_retention.py
**Line:** 332:332
**Comment:**
*Custom Rule: Add an explicit float type annotation to this backoff
constant.
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%2F41075&comment_hash=8fd7043637ee5c0b799936f64c48c7d76dd6c384de579edcf97728b0f5edbc47&reaction=like'>👍</a>
| <a
href='https://app.codeant.ai/feedback?pr_url=https%3A%2F%2Fgithub.com%2Fapache%2Fsuperset%2Fpull%2F41075&comment_hash=8fd7043637ee5c0b799936f64c48c7d76dd6c384de579edcf97728b0f5edbc47&reaction=dislike'>👎</a>
##########
superset/tasks/version_history_retention.py:
##########
@@ -0,0 +1,548 @@
+# 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.
+"""Celery task: prune old entity-version history.
+
+Retention is time-based. The task deletes parent + child shadow rows
+owned by ``version_transaction`` rows whose ``issued_at`` is older
+than ``SUPERSET_VERSION_HISTORY_RETENTION_DAYS`` (default 30, env
+overridable, non-positive to disable).
+
+One preservation rule, applied across every shadow table (parent,
+child, and the M2M association):
+
+* **Live** (``end_transaction_id IS NULL``) — never pruned.
+
+Baseline rows (``operation_type = 0``) and any closed historical row
+are subject to the same retention window as everything else. An
+entity that hasn't been edited within the window has only its live
+row remaining; the historical chain (including the synthetic
+baseline) ages out.
+
+If any shadow row anchored at a transaction is live (in a parent,
+child, or M2M shadow), the ``version_transaction`` and its
+``version_changes`` rows are preserved. Closed shadow rows whose lifecycle
+touches a pruned transaction are removed so their foreign keys cannot retain
+otherwise-expired history. Every other prunable transaction is dropped, and
+its ``version_changes`` rows cascade via the FK.
+
+Registered via ``CeleryConfig.beat_schedule`` in ``superset/config.py``.
+Idempotent: a second run prunes nothing.
+"""
+
+from __future__ import annotations
+
+import logging
+import time
+from collections.abc import Iterator
+from dataclasses import dataclass
+from datetime import datetime, timedelta, timezone
+from typing import Any
+
+import sqlalchemy as sa
+from flask import current_app
+from sqlalchemy.exc import OperationalError
+
+from superset.extensions import celery_app, db, stats_logger_manager
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass(frozen=True)
+class ShadowTables:
+ """The four Continuum-managed Table objects the prune walks.
+
+ Bundled here so the prune helper's signature stays at two arguments
+ instead of five. The shape is set once at task entry by
+ ``_resolve_shadow_tables`` and threaded through the retry loop.
+ """
+
+ parent: list[sa.Table]
+ child: list[sa.Table]
+ m2m: sa.Table | None
+ transaction: sa.Table
+
+
+def _resolve_shadow_tables(tx_table: sa.Table) -> ShadowTables:
+ """Resolve the parent / child / m2m shadow Tables from Continuum's
+ mapper registry and bundle them with the transaction Table.
+
+ ``dashboard_slices_version`` is M2M-tracked by Continuum and lives
+ in metadata under that name (Continuum auto-creates the Table; it
+ isn't registered as a versioned class). Carried separately on the
+ ``ShadowTables`` dataclass because it doesn't follow the parent /
+ child class shape.
+ """
+ # pylint: disable=import-outside-toplevel
+ from sqlalchemy_continuum import version_class
+ from sqlalchemy_continuum.exc import ClassNotVersioned
+
+ from superset.connectors.sqla.models import SqlaTable, SqlMetric,
TableColumn
+ from superset.models.dashboard import Dashboard
+ from superset.models.slice import Slice
+
+ # ``ClassNotVersioned`` is the only expected failure here — versioning
+ # init runs at startup; if it didn't, every class lookup raises this.
+ # Narrowing the catch keeps a real underlying failure (e.g. a metadata
+ # inconsistency after ``make_versioned``) from being silently swallowed
+ # into a no-op retention pass.
+ missing_tables: list[str] = []
+ parent_tables: list[sa.Table] = []
+ for cls in (Dashboard, Slice, SqlaTable):
+ try:
+ parent_tables.append(version_class(cls).__table__)
+ except ClassNotVersioned:
+ missing_tables.append(cls.__name__)
+
+ child_tables: list[sa.Table] = []
+ for cls in (TableColumn, SqlMetric):
+ try:
+ child_tables.append(version_class(cls).__table__)
+ except ClassNotVersioned:
+ missing_tables.append(cls.__name__)
+
+ metadata = parent_tables[0].metadata if parent_tables else None
+ m2m_table = (
+ metadata.tables.get("dashboard_slices_version")
+ if metadata is not None
+ else None
+ )
+ if m2m_table is None:
+ missing_tables.append("dashboard_slices_version")
+
+ if missing_tables:
+ raise RuntimeError(
+ "version-history retention requires every shadow table; missing: "
+ + ", ".join(missing_tables)
+ )
+
+ return ShadowTables(
+ parent=parent_tables,
+ child=child_tables,
+ m2m=m2m_table,
+ transaction=tx_table,
+ )
+
+
+@dataclass(frozen=True)
+class _PruneWindow:
+ """One id-ordered window of prune candidates.
+
+ ``prunable`` is the subset of the window's candidate transactions
+ that are safe to delete. ``candidate_count`` and ``max_candidate_id``
+ drive the batch loop in :func:`_prune_old_versions_impl`: the loop
+ stops when a window returns fewer than ``_MAX_PRUNE_BATCH``
+ candidates, and otherwise advances ``after_id`` to
+ ``max_candidate_id`` to fetch the next window.
+ """
+
+ prunable: list[int]
+ candidate_count: int
+ max_candidate_id: int
+
+
+def _resolve_prune_window(
+ conn: sa.engine.Connection,
+ cutoff: datetime,
+ shadow_tables: list[sa.Table],
+ after_id: int,
+ batch_size: int,
+) -> _PruneWindow:
+ """Resolve one id-ordered window of ``version_transaction`` rows
+ eligible to prune: ``issued_at < cutoff`` AND ``id > after_id``,
+ ordered by ``id`` and capped at *batch_size*. From that window,
+ ``prunable`` excludes any transaction still serving as the live row
+ (``end_transaction_id IS NULL``) of any versioned entity in *any*
+ shadow table — parent, child, or M2M.
+
+ A child (``table_columns_version`` / ``sql_metrics_version``) or the
+ M2M association (``dashboard_slices_version``) is versioned on a
+ validity lifecycle independent of its parent: after a parent-only
+ edit, an unchanged child stays live anchored at an *older*
+ transaction than the parent's current live row. That older
+ transaction must be preserved too — otherwise
+ :func:`_delete_for_transactions` would drop the still-live
+ child/association row and silently strip the surviving version of its
+ columns / metrics / slices. Scanning every shadow for live rows (not
+ just parents) is what prevents that corruption.
+
+ Windowing by ``id`` (rather than materializing the whole backlog)
+ keeps per-pass memory and lock/transaction-hold time bounded. Live
+ rows older than the cutoff are never deleted but are skipped via the
+ ``after_id`` watermark, so the batch loop still terminates.
+ """
+ # pylint: disable=import-outside-toplevel
+ from sqlalchemy_continuum import versioning_manager
+
+ tx_table = versioning_manager.transaction_cls.__table__
+ # ``ORDER BY id LIMIT`` lets PostgreSQL and MySQL use the primary key for
+ # a populated backlog, where expired transactions cluster at the low-id
+ # end. The separate ``issued_at`` index added by migration d3b9a1f6c204
+ # covers the complementary case: when no rows meet the cutoff, both
+ # engines choose it for an empty range scan instead of walking the full
+ # primary key. Planner checks with 500,000 rows confirmed that adding the
+ # cutoff index does not displace the efficient primary-key backlog plan.
+ candidate_ids: list[int] = [
+ row[0]
+ for row in conn.execute(
+ sa.select(tx_table.c.id)
+ .where(tx_table.c.issued_at < cutoff)
+ .where(tx_table.c.id > after_id)
+ .order_by(tx_table.c.id)
+ .limit(batch_size)
+ )
+ ]
+ if not candidate_ids:
+ return _PruneWindow(prunable=[], candidate_count=0,
max_candidate_id=after_id)
+
+ # The select is ordered by ``id`` ascending, so the last element is
+ # the watermark for the next window.
+ max_candidate_id = candidate_ids[-1]
+
+ # Build the set of transaction ids that still anchor a live row
+ # (``end_transaction_id IS NULL``) in some shadow table. Those
+ # transactions represent the current state of an entity (or one of
+ # its children / associations) and must be preserved regardless of
+ # age. Chunked over the candidates to keep the bind-parameter count
+ # inside SQLite's ``SQLITE_MAX_VARIABLE_NUMBER`` floor (see
+ # ``_TX_ID_CHUNK_SIZE`` below). Ids already confirmed live by an
+ # earlier table are skipped before probing the next one.
+ preserved_ids: set[int] = set()
+ for stbl in shadow_tables:
+ remaining = [tx_id for tx_id in candidate_ids if tx_id not in
preserved_ids]
+ if not remaining:
+ break
+ for chunk in _chunked(remaining, _TX_ID_CHUNK_SIZE):
+ for row in conn.execute(
+ sa.select(stbl.c.transaction_id)
+ .where(stbl.c.transaction_id.in_(chunk))
+ .where(stbl.c.end_transaction_id.is_(None))
+ .distinct()
+ ):
+ preserved_ids.add(row[0])
+
+ prunable = [tx_id for tx_id in candidate_ids if tx_id not in preserved_ids]
+ return _PruneWindow(
+ prunable=prunable,
+ candidate_count=len(candidate_ids),
+ max_candidate_id=max_candidate_id,
+ )
+
+
+# SQLite's ``SQLITE_MAX_VARIABLE_NUMBER`` defaults to 999 (lifted to
+# 32766 in 3.32+ but the older limit can still apply in shipped
+# builds). Postgres + MySQL handle tens of thousands of bind params
+# without complaint, so the chunk size is dictated by the SQLite floor.
+# 500 keeps each single-column DELETE / SELECT (see
``_delete_for_transactions``,
+# which splits the create/close predicates into two statements) well inside
+# that floor, with margin for any other bound params in the surrounding
+# statement.
+_TX_ID_CHUNK_SIZE = 500
+
+# Maximum ``version_transaction`` rows resolved + pruned per SERIALIZABLE
+# pass. The prune loops over id-ordered windows of this size (see
+# ``_prune_old_versions_impl``) so memory and lock/transaction-hold time
+# stay bounded per pass instead of scaling with the full backlog on the
+# first run after deploy.
+_MAX_PRUNE_BATCH = 1000
+
+
+def _delete_for_transactions(
+ conn: sa.engine.Connection,
+ tables: list[sa.Table],
+ tx_ids: list[int],
+) -> int:
+ """Delete shadow rows in *tables* whose lifespan touches a pruned
+ transaction — either ``transaction_id`` (created at) or
+ ``end_transaction_id`` (closed at) is in *tx_ids*. Returns total
+ rowcount across all tables.
+
+ The ``end_transaction_id`` predicate is required to keep referential
+ integrity when transactions span multiple entities. A flush that
+ saves dashboard + slice + dataset at the same ``tx=X`` produces
+ three shadow rows sharing that tx. If only the dashboard is later
+ edited at ``tx=Y``, the dashboard row at ``tx=X`` is closed
+ (``end_tx=Y``) while the slice/dataset rows stay live at
+ ``tx=X``. Retention preserves ``tx=X`` (slice/dataset are live
+ there) and prunes ``tx=Y``. Without the ``end_tx`` predicate, the
+ dashboard's closed row at ``tx=X`` survives step 1 — its
+ ``end_transaction_id=Y`` then violates the FK when step 2 deletes
+ ``version_transaction`` row ``Y``.
+
+ Live rows are never matched by either predicate
+ (``end_transaction_id IS NULL`` is not ``IN`` anything; live rows'
+ ``transaction_id`` is preserved by construction in
+ :func:`_resolve_prune_window`).
+
+ ``tx_ids`` is chunked into batches of ``_TX_ID_CHUNK_SIZE`` so the
+ bind-parameter count stays inside SQLite's ``SQLITE_MAX_VARIABLE_
+ NUMBER`` limit. Postgres and MySQL would happily accept the full
+ list, but the floor is dialect-agnostic since the retention task is
+ the only path that accumulates open-ended id batches.
+ """
+ if not tx_ids:
+ return 0
+ total = 0
+ for tbl in tables:
+ for chunk in _chunked(tx_ids, _TX_ID_CHUNK_SIZE):
+ # Two single-column DELETEs rather than one ``OR`` of both
+ # columns. The OR form binds every id twice (once for
+ # ``transaction_id``, once for ``end_transaction_id``), so a
+ # full chunk would bind ``2 * _TX_ID_CHUNK_SIZE`` params and
+ # overflow SQLite's ``SQLITE_MAX_VARIABLE_NUMBER`` floor. Each
+ # single-column statement binds at most ``_TX_ID_CHUNK_SIZE``.
+ # A row whose create- and close-tx are both pruned is removed
+ # by the first matching DELETE; the second finds nothing, so
+ # there is no double count.
+ for col in (tbl.c.transaction_id, tbl.c.end_transaction_id):
+ result = conn.execute(sa.delete(tbl).where(col.in_(chunk)))
+ total += result.rowcount or 0
+ return total
+
+
+def _chunked(items: list[int], size: int) -> Iterator[list[int]]:
+ """Yield *items* in fixed-size lists. Final chunk may be smaller."""
+ for i in range(0, len(items), size):
+ yield items[i : i + size]
+
+
+#: Maximum number of attempts the prune will make before giving up.
+#: A daily Celery beat schedule means the next chance is 24h out, so
+#: a small inline retry materially improves the recovery time for the
+#: serialization-conflict path.
+_MAX_RETRY_ATTEMPTS = 3
+
+#: Base for exponential backoff between retries (seconds). Worst-case
+#: extra latency with the 3-attempt cap above and the factor below is
+#: ``BASE + BASE * FACTOR`` = ~0.5s — well inside the prune's own
+#: typical runtime.
+_RETRY_BACKOFF_BASE_SECONDS = 0.1
+
+#: Exponential-backoff multiplier between successive retry attempts.
+#: Backoff for attempt N is ``BASE * (FACTOR ** (N - 1))``.
+_RETRY_BACKOFF_FACTOR = 4
+
+#: Statsd metric prefix for retention emissions. Mirrors the activity-view
+#: orchestrator's ``superset.activity_view.*`` namespace so a single
+#: Grafana filter (``superset.versioning.*``) catches both sides of the
+#: feature. The pruned-count gauge fires every run; the skipped counter
+#: fires for the "retention disabled" and "no versioned classes" cases;
+#: the retried counter fires when the SERIALIZABLE block tripped at
+#: least one conflict before settling.
+_METRIC_PREFIX = "superset.versioning.retention"
Review Comment:
**Suggestion:** Add an explicit string type annotation to this metric prefix
constant. [custom_rule]
**Severity Level:** Minor 🧹
<details>
<summary><b>Why it matters? ⭐ </b></summary>
This string module constant is a relevant variable that can be annotated,
but it is currently untyped. The suggestion accurately reflects the type-hint
requirement.
</details>
<details>
<summary><b>Rule source 📖 </b></summary>
.cursor/rules/dev-standard.mdc (line 28)
</details>
[](https://app.codeant.ai/fix-in-ide?tool=cursor&prompt_id=9cd0c557bf6b4be8aec0a783ccee07fc&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=9cd0c557bf6b4be8aec0a783ccee07fc&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/tasks/version_history_retention.py
**Line:** 345:345
**Comment:**
*Custom Rule: Add an explicit string type annotation to this metric
prefix constant.
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%2F41075&comment_hash=510b24be93f018b8a3560317a157ae02c6085e24083dda6b36c56c95445850fa&reaction=like'>👍</a>
| <a
href='https://app.codeant.ai/feedback?pr_url=https%3A%2F%2Fgithub.com%2Fapache%2Fsuperset%2Fpull%2F41075&comment_hash=510b24be93f018b8a3560317a157ae02c6085e24083dda6b36c56c95445850fa&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]
