feat(versioning): version-history retention cleanup job

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`.

superset/config.py

@@ -1649,6 +1649,49 @@ def sync_theme_logo_href(
     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",

superset/initialization/__init__.py

@@ -775,6 +775,14 @@ def init_versioning(self) -> None:
         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 @@ 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 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:
         """

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:

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)

superset/tasks/celery_app.py

@@ -35,7 +35,7 @@
 # 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