feat(ci): publish Python unit test results as PR check annotations

Right now a red check on the Python-Unit job just says "failed" and
sends contributors to raw pytest logs to find out which test broke.
Add EnricoMi/publish-unit-test-result-action so failing tests get
annotated inline on the PR diff instead.

pytest already writes JUnit XML for free via --junit-xml; wire that up
for all three pytest invocations in the unit-tests job and upload it as
an artifact (uploaded even on failure, since that's exactly when it's
needed). A second workflow, triggered via workflow_run (same pattern
already used for the translation-regression bot), downloads that
artifact and publishes the check run. workflow_run always runs in the
base-branch context, so it can safely be granted checks:write even for
PRs from forks or Dependabot, without ever checking out untrusted PR
code.

Scoped to Python-Unit only for now; the frontend's 8-way jest shard and
the Python integration test matrix are natural follow-ups once this
pattern proves out.

.github/workflows/superset-docs-deploy.yml

@@ -18,6 +18,16 @@ on:
 
   workflow_dispatch: {}
 
+# Serialize deploys: the action pushes to apache/superset-site without
+# rebasing, so concurrent runs race on the final push and the loser fails
+# with `! [rejected] asf-site -> asf-site (fetch first)`. Cancel any
+# in-progress run as soon as a newer one starts — the destination repo
+# isn't touched until the final push step, so canceling mid-build is safe,
+# and the freshest content always wins.
+concurrency:
+  group: docs-deploy-asf-site
+  cancel-in-progress: true
+
 permissions:
   contents: read
 
@@ -37,68 +47,17 @@ jobs:
 
         env:
           SUPERSET_SITE_BUILD: ${{ (secrets.SUPERSET_SITE_BUILD != '' && secrets.SUPERSET_SITE_BUILD != '') || '' }}
-
-  # Master gets frequent, sometimes bursty pushes, and each one can trigger a
-  # deploy attempt. Rather than let every superseded attempt get force-killed
-  # by the build-deploy concurrency group below (which shows up as a
-  # `cancelled` — i.e. red/failing-looking — check on that commit), have each
-  # run check up front whether it's still building master's current tip and,
-  # if not, skip cleanly. Deliberately outside the docs-deploy-asf-site
-  # concurrency group so it runs immediately for every trigger without
-  # blocking or being blocked by anything.
-  check-freshness:
-    runs-on: ubuntu-26.04
-    outputs:
-      is-current: ${{ steps.check.outputs.is-current }}
-    steps:
-      - name: "Check whether this is still master's current commit"
-        id: check
-        env:
-          GH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
-          BUILD_SHA: ${{ github.event.workflow_run.head_sha || github.sha }}
-          EVENT_NAME: ${{ github.event_name }}
-        run: |
-          # workflow_dispatch is a deliberate, one-off action (e.g. to deploy
-          # a specific ref) rather than something racing other triggers, so
-          # it always bypasses the freshness gate below.
-          if [ "${EVENT_NAME}" = "workflow_dispatch" ]; then
-            echo "is-current=true" >> "$GITHUB_OUTPUT"
-            exit 0
-          fi
-
-          latest_sha="$(gh api "repos/${{ github.repository }}/commits/master" --jq .sha)"
-          if [ "${latest_sha}" = "${BUILD_SHA}" ]; then
-            echo "is-current=true" >> "$GITHUB_OUTPUT"
-          else
-            echo "is-current=false" >> "$GITHUB_OUTPUT"
-            echo "::notice::master has moved on to ${latest_sha} since ${BUILD_SHA} was triggered — a newer run will deploy it, so skipping this one."
-          fi
-
   build-deploy:
-    needs: [config, check-freshness]
-    # Only the run for master's current tip proceeds; anything superseded
-    # already skipped at check-freshness above instead of landing here.
+    needs: config
     # For workflow_run triggers, only deploy when the triggering run originated
     # from this repository (not a fork), ensuring the checked-out code and any
     # local actions executed with deploy credentials are trusted.
     if: >-
       needs.config.outputs.has-secrets &&
-      needs.check-freshness.outputs.is-current == 'true' &&
       (github.event_name != 'workflow_run' ||
        github.event.workflow_run.head_repository.full_name == github.repository)
     name: Build & Deploy
     runs-on: ubuntu-26.04
-    # Serialize deploys: the action pushes to apache/superset-site without
-    # rebasing, so concurrent runs race on the final push and the loser fails
-    # with `! [rejected] asf-site -> asf-site (fetch first)`. Cancel any
-    # in-progress run as soon as a newer one starts — the destination repo
-    # isn't touched until the final push step, so canceling mid-build is safe,
-    # and the freshest content always wins. The check-freshness gate above
-    # means it should be rare for more than one run to reach this point, but
-    # this stays as a backstop against the actual push race.
-    concurrency:
-      group: docs-deploy-asf-site
-      cancel-in-progress: true
     steps:
       - name: "Checkout ${{ github.event.workflow_run.head_sha || github.sha }}"
         uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
@@ -162,29 +121,7 @@ jobs:
         working-directory: docs
         run: |
           yarn build
-      # The check-freshness job above narrows the window but doesn't close it: an
-      # older run can observe is-current=true, then sit through this build while a
-      # newer run's own freshness check also passes and it deploys and finishes
-      # first. If this (stale) run then wins entry into the concurrency group, it
-      # would overwrite the newer content that already deployed. Re-check right
-      # before the one step that actually mutates superset-site, so a stale run
-      # skips deploying instead of clobbering a fresher one that already ran.
-      - name: "Re-check freshness immediately before deploying"
-        id: recheck-freshness
-        if: github.event_name != 'workflow_dispatch'
-        env:
-          GH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
-          BUILD_SHA: ${{ github.event.workflow_run.head_sha || github.sha }}
-        run: |
-          latest_sha="$(gh api "repos/${{ github.repository }}/commits/master" --jq .sha)"
-          if [ "${latest_sha}" = "${BUILD_SHA}" ]; then
-            echo "still-current=true" >> "$GITHUB_OUTPUT"
-          else
-            echo "still-current=false" >> "$GITHUB_OUTPUT"
-            echo "::notice::master has moved on to ${latest_sha} since ${BUILD_SHA} was triggered — skipping deploy of stale content."
-          fi
       - name: deploy docs
-        if: github.event_name == 'workflow_dispatch' || steps.recheck-freshness.outputs.still-current == 'true'
         uses: ./.github/actions/github-action-push-to-another-repository
         env:
           API_TOKEN_GITHUB: ${{ secrets.SUPERSET_SITE_BUILD }}

.github/workflows/superset-python-unittest-report.yml

@@ -0,0 +1,65 @@
+name: Python Unit Test Results
+
+on:
+  # zizmor: ignore[dangerous-triggers] - runs in base-branch context and only consumes artifacts uploaded by Python-Unit; never checks out PR code (see note below)
+  workflow_run:
+    workflows: ["Python-Unit"]
+    types: [completed]
+
+# This workflow publishes a check run annotating failing Python unit tests
+# inline on the PR diff, using JUnit XML uploaded by the Python-Unit workflow.
+# It uses the workflow_run trigger so that it always runs in the base-branch
+# context and can safely be granted write permissions, even for PRs from
+# forks or Dependabot.
+#
+# IMPORTANT: This workflow must NEVER check out code from the PR branch. All
+# data comes from artifacts uploaded by the Python-Unit workflow.
+permissions:
+  contents: read
+  checks: write
+  issues: read
+  actions: read
+
+jobs:
+  report:
+    runs-on: ubuntu-26.04
+    timeout-minutes: 10
+    if: >
+      github.event.workflow_run.conclusion == 'success' ||
+      github.event.workflow_run.conclusion == 'failure'
+    steps:
+      # Fails soft (continue-on-error) because the source unit-tests job is
+      # itself gated on change detection: a docs-only PR skips it entirely,
+      # so there is nothing to download or report on.
+      - name: Download JUnit results
+        id: download
+        continue-on-error: true
+        uses: actions/download-artifact@3e5f45b2cfb9172054b4087a40e8e0b5a5461e7c # v8
+        with:
+          pattern: "junit-results-*"
+          path: artifacts
+          merge-multiple: true
+          run-id: ${{ github.event.workflow_run.id }}
+          github-token: ${{ secrets.GITHUB_TOKEN }}
+
+      - name: Download event file
+        id: download-event
+        if: steps.download.outcome == 'success'
+        continue-on-error: true
+        uses: actions/download-artifact@3e5f45b2cfb9172054b4087a40e8e0b5a5461e7c # v8
+        with:
+          name: "Event File"
+          path: event
+          run-id: ${{ github.event.workflow_run.id }}
+          github-token: ${{ secrets.GITHUB_TOKEN }}
+
+      - name: Publish test results
+        if: steps.download.outcome == 'success' && steps.download-event.outcome == 'success'
+        uses: EnricoMi/publish-unit-test-result-action@d0a4676d0e0b938bc201470d88276b7c74c712b3 # v2.24.0
+        with:
+          commit: ${{ github.event.workflow_run.head_sha }}
+          event_file: event/event.json
+          event_name: ${{ github.event.workflow_run.event }}
+          files: "artifacts/**/*.xml"
+          check_name: "Python Unit Test Results"
+          comment_mode: "off"

.github/workflows/superset-python-unittest.yml

@@ -74,14 +74,14 @@ jobs:
           SUPERSET_TESTENV: true
           SUPERSET_SECRET_KEY: not-a-secret
         run: |
-          pytest --durations-min=0.5 --cov-report= --cov=superset ./tests/common ./tests/unit_tests --cache-clear --maxfail=50
+          pytest --durations-min=0.5 --cov-report= --cov=superset ./tests/common ./tests/unit_tests --cache-clear --maxfail=50 --junit-xml=test-results/junit-unit.xml
       - name: Python 100% coverage unit tests
         env:
           SUPERSET_TESTENV: true
           SUPERSET_SECRET_KEY: not-a-secret
         run: |
-          pytest --durations-min=0.5 --cov=superset/sql/ ./tests/unit_tests/sql/ --cache-clear --cov-fail-under=100
-          pytest --durations-min=0.5 --cov=superset/semantic_layers/ ./tests/unit_tests/semantic_layers/ --cache-clear --cov-fail-under=100
+          pytest --durations-min=0.5 --cov=superset/sql/ ./tests/unit_tests/sql/ --cache-clear --cov-fail-under=100 --junit-xml=test-results/junit-sql-coverage.xml
+          pytest --durations-min=0.5 --cov=superset/semantic_layers/ ./tests/unit_tests/semantic_layers/ --cache-clear --cov-fail-under=100 --junit-xml=test-results/junit-semantic-layers-coverage.xml
       - name: Upload code coverage
         uses: codecov/codecov-action@fb8b3582c8e4def4969c97caa2f19720cb33a72f # v7.0.0
         with:
@@ -89,6 +89,33 @@ jobs:
           verbose: true
           use_oidc: true
           slug: apache/superset
+      # Uploaded even when a pytest step above fails, since that is exactly
+      # when the JUnit results are needed downstream, to annotate the PR with
+      # the failing tests. Consumed by the "Python Unit Test Results" workflow
+      # via workflow_run (see that workflow for why it can't just be a step
+      # here: it needs to run with write permissions, which this PR-triggered
+      # job can't safely have on a fork PR).
+      - name: Upload JUnit test results
+        if: always()
+        uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7
+        with:
+          name: junit-results-${{ matrix.python-version }}
+          path: test-results/
+          retention-days: 7
+
+  # Uploads the raw pull_request event payload so the "Python Unit Test
+  # Results" workflow (running via workflow_run, in base-branch context) can
+  # look up which PR/commit to annotate without checking out untrusted code.
+  event-file:
+    runs-on: ubuntu-26.04
+    timeout-minutes: 5
+    steps:
+      - name: Upload event file
+        uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7
+        with:
+          name: Event File
+          path: ${{ github.event_path }}
+          retention-days: 7
 
   # Stable required-status-check anchor. `unit-tests` is a matrix job gated on
   # change detection, so on non-Python PRs it is skipped and never produces its

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/plugins/legacy-preset-chart-nvd3/src/vendor/superset/exploreUtils.ts

@@ -1,69 +0,0 @@
-/**
- * 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.
- */
-// @ts-nocheck -- vendor file; not fully typed
-/* eslint camelcase: 0 */
-import URI from 'urijs';
-import safeStringify from 'fast-safe-stringify';
-
-const MAX_URL_LENGTH = 8000;
-
-export function getURIDirectory(formData, endpointType = 'base') {
-  // Building the directory part of the URI
-  let directory = '/explore/';
-  if (['json', 'csv', 'query', 'results', 'samples'].includes(endpointType)) {
-    directory = '/superset/explore_json/';
-  }
-
-  return directory;
-}
-
-export function getExploreLongUrl(
-  formData,
-  endpointType,
-  allowOverflow = true,
-  extraSearch = {},
-) {
-  if (!formData.datasource) {
-    return null;
-  }
-
-  const uri = new URI('/');
-  const directory = getURIDirectory(formData, endpointType);
-  const search = uri.search(true);
-  Object.keys(extraSearch).forEach(key => {
-    search[key] = extraSearch[key];
-  });
-  search.form_data = safeStringify(formData);
-  if (endpointType === 'standalone') {
-    search.standalone = 'true';
-  }
-  const url = uri.directory(directory).search(search).toString();
-  if (!allowOverflow && url.length > MAX_URL_LENGTH) {
-    const minimalFormData = {
-      datasource: formData.datasource,
-      viz_type: formData.viz_type,
-    };
-
-    return getExploreLongUrl(minimalFormData, endpointType, false, {
-      URL_IS_TOO_LONG_TO_SHARE: null,
-    });
-  }
-
-  return url;
-}

superset-frontend/plugins/preset-chart-deckgl/src/utils/explore.ts

@@ -26,7 +26,7 @@ export function getURIDirectory(endpointType = 'base') {
   // Building the directory part of the URI
   let directory = '/explore/';
   if (['json', 'csv', 'query', 'results', 'samples'].includes(endpointType)) {
-    directory = '/superset/explore_json/';
+    directory = '/explore_json/';
   }
 
   return directory;

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/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/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/mysql.py

@@ -250,12 +250,9 @@ class MySQLEngineSpec(BasicParametersMixin, BaseEngineSpec):
 
     _time_grain_expressions = {
         None: "{col}",
-        TimeGrain.SECOND: "DATE_ADD(DATE({col}), "
-        "INTERVAL (HOUR({col})*60*60 + MINUTE({col})*60"
-        " + SECOND({col})) SECOND)",
-        TimeGrain.MINUTE: "DATE_ADD(DATE({col}), "
-        "INTERVAL (HOUR({col})*60 + MINUTE({col})) MINUTE)",
-        TimeGrain.HOUR: "DATE_ADD(DATE({col}), INTERVAL HOUR({col}) HOUR)",
+        TimeGrain.SECOND: "DATE_FORMAT({col}, '%Y-%m-%d %H:%i:%s')",
+        TimeGrain.MINUTE: "DATE_FORMAT({col}, '%Y-%m-%d %H:%i:00')",
+        TimeGrain.HOUR: "DATE_FORMAT({col}, '%Y-%m-%d %H:00:00')",
         TimeGrain.DAY: "DATE({col})",
         TimeGrain.WEEK: "DATE(DATE_SUB({col}, INTERVAL DAYOFWEEK({col}) - 1 DAY))",
         TimeGrain.MONTH: "DATE(DATE_SUB({col}, INTERVAL DAYOFMONTH({col}) - 1 DAY))",

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/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/shared/native_filters.py

@@ -335,7 +335,7 @@ def migrate_dashboard(dashboard: Dashboard) -> None:  # noqa: C901
 
                 meta["code"] = dedent(
                     f"""
-                        &#9888; The <a href="/superset/slice/{slc.id}/">{slc.slice_name}
+                        &#9888; The <a href="/slice/{slc.id}/">{slc.slice_name}
                         </a> filter-box chart has been migrated to a native filter.
                         """
                 )

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/security/manager.py

@@ -568,6 +568,67 @@ def freeze_value(value: Any) -> str:
     return json.dumps(_strip_overridable_keys(value), sort_keys=True)
 
 
+# Frontend-only markers that ``normalizeTimeColumn`` adds when it synthesizes a
+# chart's x-axis into a ``BASE_AXIS`` column. Like ``timeGrain`` they decorate
+# the column without changing which data is queried, so they must not count as
+# payload tampering.
+BASE_AXIS_SYNTHETIC_KEYS = frozenset({"columnType", "isColumnReference"})
+
+
+def _denormalize_base_axis_column(value: Any) -> Any:
+    """
+    Reduce a synthesized ``BASE_AXIS`` x-axis column back to the reference it
+    stands for.
+
+    Before a chart is queried, ``normalizeTimeColumn`` (superset-ui-core,
+    ``normalizeTimeColumn.ts``) replaces the chart's saved x-axis in the query
+    with an adhoc column tagged ``columnType: "BASE_AXIS"``:
+
+    * for a *physical* x-axis it emits a pure column reference
+      (``isColumnReference: true`` with ``sqlExpression`` == ``label`` == the
+      physical column name), and
+    * for an *adhoc* x-axis it copies the saved adhoc column and adds the
+      markers (plus the time grain, already dropped by ``freeze_value``).
+
+    Neither form selects data beyond the saved x-axis, yet neither appears
+    verbatim in the chart's stored ``params``/``query_context`` (the x-axis is
+    stored under its own ``x_axis`` control, and for charts whose saved
+    ``query_context`` is NULL there is nothing to compare against at all). So a
+    guest merely loading such a chart would otherwise be rejected as a tamperer.
+
+    This collapses the synthesized column to its underlying reference, so it
+    compares equal to the stored x-axis: a physical reference becomes its column
+    name, and an adhoc reference becomes the underlying adhoc column without the
+    synthesized markers. The collapsed value must still match a value stored on
+    the chart, so tagging an unrelated column or free-form SQL as ``BASE_AXIS``
+    grants no additional access. Non-``BASE_AXIS`` values are returned unchanged.
+
+    This has to reduce the value rather than merely drop keys the way
+    ``GUEST_OVERRIDABLE_VALUE_KEYS`` handles ``timeGrain``: a physical x-axis is
+    stored as a bare string (e.g. ``"order_date"``) but sent as a dict, so no
+    amount of key-stripping makes the two compare equal; the dict must collapse
+    back to the string.
+    """
+    if not isinstance(value, dict) or value.get("columnType") != "BASE_AXIS":
+        return value
+    if value.get("isColumnReference") and isinstance(value.get("sqlExpression"), str):
+        return value["sqlExpression"]
+    return {k: v for k, v in value.items() if k not in BASE_AXIS_SYNTHETIC_KEYS}
+
+
+def _payload_value_identity(value: Any, *, is_metric: bool) -> str:
+    """
+    Comparison identity for a column or metric in the guest anti-tamper check.
+
+    Metrics compare by exact frozen value. Columns additionally collapse a
+    synthesized ``BASE_AXIS`` x-axis to the column it references, so a guest is
+    not rejected for a column the frontend derived from the chart's own x-axis.
+    """
+    if is_metric:
+        return freeze_value(value)
+    return freeze_value(_denormalize_base_axis_column(value))
+
+
 def _native_filter_allowed_targets(
     query_context: "QueryContext", form_data: dict[str, Any]
 ) -> Optional[tuple[set[str], set[str]]]:
@@ -795,6 +856,16 @@ def _collect_sortable_identifiers(
             column_config,
             is_metric=False,
         )
+    # The x-axis is a visible dimension held under its own control; a guest may
+    # legitimately sort an embedded chart by it (the request sends it as the
+    # synthesized BASE_AXIS column, reduced to its reference below).
+    if params.get("x_axis"):
+        _add_visible_sort_targets(
+            allowed,
+            [params["x_axis"]],
+            column_config,
+            is_metric=False,
+        )
     _add_visible_sort_targets(
         allowed,
         params.get("metrics"),
@@ -982,7 +1053,12 @@ def _orderby_modified(
             return True
         if freeze_value(entry) in stored_orderby_entries:
             continue
-        if not _requested_sort_target_identifiers(entry[0]) & visible_targets:
+        # Reduce a synthesized BASE_AXIS x-axis sort target back to the reference
+        # it stands for, so sorting by the chart's own temporal dimension is not
+        # read as a new expression. The reduced target must still be a visible
+        # sort target, so this grants nothing beyond the chart's own columns.
+        target = _denormalize_base_axis_column(entry[0])
+        if not _requested_sort_target_identifiers(target) & visible_targets:
             return True
     return False
 
@@ -1048,18 +1124,34 @@ def _columns_metrics_modified(
     Whether the requested columns/metrics/group-by read beyond what the stored
     chart exposes. Each requested set must be a subset of the values stored on
     the chart (params and, when present, the stored query context).
+
+    Column-valued keys compare by an identity that first collapses a
+    frontend-synthesized ``BASE_AXIS`` x-axis back to the column it references
+    (see ``_denormalize_base_axis_column``), and additionally allow the chart's
+    stored ``x_axis``: the x-axis is a saved dimension the query carries in
+    ``columns`` but that is not itself listed under ``columns``/``groupby``.
     """
     for key, stored_params_keys, equivalent in [
         ("metrics", _STORED_METRIC_PARAMS, ["metrics"]),
         ("columns", _STORED_COLUMN_PARAMS, ["columns", "groupby"]),
         ("groupby", _STORED_COLUMN_PARAMS, ["columns", "groupby"]),
     ]:
-        requested_values = {freeze_value(value) for value in form_data.get(key) or []}
+        is_metric = key == "metrics"
+
+        # Requested column values additionally collapse a frontend-synthesized
+        # ``BASE_AXIS`` x-axis to the column it references (see
+        # ``_payload_value_identity``); metrics compare by exact frozen value.
+        requested_values = {
+            _payload_value_identity(value, is_metric=is_metric)
+            for value in form_data.get(key) or []
+        }
         # Stored params are read across every control name that can hold a
         # metric or column for some chart type: charts whose query is built
         # from e.g. ``metric``/``entity``/``groupby`` never store a literal
         # ``metrics``/``columns`` key, yet their generated payload uses those,
-        # and a guest replaying the chart's own values is not tampering.
+        # and a guest replaying the chart's own values is not tampering. This
+        # set already includes the temporal ``x_axis``; stored raw params never
+        # carry the synthesized ``BASE_AXIS`` shape, so exact matching is right.
         stored_values = _stored_param_values(
             stored_chart.params_dict, stored_params_keys
         )
@@ -1068,7 +1160,7 @@ def _columns_metrics_modified(
 
         # compare queries in query_context
         queries_values = {
-            freeze_value(value)
+            _payload_value_identity(value, is_metric=is_metric)
             for query in query_context.queries
             for value in getattr(query, key, []) or []
         }
@@ -1076,7 +1168,8 @@ def _columns_metrics_modified(
             for query in stored_query_context.get("queries") or []:
                 for equiv_key in equivalent:
                     stored_values.update(
-                        {freeze_value(value) for value in query.get(equiv_key) or []}
+                        _payload_value_identity(value, is_metric=is_metric)
+                        for value in query.get(equiv_key) or []
                     )
 
         if not queries_values.issubset(stored_values):
@@ -1106,6 +1199,13 @@ def query_context_modified(query_context: "QueryContext") -> bool:
         return False
 
     if form_data.get("slice_id") != stored_chart.id:
+        # Only the chart's own (server-side) id is logged; the guest-supplied
+        # slice_id is not echoed, to avoid logging request-controlled values.
+        logger.warning(
+            "Guest chart payload rejected for slice %s: requested slice_id does "
+            "not match the chart",
+            stored_chart.id,
+        )
         return True
 
     stored_query_context = (
@@ -1114,12 +1214,26 @@ def query_context_modified(query_context: "QueryContext") -> bool:
         else None
     )
 
+    # A rejected guest load is most often a chart whose saved query_context is
+    # NULL or stale rather than genuine tampering, and the generic 403 gives no
+    # way to tell which comparator objected. Log that here (server-side only, no
+    # payload values) so the failure is diagnosable; whether the stored
+    # query_context was present is the key signal for the missing/stale case.
+    # Use ``is not None`` so an empty-but-present stored context reads as present.
+    stored_context_state = "present" if stored_query_context is not None else "missing"
+
     # compare columns and metrics in form_data with stored values. Order-by is
     # handled separately: a strict subset check there would reject a guest
     # legitimately sorting an embedded chart by one of its existing columns.
     if _columns_metrics_modified(
         query_context, form_data, stored_chart, stored_query_context
     ):
+        logger.warning(
+            "Guest chart payload rejected for slice %s: columns/metrics/group-by "
+            "not a subset of the stored chart (stored query_context %s)",
+            stored_chart.id,
+            stored_context_state,
+        )
         return True
 
     if _series_limit_metric_modified(
@@ -1128,11 +1242,23 @@ def query_context_modified(query_context: "QueryContext") -> bool:
         stored_chart,
         stored_query_context,
     ):
+        logger.warning(
+            "Guest chart payload rejected for slice %s: series-limit metric not "
+            "on the stored chart (stored query_context %s)",
+            stored_chart.id,
+            stored_context_state,
+        )
         return True
 
     # Order-by may sort only by columns/metrics already present in the stored
     # chart; new expressions (e.g. ``random()``) are still rejected.
     if _orderby_modified(query_context, stored_chart, stored_query_context):
+        logger.warning(
+            "Guest chart payload rejected for slice %s: order-by references a "
+            "term not on the stored chart (stored query_context %s)",
+            stored_chart.id,
+            stored_context_state,
+        )
         return True
 
     return False

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/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_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_mysql.py

@@ -23,7 +23,7 @@ from typing import Any, Optional
 from unittest.mock import Mock, patch
 
 import pytest
-from sqlalchemy import types
+from sqlalchemy import column, types
 from sqlalchemy.dialects.mysql import (
     BIT,
     DECIMAL,
@@ -38,6 +38,8 @@ from sqlalchemy.dialects.mysql import (
 )
 from sqlalchemy.engine.url import make_url, URL  # noqa: F401
 
+from superset.constants import TimeGrain
+from superset.db_engine_specs.base import TimestampExpression
 from superset.utils.core import GenericDataType
 from tests.unit_tests.db_engine_specs.utils import (
     assert_column_spec,
@@ -304,3 +306,87 @@ def test_get_datatype_pymysql_fallback():
     finally:
         # Restore original state
         MySQLEngineSpec.type_code_map = original_type_code_map
+
+
+@pytest.mark.parametrize(
+    ("grain", "expected_expression"),
+    [
+        (None, "my_col"),
+        (
+            TimeGrain.SECOND,
+            "DATE_FORMAT(my_col, '%Y-%m-%d %H:%i:%s')",
+        ),
+        (
+            TimeGrain.MINUTE,
+            "DATE_FORMAT(my_col, '%Y-%m-%d %H:%i:00')",
+        ),
+        (
+            TimeGrain.HOUR,
+            "DATE_FORMAT(my_col, '%Y-%m-%d %H:00:00')",
+        ),
+        (TimeGrain.DAY, "DATE(my_col)"),
+        (
+            TimeGrain.WEEK,
+            "DATE(DATE_SUB(my_col, INTERVAL DAYOFWEEK(my_col) - 1 DAY))",
+        ),
+        (
+            TimeGrain.MONTH,
+            "DATE(DATE_SUB(my_col, INTERVAL DAYOFMONTH(my_col) - 1 DAY))",
+        ),
+        (
+            TimeGrain.QUARTER,
+            "MAKEDATE(YEAR(my_col), 1) "
+            "+ INTERVAL QUARTER(my_col) QUARTER - INTERVAL 1 QUARTER",
+        ),
+        (
+            TimeGrain.YEAR,
+            "DATE(DATE_SUB(my_col, INTERVAL DAYOFYEAR(my_col) - 1 DAY))",
+        ),
+        (
+            TimeGrain.WEEK_STARTING_MONDAY,
+            "DATE(DATE_SUB(my_col, "
+            "INTERVAL DAYOFWEEK(DATE_SUB(my_col, "
+            "INTERVAL 1 DAY)) - 1 DAY))",
+        ),
+    ],
+)
+def test_time_grain_expressions(
+    grain: Optional[TimeGrain], expected_expression: str
+) -> None:
+    """
+    Test that MySQL time grain expression templates produce the expected SQL.
+    Guards against the bare DATE() call being dropped by SQLGlot sanitization
+    or SQLAlchemy proxying for the SECOND/MINUTE/HOUR grains, which used to
+    truncate to a bare `DATE({col})`.
+    """
+    from superset.db_engine_specs.mysql import MySQLEngineSpec
+
+    actual = MySQLEngineSpec._time_grain_expressions[grain].replace("{col}", "my_col")
+    assert actual == expected_expression
+
+
+def test_compile_timegrain_expression_preserves_date_truncation() -> None:
+    """
+    Test that compile_timegrain_expression preserves the full DATE_FORMAT
+    truncation in the MySQL HOUR time grain expression, including when the
+    expression is proxied through a subquery (series-limit path).
+
+    Regression test for: ECharts HOUR grain generates invalid SQL (DATE()
+    dropped by sanitization/proxying).
+    """
+    from sqlalchemy import select
+
+    from superset.db_engine_specs.mysql import MySQLEngineSpec
+
+    col = column("my_col")
+    template = MySQLEngineSpec._time_grain_expressions[TimeGrain.HOUR]
+    expr = TimestampExpression(template, col)
+    expected = "DATE_FORMAT(my_col, '%Y-%m-%d %H:00:00')"
+
+    compiled = str(expr)
+    assert compiled == expected, f"DATE_FORMAT truncation was dropped. Got: {compiled}"
+
+    proxied = str(select(select(expr.label("bucket")).subquery().c.bucket))
+    assert expected in proxied, (
+        f"DATE_FORMAT truncation was dropped in proxied expression. Got: {proxied}"
+    )

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_starrocks.py

@@ -22,6 +22,7 @@ from pytest_mock import MockerFixture
 from sqlalchemy import JSON, types
 from sqlalchemy.engine.url import make_url
 
+from superset.constants import TimeGrain
 from superset.db_engine_specs.starrocks import (
     ARRAY,
     BITMAP,
@@ -231,8 +232,8 @@ def test_get_catalog_names(mocker: MockerFixture) -> None:
     # StarRocks returns rows with keys: ['Catalog', 'Type', 'Comment']
     mock_row_1 = mocker.MagicMock()
     mock_row_1.keys.return_value = ["Catalog", "Type", "Comment"]
-    mock_row_1.__getitem__ = (
-        lambda self, key: "default_catalog" if key == "Catalog" else None
+    mock_row_1.__getitem__ = lambda self, key: (
+        "default_catalog" if key == "Catalog" else None
     )
 
     mock_row_2 = mocker.MagicMock()
@@ -364,3 +365,23 @@ def test_get_prequeries_with_email_prefix_from_user_email_when_effective_user_di
     assert StarRocksEngineSpec.get_prequeries(database) == [
         'EXECUTE AS "alice.doe" WITH NO REVERT;'
     ]
+
+
+def test_time_grain_expressions_inherit_mysql() -> None:
+    """
+    Test that StarRocksEngineSpec inherits MySQL time grain expressions and
+    that the HOUR grain renders to the correct DATE_FORMAT truncation SQL.
+
+    StarRocks does not override _time_grain_expressions, so it reuses MySQL's
+    templates, including the HOUR grain. This asserts the rendered HOUR output
+    (not object identity) to guard against the HOUR-grain truncation regression.
+
+    Regression test for: ECharts HOUR grain generated invalid, over-truncated
+    SQL when the grain expression was normalized or proxied (#36798).
+    """
+    from superset.db_engine_specs.starrocks import StarRocksEngineSpec
+
+    actual = StarRocksEngineSpec._time_grain_expressions[TimeGrain.HOUR].replace(
+        "{col}", "my_col"
+    )
+    assert actual == "DATE_FORMAT(my_col, '%Y-%m-%d %H:00:00')"

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/security/manager_test.py

@@ -18,6 +18,7 @@
 # pylint: disable=invalid-name, unused-argument, redefined-outer-name
 
 import json  # noqa: TID251
+import logging
 from types import SimpleNamespace
 from typing import Any, Optional
 from unittest.mock import MagicMock
@@ -348,6 +349,572 @@ def test_raise_for_access_guest_user_tampered_queries_columns(
         sm.raise_for_access(query_context=query_context)
 
 
+def _base_axis_physical_column(name: str, time_grain: str = "P1D") -> dict[str, Any]:
+    """
+    The synthesized x-axis column ``normalizeTimeColumn`` emits for a *physical*
+    x-axis (see superset-ui-core ``normalizeTimeColumn.ts``): a pure column
+    reference wrapped with the BASE_AXIS markers and the chart's time grain.
+    """
+    return {
+        "timeGrain": time_grain,
+        "columnType": "BASE_AXIS",
+        "sqlExpression": name,
+        "label": name,
+        "expressionType": "SQL",
+        "isColumnReference": True,
+    }
+
+
+def _guest_query(columns: list[Any], metrics: list[AdhocMetric]) -> QueryObject:
+    """A QueryObject carrying the given columns/metrics, as a guest request would."""
+    return QueryObject(columns=columns, metrics=metrics)  # type: ignore[arg-type]
+
+
+def test_raise_for_access_guest_user_ok_base_axis_physical_x_axis(
+    mocker: MockerFixture,
+    app_context: None,
+    stored_metrics: list[AdhocMetric],
+) -> None:
+    """
+    A guest may load a chart whose saved ``query_context`` is NULL and whose
+    x-axis is a physical column. ``normalizeTimeColumn`` sends the x-axis as a
+    synthesized ``BASE_AXIS`` column that never appears verbatim in the stored
+    ``params`` (the x-axis lives under its own ``x_axis`` control), so before the
+    carve-out this legitimate load was rejected with a 403.
+    """
+    sm = SupersetSecurityManager(appbuilder)
+    mocker.patch.object(sm, "is_guest_user", return_value=True)
+    mocker.patch.object(sm, "can_access", return_value=True)
+
+    query_context = mocker.MagicMock()
+    query_context.slice_.id = 42
+    query_context.slice_.query_context = None
+    query_context.slice_.params_dict = {
+        "x_axis": "order_date",
+        "metrics": stored_metrics,
+        "columns": [],
+    }
+    query_context.form_data = {
+        "slice_id": 42,
+        "x_axis": "order_date",
+        "metrics": stored_metrics,
+        "columns": [],
+    }
+    query_context.queries = [
+        _guest_query([_base_axis_physical_column("order_date")], stored_metrics)
+    ]
+    sm.raise_for_access(query_context=query_context)
+
+
+def test_raise_for_access_guest_user_ok_base_axis_adhoc_x_axis(
+    mocker: MockerFixture,
+    app_context: None,
+    stored_metrics: list[AdhocMetric],
+) -> None:
+    """
+    Same as above for an *adhoc* x-axis: ``normalizeTimeColumn`` copies the saved
+    adhoc column and adds the BASE_AXIS markers, so the request carries the saved
+    x-axis plus synthesized decorations and must not read as tampering.
+    """
+    sm = SupersetSecurityManager(appbuilder)
+    mocker.patch.object(sm, "is_guest_user", return_value=True)
+    mocker.patch.object(sm, "can_access", return_value=True)
+
+    adhoc_x_axis: dict[str, Any] = {
+        "label": "order month",
+        "sqlExpression": "DATE_TRUNC('month', order_date)",
+        "expressionType": "SQL",
+    }
+    query_context = mocker.MagicMock()
+    query_context.slice_.id = 42
+    query_context.slice_.query_context = None
+    query_context.slice_.params_dict = {
+        "x_axis": adhoc_x_axis,
+        "metrics": stored_metrics,
+        "columns": [],
+    }
+    query_context.form_data = {
+        "slice_id": 42,
+        "x_axis": adhoc_x_axis,
+        "metrics": stored_metrics,
+        "columns": [],
+    }
+    synthesized = {"timeGrain": "P1M", "columnType": "BASE_AXIS", **adhoc_x_axis}
+    query_context.queries = [_guest_query([synthesized], stored_metrics)]
+    sm.raise_for_access(query_context=query_context)
+
+
+def test_raise_for_access_guest_user_base_axis_forged_column_reference(
+    mocker: MockerFixture,
+    app_context: None,
+    stored_metrics: list[AdhocMetric],
+) -> None:
+    """
+    The BASE_AXIS carve-out must not become a smuggling channel: a column tagged
+    ``BASE_AXIS`` that references a column the chart never exposed is still
+    rejected. Here the forged reference points at ``secret_col`` while the stored
+    x-axis is ``order_date``.
+    """
+    sm = SupersetSecurityManager(appbuilder)
+    mocker.patch.object(sm, "is_guest_user", return_value=True)
+    mocker.patch.object(sm, "can_access", return_value=True)
+
+    query_context = mocker.MagicMock()
+    query_context.slice_.id = 42
+    query_context.slice_.query_context = None
+    query_context.slice_.params_dict = {
+        "x_axis": "order_date",
+        "metrics": stored_metrics,
+        "columns": [],
+    }
+    query_context.form_data = {
+        "slice_id": 42,
+        "x_axis": "order_date",
+        "metrics": stored_metrics,
+        "columns": [],
+    }
+    query_context.queries = [
+        _guest_query([_base_axis_physical_column("secret_col")], stored_metrics)
+    ]
+    with pytest.raises(SupersetSecurityException):
+        sm.raise_for_access(query_context=query_context)
+
+
+def test_raise_for_access_guest_user_base_axis_forged_adhoc_expression(
+    mocker: MockerFixture,
+    app_context: None,
+    stored_metrics: list[AdhocMetric],
+) -> None:
+    """
+    A BASE_AXIS-tagged adhoc column whose SQL body differs from the saved x-axis
+    is rejected: tagging free-form SQL as BASE_AXIS must not authorize it.
+    """
+    sm = SupersetSecurityManager(appbuilder)
+    mocker.patch.object(sm, "is_guest_user", return_value=True)
+    mocker.patch.object(sm, "can_access", return_value=True)
+
+    adhoc_x_axis: dict[str, Any] = {
+        "label": "order month",
+        "sqlExpression": "DATE_TRUNC('month', order_date)",
+        "expressionType": "SQL",
+    }
+    forged = {
+        "timeGrain": "P1M",
+        "columnType": "BASE_AXIS",
+        "label": "order month",
+        "sqlExpression": "list_secret()",
+        "expressionType": "SQL",
+    }
+    query_context = mocker.MagicMock()
+    query_context.slice_.id = 42
+    query_context.slice_.query_context = None
+    query_context.slice_.params_dict = {
+        "x_axis": adhoc_x_axis,
+        "metrics": stored_metrics,
+        "columns": [],
+    }
+    query_context.form_data = {
+        "slice_id": 42,
+        "x_axis": adhoc_x_axis,
+        "metrics": stored_metrics,
+        "columns": [],
+    }
+    query_context.queries = [_guest_query([forged], stored_metrics)]
+    with pytest.raises(SupersetSecurityException):
+        sm.raise_for_access(query_context=query_context)
+
+
+def test_raise_for_access_guest_user_ok_base_axis_stale_query_context(
+    mocker: MockerFixture,
+    app_context: None,
+    stored_metrics: list[AdhocMetric],
+) -> None:
+    """
+    A chart with a *stale* (non-NULL) saved ``query_context`` from an older
+    frontend that predates the BASE_AXIS column still loads: the x-axis carve-out
+    is sourced from ``params`` rather than the stored ``query_context``, so a
+    stored context that lacks the synthesized column does not reject the guest.
+    """
+    sm = SupersetSecurityManager(appbuilder)
+    mocker.patch.object(sm, "is_guest_user", return_value=True)
+    mocker.patch.object(sm, "can_access", return_value=True)
+
+    query_context = mocker.MagicMock()
+    query_context.slice_.id = 42
+    query_context.slice_.query_context = json.dumps(
+        {"queries": [{"columns": [], "metrics": stored_metrics}]}
+    )
+    query_context.slice_.params_dict = {
+        "x_axis": "order_date",
+        "metrics": stored_metrics,
+        "columns": [],
+    }
+    query_context.form_data = {
+        "slice_id": 42,
+        "x_axis": "order_date",
+        "metrics": stored_metrics,
+        "columns": [],
+    }
+    query_context.queries = [
+        _guest_query([_base_axis_physical_column("order_date")], stored_metrics)
+    ]
+    sm.raise_for_access(query_context=query_context)
+
+
+def test_raise_for_access_guest_user_ok_base_axis_orderby(
+    mocker: MockerFixture,
+    app_context: None,
+    stored_metrics: list[AdhocMetric],
+) -> None:
+    """
+    A guest may sort an embedded chart by its own temporal x-axis, which the
+    frontend sends as the synthesized BASE_AXIS column in the order-by term.
+    """
+    sm = SupersetSecurityManager(appbuilder)
+    mocker.patch.object(sm, "is_guest_user", return_value=True)
+    mocker.patch.object(sm, "can_access", return_value=True)
+
+    base_axis = _base_axis_physical_column("order_date")
+    query_context = mocker.MagicMock()
+    query_context.slice_.id = 42
+    query_context.slice_.query_context = None
+    query_context.slice_.params_dict = {
+        "x_axis": "order_date",
+        "metrics": stored_metrics,
+        "columns": [],
+    }
+    query_context.form_data = {
+        "slice_id": 42,
+        "x_axis": "order_date",
+        "metrics": stored_metrics,
+        "columns": [],
+        "orderby": [[base_axis, True]],
+    }
+    query_context.queries = [_guest_query([base_axis], stored_metrics)]
+    sm.raise_for_access(query_context=query_context)
+
+
+def test_raise_for_access_guest_user_base_axis_orderby_forged(
+    mocker: MockerFixture,
+    app_context: None,
+    stored_metrics: list[AdhocMetric],
+) -> None:
+    """
+    The order-by carve-out is bounded: sorting by a BASE_AXIS-tagged column that
+    references a column the chart never exposed is still rejected.
+    """
+    sm = SupersetSecurityManager(appbuilder)
+    mocker.patch.object(sm, "is_guest_user", return_value=True)
+    mocker.patch.object(sm, "can_access", return_value=True)
+
+    query_context = mocker.MagicMock()
+    query_context.slice_.id = 42
+    query_context.slice_.query_context = None
+    query_context.slice_.params_dict = {
+        "x_axis": "order_date",
+        "metrics": stored_metrics,
+        "columns": [],
+    }
+    query_context.form_data = {
+        "slice_id": 42,
+        "x_axis": "order_date",
+        "metrics": stored_metrics,
+        "columns": [],
+        "orderby": [[_base_axis_physical_column("secret_col"), True]],
+    }
+    query_context.queries = [
+        _guest_query([_base_axis_physical_column("order_date")], stored_metrics)
+    ]
+    with pytest.raises(SupersetSecurityException):
+        sm.raise_for_access(query_context=query_context)
+
+
+def test_raise_for_access_guest_user_ok_base_axis_multiple_queries(
+    mocker: MockerFixture,
+    app_context: None,
+    stored_metrics: list[AdhocMetric],
+) -> None:
+    """
+    The carve-out applies across every query in the request: a chart that issues
+    more than one query (each carrying the synthesized BASE_AXIS x-axis) loads.
+    """
+    sm = SupersetSecurityManager(appbuilder)
+    mocker.patch.object(sm, "is_guest_user", return_value=True)
+    mocker.patch.object(sm, "can_access", return_value=True)
+
+    base_axis = _base_axis_physical_column("order_date")
+    query_context = mocker.MagicMock()
+    query_context.slice_.id = 42
+    query_context.slice_.query_context = None
+    query_context.slice_.params_dict = {
+        "x_axis": "order_date",
+        "metrics": stored_metrics,
+        "columns": [],
+    }
+    query_context.form_data = {
+        "slice_id": 42,
+        "x_axis": "order_date",
+        "metrics": stored_metrics,
+        "columns": [],
+    }
+    query_context.queries = [
+        _guest_query([base_axis], stored_metrics),
+        _guest_query([base_axis], stored_metrics),
+    ]
+    sm.raise_for_access(query_context=query_context)
+
+
+def test_raise_for_access_guest_user_ok_x_axis_reused_in_columns(
+    mocker: MockerFixture,
+    app_context: None,
+    stored_metrics: list[AdhocMetric],
+) -> None:
+    """
+    The stored x-axis is an allowed column even when requested verbatim (not
+    BASE_AXIS-wrapped): the x-axis dimension is not listed under ``columns`` in
+    ``params`` but is a legitimate selectable dimension.
+    """
+    sm = SupersetSecurityManager(appbuilder)
+    mocker.patch.object(sm, "is_guest_user", return_value=True)
+    mocker.patch.object(sm, "can_access", return_value=True)
+
+    adhoc_x_axis: dict[str, Any] = {
+        "label": "order month",
+        "sqlExpression": "DATE_TRUNC('month', order_date)",
+        "expressionType": "SQL",
+    }
+    query_context = mocker.MagicMock()
+    query_context.slice_.id = 42
+    query_context.slice_.query_context = None
+    query_context.slice_.params_dict = {
+        "x_axis": adhoc_x_axis,
+        "metrics": stored_metrics,
+        "columns": [],
+    }
+    query_context.form_data = {
+        "slice_id": 42,
+        "x_axis": adhoc_x_axis,
+        "metrics": stored_metrics,
+        "columns": [adhoc_x_axis],
+    }
+    query_context.queries = [_guest_query([adhoc_x_axis], stored_metrics)]
+    sm.raise_for_access(query_context=query_context)
+
+
+def test_raise_for_access_guest_user_base_axis_non_string_sql_rejected(
+    mocker: MockerFixture,
+    app_context: None,
+    stored_metrics: list[AdhocMetric],
+) -> None:
+    """
+    A BASE_AXIS column with a non-string ``sqlExpression`` does not take the
+    physical-reference shortcut; it falls back to the dict comparison, matches no
+    stored value, and is rejected without raising an unexpected error.
+    """
+    sm = SupersetSecurityManager(appbuilder)
+    mocker.patch.object(sm, "is_guest_user", return_value=True)
+    mocker.patch.object(sm, "can_access", return_value=True)
+
+    forged = {
+        "columnType": "BASE_AXIS",
+        "isColumnReference": True,
+        "sqlExpression": ["order_date"],
+        "label": "order_date",
+    }
+    query_context = mocker.MagicMock()
+    query_context.slice_.id = 42
+    query_context.slice_.query_context = None
+    query_context.slice_.params_dict = {
+        "x_axis": "order_date",
+        "metrics": stored_metrics,
+        "columns": [],
+    }
+    query_context.form_data = {
+        "slice_id": 42,
+        "x_axis": "order_date",
+        "metrics": stored_metrics,
+        "columns": [],
+    }
+    query_context.queries = [_guest_query([forged], stored_metrics)]
+    with pytest.raises(SupersetSecurityException):
+        sm.raise_for_access(query_context=query_context)
+
+
+def test_raise_for_access_guest_user_logs_rejection_reason(
+    mocker: MockerFixture,
+    app_context: None,
+    caplog: pytest.LogCaptureFixture,
+    stored_metrics: list[AdhocMetric],
+) -> None:
+    """
+    A rejected guest payload is logged server-side with which comparator objected
+    and whether the stored query_context was present, and without any payload
+    values (so column names / SQL are not leaked into logs).
+    """
+    sm = SupersetSecurityManager(appbuilder)
+    mocker.patch.object(sm, "is_guest_user", return_value=True)
+    mocker.patch.object(sm, "can_access", return_value=True)
+
+    query_context = mocker.MagicMock()
+    query_context.slice_.id = 42
+    query_context.slice_.query_context = None
+    query_context.slice_.params_dict = {"columns": [], "metrics": stored_metrics}
+    query_context.form_data = {
+        "slice_id": 42,
+        "columns": [
+            {
+                "label": "secret",
+                "sqlExpression": "secret_col",
+                "expressionType": "SQL",
+            }
+        ],
+        "metrics": stored_metrics,
+    }
+    query_context.queries = [_guest_query([], stored_metrics)]
+
+    with caplog.at_level(logging.WARNING, logger="superset.security.manager"):
+        with pytest.raises(SupersetSecurityException):
+            sm.raise_for_access(query_context=query_context)
+
+    assert "columns/metrics/group-by" in caplog.text
+    assert "stored query_context missing" in caplog.text
+    # The rejected column name / SQL must not be logged.
+    assert "secret_col" not in caplog.text
+
+
+def test_raise_for_access_guest_user_ok_base_axis_in_form_data(
+    mocker: MockerFixture,
+    app_context: None,
+    stored_metrics: list[AdhocMetric],
+) -> None:
+    """
+    The carve-out applies to the ``form_data`` comparison too, not only the
+    per-query one: a BASE_AXIS x-axis carried in ``form_data["columns"]`` is
+    recognized as the stored x-axis.
+    """
+    sm = SupersetSecurityManager(appbuilder)
+    mocker.patch.object(sm, "is_guest_user", return_value=True)
+    mocker.patch.object(sm, "can_access", return_value=True)
+
+    base_axis = _base_axis_physical_column("order_date")
+    query_context = mocker.MagicMock()
+    query_context.slice_.id = 42
+    query_context.slice_.query_context = None
+    query_context.slice_.params_dict = {
+        "x_axis": "order_date",
+        "metrics": stored_metrics,
+        "columns": [],
+    }
+    query_context.form_data = {
+        "slice_id": 42,
+        "x_axis": "order_date",
+        "metrics": stored_metrics,
+        "columns": [base_axis],
+    }
+    query_context.queries = [_guest_query([base_axis], stored_metrics)]
+    sm.raise_for_access(query_context=query_context)
+
+
+def test_raise_for_access_guest_user_ok_base_axis_groupby(
+    mocker: MockerFixture,
+    app_context: None,
+    stored_metrics: list[AdhocMetric],
+) -> None:
+    """
+    The x-axis carve-out also applies to the ``groupby`` key, not only
+    ``columns``.
+    """
+    sm = SupersetSecurityManager(appbuilder)
+    mocker.patch.object(sm, "is_guest_user", return_value=True)
+    mocker.patch.object(sm, "can_access", return_value=True)
+
+    base_axis = _base_axis_physical_column("order_date")
+    query_context = mocker.MagicMock()
+    query_context.slice_.id = 42
+    query_context.slice_.query_context = None
+    query_context.slice_.params_dict = {
+        "x_axis": "order_date",
+        "metrics": stored_metrics,
+        "groupby": [],
+    }
+    query_context.form_data = {
+        "slice_id": 42,
+        "x_axis": "order_date",
+        "metrics": stored_metrics,
+        "groupby": [base_axis],
+    }
+    query_context.queries = [_guest_query([], stored_metrics)]
+    sm.raise_for_access(query_context=query_context)
+
+
+def test_raise_for_access_guest_user_ok_bare_string_x_axis_in_columns(
+    mocker: MockerFixture,
+    app_context: None,
+    stored_metrics: list[AdhocMetric],
+) -> None:
+    """
+    A physical (bare-string) x-axis requested verbatim as a column is allowed by
+    the x_axis carve-out, without any BASE_AXIS wrapping.
+    """
+    sm = SupersetSecurityManager(appbuilder)
+    mocker.patch.object(sm, "is_guest_user", return_value=True)
+    mocker.patch.object(sm, "can_access", return_value=True)
+
+    query_context = mocker.MagicMock()
+    query_context.slice_.id = 42
+    query_context.slice_.query_context = None
+    query_context.slice_.params_dict = {
+        "x_axis": "order_date",
+        "metrics": stored_metrics,
+        "columns": [],
+    }
+    query_context.form_data = {
+        "slice_id": 42,
+        "x_axis": "order_date",
+        "metrics": stored_metrics,
+        "columns": ["order_date"],
+    }
+    query_context.queries = [_guest_query(["order_date"], stored_metrics)]
+    sm.raise_for_access(query_context=query_context)
+
+
+def test_raise_for_access_guest_user_base_axis_in_metrics_rejected(
+    mocker: MockerFixture,
+    app_context: None,
+    stored_metrics: list[AdhocMetric],
+) -> None:
+    """
+    The carve-out is column-only: a metric tagged BASE_AXIS is not collapsed, so
+    it must exact-match a stored metric and an unrelated one is still rejected.
+    """
+    sm = SupersetSecurityManager(appbuilder)
+    mocker.patch.object(sm, "is_guest_user", return_value=True)
+    mocker.patch.object(sm, "can_access", return_value=True)
+
+    forged_metric = {
+        "columnType": "BASE_AXIS",
+        "isColumnReference": True,
+        "sqlExpression": "SUM(revenue)",
+        "label": "SUM(revenue)",
+        "expressionType": "SQL",
+    }
+    query_context = mocker.MagicMock()
+    query_context.slice_.id = 42
+    query_context.slice_.query_context = None
+    query_context.slice_.params_dict = {
+        "x_axis": "order_date",
+        "metrics": stored_metrics,
+        "columns": [],
+    }
+    query_context.form_data = {
+        "slice_id": 42,
+        "metrics": [forged_metric],
+    }
+    query_context.queries = [_guest_query([], stored_metrics)]
+    with pytest.raises(SupersetSecurityException):
+        sm.raise_for_access(query_context=query_context)
+
+
 def test_raise_for_access_query_default_schema(
     mocker: MockerFixture,
     app_context: None,

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