This is an automated email from the ASF dual-hosted git repository.
rusackas pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/superset.git
The following commit(s) were added to refs/heads/master by this push:
new a73e2485de8 feat(versioning): version-history retention cleanup job
(#41075)
a73e2485de8 is described below
commit a73e2485de8e587a17c3ecb33720762ffcf68709
Author: Mike Bridge <[email protected]>
AuthorDate: Tue Jul 28 03:59:11 2026 +0100
feat(versioning): version-history retention cleanup job (#41075)
Co-authored-by: Mike Bridge <[email protected]>
Co-authored-by: Claude Opus 4.8 (1M context) <[email protected]>
---
UPDATING.md | 10 +
superset/config.py | 51 ++
superset/initialization/__init__.py | 73 ++-
superset/mcp_service/sql_lab/tool/execute_sql.py | 37 +-
...a1f6c204_version_transaction_issued_at_index.py | 48 ++
superset/tasks/celery_app.py | 2 +-
superset/tasks/version_history_retention.py | 548 +++++++++++++++++++++
superset/versioning/queries.py | 19 +-
.../versioning/retention_prune_tests.py | 510 +++++++++++++++++++
tests/unit_tests/config_test.py | 26 +
tests/unit_tests/initialization_test.py | 161 ++++++
.../test_version_transaction_issued_at_index.py | 90 ++++
tests/unit_tests/semantic_layers/mapper_test.py | 4 +-
.../tasks/test_version_history_retention.py | 200 ++++++++
14 files changed, 1749 insertions(+), 30 deletions(-)
diff --git a/UPDATING.md b/UPDATING.md
index 1f2e73765e2..aac6792c8fb 100644
--- a/UPDATING.md
+++ b/UPDATING.md
@@ -358,6 +358,16 @@ A read-only companion to the version-history endpoints:
each entity type gains a
Authorization reuses the resource's `can_read` permission and per-object
`raise_for_access`; related-entity rows are visibility-filtered to what the
caller may see. The stream is empty unless version capture is on
(`ENABLE_VERSIONING_CAPTURE`).
+### Version-history retention (pruning)
+
+Entity version history (the `version_transaction` / `*_version` shadow tables
that back version capture) is aged out by a nightly Celery beat task,
`version_history.prune_old_versions`
(`superset.tasks.version_history_retention`).
+
+| Key | Default | Purpose |
+|---|---|---|
+| `SUPERSET_VERSION_HISTORY_RETENTION_DAYS` | `30` | Version rows whose owning
`version_transaction.issued_at` is older than this many days are pruned. Each
entity's live row (`end_transaction_id IS NULL`) is always preserved, as are
the live rows of its children and associations; closed historical rows
(including the baseline) age out. Set to `0` or a negative value to disable
pruning. |
+
+The task ships in the default `CeleryConfig.beat_schedule`; a deployment that
overrides `CELERY_CONFIG` without inheriting the default will log a startup
warning that the prune task is absent (so it never silently stops running).
Retention only prunes whatever history exists — capture itself is gated
separately by `ENABLE_VERSIONING_CAPTURE` (ships off).
+
### Webhook alerts/reports block private/internal hosts by default
Webhook alert/report dispatch (`WebhookNotification.send`) now validates the
target URL's host against the same private/internal-IP block applied to dataset
import URLs. If the resolved host is in a loopback, link-local, private
(RFC-1918), shared-CGNAT, or multicast range, the webhook is rejected with
`NotificationParamException`.
diff --git a/superset/config.py b/superset/config.py
index 4eb87cea2e4..b7c71f6b610 100644
--- a/superset/config.py
+++ b/superset/config.py
@@ -1649,6 +1649,49 @@ ENABLE_VERSIONING_CAPTURE: bool =
utils.parse_boolean_string(
os.environ.get("ENABLE_VERSIONING_CAPTURE", "false")
)
+# Retention window (days) for entity version history. Version rows
+# whose owning ``version_transaction.issued_at`` is older than this
+# value are pruned by the ``version_history.prune_old_versions``
+# Celery beat task (registered below in ``CeleryConfig.beat_schedule``).
+# If any row anchored at a transaction is live
+# (``end_transaction_id IS NULL``), that entire transaction is preserved.
+# Baseline rows (``operation_type=0``) and closed historical rows otherwise
+# age out alongside the rest. Any non-positive value disables pruning.
+# Read from environment variable of the same name.
+_DEFAULT_VERSION_HISTORY_RETENTION_DAYS: int = 30
+# Keep cutoff arithmetic comfortably inside ``datetime``'s supported range
+# while allowing retention windows far beyond any practical deployment age.
+_MAX_VERSION_HISTORY_RETENTION_DAYS: int = 36_500
+
+
+def _parse_version_history_retention_days() -> int:
+ """Parse the retention window without making invalid input fatal."""
+ value: str | None =
os.environ.get("SUPERSET_VERSION_HISTORY_RETENTION_DAYS")
+ if value is None:
+ return _DEFAULT_VERSION_HISTORY_RETENTION_DAYS
+ try:
+ retention_days = int(value)
+ except ValueError:
+ logger.warning(
+ "Invalid SUPERSET_VERSION_HISTORY_RETENTION_DAYS=%r; using %d",
+ value,
+ _DEFAULT_VERSION_HISTORY_RETENTION_DAYS,
+ )
+ return _DEFAULT_VERSION_HISTORY_RETENTION_DAYS
+ if retention_days > _MAX_VERSION_HISTORY_RETENTION_DAYS:
+ logger.warning(
+ "SUPERSET_VERSION_HISTORY_RETENTION_DAYS=%r exceeds the maximum "
+ "of %d; using %d",
+ value,
+ _MAX_VERSION_HISTORY_RETENTION_DAYS,
+ _DEFAULT_VERSION_HISTORY_RETENTION_DAYS,
+ )
+ return _DEFAULT_VERSION_HISTORY_RETENTION_DAYS
+ return retention_days
+
+
+SUPERSET_VERSION_HISTORY_RETENTION_DAYS: int =
_parse_version_history_retention_days()
+
# Adds a warning message on sqllab save query and schedule query modals.
SQLLAB_SAVE_WARNING_MESSAGE = None
SQLLAB_SCHEDULE_WARNING_MESSAGE = None
@@ -1702,6 +1745,7 @@ class CeleryConfig: # pylint:
disable=too-few-public-methods
"superset.tasks.cache",
"superset.tasks.slack",
"superset.tasks.export_dashboard_excel",
+ "superset.tasks.version_history_retention",
)
result_backend = "db+sqlite:///celery_results.sqlite"
worker_prefetch_multiplier = 1
@@ -1721,6 +1765,13 @@ class CeleryConfig: # pylint:
disable=too-few-public-methods
"task": "reports.prune_log",
"schedule": crontab(minute=0, hour=0),
},
+ # Entity version-history retention. Daily at 03:00; the task
+ # itself short-circuits when SUPERSET_VERSION_HISTORY_RETENTION_DAYS
+ # is non-positive (disabled).
+ "version_history.prune_old_versions": {
+ "task": "version_history.prune_old_versions",
+ "schedule": crontab(minute=0, hour=3),
+ },
# Uncomment to enable pruning of the query table
# "prune_query": {
# "task": "prune_query",
diff --git a/superset/initialization/__init__.py
b/superset/initialization/__init__.py
index 7cf9e49a0c8..36a5fc31bd8 100644
--- a/superset/initialization/__init__.py
+++ b/superset/initialization/__init__.py
@@ -775,6 +775,14 @@ class SupersetAppInitializer: # pylint:
disable=too-many-public-methods
does not load ``superset.config`` (some test factories, embedded
use) stays inert by default rather than silently enabling capture.
"""
+ # Beat-schedule check first: the retention task is independent of
+ # save-path capture and remains useful for ageing-out rows already
+ # written by prior deploys. An operator hitting the kill-switch in
+ # anger may also be running a hand-rolled ``CeleryConfig`` that
+ # silently dropped the prune entry; surfacing both misconfigurations
+ # at the same restart is the cheap, observability-positive shape.
+ self._warn_if_retention_beat_missing()
+
if not self.config.get("ENABLE_VERSIONING_CAPTURE", False):
logger.warning(
"versioning: ENABLE_VERSIONING_CAPTURE is False; "
@@ -861,10 +869,67 @@ class SupersetAppInitializer: # pylint:
disable=too-many-public-methods
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 four 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``.
+ * a dotted import string — also accepted by Celery, but deliberately
+ skipped here because resolving operator code solely for this warning
+ would duplicate Celery loader behavior and could add startup side
+ effects.
+ """
+ celery_config: Any = self.config.get("CELERY_CONFIG")
+ if celery_config is None:
+ return # Celery disabled entirely; no retention task to warn
about.
+ if isinstance(celery_config, str):
+ return # Celery resolves dotted config references in its loader.
+ 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: set[Any] = {
+ entry.get("task")
+ for entry in (beat_schedule or {}).values()
+ if isinstance(entry, dict)
+ }
+ registered_tasks.update(beat_schedule or {}) # tolerate key == task
name
+ if not beat_schedule or self._RETENTION_TASK_NAME not in
registered_tasks:
+ logger.warning(
+ "versioning: CELERY_CONFIG.beat_schedule is missing the "
+ "%r entry — the retention task will not fire and shadow "
+ "tables will grow unbounded. Either inherit from the "
+ "default CeleryConfig or add the entry to your override.",
+ self._RETENTION_TASK_NAME,
+ )
def init_app_in_ctx(self) -> None:
"""
diff --git a/superset/mcp_service/sql_lab/tool/execute_sql.py
b/superset/mcp_service/sql_lab/tool/execute_sql.py
index 878a46c8b18..12351827b4b 100644
--- a/superset/mcp_service/sql_lab/tool/execute_sql.py
+++ b/superset/mcp_service/sql_lab/tool/execute_sql.py
@@ -210,21 +210,7 @@ async def execute_sql(request: ExecuteSqlRequest, ctx:
Context) -> ExecuteSqlRes
"template_params supplied but ENABLE_TEMPLATE_PROCESSING is
off"
)
- # Log successful execution
- if response.success:
- await ctx.info(
- "SQL execution completed successfully: rows_returned=%s, "
- "execution_time=%s"
- % (
- response.row_count,
- response.execution_time,
- )
- )
- else:
- await ctx.info(
- "SQL execution failed: error=%s, error_type=%s"
- % (response.error, response.error_type)
- )
+ await _log_execution_result(response, ctx)
return response
@@ -258,6 +244,27 @@ async def execute_sql(request: ExecuteSqlRequest, ctx:
Context) -> ExecuteSqlRes
raise
+async def _log_execution_result(
+ response: ExecuteSqlResponse,
+ ctx: Context,
+) -> None:
+ """Log the outcome of an SQL execution."""
+ if response.success:
+ await ctx.info(
+ "SQL execution completed successfully: rows_returned=%s, "
+ "execution_time=%s"
+ % (
+ response.row_count,
+ response.execution_time,
+ )
+ )
+ else:
+ await ctx.info(
+ "SQL execution failed: error=%s, error_type=%s"
+ % (response.error, response.error_type)
+ )
+
+
def _sanitize_row_values(rows: list[dict[str, Any]]) -> None:
"""Sanitize non-serializable values in rows for JSON serialization."""
for row in rows:
diff --git
a/superset/migrations/versions/2026-07-27_10-00_d3b9a1f6c204_version_transaction_issued_at_index.py
b/superset/migrations/versions/2026-07-27_10-00_d3b9a1f6c204_version_transaction_issued_at_index.py
new file mode 100644
index 00000000000..f59e1269ba8
--- /dev/null
+++
b/superset/migrations/versions/2026-07-27_10-00_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: e5f6a7b8c9d0
+Create Date: 2026-07-27 10:00:00.000000
+"""
+
+from superset.migrations.shared.utils import create_index, drop_index
+
+revision: str = "d3b9a1f6c204"
+down_revision: str = "e5f6a7b8c9d0"
+
+INDEX_NAME: str = "ix_version_transaction_issued_at"
+TABLE_NAME: str = "version_transaction"
+
+
+def upgrade() -> None:
+ """Create the retention cutoff index if it does not exist."""
+ create_index(TABLE_NAME, INDEX_NAME, ["issued_at"], unique=False)
+
+
+def downgrade() -> None:
+ """Remove the retention cutoff index."""
+ drop_index(TABLE_NAME, INDEX_NAME)
diff --git a/superset/tasks/celery_app.py b/superset/tasks/celery_app.py
index 96d64896895..4189a9a69f4 100644
--- a/superset/tasks/celery_app.py
+++ b/superset/tasks/celery_app.py
@@ -35,7 +35,7 @@ flask_app = create_app()
# Need to import late, as the celery_app will have been setup by "create_app()"
# ruff: noqa: E402, F401
# pylint: disable=wrong-import-position, unused-import
-from . import cache, scheduler
+from . import cache, scheduler, version_history_retention
# Export the celery app globally for Celery (as run on the cmd line) to find
app = celery_app
diff --git a/superset/tasks/version_history_retention.py
b/superset/tasks/version_history_retention.py
new file mode 100644
index 00000000000..b9aca62fa84
--- /dev/null
+++ b/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.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: int = 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: int = 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: int = 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: float = 0.1
+
+#: Exponential-backoff multiplier between successive retry attempts.
+#: Backoff for attempt N is ``BASE * (FACTOR ** (N - 1))``.
+_RETRY_BACKOFF_FACTOR: int = 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: str = "superset.versioning.retention"
+
+
+def _run_prune_pass(
+ cutoff: datetime, tables: ShadowTables, after_id: int = 0
+) -> dict[str, Any]:
+ """One SERIALIZABLE pass over a single id-ordered window of candidate
+ transactions starting after ``after_id``. The caller wraps this in
+ the retry loop (serialization conflict → fresh connection from a
+ clean snapshot) and the batch loop (advance ``after_id`` until the
+ backlog drains). The returned ``candidate_count`` / ``max_candidate_id``
+ drive that batch loop."""
+ # Scan every shadow (parent + child + M2M) for live rows, not just
+ # parents: children and the M2M association live on independent
+ # validity lifecycles and may anchor a still-live row at an older
+ # transaction than the parent's current live row.
+ live_bearing_tables: list[sa.Table] = [*tables.parent, *tables.child]
+ if tables.m2m is not None:
+ live_bearing_tables.append(tables.m2m)
+
+ # The Celery task runs outside the request-bound DB session, so we
+ # use a fresh connection rather than ``db.session`` to avoid stepping
+ # on web-request state.
+ with (
+ db.engine.connect().execution_options(isolation_level="SERIALIZABLE")
as conn,
+ conn.begin(),
+ ):
+ window = _resolve_prune_window(
+ conn, cutoff, live_bearing_tables, after_id, _MAX_PRUNE_BATCH
+ )
+ tx_ids = window.prunable
+
+ parent_rows = _delete_for_transactions(conn, tables.parent, tx_ids)
+ child_rows = _delete_for_transactions(conn, tables.child, tx_ids)
+ m2m_rows = (
+ _delete_for_transactions(conn, [tables.m2m], tx_ids)
+ if tables.m2m is not None
+ else 0
+ )
+
+ # Drop the version_transaction rows themselves. ON DELETE
+ # CASCADE on version_changes.transaction_id removes the
+ # associated change records automatically. Same SQLite bind-
+ # parameter chunking applies as the shadow deletes above.
+ tx_rows = 0
+ for chunk in _chunked(tx_ids, _TX_ID_CHUNK_SIZE):
+ tx_rows += (
+ conn.execute(
+ sa.delete(tables.transaction).where(
+ tables.transaction.c.id.in_(chunk)
+ )
+ ).rowcount
+ or 0
+ )
+
+ return {
+ "cutoff": cutoff.isoformat(),
+ "candidate_count": window.candidate_count,
+ "max_candidate_id": window.max_candidate_id,
+ "pruned_transactions": tx_rows,
+ "pruned_parent_shadows": parent_rows,
+ "pruned_child_shadows": child_rows,
+ "pruned_m2m_shadows": m2m_rows,
+ }
+
+
+def _run_pass_with_retry(
+ cutoff: datetime, tables: ShadowTables, after_id: int
+) -> tuple[dict[str, Any], int]:
+ """Run one window pass, retrying on serialization conflict. Returns
+ ``(stats, retries_used)``; re-raises the ``OperationalError`` if all
+ ``_MAX_RETRY_ATTEMPTS`` attempts conflict.
+
+ Postgres surfaces conflicts as ``SerializationFailure`` (a subclass
+ of ``sqlalchemy.exc.OperationalError``). The catch is deliberately the
+ broader ``OperationalError`` — it also covers transient faults such as
+ SQLite's "database is locked" and dropped connections, all of which are
+ safe to retry because each pass is idempotent and runs in its own fresh
+ transaction. Without the inline retry a single conflict pushes the next
+ attempt 24h out (daily Celery beat), so under sustained write pressure
+ the prune could silently fail for days in a row.
+ """
+ for attempt in range(1, _MAX_RETRY_ATTEMPTS + 1):
+ try:
+ return _run_prune_pass(cutoff, tables, after_id), attempt - 1
+ except OperationalError as exc:
+ stats_logger_manager.instance.incr(f"{_METRIC_PREFIX}.retried")
+ if attempt == _MAX_RETRY_ATTEMPTS:
+ logger.warning(
+ "version_history_retention: gave up after %d attempts: %s",
+ _MAX_RETRY_ATTEMPTS,
+ exc,
+ )
+ raise
+ backoff = _RETRY_BACKOFF_BASE_SECONDS * (
+ _RETRY_BACKOFF_FACTOR ** (attempt - 1)
+ )
+ logger.info(
+ "version_history_retention: attempt %d hit %s; retrying in
%.2fs",
+ attempt,
+ type(exc).__name__,
+ backoff,
+ )
+ time.sleep(backoff)
+ raise AssertionError("unreachable") # pragma: no cover
+
+
+def _prune_old_versions_impl(retention_days: int) -> dict[str, Any]:
+ """Pure-Python implementation of the prune. Split out from the
+ Celery task wrapper so unit tests can call it directly without the
+ Celery harness.
+
+ Returns a stats dict for logging / test assertions.
+
+ Isolation level: SERIALIZABLE. The prune is logically a multi-step
+ read-then-write (candidate-vs-preserved SELECTs feeding the shadow
+ DELETEs). At READ COMMITTED there is a TOCTOU window — a save
+ committing between the preserved-ids snapshot and the DELETEs can
+ leave a stale view of which transaction ids are still serving as
+ the live row of some entity, and a shadow row that became live
+ mid-task can be silently dropped. SERIALIZABLE makes the prune
+ atomic against concurrent writers. SQLite is single-writer so
+ SERIALIZABLE is the only level available; MySQL InnoDB and Postgres
+ both support it natively.
+
+ Postgres surfaces conflicts as ``SerializationFailure`` (a subclass
+ of ``sqlalchemy.exc.OperationalError``). The prune retries up to
+ ``_MAX_RETRY_ATTEMPTS`` with exponential backoff before giving up
+ and letting the outer Celery wrapper log + return ``{"error": 1}``.
+ Without the inline retry, a single conflict pushes the next attempt
+ 24 hours out (daily Celery beat), and under sustained write
+ pressure the prune can silently fail for many days in a row.
+ """
+ if retention_days <= 0:
+ logger.info(
+ "version_history_retention:
SUPERSET_VERSION_HISTORY_RETENTION_DAYS "
+ "<= 0; skipping",
+ )
+ stats_logger_manager.instance.incr(f"{_METRIC_PREFIX}.skipped")
+ return {"skipped": 1}
+
+ # pylint: disable=import-outside-toplevel
+ from sqlalchemy_continuum import versioning_manager
+
+ tables =
_resolve_shadow_tables(versioning_manager.transaction_cls.__table__)
+ # Naive-UTC to match ``version_transaction.issued_at`` (Continuum stores
+ # it tz-naive via ``utc_now()``); ``datetime.utcnow()`` is deprecated on
+ # 3.12+, so derive the same value from the tz-aware clock and drop tzinfo.
+ cutoff = datetime.now(timezone.utc).replace(tzinfo=None) - timedelta(
+ days=retention_days
+ )
+
+ # Drain the backlog one bounded, id-ordered window at a time. Each
+ # window is its own retried SERIALIZABLE pass, so memory and
+ # lock/transaction-hold time stay bounded per pass even on the first
+ # run after deploy. The loop stops once a window returns fewer than a
+ # full batch of candidates (nothing left older than the cutoff).
+ totals: dict[str, int] = {
+ "pruned_transactions": 0,
+ "pruned_parent_shadows": 0,
+ "pruned_child_shadows": 0,
+ "pruned_m2m_shadows": 0,
+ }
+ total_retried = 0
+ after_id = 0
+ while True:
+ pass_stats, retries = _run_pass_with_retry(cutoff, tables, after_id)
+ total_retried += retries
+ for key in totals:
+ totals[key] += pass_stats.get(key, 0)
+ if pass_stats.get("candidate_count", 0) < _MAX_PRUNE_BATCH:
+ break
+ after_id = pass_stats.get("max_candidate_id", after_id)
+
+ stats: dict[str, Any] = {"cutoff": cutoff.isoformat(), **totals}
+ if total_retried:
+ stats["retried"] = total_retried
+ stats_logger_manager.instance.gauge(
+ f"{_METRIC_PREFIX}.pruned_transactions", stats["pruned_transactions"]
+ )
+ logger.info("version_history_retention: %s", stats)
+ return stats
+
+
+@celery_app.task(name="version_history.prune_old_versions")
+def prune_old_versions() -> dict[str, Any]:
+ """Celery beat task entry point. Wraps the implementation with
+ config lookup + broad exception handling so a single failed run
+ doesn't poison the schedule (the next firing retries from a clean
+ slate).
+ """
+ try:
+ retention_days = int(
+ current_app.config.get("SUPERSET_VERSION_HISTORY_RETENTION_DAYS",
30)
+ )
+ return _prune_old_versions_impl(retention_days)
+ except Exception: # pylint: disable=broad-except
+ logger.exception("version_history.prune_old_versions: task failed")
+ # Emit a failure counter so a prune that fails every night (e.g. an
+ # exhausted-retry serialization storm, or a non-OperationalError
+ # fault) is alertable via statsd rather than only via log inspection —
+ # this is the destructive job's primary failure mode.
+ stats_logger_manager.instance.incr(f"{_METRIC_PREFIX}.failed")
+ return {"error": 1}
diff --git a/superset/versioning/queries.py b/superset/versioning/queries.py
index 55fb5c830fe..29ffd103cb5 100644
--- a/superset/versioning/queries.py
+++ b/superset/versioning/queries.py
@@ -180,9 +180,10 @@ def current_version_number(model_cls: type[Model],
entity_id: int) -> int | None
version rows yet.
Note: this index is *unstable under retention pruning*. The scheduled
- retention task drops shadow rows older than the configured
- retention window, so the same integer can refer to different rows
- before and after a prune cycle. Use
+ :func:`prune_old_versions` task drops shadow rows whose owning
+ ``version_transaction`` is older than
+ :envvar:`SUPERSET_VERSION_HISTORY_RETENTION_DAYS`, so the same integer
+ can refer to different rows before and after a prune cycle. Use
:func:`current_live_transaction_id` for a stable identifier.
"""
count = _get_version_count(model_cls, entity_id)
@@ -381,11 +382,13 @@ def resolve_version_uuid(
transaction in Python because there's no portable SQL form for a
UUIDv5 derivation across PostgreSQL / MySQL / SQLite (Postgres has
``uuid_generate_v5``; the other two do not). The iteration count is
- bounded by the configured retention window worth of edits — the
- retention task ages older shadow rows out — so the
- practical N is at most a few hundred. If retention is ever
- disabled on a heavily-edited entity, this loop is the
- place to revisit.
+ roughly bounded by ``SUPERSET_VERSION_HISTORY_RETENTION_DAYS`` worth
+ of edits — the retention task ages older shadow rows out, though
+ transactions still anchoring a live row survive past the window
+ (their count is bounded by the entity's current size, not its edit
+ volume) — so the practical N is at most a few hundred. If retention
+ is ever disabled (``= 0``) on a heavily-edited entity, this loop is
+ the place to revisit.
Pass *entity* to skip the ``find_active_by_uuid`` lookup; see
:func:`list_versions` for the rationale.
diff --git a/tests/integration_tests/versioning/retention_prune_tests.py
b/tests/integration_tests/versioning/retention_prune_tests.py
new file mode 100644
index 00000000000..1e57b6dfe06
--- /dev/null
+++ b/tests/integration_tests/versioning/retention_prune_tests.py
@@ -0,0 +1,510 @@
+# 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 collections.abc import Iterator
+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
+ ) -> Iterator[None]:
+ """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.
+ """
+ import sqlalchemy as sa
+ from sqlalchemy_continuum import versioning_manager
+
+ from superset.initialization import SupersetAppInitializer
+
+ previous_versioning: bool = bool(
+ versioning_manager.options.get("versioning", True)
+ )
+ listeners_were_attached: bool = sa.event.contains(
+ sa.orm.Mapper,
+ "after_insert",
+ versioning_manager.track_inserts,
+ )
+
+ versioning_manager.options["versioning"] = True
+ SupersetAppInitializer._add_continuum_write_listeners()
+ try:
+ yield
+ finally:
+ if listeners_were_attached:
+ SupersetAppInitializer._add_continuum_write_listeners()
+ else:
+ SupersetAppInitializer._remove_continuum_write_listeners()
+ versioning_manager.options["versioning"] = previous_versioning
+
+ 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, timezone
+
+ 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.now(timezone.utc).replace(tzinfo=None)
+ - 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, timezone
+
+ 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: sa.Table = 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.now(timezone.utc).replace(tzinfo=None)
+ - 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: Dashboard = (
+ db.session.query(Dashboard)
+ .filter(Dashboard.dashboard_title == "USA Births Names")
+ .one()
+ )
+ chart: Slice = 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: set[int] = {row.transaction_id for row in
change_rows}
+ assert len(transaction_ids) == 1
+ transaction_id: int = 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, timezone
+ 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.now(timezone.utc).replace(tzinfo=None)
+ - 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: int = 0
+
+ def counting_fail(*args: Any, **kwargs: Any) -> dict[str, Any]:
+ nonlocal call_count
+ call_count += 1
+ return always_fail(*args, **kwargs)
+
+ with patch.object(
+ version_history_retention,
+ "_run_prune_pass",
+ side_effect=counting_fail,
+ ):
+ with patch.object(version_history_retention.time, "sleep"):
+ with pytest.raises(OperationalError):
+ _prune_old_versions_impl(retention_days=30)
+
+ assert call_count == _MAX_RETRY_ATTEMPTS, (
+ f"Expected exactly {_MAX_RETRY_ATTEMPTS} attempts; got
{call_count}"
+ )
diff --git a/tests/unit_tests/config_test.py b/tests/unit_tests/config_test.py
index 705f7f4073f..93cc6f9826d 100644
--- a/tests/unit_tests/config_test.py
+++ b/tests/unit_tests/config_test.py
@@ -47,6 +47,32 @@ FULL_DTTM_DEFAULTS_EXAMPLE = {
}
+def test_invalid_version_history_retention_env_uses_default(
+ monkeypatch: pytest.MonkeyPatch,
+ caplog: pytest.LogCaptureFixture,
+) -> None:
+ """Invalid retention input does not prevent configuration from loading."""
+ from superset import config
+
+ monkeypatch.setenv("SUPERSET_VERSION_HISTORY_RETENTION_DAYS", "30d")
+
+ assert config._parse_version_history_retention_days() == 30
+ assert "Invalid SUPERSET_VERSION_HISTORY_RETENTION_DAYS='30d'" in
caplog.text
+
+
+def test_oversized_version_history_retention_env_uses_default(
+ monkeypatch: pytest.MonkeyPatch,
+ caplog: pytest.LogCaptureFixture,
+) -> None:
+ """An oversized retention window cannot overflow cutoff arithmetic."""
+ from superset import config
+
+ monkeypatch.setenv("SUPERSET_VERSION_HISTORY_RETENTION_DAYS", "1000000000")
+
+ assert config._parse_version_history_retention_days() == 30
+ assert "exceeds the maximum" in caplog.text
+
+
def apply_dttm_defaults(table: "SqlaTable", dttm_defaults: dict[str, Any]) ->
None:
"""Applies dttm defaults to the table, mutates in place."""
for dbcol in table.columns:
diff --git a/tests/unit_tests/initialization_test.py
b/tests/unit_tests/initialization_test.py
index e3e6d24c6e3..860063905a0 100644
--- a/tests/unit_tests/initialization_test.py
+++ b/tests/unit_tests/initialization_test.py
@@ -16,6 +16,7 @@
# under the License.
import os
+from typing import Any
from unittest.mock import MagicMock, patch, PropertyMock
from sqlalchemy.exc import OperationalError
@@ -518,3 +519,163 @@ class TestAppRootMiddlewareBoundary:
assert status.startswith("200")
assert captured["PATH_INFO"] == "/welcome/"
assert captured["SCRIPT_NAME"] == "/myapp"
+
+
+class TestRetentionBeatWarning:
+ """Cover ``_warn_if_retention_beat_missing`` — the startup check that
+ surfaces a missing ``version_history.prune_old_versions`` beat entry.
+ (The ``ENABLE_VERSIONING_CAPTURE`` kill-switch branch of
+ ``init_versioning`` is covered by ``TestInitVersioning`` above.)
+
+ Operators who redefine ``CeleryConfig`` instead of subclassing or
+ merging the default silently lose the retention task; this pins that
+ the misconfiguration is logged at startup rather than discovered when
+ disk fills."""
+
+ def _initializer(self, config: dict[str, Any]) -> SupersetAppInitializer:
+ """Build a ``SupersetAppInitializer`` against a minimal mock app
+ whose only meaningful attribute is the config dict;
+ ``_warn_if_retention_beat_missing`` only reads from ``self.config``."""
+ app = MagicMock()
+ app.config = config
+ return SupersetAppInitializer(app)
+
+ @patch("superset.initialization.logger")
+ def test_warn_when_celery_beat_schedule_missing_retention_entry(
+ self, mock_logger: MagicMock
+ ) -> None:
+ """When ``CELERY_CONFIG.beat_schedule`` is present but lacks the
+ ``version_history.prune_old_versions`` entry, the helper emits
+ a WARNING. This guards the silent-failure mode where capture writes
+ rows but the prune never fires."""
+
+ class _PartialCeleryConfig:
+ beat_schedule: dict[str, dict[str, str]] = {
+ "reports.scheduler": {"task": "reports.scheduler"}
+ }
+
+ initializer = self._initializer({"CELERY_CONFIG":
_PartialCeleryConfig})
+ initializer._warn_if_retention_beat_missing()
+
+ assert any(
+ "version_history.prune_old_versions" in str(call)
+ for call in mock_logger.warning.call_args_list
+ ), (
+ "Expected a WARNING naming the missing retention entry; "
+ f"got {mock_logger.warning.call_args_list}"
+ )
+
+ @patch("superset.initialization.logger")
+ def test_no_warn_when_celery_beat_schedule_includes_retention_entry(
+ self, mock_logger: MagicMock
+ ) -> None:
+ """When the default ``CeleryConfig`` (or any class with the
+ entry) is in play, no warning fires. The happy path."""
+
+ class _CompleteCeleryConfig:
+ beat_schedule: dict[str, dict[str, str]] = {
+ "version_history.prune_old_versions": {
+ "task": "version_history.prune_old_versions",
+ },
+ }
+
+ initializer = self._initializer({"CELERY_CONFIG":
_CompleteCeleryConfig})
+ initializer._warn_if_retention_beat_missing()
+
+ mock_logger.warning.assert_not_called()
+
+ @patch("superset.initialization.logger")
+ def test_no_warn_when_retention_task_registered_under_other_key(
+ self, mock_logger: MagicMock
+ ) -> None:
+ """The retention task registered under a non-matching schedule key
+ (a valid Celery config) MUST NOT warn: the check matches on each
+ entry's ``task``, not the schedule key."""
+
+ class _RenamedKeyCeleryConfig:
+ beat_schedule: dict[str, dict[str, str]] = {
+ "prune_versions": {
+ "task": "version_history.prune_old_versions",
+ },
+ }
+
+ initializer = self._initializer({"CELERY_CONFIG":
_RenamedKeyCeleryConfig})
+ initializer._warn_if_retention_beat_missing()
+
+ mock_logger.warning.assert_not_called()
+
+ @patch("superset.initialization.logger")
+ def test_no_warn_when_celery_config_is_none(self, mock_logger: MagicMock)
-> None:
+ """``CELERY_CONFIG = None`` is the documented "disable Celery
+ entirely" path. The warn-log MUST NOT fire — the operator made
+ a deliberate choice; complaining about a missing retention entry
+ on a Celery-disabled deployment trains operators to ignore the
+ warning."""
+ initializer = self._initializer({"CELERY_CONFIG": None})
+ initializer._warn_if_retention_beat_missing()
+ mock_logger.warning.assert_not_called()
+
+ @patch("superset.initialization.logger")
+ def test_dict_form_celery_config_with_entry_does_not_warn(
+ self, mock_logger: MagicMock
+ ) -> None:
+ """Celery accepts a dict-shaped config via
+ ``config_from_object``. The warn-log MUST discriminate by
+ ``isinstance(dict)`` so an operator who supplies a dict with the
+ entry doesn't see a false-positive warning."""
+ initializer = self._initializer(
+ {
+ "CELERY_CONFIG": {
+ "broker_url": "redis://localhost",
+ "beat_schedule": {
+ "version_history.prune_old_versions": {
+ "task": "version_history.prune_old_versions",
+ },
+ },
+ },
+ }
+ )
+ initializer._warn_if_retention_beat_missing()
+ mock_logger.warning.assert_not_called()
+
+ @patch("superset.initialization.logger")
+ def test_dict_form_celery_config_without_entry_warns(
+ self, mock_logger: MagicMock
+ ) -> None:
+ """The dict-shape symmetry of the previous test: a dict without
+ the entry MUST emit the warning, same as a class without it."""
+ initializer = self._initializer(
+ {
+ "CELERY_CONFIG": {
+ "broker_url": "redis://localhost",
+ "beat_schedule": {
+ "reports.scheduler": {"task": "reports.scheduler"},
+ },
+ },
+ }
+ )
+ initializer._warn_if_retention_beat_missing()
+
+ assert any(
+ "version_history.prune_old_versions" in str(call)
+ for call in mock_logger.warning.call_args_list
+ ), (
+ "Expected a WARNING for dict-form CELERY_CONFIG missing the "
+ f"entry; got {mock_logger.warning.call_args_list}"
+ )
+
+ @patch("superset.initialization.logger")
+ def test_string_celery_config_reference_does_not_warn(
+ self, mock_logger: MagicMock
+ ) -> None:
+ """Dotted config references are resolved by Celery's own loader.
+
+ The startup diagnostic must not treat an unresolved reference as an
+ empty schedule and emit a false warning.
+ """
+ initializer = self._initializer(
+ {"CELERY_CONFIG": "custom_celery_config:CeleryConfig"}
+ )
+ initializer._warn_if_retention_beat_missing()
+
+ mock_logger.warning.assert_not_called()
diff --git
a/tests/unit_tests/migrations/test_version_transaction_issued_at_index.py
b/tests/unit_tests/migrations/test_version_transaction_issued_at_index.py
new file mode 100644
index 00000000000..5e429984d86
--- /dev/null
+++ b/tests/unit_tests/migrations/test_version_transaction_issued_at_index.py
@@ -0,0 +1,90 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""Tests for the version-transaction retention index migration."""
+
+from importlib import import_module
+from types import ModuleType
+
+import pytest
+from alembic.migration import MigrationContext
+from alembic.operations import Operations
+from sqlalchemy import (
+ Column,
+ create_engine,
+ DateTime,
+ inspect,
+ Integer,
+ MetaData,
+ Table,
+)
+from sqlalchemy.engine import Engine
+
+migration: ModuleType = import_module(
+ "superset.migrations.versions."
+ "2026-07-27_10-00_d3b9a1f6c204_version_transaction_issued_at_index"
+)
+
+
[email protected]
+def engine() -> Engine:
+ """Create a minimal pre-migration version_transaction table."""
+ engine = create_engine("sqlite:///:memory:")
+ metadata = MetaData()
+ Table(
+ migration.TABLE_NAME,
+ metadata,
+ Column("id", Integer, primary_key=True),
+ Column("issued_at", DateTime, nullable=False),
+ )
+ metadata.create_all(engine)
+ return engine
+
+
+def _indexes(engine: Engine) -> dict[str, list[str]]:
+ return {
+ index["name"]: index["column_names"]
+ for index in inspect(engine).get_indexes(migration.TABLE_NAME)
+ }
+
+
+def test_upgrade_creates_issued_at_index(engine: Engine) -> None:
+ with engine.begin() as connection:
+ context = MigrationContext.configure(connection)
+ with Operations.context(context):
+ migration.upgrade()
+
+ assert _indexes(engine)[migration.INDEX_NAME] == ["issued_at"]
+
+
+def test_downgrade_drops_issued_at_index(engine: Engine) -> None:
+ with engine.begin() as connection:
+ context = MigrationContext.configure(connection)
+ with Operations.context(context):
+ migration.upgrade()
+ migration.downgrade()
+
+ assert migration.INDEX_NAME not in _indexes(engine)
+
+
+def test_upgrade_is_idempotent(engine: Engine) -> None:
+ with engine.begin() as connection:
+ context = MigrationContext.configure(connection)
+ with Operations.context(context):
+ migration.upgrade()
+ migration.upgrade()
+
+ assert _indexes(engine)[migration.INDEX_NAME] == ["issued_at"]
diff --git a/tests/unit_tests/semantic_layers/mapper_test.py
b/tests/unit_tests/semantic_layers/mapper_test.py
index 3821455aca0..32d6f7d0eb4 100644
--- a/tests/unit_tests/semantic_layers/mapper_test.py
+++ b/tests/unit_tests/semantic_layers/mapper_test.py
@@ -3868,14 +3868,14 @@ def
test_map_query_object_shifts_time_offset_via_temporal_range_filter(
assert query.filters is not None
gte = next(
f.value
- for f in query.filters
+ for f in query.filters or ()
if f.column is not None
and f.column.name == "order_date"
and f.operator == Operator.GREATER_THAN_OR_EQUAL
)
lt = next(
f.value
- for f in query.filters
+ for f in query.filters or ()
if f.column is not None
and f.column.name == "order_date"
and f.operator == Operator.LESS_THAN
diff --git a/tests/unit_tests/tasks/test_version_history_retention.py
b/tests/unit_tests/tasks/test_version_history_retention.py
new file mode 100644
index 00000000000..bc72fcc3cc4
--- /dev/null
+++ b/tests/unit_tests/tasks/test_version_history_retention.py
@@ -0,0 +1,200 @@
+# 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.
+"""Unit tests for the operational instrumentation in
+``superset.tasks.version_history_retention``.
+
+Covers the branches that emit statsd counters: the ``retention_days <= 0``
+short-circuit, incomplete shadow-table resolution, the ``OperationalError``
+retry path, and the terminal failure counter. The
+"happy path" / SERIALIZABLE retry behaviour against a real database is
+exercised by ``tests/integration_tests/versioning/retention_prune_tests.py``;
+this file pins the metric-emission contract that is load-bearing for
+operator alerting.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Iterator
+from unittest.mock import MagicMock, patch
+
+import pytest
+from sqlalchemy.exc import OperationalError
+from sqlalchemy_continuum.exc import ClassNotVersioned
+
+from superset.tasks import version_history_retention
+
+
[email protected](name="stats")
+def _stats_fixture() -> Iterator[MagicMock]:
+ """Patch the shared stats logger so every test can assert on
+ emissions without standing up the real statsd backend."""
+ with patch.object(
+ version_history_retention, "stats_logger_manager"
+ ) as mock_manager:
+ mock_manager.instance = MagicMock()
+ yield mock_manager.instance
+
+
+def test_retention_disabled_emits_skipped_metric(stats: MagicMock) -> None:
+ """``retention_days <= 0`` is the documented "disable retention"
+ config. The early-return must emit
``superset.versioning.retention.skipped``
+ so a dashboard can tell "operator disabled it" apart from "scheduler
+ isn't running"."""
+ result =
version_history_retention._prune_old_versions_impl(retention_days=0)
+ assert result == {"skipped": 1}
+ stats.incr.assert_called_once_with("superset.versioning.retention.skipped")
+ stats.gauge.assert_not_called()
+
+
+def test_task_normalizes_string_retention_config(stats: MagicMock) -> None:
+ """String values from custom config modules are normalized to integers."""
+ mock_app: MagicMock = MagicMock()
+ mock_app.config = {"SUPERSET_VERSION_HISTORY_RETENTION_DAYS": "30"}
+ with (
+ patch.object(version_history_retention, "current_app", mock_app),
+ patch.object(
+ version_history_retention,
+ "_prune_old_versions_impl",
+ return_value={"pruned_transactions": 0},
+ ) as prune,
+ ):
+ result = version_history_retention.prune_old_versions()
+
+ assert result == {"pruned_transactions": 0}
+ prune.assert_called_once_with(30)
+ stats.incr.assert_not_called()
+
+
+def test_incomplete_shadow_table_resolution_fails_closed(
+ stats: MagicMock,
+) -> None:
+ """Missing shadow metadata must abort before the destructive pass."""
+ with (
+ patch.object(
+ version_history_retention,
+ "_resolve_shadow_tables",
+ side_effect=RuntimeError("missing shadow"),
+ ),
+ pytest.raises(RuntimeError, match="missing shadow"),
+ ):
+ version_history_retention._prune_old_versions_impl(retention_days=30)
+ stats.incr.assert_not_called()
+
+
+def test_resolve_shadow_tables_rejects_partial_registry() -> None:
+ """One missing versioned mapper makes the complete registry unsafe."""
+ resolved_table: MagicMock = MagicMock()
+
+ def resolve_version_class(model: type[object]) -> MagicMock:
+ if model.__name__ == "TableColumn":
+ raise ClassNotVersioned(model)
+ version_model = MagicMock()
+ version_model.__table__ = resolved_table
+ return version_model
+
+ with patch("sqlalchemy_continuum.version_class",
side_effect=resolve_version_class):
+ with pytest.raises(RuntimeError, match="TableColumn"):
+ version_history_retention._resolve_shadow_tables(MagicMock())
+
+
+def test_serialization_failure_then_success_increments_retried_once(
+ stats: MagicMock,
+) -> None:
+ """A single ``OperationalError`` on attempt 1 should:
+ * fire ``.retried`` once (one retry happened),
+ * sleep for ``_RETRY_BACKOFF_BASE_SECONDS`` (patched away in tests),
+ * succeed on attempt 2 with ``stats["retried"] == 1``,
+ * fire ``.pruned_transactions`` gauge with the success count.
+
+ The contract on ``.retried`` is "fires per retry attempt observed"
+ (per-attempt, not per-session). This test pins the per-attempt shape so
+ a future refactor doesn't silently change the metric semantics."""
+ pass_fn: MagicMock = MagicMock(
+ side_effect=[
+ OperationalError("SELECT 1", {}, Exception("could not serialize
access")),
+ {"pruned_transactions": 7, "cutoff": "2026-01-01T00:00:00"},
+ ]
+ )
+ tables = version_history_retention.ShadowTables(
+ parent=[MagicMock()], child=[MagicMock()], m2m=None,
transaction=MagicMock()
+ )
+ with (
+ patch.object(
+ version_history_retention, "_resolve_shadow_tables",
return_value=tables
+ ),
+ patch.object(version_history_retention, "_run_prune_pass", pass_fn),
+ patch.object(version_history_retention.time, "sleep"),
+ ):
+ result =
version_history_retention._prune_old_versions_impl(retention_days=30)
+
+ assert result["retried"] == 1
+ assert result["pruned_transactions"] == 7
+ incr_calls = [call.args[0] for call in stats.incr.call_args_list]
+ assert incr_calls == ["superset.versioning.retention.retried"], (
+ f"Expected exactly one .retried emission; got {incr_calls}"
+ )
+ stats.gauge.assert_called_once_with(
+ "superset.versioning.retention.pruned_transactions", 7
+ )
+
+
+def test_all_attempts_fail_reraises_after_max_retries(stats: MagicMock) ->
None:
+ """When every attempt raises ``OperationalError``, the task re-raises
+ after ``_MAX_RETRY_ATTEMPTS`` so the outer Celery wrapper logs it.
+ The retry counter fires once per attempt that hit the exception."""
+ exc: OperationalError = OperationalError("SELECT 1", {},
Exception("conflict"))
+ tables = version_history_retention.ShadowTables(
+ parent=[MagicMock()], child=[MagicMock()], m2m=None,
transaction=MagicMock()
+ )
+ with (
+ patch.object(
+ version_history_retention, "_resolve_shadow_tables",
return_value=tables
+ ),
+ patch.object(version_history_retention, "_run_prune_pass",
side_effect=exc),
+ patch.object(version_history_retention.time, "sleep"),
+ pytest.raises(OperationalError),
+ ):
+ version_history_retention._prune_old_versions_impl(retention_days=30)
+
+ incr_calls = [call.args[0] for call in stats.incr.call_args_list]
+ assert (
+ incr_calls.count("superset.versioning.retention.retried")
+ == version_history_retention._MAX_RETRY_ATTEMPTS
+ ), (
+ f"Expected {version_history_retention._MAX_RETRY_ATTEMPTS} "
+ f".retried emissions (one per attempt); got {incr_calls}"
+ )
+
+
+def test_terminal_failure_emits_failed_metric_and_swallows(stats: MagicMock)
-> None:
+ """The Celery wrapper catches a terminal failure, returns ``{"error": 1}``
+ (so the schedule isn't poisoned), AND emits a ``.failed`` counter so the
+ destructive job's primary failure mode is alertable, not just logged."""
+ mock_app: MagicMock = MagicMock()
+ mock_app.config = {"SUPERSET_VERSION_HISTORY_RETENTION_DAYS": 30}
+ with (
+ patch.object(version_history_retention, "current_app", mock_app),
+ patch.object(
+ version_history_retention,
+ "_prune_old_versions_impl",
+ side_effect=RuntimeError("boom"),
+ ),
+ ):
+ result = version_history_retention.prune_old_versions()
+
+ assert result == {"error": 1}
+ stats.incr.assert_called_once_with("superset.versioning.retention.failed")
