chore(deps): bump github/codeql-action/init from 4.37.1 to 4.37.2

Bumps [github/codeql-action/init](https://github.com/github/codeql-action) from 4.37.1 to 4.37.2.
- [Release notes](https://github.com/github/codeql-action/releases)
- [Changelog](https://github.com/github/codeql-action/blob/main/CHANGELOG.md)
- [Commits](7188fc3636...e0647621c2)

---
updated-dependencies:
- dependency-name: github/codeql-action/init
  dependency-version: 4.37.2
  dependency-type: direct:production
  update-type: version-update:semver-patch
...

Signed-off-by: dependabot[bot] <support@github.com>

.github/workflows/codeql-analysis.yml

@@ -64,7 +64,7 @@ jobs:
 
       # Initializes the CodeQL tools for scanning.
       - name: Initialize CodeQL
-        uses: github/codeql-action/init@7188fc363630916deb702c7fdcf4e481b751f97a # v4.37.1
+        uses: github/codeql-action/init@e0647621c2984b5ed2f768cb892365bf2a616ad1 # v4.37.2
         with:
           languages: ${{ matrix.language }}
           # If you wish to specify custom queries, you can do so here or in a config file.

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-frontend/packages/superset-ui-core/src/components/MetadataBar/MetadataBar.tsx

@@ -89,6 +89,7 @@ const StyledItem = styled.div<{
     & .metadata-text {
       color: ${theme.colorTextSecondary};
       min-width: ${TEXT_MIN_WIDTH}px;
+      max-width: ${TEXT_MAX_WIDTH}px;
       overflow: hidden;
       text-overflow: ${collapsed ? 'unset' : 'ellipsis'};
       white-space: nowrap;

superset-frontend/packages/superset-ui-core/src/components/Select/Select.test.tsx

@@ -1032,6 +1032,35 @@ test('do not count unselected disabled options in "Select all"', async () => {
   ).toBeInTheDocument();
 });
 
+test('"Select all" does not count null-valued options', async () => {
+  // A falsy-valued option (e.g. <NULL>, value: null) is skipped by
+  // handleSelectAll, so it must not be counted in the "Select all" badge or
+  // the count overstates the selection. Regression test for #40228. Uses a
+  // local options array to stay isolated from tests that mutate OPTIONS.
+  const localOptions = [
+    { label: 'Alpha', value: 1 },
+    { label: 'Bravo', value: 2 },
+  ];
+  render(
+    <Select
+      {...defaultProps}
+      options={[...localOptions, NULL_OPTION]}
+      mode="multiple"
+      maxTagCount={0}
+    />,
+  );
+  await open();
+  // Three options are visible, but the <NULL> option is not bulk-selectable,
+  // so the badge must count only the two real options (would be 3 before fix).
+  await userEvent.click(
+    await screen.findByText(selectAllButtonText(localOptions.length)),
+  );
+  // And Select all selects exactly those two — the null option is skipped.
+  const values = await findAllSelectValues();
+  expect(values.length).toBe(1);
+  expect(values[0]).toHaveTextContent(`+ ${localOptions.length} ...`);
+});
+
 test('"Deselect all" counts all selected options', async () => {
   render(<Select {...defaultProps} allowNewOptions mode="multiple" />);
   await open();

superset-frontend/packages/superset-ui-core/src/components/Select/Select.tsx

@@ -332,7 +332,12 @@ const Select = forwardRef(
           const isDisabled = option.disabled;
           const isNew = option.isNewOption;
 
+          // Mirror handleSelectAll, which skips falsy-valued options (e.g. the
+          // <NULL> option whose value is null): they are not bulk-selectable,
+          // so counting them here makes the "Select all" badge overstate what
+          // gets selected.
           if (
+            option.value &&
             (!isDisabled || isSelected) &&
             ((isNew && isSelected) || !isNew)
           ) {

superset-frontend/src/dashboard/actions/hydrate.ts

@@ -27,7 +27,7 @@ import { initSliceEntities } from 'src/dashboard/reducers/sliceEntities';
 import { getInitialState as getInitialNativeFilterState } from 'src/dashboard/reducers/nativeFilters';
 import { applyDefaultFormData } from 'src/explore/store';
 import { buildActiveFilters } from 'src/dashboard/util/activeDashboardFilters';
-import { findPermission } from 'src/utils/findPermission';
+import { canDownloadData, findPermission } from 'src/utils/findPermission';
 import {
   canUserEditDashboard,
   canUserSaveAsDashboard,
@@ -369,7 +369,7 @@ export const hydrateDashboard =
             'Superset',
             roles,
           ),
-          superset_can_download: findPermission('can_csv', 'Superset', roles),
+          superset_can_download: canDownloadData(roles),
           common: {
             // legacy, please use state.common instead
             conf: common?.conf,

superset-frontend/src/utils/findPermission.test.ts

@@ -16,7 +16,16 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-import { findPermission } from './findPermission';
+import { isFeatureEnabled } from '@superset-ui/core';
+import { UserRoles } from 'src/types/bootstrapTypes';
+import { canDownloadData, findPermission } from './findPermission';
+
+jest.mock('@superset-ui/core', () => ({
+  ...jest.requireActual('@superset-ui/core'),
+  isFeatureEnabled: jest.fn(),
+}));
+
+const mockIsFeatureEnabled = isFeatureEnabled as jest.Mock;
 
 test('findPermission for single role', () => {
   expect(findPermission('abc', 'def', { role: [['abc', 'def']] })).toEqual(
@@ -61,3 +70,44 @@ test('findPermission for multiple roles', () => {
 test('handles nonexistent roles', () => {
   expect(findPermission('abc', 'def', null)).toEqual(false);
 });
+
+describe('canDownloadData', () => {
+  const csvOnly: UserRoles = { role: [['can_csv', 'Superset']] };
+  const exportOnly: UserRoles = { role: [['can_export_data', 'Superset']] };
+  const both: UserRoles = {
+    role: [
+      ['can_csv', 'Superset'],
+      ['can_export_data', 'Superset'],
+    ],
+  };
+  const neither: UserRoles = { role: [['can_write', 'Chart']] };
+
+  afterEach(() => {
+    mockIsFeatureEnabled.mockReset();
+  });
+
+  test('checks can_csv when GranularExportControls is off', () => {
+    mockIsFeatureEnabled.mockReturnValue(false);
+    expect(canDownloadData(csvOnly)).toEqual(true);
+    expect(canDownloadData(exportOnly)).toEqual(false);
+    expect(canDownloadData(both)).toEqual(true);
+    expect(canDownloadData(neither)).toEqual(false);
+  });
+
+  test('checks can_export_data only when GranularExportControls is on, matching the backend', () => {
+    mockIsFeatureEnabled.mockReturnValue(true);
+    expect(canDownloadData(exportOnly)).toEqual(true);
+    // no can_csv fallback: a can_csv-only user would 403 at the backend, so
+    // the button must not show
+    expect(canDownloadData(csvOnly)).toEqual(false);
+    expect(canDownloadData(both)).toEqual(true);
+    expect(canDownloadData(neither)).toEqual(false);
+  });
+
+  test('handles nonexistent roles', () => {
+    mockIsFeatureEnabled.mockReturnValue(true);
+    expect(canDownloadData(null)).toEqual(false);
+    mockIsFeatureEnabled.mockReturnValue(false);
+    expect(canDownloadData(undefined)).toEqual(false);
+  });
+});

superset-frontend/src/utils/findPermission.ts

@@ -17,6 +17,7 @@
  * under the License.
  */
 import memoizeOne from 'memoize-one';
+import { FeatureFlag, isFeatureEnabled } from '@superset-ui/core';
 import { UserRoles } from 'src/types/bootstrapTypes';
 
 export const findPermission = memoizeOne(
@@ -26,3 +27,15 @@ export const findPermission = memoizeOne(
       permissions.some(([perm_, view_]) => perm_ === perm && view_ === view),
     ),
 );
+
+/**
+ * Whether the user may download chart data (CSV, Excel). Mirrors what the
+ * backend enforces: with GranularExportControls enabled it checks the granular
+ * can_export_data permission, otherwise can_csv. The same shape as
+ * hydrateExplore and usePermissions, so the download button never shows for a
+ * user the backend would 403.
+ */
+export const canDownloadData = (roles?: UserRoles | null): boolean =>
+  isFeatureEnabled(FeatureFlag.GranularExportControls)
+    ? findPermission('can_export_data', 'Superset', roles)
+    : findPermission('can_csv', 'Superset', roles);

superset/commands/dataset/importers/v0.py

@@ -298,7 +298,7 @@ class ImportDatasetsCommand(BaseCommand):
         for file_name, content in self.contents.items():
             try:
                 config = yaml.safe_load(content)
-            except yaml.parser.ParserError as ex:
+            except yaml.YAMLError as ex:
                 logger.exception("Invalid YAML file")
                 raise IncorrectVersionError(
                     f"{file_name} is not a valid YAML file"

superset/commands/importers/v1/utils.py

@@ -59,7 +59,7 @@ def load_yaml(file_name: str, content: str) -> dict[str, Any]:
     """Try to load a YAML file"""
     try:
         return yaml.safe_load(content)
-    except yaml.parser.ParserError as ex:
+    except yaml.YAMLError as ex:
         logger.exception("Invalid YAML in %s", file_name)
         raise ValidationError({file_name: "Not a valid YAML file"}) from ex
 

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",

superset/daos/dashboard.py

@@ -397,7 +397,8 @@ class DashboardDAO(BaseDAO[Dashboard]):
             md["color_namespace"] = data.get("color_namespace")
 
         md["expanded_slices"] = data.get("expanded_slices", {})
-        md["refresh_frequency"] = data.get("refresh_frequency", 0)
+        if "refresh_frequency" in data:
+            md["refresh_frequency"] = data["refresh_frequency"]
         md["color_scheme"] = data.get("color_scheme", "")
         md["label_colors"] = data.get("label_colors", {})
         md["shared_label_colors"] = data.get("shared_label_colors", [])

superset/datasets/datetime_format_detector.py

@@ -122,8 +122,23 @@ class DatetimeFormatDetector:
             # This handles different SQL dialects (LIMIT, TOP, FETCH FIRST, etc.)
             sql = database.apply_limit_to_sql(sql, limit=self.sample_size, force=True)
 
-            # Execute query and get results
-            df = database.get_df(sql, dataset.schema)
+            # Execute query and get results. Failures here come from the
+            # target database itself (bad connection config, transient
+            # outage, permission errors, etc.), not from Superset's own
+            # logic. Format detection is a best-effort optimization with no
+            # user-facing impact when it's skipped, so log at WARNING
+            # instead of capturing an ERROR-level exception for every sample
+            # query a misconfigured/unreachable database rejects.
+            try:
+                df = database.get_df(sql, dataset.schema)
+            except Exception as ex:
+                logger.warning(
+                    "Could not query column %s.%s for format detection: %s",
+                    dataset.table_name,
+                    column.column_name,
+                    str(ex),
+                )
+                return None
 
             if df.empty or column.column_name not in df.columns:
                 logger.warning(

superset/db_engine_specs/druid.py

@@ -53,11 +53,6 @@ class DruidEngineSpec(BaseEngineSpec):
     type_probe_needs_row = True
     requires_column_value_normalization = True
 
-    encrypted_extra_sensitive_fields = {
-        "$.connect_args.jwt": "JWT Token",
-        "$.connect_args.password": "Password",
-    }
-
     metadata = {
         "description": (
             "Apache Druid is a high performance real-time analytics database."

superset/db_engine_specs/hive.py

@@ -308,11 +308,38 @@ class HiveEngineSpec(PrestoEngineSpec):
         catalog: str | None = None,
         schema: str | None = None,
     ) -> tuple[URL, dict[str, Any]]:
-        if schema:
-            uri = uri.set(database=parse.quote(schema, safe=""))
+        """
+        Return the URI and connection arguments unchanged.
 
+        This overrides the Presto implementation, whose ``catalog/schema`` URI
+        convention doesn't apply to PyHive URLs. The URI must also not be
+        rewritten to point at the selected schema: PyHive issues ``USE`` on the
+        URI database at connect time, so on backends where the URI database
+        selects a catalog (e.g. Spark Thrift Server) rewriting it would be seen
+        as a catalog change and break table resolution. Schema selection is
+        handled by :meth:`get_prequeries` instead.
+        """
         return uri, connect_args
 
+    @classmethod
+    def get_prequeries(
+        cls,
+        database: Database,
+        catalog: str | None = None,
+        schema: str | None = None,
+    ) -> list[str]:
+        """
+        Return a ``USE`` statement to select the schema for new connections.
+
+        A plain single-identifier ``USE`` is valid on both HiveServer2 and
+        Spark Thrift Server, and on the latter it resolves within the current
+        catalog rather than replacing it.
+        """
+        if schema:
+            escaped_schema = schema.replace("`", "``")
+            return [f"USE `{escaped_schema}`"]
+        return []
+
     @classmethod
     def get_schema_from_engine_params(
         cls,

superset/db_engine_specs/postgres.py

@@ -29,6 +29,7 @@ from sqlalchemy.dialects.postgresql import DOUBLE_PRECISION, ENUM, INTERVAL, JSO
 from sqlalchemy.dialects.postgresql.base import PGInspector
 from sqlalchemy.engine.reflection import Inspector
 from sqlalchemy.engine.url import URL
+from sqlalchemy.sql.expression import ColumnClause
 from sqlalchemy.types import Date, DateTime, String
 
 from superset.constants import TimeGrain
@@ -36,6 +37,7 @@ from superset.db_engine_specs.base import (
     BaseEngineSpec,
     BasicParametersMixin,
     DatabaseCategory,
+    TimestampExpression,
 )
 from superset.errors import ErrorLevel, SupersetError, SupersetErrorType
 from superset.exceptions import SupersetException, SupersetSecurityException
@@ -256,6 +258,31 @@ class PostgresBaseEngineSpec(BaseEngineSpec):
     def epoch_to_dttm(cls) -> str:
         return "(timestamp 'epoch' + {col} * interval '1 second')"
 
+    @classmethod
+    def get_timestamp_expr(
+        cls,
+        col: ColumnClause,
+        pdf: str | None,
+        time_grain: str | None,
+    ) -> TimestampExpression:
+        """
+        Construct a timestamp expression while preserving pure ``DATE`` semantics.
+
+        Applying ``DATE_TRUNC`` to a ``DATE`` column implicitly casts the value to
+        ``TIMESTAMP``, which can trigger unwanted timezone conversion on the client
+        and shift the displayed date by a day. To avoid this, the truncated value is
+        cast back to ``DATE`` when the source column is a pure ``DATE`` type.
+
+        See https://github.com/apache/superset/issues/42254.
+        """
+        expr = super().get_timestamp_expr(col, pdf, time_grain)
+        col_type = getattr(col, "type", None)
+        # ``DateTime``/``TIMESTAMP`` are distinct SQLAlchemy types (not subclasses
+        # of ``Date``), so this only matches pure ``DATE`` columns.
+        if time_grain and isinstance(col_type, Date):
+            return TimestampExpression(f"CAST({expr.name} AS DATE)", col, type_=Date())
+        return expr
+
     @classmethod
     def convert_dttm(
         cls, target_type: str, dttm: datetime, db_extra: dict[str, Any] | None = None

superset/db_engine_specs/presto.py

@@ -166,12 +166,6 @@ class PrestoBaseEngineSpec(BaseEngineSpec, metaclass=ABCMeta):
 
     supports_dynamic_schema = True
     supports_catalog = supports_dynamic_catalog = supports_cross_catalog_queries = True
-
-    encrypted_extra_sensitive_fields = {
-        "$.auth_params.password": "Password",
-        "$.auth_params.token": "JWT Token",
-        "$.connect_args.requests_kwargs.jwt": "JWT Token",
-    }
     # Not set here: GROUPING SETS support is opted in per-concrete-engine
     # (``PrestoEngineSpec``, ``TrinoEngineSpec``) rather than on this shared
     # base, since Hive-family descendants (``HiveEngineSpec``, ``SparkEngineSpec``,

superset/db_engine_specs/trino.py

@@ -73,11 +73,6 @@ class TrinoEngineSpec(PrestoBaseEngineSpec):
     allows_alias_to_source_column = False
     supports_grouping_sets = True
 
-    encrypted_extra_sensitive_fields = {
-        **PrestoBaseEngineSpec.encrypted_extra_sensitive_fields,
-        "$.oauth2_client_info.secret": "OAuth2 client secret",
-    }
-
     # The full set of columns Trino's "<table>$partitions" exposes for an
     # Iceberg table. The real partition keys are nested in the "partition" ROW,
     # so none of these are user partition columns.

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:
         """

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 @@ 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

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}

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.

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}"
+        )

tests/unit_tests/commands/importers/v1/utils_test.py

@@ -117,3 +117,24 @@ class TestConvertTemporalColumns:
 
         call_args = mock_logger.warning.call_args[0]
         assert call_args[1] == 2  # 2 out-of-bounds, 1 pre-existing null
+
+
+class TestLoadYaml:
+    def test_parser_error_raises_validation_error(self) -> None:
+        """A malformed flow sequence raises yaml.parser.ParserError."""
+        from marshmallow.exceptions import ValidationError
+
+        from superset.commands.importers.v1.utils import load_yaml
+
+        with pytest.raises(ValidationError):
+            load_yaml("test.yaml", "key: [unclosed")
+
+    def test_scanner_error_raises_validation_error(self) -> None:
+        """An unterminated quoted scalar raises yaml.scanner.ScannerError,
+        a sibling of ParserError under yaml.error.YAMLError."""
+        from marshmallow.exceptions import ValidationError
+
+        from superset.commands.importers.v1.utils import load_yaml
+
+        with pytest.raises(ValidationError):
+            load_yaml("test.yaml", 'key: "unterminated string')

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:

tests/unit_tests/dao/dashboard_test.py

@@ -24,6 +24,7 @@ from superset.connectors.sqla.models import Database, SqlaTable
 from superset.daos.dashboard import DashboardDAO
 from superset.models.dashboard import Dashboard
 from superset.models.slice import Slice
+from superset.utils import json
 from tests.unit_tests.conftest import with_feature_flags
 
 
@@ -117,3 +118,53 @@ def test_set_dash_metadata_preserves_soft_deleted_members(
     )
     # And the position slot kept its UUID rather than being nulled.
     assert positions["CHART-trashed"]["meta"]["uuid"] == str(trashed_chart.uuid)
+
+
+def test_set_dash_metadata_preserves_refresh_frequency(session: Session) -> None:
+    """set_dash_metadata must not reset refresh_frequency when absent from data.
+
+    Regression test for #42116: ``data.get("refresh_frequency", 0)`` would
+    unconditionally overwrite the existing value with 0 whenever the caller
+    did not include ``refresh_frequency`` in the data dict.
+    """
+    Dashboard.metadata.create_all(session.get_bind())
+
+    dashboard = Dashboard(
+        dashboard_title="refresh_test_dash",
+        json_metadata=json.dumps({"refresh_frequency": 30}),
+    )
+    db.session.add(dashboard)
+    db.session.flush()
+
+    # Simulate a save that does NOT include refresh_frequency
+    # (e.g. changing only the title via the PropertiesModal).
+    DashboardDAO.set_dash_metadata(dashboard, {"color_scheme": "superset"})
+
+    md = json.loads(dashboard.json_metadata)
+    assert md["refresh_frequency"] == 30, (
+        "refresh_frequency should be preserved when not present in data"
+    )
+
+
+def test_set_dash_metadata_updates_refresh_frequency_when_present(
+    session: Session,
+) -> None:
+    """set_dash_metadata must update refresh_frequency when it IS in data."""
+    Dashboard.metadata.create_all(session.get_bind())
+
+    dashboard = Dashboard(
+        dashboard_title="refresh_test_dash_2",
+        json_metadata=json.dumps({"refresh_frequency": 30}),
+    )
+    db.session.add(dashboard)
+    db.session.flush()
+
+    # Simulate a save that explicitly sets refresh_frequency to 0.
+    DashboardDAO.set_dash_metadata(
+        dashboard, {"refresh_frequency": 0, "color_scheme": "superset"}
+    )
+
+    md = json.loads(dashboard.json_metadata)
+    assert md["refresh_frequency"] == 0, (
+        "refresh_frequency should be updated when present in data"
+    )

tests/unit_tests/datasets/commands/importers/v0/import_test.py

@@ -75,3 +75,33 @@ def test_import_from_dict_skips_check_without_user(mocker: MockerFixture) -> Non
 
     can_access.assert_not_called()
     imported.assert_called_once()
+
+
+def test_validate_parser_error_raises_incorrect_version_error() -> None:
+    """
+    A YAML file malformed in a way that raises ``yaml.parser.ParserError``
+    (an unterminated flow sequence) is reported as an invalid version.
+    """
+    from superset.commands.dataset.importers import v0
+    from superset.commands.importers.exceptions import IncorrectVersionError
+
+    command = v0.ImportDatasetsCommand({"broken.yaml": "[1, 2"})
+
+    with pytest.raises(IncorrectVersionError):
+        command.validate()
+
+
+def test_validate_scanner_error_raises_incorrect_version_error() -> None:
+    """
+    A YAML file malformed in a way that raises ``yaml.scanner.ScannerError``
+    (an unterminated quoted string) is a sibling of ``ParserError`` under
+    ``yaml.YAMLError`` and must also be reported as an invalid version rather
+    than propagating as a raw, uncaught exception.
+    """
+    from superset.commands.dataset.importers import v0
+    from superset.commands.importers.exceptions import IncorrectVersionError
+
+    command = v0.ImportDatasetsCommand({"broken.yaml": 'key: "unterminated string'})
+
+    with pytest.raises(IncorrectVersionError):
+        command.validate()

tests/unit_tests/datasets/test_datetime_format_detector.py

@@ -16,6 +16,7 @@
 # under the License.
 """Tests for datetime format detector."""
 
+import logging
 from unittest.mock import MagicMock
 
 import pandas as pd
@@ -108,16 +109,48 @@ def test_detect_column_format_empty_data(
 
 
 def test_detect_column_format_error_handling(
-    mock_dataset: MagicMock, mock_column: MagicMock
+    mock_dataset: MagicMock, mock_column: MagicMock, caplog: pytest.LogCaptureFixture
 ) -> None:
-    """Test error handling during format detection."""
+    """Test error handling during format detection.
+
+    A failure while querying the target database (bad connection config,
+    transient outage, permission errors, etc.) is expected and already
+    fully handled -- it must not be captured as an ERROR-level exception,
+    since that floods Sentry with noise for every sample query a
+    misconfigured/unreachable database rejects.
+    """
     # Simulate database error
     mock_dataset.database.get_df.side_effect = Exception("Database error")
 
     detector = DatetimeFormatDetector()
-    detected_format = detector.detect_column_format(mock_dataset, mock_column)
+    with caplog.at_level(logging.WARNING):
+        detected_format = detector.detect_column_format(mock_dataset, mock_column)
 
     assert detected_format is None
+    assert not any(record.levelno >= logging.ERROR for record in caplog.records)
+    assert any(
+        record.levelno == logging.WARNING and "Could not query column" in record.message
+        for record in caplog.records
+    )
+
+
+def test_detect_column_format_internal_error_still_logs_at_error(
+    mock_dataset: MagicMock, mock_column: MagicMock, caplog: pytest.LogCaptureFixture
+) -> None:
+    """A genuine internal bug (not a database-query failure) should still be
+    logged at ERROR so it remains visible/actionable, unlike the expected
+    database-query failure case above."""
+    mock_dataset.database.get_sqla_engine.side_effect = RuntimeError(
+        "unexpected internal error"
+    )
+
+    detector = DatetimeFormatDetector()
+    with caplog.at_level(logging.WARNING):
+        detected_format = detector.detect_column_format(mock_dataset, mock_column)
+
+    assert detected_format is None
+    assert any(record.levelno >= logging.ERROR for record in caplog.records)
+    mock_dataset.database.get_df.assert_not_called()
 
 
 def test_detect_all_formats(mock_dataset: MagicMock) -> None:

tests/unit_tests/db_engine_specs/test_druid.py

@@ -211,50 +211,3 @@ def test_non_string_cursor_type_unaffected_by_druid_spec() -> None:
 
     col = result_set.columns[0]
     assert col["type"] == "INT"
-
-
-def test_mask_encrypted_extra() -> None:
-    """
-    Only the credentials inside `connect_args` should be masked, not the whole object.
-    """
-    from superset.db_engine_specs.druid import DruidEngineSpec
-    from superset.utils import json
-
-    config = json.dumps(
-        {
-            "connect_args": {
-                "scheme": "https",
-                "jwt": "my-secret-token",
-                "password": "my-password",
-            },
-        }
-    )
-
-    assert DruidEngineSpec.mask_encrypted_extra(config) == json.dumps(
-        {
-            "connect_args": {
-                "scheme": "https",
-                "jwt": "XXXXXXXXXX",
-                "password": "XXXXXXXXXX",
-            },
-        }
-    )
-
-
-def test_unmask_encrypted_extra() -> None:
-    """
-    Masked credentials are reused from the previous value; edited ones are kept.
-    """
-    from superset.db_engine_specs.druid import DruidEngineSpec
-    from superset.utils import json
-
-    old = json.dumps(
-        {"connect_args": {"scheme": "https", "jwt": "old-token", "password": "old"}}
-    )
-    new = json.dumps(
-        {"connect_args": {"scheme": "http", "jwt": "XXXXXXXXXX", "password": "new"}}
-    )
-
-    assert DruidEngineSpec.unmask_encrypted_extra(old, new) == json.dumps(
-        {"connect_args": {"scheme": "http", "jwt": "old-token", "password": "new"}}
-    )

tests/unit_tests/db_engine_specs/test_hive.py

@@ -64,6 +64,74 @@ def test_get_schema_from_engine_params() -> None:
     )
 
 
+@pytest.mark.parametrize(
+    "catalog,schema",
+    [
+        (None, None),
+        (None, "new_schema"),
+        ("catalog", None),
+        ("catalog", "new_schema"),
+    ],
+)
+def test_adjust_engine_params(catalog: Optional[str], schema: Optional[str]) -> None:
+    """
+    Test that ``adjust_engine_params`` leaves the URI untouched.
+
+    The URI database must not be rewritten to the selected schema: PyHive runs
+    ``USE`` on it at connect time, and on Spark Thrift Server it can select a
+    catalog, so overwriting it with the schema breaks table resolution (see
+    issue #30208). Schema selection happens via ``get_prequeries`` instead.
+    """
+    from superset.db_engine_specs.hive import HiveEngineSpec
+
+    uri = make_url("hive://localhost:10000/default")
+    connect_args = {"foo": "bar"}
+
+    adjusted_uri, adjusted_connect_args = HiveEngineSpec.adjust_engine_params(
+        uri,
+        connect_args,
+        catalog=catalog,
+        schema=schema,
+    )
+    assert adjusted_uri is uri
+    assert adjusted_connect_args is connect_args
+    assert connect_args == {"foo": "bar"}
+
+
+@pytest.mark.parametrize(
+    "catalog,schema,expected",
+    [
+        (None, None, []),
+        ("catalog", None, []),
+        (None, "new_schema", ["USE `new_schema`"]),
+        ("catalog", "new_schema", ["USE `new_schema`"]),
+        (None, "evil`schema", ["USE `evil``schema`"]),
+    ],
+)
+def test_get_prequeries(
+    mocker: MockerFixture,
+    catalog: Optional[str],
+    schema: Optional[str],
+    expected: list[str],
+) -> None:
+    """
+    Test that ``get_prequeries`` selects the schema with a ``USE`` statement.
+
+    Together with ``supports_dynamic_schema`` this implements per-query schema
+    selection, so unqualified table names resolve in the schema Superset
+    attributes the query to.
+    """
+    from superset.db_engine_specs.hive import HiveEngineSpec
+
+    assert HiveEngineSpec.supports_dynamic_schema
+
+    database = mocker.MagicMock()
+    assert (
+        HiveEngineSpec.get_prequeries(database, catalog=catalog, schema=schema)
+        == expected
+    )
+
+
 def test_select_star(mocker: MockerFixture) -> None:
     """
     Test the ``select_star`` method.

tests/unit_tests/db_engine_specs/test_postgres.py

@@ -22,6 +22,7 @@ from unittest.mock import MagicMock
 import pytest
 from pytest_mock import MockerFixture
 from sqlalchemy import column, types
+from sqlalchemy.dialects import postgresql
 from sqlalchemy.dialects.postgresql import DOUBLE_PRECISION, ENUM, INTERVAL, JSON
 from sqlalchemy.engine.interfaces import Dialect
 from sqlalchemy.engine.url import make_url
@@ -370,6 +371,52 @@ class TestRedshiftDetection:
         assert "pool_events" not in params
 
 
+def _compile(expr: Any) -> str:
+    return str(expr.compile(None, dialect=postgresql.dialect()))
+
+
+def test_get_timestamp_expr_date_column_casts_back_to_date() -> None:
+    """
+    DB Eng Specs (postgres): a time grain on a pure DATE column casts the
+    ``DATE_TRUNC`` result back to DATE to avoid timezone-driven date shifts.
+
+    See https://github.com/apache/superset/issues/42254.
+    """
+    col = column("event_date", type_=types.Date())
+    expr = spec.get_timestamp_expr(col, None, "P1D")
+    assert _compile(expr) == "CAST(DATE_TRUNC('day', event_date) AS DATE)"
+
+
+def test_get_timestamp_expr_datetime_column_not_cast() -> None:
+    """
+    DB Eng Specs (postgres): DATETIME/TIMESTAMP columns keep their timestamp
+    semantics and are not cast back to DATE.
+    """
+    col = column("event_ts", type_=types.DateTime())
+    expr = spec.get_timestamp_expr(col, None, "P1D")
+    assert _compile(expr) == "DATE_TRUNC('day', event_ts)"
+
+
+def test_get_timestamp_expr_date_column_without_grain_not_cast() -> None:
+    """
+    DB Eng Specs (postgres): without a time grain there is no DATE_TRUNC, so the
+    column is left untouched.
+    """
+    col = column("event_date", type_=types.Date())
+    expr = spec.get_timestamp_expr(col, None, None)
+    assert _compile(expr) == "event_date"
+
+
+def test_get_timestamp_expr_untyped_column_not_cast() -> None:
+    """
+    DB Eng Specs (postgres): columns without a known type (e.g. raw expressions)
+    are not cast to DATE.
+    """
+    col = column("some_expr")
+    expr = spec.get_timestamp_expr(col, None, "P1Y")
+    assert _compile(expr) == "DATE_TRUNC('year', some_expr)"
+
+
 def test_interval_type_mutator() -> None:
     """
     DB Eng Specs (postgres): Test INTERVAL type mutator

tests/unit_tests/db_engine_specs/test_presto.py

@@ -476,68 +476,3 @@ def test_partition_query_escapes_single_quote_in_filter_value(
     # by injected SQL) must NOT appear anywhere in the output — that would
     # mean the payload broke out of the literal.
     assert "'2024-01-01' UNION SELECT" not in sql
-
-
-def test_mask_encrypted_extra() -> None:
-    """
-    The sensitive `auth_params` values are masked, while `auth_method` and
-    non-sensitive fields such as `username` stay visible.
-    """
-    from superset.db_engine_specs.presto import PrestoEngineSpec
-    from superset.utils import json
-
-    config = json.dumps(
-        {
-            "auth_method": "basic",
-            "auth_params": {"username": "alice", "password": "my-password"},
-        }
-    )
-
-    assert PrestoEngineSpec.mask_encrypted_extra(config) == json.dumps(
-        {
-            "auth_method": "basic",
-            "auth_params": {"username": "alice", "password": "XXXXXXXXXX"},
-        }
-    )
-
-
-def test_mask_encrypted_extra_jwt_in_connect_args() -> None:
-    """
-    A JWT passed via `connect_args.requests_kwargs` is masked without touching
-    the surrounding connection settings.
-    """
-    from superset.db_engine_specs.presto import PrestoEngineSpec
-    from superset.utils import json
-
-    config = json.dumps(
-        {
-            "connect_args": {
-                "protocol": "https",
-                "requests_kwargs": {"jwt": "my-secret-token"},
-            },
-        }
-    )
-
-    assert PrestoEngineSpec.mask_encrypted_extra(config) == json.dumps(
-        {
-            "connect_args": {
-                "protocol": "https",
-                "requests_kwargs": {"jwt": "XXXXXXXXXX"},
-            },
-        }
-    )
-
-
-def test_unmask_encrypted_extra() -> None:
-    """
-    Masked credentials are reused from the previous value; edited ones are kept.
-    """
-    from superset.db_engine_specs.presto import PrestoEngineSpec
-    from superset.utils import json
-
-    old = json.dumps({"auth_method": "jwt", "auth_params": {"token": "old-token"}})
-    new = json.dumps({"auth_method": "jwt", "auth_params": {"token": "XXXXXXXXXX"}})
-
-    assert PrestoEngineSpec.unmask_encrypted_extra(old, new) == json.dumps(
-        {"auth_method": "jwt", "auth_params": {"token": "old-token"}}
-    )

tests/unit_tests/db_engine_specs/test_trino.py

@@ -1650,82 +1650,3 @@ def test_handle_boolean_filter() -> None:
         str(result_computed.compile(compile_kwargs={"literal_binds": True}))
         == "(expiration = 1) = true"
     )
-
-
-def test_mask_encrypted_extra() -> None:
-    """
-    All `auth_params` values and the OAuth2 client secret are masked, while
-    `auth_method` and other non-sensitive fields stay visible.
-    """
-    from superset.db_engine_specs.trino import TrinoEngineSpec
-
-    config = json.dumps(
-        {
-            "auth_method": "jwt",
-            "auth_params": {"token": "my-secret-token"},
-            "oauth2_client_info": {"id": "client-id", "secret": "my-secret"},
-        }
-    )
-
-    assert TrinoEngineSpec.mask_encrypted_extra(config) == json.dumps(
-        {
-            "auth_method": "jwt",
-            "auth_params": {"token": "XXXXXXXXXX"},
-            "oauth2_client_info": {"id": "client-id", "secret": "XXXXXXXXXX"},
-        }
-    )
-
-
-def test_mask_encrypted_extra_jwt_in_connect_args() -> None:
-    """
-    A JWT passed via `connect_args.requests_kwargs` is masked without touching
-    the surrounding connection settings.
-    """
-    from superset.db_engine_specs.trino import TrinoEngineSpec
-
-    config = json.dumps(
-        {
-            "connect_args": {
-                "protocol": "https",
-                "requests_kwargs": {"jwt": "my-secret-token"},
-            },
-        }
-    )
-
-    assert TrinoEngineSpec.mask_encrypted_extra(config) == json.dumps(
-        {
-            "connect_args": {
-                "protocol": "https",
-                "requests_kwargs": {"jwt": "XXXXXXXXXX"},
-            },
-        }
-    )
-
-
-def test_unmask_encrypted_extra() -> None:
-    """
-    Masked credentials are reused from the previous value; edited ones are kept.
-    """
-    from superset.db_engine_specs.trino import TrinoEngineSpec
-
-    old = json.dumps(
-        {
-            "auth_method": "basic",
-            "auth_params": {"username": "alice", "password": "old-password"},
-        }
-    )
-    # `username` is not masked on read, so it comes back in cleartext; only the
-    # masked `password` is revealed from the previous value.
-    new = json.dumps(
-        {
-            "auth_method": "basic",
-            "auth_params": {"username": "alice", "password": "XXXXXXXXXX"},
-        }
-    )
-
-    assert TrinoEngineSpec.unmask_encrypted_extra(old, new) == json.dumps(
-        {
-            "auth_method": "basic",
-            "auth_params": {"username": "alice", "password": "old-password"},
-        }
-    )

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()

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"]

tests/unit_tests/result_set_test.py

@@ -571,3 +571,53 @@ def test_stringify_values_non_serializable_dict_falls_back_to_str() -> None:
     # Must not raise — falls back to str()
     result = stringify_values(data)
     assert result[0] == str({"key": _Unserializable()})
+
+
+def test_empty_result_set_preserves_column_metadata() -> None:
+    """
+    Test that column metadata is preserved when query returns zero rows.
+
+    When a query returns no data but has a valid cursor description, the
+    column names and types from cursor_description should be preserved
+    in the result set. This allows downstream consumers (like the UI)
+    to display column headers even for empty result sets.
+    """
+    data: DbapiResult = []
+    description = [
+        ("id", "int", None, None, None, None, True),
+        ("name", "varchar", None, None, None, None, True),
+        ("created_at", "timestamp", None, None, None, None, True),
+    ]
+
+    result_set = SupersetResultSet(
+        data,
+        description,  # type: ignore
+        BaseEngineSpec,
+    )
+
+    # Verify column count
+    assert len(result_set.columns) == 3
+
+    # Verify column names are preserved
+    column_names = [col["column_name"] for col in result_set.columns]
+    assert column_names == ["id", "name", "created_at"]
+
+    assert result_set.columns[0]["type"] == BaseEngineSpec.get_datatype(
+        description[0][1]
+    )
+    assert result_set.columns[1]["type"] == BaseEngineSpec.get_datatype(
+        description[1][1]
+    )
+    assert result_set.columns[2]["type"] == BaseEngineSpec.get_datatype(
+        description[2][1]
+    )
+
+    # Verify the PyArrow table has the correct schema
+    assert result_set.table.num_rows == 0
+    assert len(result_set.table.column_names) == 3
+    assert list(result_set.table.column_names) == ["id", "name", "created_at"]
+
+    # Verify DataFrame conversion works
+    df = result_set.to_pandas_df()
+    assert len(df) == 0
+    assert list(map(str, df.columns)) == ["id", "name", "created_at"]

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

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")