diff --git a/UPDATING.md b/UPDATING.md index 1f2e73765e2d..aac6792c8fb8 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 4eb87cea2e4c..b7c71f6b610c 100644 --- a/superset/config.py +++ b/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", diff --git a/superset/initialization/__init__.py b/superset/initialization/__init__.py index 7cf9e49a0c8c..36a5fc31bd89 100644 --- a/superset/initialization/__init__.py +++ b/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: """ diff --git a/superset/mcp_service/sql_lab/tool/execute_sql.py b/superset/mcp_service/sql_lab/tool/execute_sql.py index 878a46c8b188..12351827b4b2 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 000000000000..f59e1269ba86 --- /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 96d648968953..4189a9a69f47 100644 --- a/superset/tasks/celery_app.py +++ b/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 diff --git a/superset/tasks/version_history_retention.py b/superset/tasks/version_history_retention.py new file mode 100644 index 000000000000..b9aca62fa84f --- /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 55fb5c830fed..29ffd103cb50 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 000000000000..1e57b6dfe06f --- /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 705f7f4073fb..93cc6f9826d2 100644 --- a/tests/unit_tests/config_test.py +++ b/tests/unit_tests/config_test.py @@ -47,6 +47,32 @@ } +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 e3e6d24c6e36..860063905a05 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 @@ def test_trailing_slash_app_root_is_normalized(self): 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 000000000000..5e429984d863 --- /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" +) + + +@pytest.fixture +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 3821455aca07..32d6f7d0eb46 100644 --- a/tests/unit_tests/semantic_layers/mapper_test.py +++ b/tests/unit_tests/semantic_layers/mapper_test.py @@ -3868,14 +3868,14 @@ def time_filter_bounds(query: SemanticQuery) -> tuple[datetime, datetime]: 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 000000000000..bc72fcc3cc46 --- /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 + + +@pytest.fixture(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")