fix: give embedded dashboard retry a budget that fits the test timeout

A single worst-case render() attempt (30s iframe wait + 30s
dashboard-content wait) could consume the entire 60s test timeout,
leaving no room for the retry the code is meant to perform.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

.github/CODEOWNERS

@@ -1,22 +1,3 @@
-# Notify all committers of DB migration changes, per SIP-59
-
-/superset/migrations/ @mistercrunch @michael-s-molina @betodealmeida @eschutho @sadpandajoe @rusackas
-
-# Notify some committers of changes in the components
-
-/superset-frontend/src/components/Select/ @michael-s-molina @geido @kgabryje
-/superset-frontend/src/components/MetadataBar/ @michael-s-molina @geido @kgabryje
-/superset-frontend/src/components/DropdownContainer/ @michael-s-molina @geido @kgabryje
-
-# Notify Helm Chart maintainers about changes in it
-
-/helm/superset/ @dpgaspar @villebro @nytai @michael-s-molina @mistercrunch @rusackas @Antonio-RiveroMartnez @hainenber
-
-# Notify E2E test maintainers of changes
-
-/superset-frontend/playwright/ @sadpandajoe @geido @eschutho @rusackas @mistercrunch
-/superset-frontend/cypress-base/ @sadpandajoe @geido @eschutho @rusackas @mistercrunch
-
 # Notify PMC members of changes to GitHub Actions
 
 /.github/ @villebro @geido @eschutho @rusackas @betodealmeida @nytai @mistercrunch @kgabryje @sha174n @dpgaspar @sadpandajoe @hainenber
@@ -30,16 +11,12 @@
 
 /.asf.yaml @villebro @geido @eschutho @rusackas @betodealmeida @nytai @mistercrunch @kgabryje @dpgaspar @sha174n @Antonio-RiveroMartnez
 
-# Maps are a finicky contribution process we care about
+# Maps are fragile and political. GeoJson edits MUST be made in the Jupyter notebook or they'll be overwritten.
 
 **/*.geojson @villebro @rusackas
 **/*.ipynb @villebro @rusackas
 /superset-frontend/plugins/plugin-chart-country-map/ @villebro @rusackas
 
-# Notify translation maintainers of changes to translations
-
-/superset/translations/ @sfirke @rusackas @villebro @sadpandajoe @hainenber
-
 # Notify PMC members of changes to extension-related files
 
 /docs/developer_portal/extensions/ @michael-s-molina @villebro @rusackas

UPDATING.md

@@ -54,6 +54,18 @@ pip install playwright && playwright install chromium
 2. Remove any references to the removed config keys from custom `superset_config.py`
 3. If you subclassed `MachineAuthProvider`, remove any `authenticate_webdriver` override and migrate auth logic to `authenticate_browser_context`
 
+### CSV/XLSX report exports of Table charts keep raw numeric values
+
+Table and Pivot Table charts sent as text in a report email now apply the
+chart's number and currency formatting so the values match what a user sees in
+Explore. As part of this, the CSV and XLSX result formats return early before
+formatting: previously the Table post-processor applied `d3NumberFormat` to
+every result format, so CSV/XLSX exports contained pre-formatted strings.
+CSV/XLSX exports now preserve numeric values and column types, which is better
+for downstream analysis but is a visible change for anyone who relied on the
+formatted text in those files. The rendered email body (the only place the
+formatting is intended for) is unaffected.
+
 ### Soft delete is on by default, and purging is live
 
 `SOFT_DELETE` now ships **on** (`DEFAULT_FEATURE_FLAGS`), so deleting a
@@ -159,6 +171,36 @@ will now get a TypeScript error and must remove the prop; keeping a manual
 override was exactly the footgun this change removes (see #42510). No
 callers in the Superset frontend codebase itself passed this prop.
 
+### Row-level security now filters table reads a same-named CTE used to hide
+
+`extract_tables_from_statement()` decided whether a reference was a CTE by matching its
+bare name against the enclosing scope's CTE names; it now resolves the name through
+`Scope.cte_sources`. Three kinds of real table read whose bare name collided with a CTE's
+were mistaken for the CTE and dropped from a statement's tables, so they were neither
+RLS-filtered nor access-checked: a schema- or catalog-qualified reference, a non-recursive
+CTE's own name inside its body, and a forward reference to a later `WITH` item.
+
+```sql
+WITH orders AS (SELECT 1 AS d) SELECT * FROM (SELECT * FROM public.orders) AS z
+WITH orders AS (SELECT * FROM orders) SELECT * FROM orders
+WITH q1 AS (SELECT key FROM q2), q2 AS (SELECT 1 AS key) SELECT * FROM q1
+```
+
+Each read is now reported, so it is filtered when `RLS_IN_SQLLAB` is enabled, matched
+against `DISALLOWED_SQL_TABLES`, and requires dataset access under
+`raise_for_access(force_dataset_match=True)`. A query that previously ran, reading those
+rows unfiltered, may now be filtered or rejected. There is no opt-out — the previous
+behavior was a row-level-security bypass.
+
+### Table aliases keep their quoting through the row-level security rewrite
+
+Both RLS transformers took the table alias as a string with its quoting stripped and
+emitted it verbatim; they now carry the parsed identifier. Emitted SQL is unchanged for an
+unquoted identifier; a quoted one keeps its quoting, and a column-alias list
+(`FROM t AS x (c1, c2)`) survives the rewrite instead of being dropped. This repairs
+row-level security for any aliased table on Snowflake, and for at least one statement shape
+on MSSQL where the rewrite previously raised `AttributeError`.
+
 ### Principal listing APIs now honour related-field filters
 
 Two authorization-related listing behaviors changed for API clients. Neither

superset-frontend/package-lock.json

@@ -272,7 +272,7 @@
         "swc-loader": "^0.2.7",
         "ts-jest": "^29.4.12",
         "tscw-config": "^1.1.2",
-        "tsx": "^4.23.5",
+        "tsx": "^4.23.7",
         "typescript": "5.4.5",
         "unzipper": "^0.12.5",
         "wait-on": "^9.1.0",
@@ -39964,9 +39964,9 @@
       "license": "0BSD"
     },
     "node_modules/tsx": {
-      "version": "4.23.5",
-      "resolved": "https://registry.npmjs.org/tsx/-/tsx-4.23.5.tgz",
-      "integrity": "sha512-rw55FUaqOoI7RvlQwLbhO4nSDApnQ4/CykPuiQ/EPvtrX3WA9Ig55jIt9VvbBJbzJuj12ueRu4PMZ2SxPVbihg==",
+      "version": "4.23.7",
+      "resolved": "https://registry.npmjs.org/tsx/-/tsx-4.23.7.tgz",
+      "integrity": "sha512-3f/u/+UDCNQ7iwUZW9FCMnNGIHzElGJYh0S/yy8IvWSsn5O7fEO/897FaG7FA2W8yryiRyuwXZ1PYLAKYaqSuQ==",
       "dev": true,
       "license": "MIT",
       "dependencies": {
@@ -43351,6 +43351,15 @@
         "node": ">=12"
       }
     },
+    "packages/superset-ui-core/node_modules/dompurify": {
+      "version": "3.4.13",
+      "resolved": "https://registry.npmjs.org/dompurify/-/dompurify-3.4.13.tgz",
+      "integrity": "sha512-2vmYIoqjze2d+kakP8S/nS5shfsl587kzwEjcGlTdiksUVgFHnFCsLYDVj/JNqJVOQZGSYBTmuycv0PodwmnMQ==",
+      "license": "(MPL-2.0 OR Apache-2.0)",
+      "optionalDependencies": {
+        "@types/trusted-types": "^2.0.7"
+      }
+    },
     "packages/superset-ui-core/node_modules/react-ace": {
       "version": "14.0.1",
       "resolved": "https://registry.npmjs.org/react-ace/-/react-ace-14.0.1.tgz",

superset-frontend/package.json

@@ -349,7 +349,7 @@
     "swc-loader": "^0.2.7",
     "ts-jest": "^29.4.12",
     "tscw-config": "^1.1.2",
-    "tsx": "^4.23.5",
+    "tsx": "^4.23.7",
     "typescript": "5.4.5",
     "unzipper": "^0.12.5",
     "wait-on": "^9.1.0",

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

@@ -989,6 +989,52 @@ test('shows all options when filterOption is false', async () => {
   expect(options[0]).toHaveTextContent('Server 0');
 });
 
+test('renders a server-matched option whose label diverges from the search term when filterOption is false (regression for #42041)', async () => {
+  // Mirrors the real permissions-search bug: the remote fetch legitimately
+  // matches the raw, underscore-containing value (e.g. a schema name like
+  // "stg_silver"), but the returned option's displayed label has had
+  // underscores replaced with spaces (see formatPermissionLabel in
+  // features/roles/utils.ts). filterOption defaults to true, which
+  // re-filters already-matched options against that same relabeled text
+  // client-side, so the underscore search term never matches and the
+  // legitimately fetched option gets hidden -- this is why
+  // PermissionsField (features/roles/RoleFormItems.tsx) sets
+  // filterOption={false}: the loader is already the authoritative filter,
+  // and its match doesn't depend on the label used to render the option.
+  const searchData = [{ label: 'stg silver', value: 100 }];
+  const loadOptions = jest.fn(async (search: string) =>
+    // totalCount must exceed the empty initial page here, otherwise
+    // AsyncSelect marks allValuesLoaded and short-circuits every later
+    // fetch, including the search request this test depends on.
+    search === ''
+      ? { data: [], totalCount: 1 }
+      : { data: searchData, totalCount: 1 },
+  );
+
+  render(
+    <AsyncSelect
+      {...defaultProps}
+      options={loadOptions}
+      filterOption={false}
+    />,
+  );
+  await open();
+
+  await type('stg_silver');
+  await waitFor(() =>
+    expect(loadOptions).toHaveBeenCalledWith(
+      'stg_silver',
+      expect.anything(),
+      expect.anything(),
+    ),
+  );
+
+  // The backend legitimately matched and returned this option (asserted
+  // above); it should render in the dropdown despite the search term using
+  // underscores while the label uses spaces.
+  expect(await findSelectOption('stg silver')).toBeInTheDocument();
+});
+
 test('preserves new option entry across search fetch when allowNewOptions is on', async () => {
   const page0Data = Array.from({ length: 10 }, (_, i) => ({
     label: `Option ${i}`,

superset-frontend/playwright/pages/EmbeddedPage.ts

@@ -103,16 +103,69 @@ export class EmbeddedPage {
   /**
    * Wait for dashboard content to render inside the iframe.
    * Looks for the grid-container which indicates charts are loading/loaded.
+   *
+   * Races the grid against the test app's `#error` box so an embed failure
+   * surfaces its message immediately, instead of blindly timing out on the
+   * grid selector and hiding the real reason.
    */
   async waitForDashboardContent(options?: { timeout?: number }): Promise<void> {
-    const frame = this.iframe;
-    await frame
+    const timeout = options?.timeout ?? EMBEDDED.DASHBOARD_RENDER;
+    const grid = this.iframe
       .locator('.grid-container, [data-test="grid-container"]')
-      .first()
-      .waitFor({
-        state: 'visible',
-        timeout: options?.timeout ?? EMBEDDED.DASHBOARD_RENDER,
-      });
+      .first();
+    const errorBox = this.page.locator(EmbeddedPage.SELECTORS.ERROR);
+
+    const ready = grid
+      .waitFor({ state: 'visible', timeout })
+      .then(() => 'ready' as const)
+      .catch(() => 'gridTimeout' as const);
+    const failed = errorBox
+      .waitFor({ state: 'visible', timeout })
+      .then(() => 'error' as const)
+      .catch(() => 'errorTimeout' as const);
+
+    const outcome = await Promise.race([ready, failed]);
+    if (outcome === 'ready') return;
+    if (outcome === 'error') {
+      const message = (await errorBox.textContent())?.trim() || 'unknown error';
+      throw new Error(`Embedded dashboard failed to render: ${message}`);
+    }
+    const status = (
+      await this.page.locator(EmbeddedPage.SELECTORS.STATUS).textContent()
+    )?.trim();
+    throw new Error(
+      `Embedded dashboard did not render within ${timeout}ms ` +
+        `(status: ${status ?? 'unknown'})`,
+    );
+  }
+
+  /**
+   * Navigate to the test app and wait for the dashboard to render, retrying
+   * the whole sequence once on failure. The embedded guest-token / SDK
+   * postMessage handshake can intermittently drop on a cold CI start, leaving
+   * the dashboard stuck before the grid renders; a fresh load clears it.
+   */
+  async render(
+    params: Parameters<EmbeddedPage['goto']>[0],
+    options?: { attempts?: number; timeout?: number },
+  ): Promise<void> {
+    const attempts = options?.attempts ?? 2;
+    for (let attempt = 1; ; attempt += 1) {
+      try {
+        await this.goto(params);
+        await this.waitForIframe(options);
+        await this.waitForDashboardContent(options);
+        return;
+      } catch (err) {
+        if (attempt >= attempts) throw err;
+        // eslint-disable-next-line no-console
+        console.warn(
+          `[embedded] dashboard render attempt ${attempt} failed, retrying: ${
+            err instanceof Error ? err.message : String(err)
+          }`,
+        );
+      }
+    }
   }
 
   /**

superset-frontend/playwright/tests/embedded/embedded-dashboard.spec.ts

@@ -149,7 +149,11 @@ test.describe('Embedded Dashboard E2E', () => {
 
   // The full embedded chain (login → guest token → iframe → dashboard render
   // → chart render) routinely exceeds the 30s default on cold CI starts.
-  test.setTimeout(60000);
+  // EmbeddedPage.render() retries once on failure, and a single worst-case
+  // attempt (30s iframe wait + 30s dashboard-content wait) can consume up to
+  // 60s on its own, so the budget must cover two full attempts or the retry
+  // never gets a chance to run before the test times out.
+  test.setTimeout(120000);
 
   let appServer: EmbedAppServer;
   let accessToken: string;
@@ -165,13 +169,11 @@ test.describe('Embedded Dashboard E2E', () => {
     await embeddedPage.exposeTokenFetcher(async () =>
       getGuestToken(page, dashboardId, { accessToken }),
     );
-    await embeddedPage.goto({
+    await embeddedPage.render({
       appUrl: appServer.url,
       uuid: embedUuid,
       supersetDomain: SUPERSET_DOMAIN,
     });
-    await embeddedPage.waitForIframe();
-    await embeddedPage.waitForDashboardContent();
     return embeddedPage;
   }
 
@@ -251,14 +253,12 @@ test.describe('Embedded Dashboard E2E', () => {
     await embeddedPage.exposeTokenFetcher(async () =>
       getGuestToken(page, dashboardId, { accessToken }),
     );
-    await embeddedPage.goto({
+    await embeddedPage.render({
       appUrl: appServer.url,
       uuid: embedUuid,
       supersetDomain: SUPERSET_DOMAIN,
       hideTitle: true,
     });
-    await embeddedPage.waitForIframe();
-    await embeddedPage.waitForDashboardContent();
 
     // The iframe URL should include uiConfig parameter
     await expect(

superset-frontend/plugins/plugin-chart-echarts/src/Timeseries/transformProps.ts

@@ -182,6 +182,50 @@ function getSymbolMarker(symbol: string, color: string) {
   }
 }
 
+/**
+ * Given the fully-built ECharts series (each already carrying its resolved
+ * `stack` id, see `getTimeCompareStackId`), find the largest per-index total
+ * across all series sharing a stack, then return the largest such total
+ * across all stacks. Series without a `stack` id (e.g. annotation layers)
+ * are ignored since they aren't part of any stacked total.
+ */
+function getMaxStackedValueByStack(
+  series: SeriesOption[],
+  isHorizontal: boolean,
+): number {
+  const totalsByStack = new Map<string, number[]>();
+  series.forEach(entry => {
+    const rawStackId = (entry as { stack?: unknown }).stack;
+    const stackId = typeof rawStackId === 'string' ? rawStackId : undefined;
+    if (!stackId || !Array.isArray(entry.data)) return;
+    const totals = totalsByStack.get(stackId) ?? [];
+    (entry.data as unknown[]).forEach((datum, idx) => {
+      let value: unknown = datum;
+      if (Array.isArray(datum)) {
+        value = isHorizontal ? datum[0] : datum[1];
+      } else if (datum && typeof datum === 'object' && 'value' in datum) {
+        const rawValue = (datum as { value: unknown }).value;
+        if (Array.isArray(rawValue)) {
+          value = isHorizontal ? rawValue[0] : rawValue[1];
+        } else {
+          value = rawValue;
+        }
+      }
+      if (typeof value === 'number' && !Number.isNaN(value)) {
+        totals[idx] = (totals[idx] ?? 0) + value;
+      }
+    });
+    totalsByStack.set(stackId, totals);
+  });
+  let max = Number.NEGATIVE_INFINITY;
+  totalsByStack.forEach(totals => {
+    totals.forEach(value => {
+      if (value > max) max = value;
+    });
+  });
+  return max;
+}
+
 export default function transformProps(
   chartProps: EchartsTimeseriesChartProps,
 ): TimeseriesChartTransformedProps {
@@ -875,7 +919,38 @@ export default function transformProps(
   // default to 0-100% range when doing row-level contribution chart
   if ((contributionMode === 'row' || isAreaExpand) && stack) {
     if (yAxisMin === undefined) yAxisMin = 0;
-    if (yAxisMax === undefined) yAxisMax = 1;
+    if (yAxisMax === undefined) {
+      if (contributionMode === 'row') {
+        // Contribution percentages are normalized so each stacked row should
+        // sum to 1, but floating point rounding can push the actual stacked
+        // total fractionally above 1 (e.g. 1.0000000000000002). Hard-capping
+        // the axis max at exactly 1 in that case causes echarts to clip the
+        // topmost stacked segment entirely rather than just rounding the
+        // pixel width, which is most visible in horizontal orientation where
+        // this axis is swapped onto the x-axis. Pad the max up to the actual
+        // stacked total when it exceeds 1 so no segment gets clipped.
+        //
+        // This padding only applies in row-contribution mode: for an Expand
+        // ("100% stacked") chart, `sortedTotalValues` holds the raw,
+        // pre-normalization row totals (e.g. 100), not values near 1, so
+        // padding against them here would stretch the axis out to the raw
+        // total instead of the intended 0-1 range.
+        //
+        // `sortedTotalValues` sums every series value per row regardless of
+        // which ECharts stack it belongs to, but with time_compare each
+        // comparison period is its own independently-normalized stack (see
+        // getTimeCompareStackId), so a chart with N comparison periods would
+        // sum to ~N instead of ~1. Compute the max per stack instead, using
+        // the already-built series (which carry the resolved stack ids).
+        const stackedTotalMax = getMaxStackedValueByStack(series, isHorizontal);
+        yAxisMax =
+          Number.isFinite(stackedTotalMax) && stackedTotalMax > 1
+            ? stackedTotalMax
+            : 1;
+      } else {
+        yAxisMax = 1;
+      }
+    }
   } else if (
     logAxis &&
     yAxisMin === undefined &&

superset-frontend/plugins/plugin-chart-echarts/test/Timeseries/transformProps.test.ts

@@ -23,6 +23,7 @@ import {
   AxisType,
   ChartProps,
   ComparisonType,
+  ContributionType,
   DataRecord,
   EventAnnotationLayer,
   FormulaAnnotationLayer,
@@ -1442,6 +1443,118 @@ test('should not apply axis bounds calculation when seriesType is not Bar for ho
   expect(xAxisRaw.max).toBeUndefined();
 });
 
+test('should not clip small segments when row-contribution percentages float above 1 in horizontal stacked bar charts', () => {
+  // These three shares are individually normalized (each column sums to 1),
+  // but due to floating point rounding their sum can land fractionally
+  // over 1. See https://github.com/apache/superset/issues/30914
+  //
+  // The margin above 1 is chosen large enough (~1e-7) that the sum stays
+  // above 1 no matter which order the underlying series get summed in
+  // (series are sorted by name for stacking, not in the order declared
+  // here), unlike a single-ULP overflow which can round differently
+  // depending on summation order and make this assertion order-dependent.
+  const shareA = 0.42;
+  const shareB = 0.38;
+  const shareC = 0.2000001;
+  expect(shareA + shareB + shareC).toBeGreaterThan(1);
+
+  const queriesData: ChartDataResponseResult[] = [
+    createTestQueryData(
+      createTestData(
+        [{ 'Series A': shareA, 'Series B': shareB, 'Series C': shareC }],
+        { intervalMs: 300000000 },
+      ),
+    ),
+  ];
+
+  const chartProps = createTestChartProps({
+    formData: {
+      ...baseFormDataHorizontalBar,
+      contributionMode: ContributionType.Row,
+      stack: StackControlsValue.Stack,
+    },
+    queriesData,
+  });
+
+  const transformedProps = transformProps(chartProps);
+
+  // In horizontal orientation, axes are swapped, so yAxis becomes xAxis.
+  // The axis max must not be hard-capped at exactly 1, otherwise echarts
+  // clips the topmost stacked segment entirely instead of just rendering
+  // a negligible sub-pixel overflow.
+  const xAxisRaw = transformedProps.echartOptions.xAxis as any;
+  expect(xAxisRaw.max).toBeGreaterThanOrEqual(shareA + shareB + shareC);
+});
+
+test('keeps the 0-1 axis range for Expand (100% stacked) charts instead of padding to the raw row total', () => {
+  // Unlike row-contribution mode, an Expand stack is not pre-normalized in
+  // the query result -- these are raw values (summing to 100, not 1) that
+  // get divided down to a 0-1 range internally. The un-normalized row total
+  // must not be used to pad the axis max, or the chart would only occupy a
+  // sliver of the plot.
+  const queriesData: ChartDataResponseResult[] = [
+    createTestQueryData(
+      createTestData([{ 'Series A': 42, 'Series B': 38, 'Series C': 20 }], {
+        intervalMs: 300000000,
+      }),
+    ),
+  ];
+
+  const chartProps = createTestChartProps({
+    formData: {
+      ...baseFormDataHorizontalBar,
+      stack: StackControlsValue.Expand,
+    },
+    queriesData,
+  });
+
+  const transformedProps = transformProps(chartProps);
+
+  const xAxisRaw = transformedProps.echartOptions.xAxis as any;
+  expect(xAxisRaw.max).toBe(1);
+});
+
+test('computes row-contribution axis padding per stack when time_compare splits a row into multiple normalized stacks', () => {
+  // With time_compare, each comparison period is normalized and stacked
+  // independently (see getTimeCompareStackId), so the current-period
+  // columns sum to ~1 in their own stack and the comparison-period columns
+  // (suffixed with the offset) sum to ~1 in a separate stack. The combined
+  // row total across both stacks is therefore ~2, but the axis max must be
+  // computed per stack, not from that combined total, or a 100% bar would
+  // only occupy about half the plot.
+  const queriesData: ChartDataResponseResult[] = [
+    createTestQueryData(
+      createTestData(
+        [
+          {
+            'Series A': 0.6,
+            'Series B': 0.4,
+            'Series A__1 year ago': 0.55,
+            'Series B__1 year ago': 0.45,
+          },
+        ],
+        { intervalMs: 300000000 },
+      ),
+    ),
+  ];
+
+  const chartProps = createTestChartProps({
+    formData: {
+      ...baseFormDataHorizontalBar,
+      contributionMode: ContributionType.Row,
+      stack: StackControlsValue.Stack,
+      time_compare: ['1 year ago'],
+    },
+    queriesData,
+  });
+
+  const transformedProps = transformProps(chartProps);
+
+  const xAxisRaw = transformedProps.echartOptions.xAxis as any;
+  expect(xAxisRaw.max).toBeGreaterThanOrEqual(1);
+  expect(xAxisRaw.max).toBeLessThan(1.5);
+});
+
 test('legend is visible on tall charts when enabled by the user', () => {
   const chartProps = createTestChartProps({
     height: 400,

superset-frontend/plugins/preset-chart-deckgl/src/layers/common.test.ts

@@ -17,6 +17,7 @@
  * under the License.
  */
 
+import { createElement } from 'react';
 import { PickingInfo } from '@deck.gl/core';
 import { JsonObject, QueryFormData } from '@superset-ui/core';
 import {
@@ -131,6 +132,32 @@ describe('commonLayerProps', () => {
     });
   });
 
+  test('clears a custom tooltip on hover-out instead of trailing the cursor', () => {
+    // Regression test for a custom (Handlebars) deck.gl tooltip that stayed
+    // visible and followed the mouse after leaving a feature.
+    const setTooltip = jest.fn();
+    const customContent = createElement('div', {
+      'data-tooltip-type': 'custom',
+    });
+    const props = commonLayerProps({
+      formData: { ...partialformData } as QueryFormData,
+      setTooltip: setTooltip as any,
+      setTooltipContent: (() => customContent) as any,
+    });
+
+    // Hovering a feature shows the custom tooltip.
+    props.onHover?.({ picked: true, x: 10, y: 20 } as any);
+    expect(setTooltip).toHaveBeenLastCalledWith({
+      content: customContent,
+      x: 10,
+      y: 20,
+    });
+
+    // Moving off the feature must dismiss it, not keep repositioning it.
+    props.onHover?.({ picked: false, x: 30, y: 40 } as any);
+    expect(setTooltip).toHaveBeenLastCalledWith(null);
+  });
+
   test('calls onSelect when table_filter is enabled', () => {
     const formData = {
       ...partialformData,

superset-frontend/plugins/preset-chart-deckgl/src/layers/common.tsx

@@ -16,7 +16,7 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-import { ReactNode, isValidElement } from 'react';
+import { ReactNode } from 'react';
 import {
   ascending as d3ascending,
   quantile as d3quantile,
@@ -70,19 +70,16 @@ export function commonLayerProps({
   if (setTooltipContent) {
     let currentTooltipContent: ReactNode = null;
 
-    const isCustomTooltip = (content: ReactNode): boolean =>
-      isValidElement(content) &&
-      content.props?.['data-tooltip-type'] === 'custom';
-
     onHover = (o: JsonObject) => {
       if (o.picked) {
         currentTooltipContent = setTooltipContent(o);
       }
 
-      if (
-        currentTooltipContent &&
-        (o.picked || isCustomTooltip(currentTooltipContent))
-      ) {
+      // Only show the tooltip while a feature is actually hovered. Custom
+      // (Handlebars) tooltips used to stay visible and follow the cursor
+      // after hover-out because their content was kept on screen even when
+      // nothing was picked.
+      if (o.picked && currentTooltipContent) {
         setTooltip({
           content: currentTooltipContent,
           x: o.x,

superset-frontend/src/features/roles/RoleFormItems.test.tsx

@@ -16,13 +16,15 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-import { render, screen } from 'spec/helpers/testing-library';
+import { render, screen, waitFor, within } from 'spec/helpers/testing-library';
+import userEvent from '@testing-library/user-event';
 import {
   RoleNameField,
   PermissionsField,
   UsersField,
   GroupsField,
 } from './RoleFormItems';
+import { fetchPermissionOptions } from './utils';
 
 jest.mock('./utils', () => ({
   fetchPermissionOptions: jest.fn(),
@@ -53,6 +55,45 @@ test('PermissionsField renders loading state', () => {
   expect(screen.getByTestId('permissions-select')).toBeInTheDocument();
 });
 
+test('PermissionsField shows a permission matched by its raw name even though the label uses spaces (regression for #42041)', async () => {
+  // fetchPermissionOptions matches the raw, underscore-containing name
+  // server-side; the returned label has already gone through
+  // formatPermissionLabel (underscores replaced with spaces for display).
+  // PermissionsField's normalizing filterOption must match the raw search
+  // term against that space-formatted label, or the option the server
+  // legitimately returned gets hidden by client-side re-filtering.
+  jest
+    .mocked(fetchPermissionOptions)
+    .mockImplementation(async (filterValue: string) =>
+      filterValue === 'stg_silver'
+        ? { data: [{ value: 1, label: 'stg silver' }], totalCount: 1 }
+        : // totalCount must exceed the empty initial page here, otherwise
+          // AsyncSelect marks allValuesLoaded and short-circuits every
+          // later fetch, including the search request this test depends on.
+          { data: [], totalCount: 1 },
+    );
+
+  render(<PermissionsField addDangerToast={addDangerToast} />);
+  const combobox = screen.getByRole('combobox');
+  await waitFor(() => userEvent.click(combobox));
+  await userEvent.clear(combobox);
+  await userEvent.type(combobox, 'stg_silver', { delay: 10 });
+
+  await waitFor(() =>
+    expect(fetchPermissionOptions).toHaveBeenCalledWith(
+      'stg_silver',
+      expect.anything(),
+      expect.anything(),
+      addDangerToast,
+    ),
+  );
+  expect(
+    await within(document.querySelector('.rc-virtual-list')!).findByText(
+      'stg silver',
+    ),
+  ).toBeInTheDocument();
+});
+
 test('UsersField renders label and select', () => {
   render(<UsersField addDangerToast={addDangerToast} loading={false} />);
   expect(screen.getByText('Users')).toBeInTheDocument();

superset-frontend/src/features/roles/RoleFormItems.tsx

@@ -60,6 +60,18 @@ export const PermissionsField = ({
         placeholder={t('Select permissions')}
         options={options}
         loading={loading}
+        // formatPermissionLabel renders the raw permission/view_menu name with
+        // underscores replaced by spaces, so AsyncSelect's default client-side
+        // re-filter never matches a raw-name search term (e.g. "stg_silver")
+        // against the displayed label ("stg silver") and hides the
+        // server-matched option. Normalize both sides so client-side narrowing
+        // still works without hiding valid matches. See #42041.
+        filterOption={(input, option) =>
+          String(option?.label ?? '')
+            .toLowerCase()
+            .replace(/_/g, ' ')
+            .includes(input.toLowerCase().replace(/_/g, ' '))
+        }
         getPopupContainer={trigger => trigger.closest('.ant-modal-container')}
         data-test="permissions-select"
       />

superset/charts/client_processing.py

@@ -25,6 +25,7 @@ In order to do that, we reproduce the post-processing in Python for these chart
 """
 
 import logging
+from functools import partial
 from io import BytesIO, StringIO
 from typing import Any, Optional, TYPE_CHECKING, Union
 
@@ -41,6 +42,12 @@ from superset.utils.core import (
     get_column_names,
     get_metric_names,
 )
+from superset.utils.number_format import (
+    AUTO_CURRENCY,
+    format_number_with_config,
+    resolve_auto_currency,
+    SMART_NUMBER,
+)
 
 if TYPE_CHECKING:
     from superset.connectors.sqla.models import BaseDatasource
@@ -49,6 +56,11 @@ if TYPE_CHECKING:
 
 logger = logging.getLogger(__name__)
 
+# Default d3 format the Table plugin applies to percent metrics, mirroring
+# ``number-format/NumberFormats.ts::PERCENT_3_POINT`` as used in the frontend's
+# ``transformProps`` formatter selection.
+PERCENT_3_POINT = ",.3%"
+
 
 def get_column_key(label: tuple[str, ...], metrics: list[str]) -> tuple[Any, ...]:
     """
@@ -74,8 +86,9 @@ def pivot_df(  # pylint: disable=too-many-locals, too-many-arguments, too-many-s
     show_rows_total: bool = False,
     show_columns_total: bool = False,
     apply_metrics_on_rows: bool = False,
+    metric_name_aggfunc: Optional[str] = None,
 ) -> pd.DataFrame:
-    metric_name = __("Total (%(aggfunc)s)", aggfunc=aggfunc)
+    metric_name = __("Total (%(aggfunc)s)", aggfunc=metric_name_aggfunc or aggfunc)
 
     if transpose_pivot:
         rows, columns = columns, rows
@@ -198,9 +211,12 @@ def pivot_df(  # pylint: disable=too-many-locals, too-many-arguments, too-many-s
                     )
                     raise
 
-                subtotal = pivot_v2_aggfunc_map[aggfunc](
-                    df.iloc[slice_, :].apply(pd.to_numeric, errors="coerce"), axis=0
-                )
+                subtotal_values = df.iloc[slice_, :]
+                if aggfunc != CURRENCY_CONTEXT_AGGREGATION:
+                    subtotal_values = subtotal_values.apply(
+                        pd.to_numeric, errors="coerce"
+                    )
+                subtotal = pivot_v2_aggfunc_map[aggfunc](subtotal_values, axis=0)
                 depth = groups.nlevels - len(subgroup) - 1
                 total = metric_name if level == 0 else __("Subtotal")
                 subtotal.name = tuple([*subgroup, total, *([""] * depth)])  # noqa: C409
@@ -232,6 +248,41 @@ def list_unique_values(series: pd.Series) -> str:
     return ", ".join({str(v) for v in pd.Series.unique(series)})
 
 
+def union_currency_context(
+    values: Union[pd.Series, pd.DataFrame], axis: int = 0
+) -> Union[tuple[str, ...], pd.Series]:
+    """Union the currency sets contributing to a Pivot Table cell or total."""
+    if isinstance(values, pd.DataFrame):
+        contexts = (
+            [union_currency_context(values[column]) for column in values.columns]
+            if axis == 0
+            else [union_currency_context(values.loc[index]) for index in values.index]
+        )
+        index = values.columns if axis == 0 else values.index
+        return pd.Series(contexts, index=index, dtype=object)
+
+    currencies: dict[str, None] = {}
+    for value in values:
+        if isinstance(value, (list, set, frozenset, tuple)):
+            currencies.update((str(currency), None) for currency in value)
+    return tuple(currencies)
+
+
+CURRENCY_CONTEXT_AGGREGATION = "__currency_context__"
+
+# The frontend's plain Count aggregator is the only Pivot Table aggregator
+# without ``getCurrencies()``. Fraction wrappers inherit that absence, so these
+# modes use the query-wide detected fallback instead of per-cell context.
+PIVOT_AGGREGATIONS_WITHOUT_CURRENCY_CONTEXT = frozenset(
+    {
+        "Count",
+        "Count as Fraction of Total",
+        "Count as Fraction of Rows",
+        "Count as Fraction of Columns",
+    }
+)
+
+
 pivot_v2_aggfunc_map = {
     "Count": pd.Series.count,
     "Count Unique Values": pd.Series.nunique,
@@ -253,53 +304,403 @@ pivot_v2_aggfunc_map = {
     "Count as Fraction of Total": pd.Series.count,
     "Count as Fraction of Rows": pd.Series.count,
     "Count as Fraction of Columns": pd.Series.count,
+    CURRENCY_CONTEXT_AGGREGATION: union_currency_context,
 }
 
 
+def format_column(
+    df: pd.DataFrame,
+    column: Any,
+    d3_format: Optional[str],
+    currency: dict[str, Any],
+    detected_currency: Optional[str] = None,
+    currency_context: Optional[pd.Series] = None,
+    fallback_to_detected: bool = True,
+) -> None:
+    """
+    Format a column in place when a number or currency format is configured.
+
+    ``detected_currency`` represents the query-wide single currency. When a
+    parallel ``currency_context`` series is present, AUTO uses each row/cell's
+    contributing currencies first. Mixed context renders a neutral number;
+    empty context optionally falls back to query-wide detection.
+    """
+    if d3_format or currency.get("symbol"):
+        if currency_context is None:
+            resolved_currency = resolve_auto_currency(currency, detected_currency)
+            df[column] = df[column].apply(
+                partial(format_number_with_config, d3_format, resolved_currency)
+            )
+            return
+
+        contexts = currency_context.reindex(df.index)
+        df[column] = [
+            format_number_with_config(
+                d3_format,
+                resolve_auto_currency(
+                    currency,
+                    detected_currency,
+                    context,
+                    fallback_to_detected,
+                ),
+                value,
+            )
+            for value, context in zip(df[column], contexts, strict=True)
+        ]
+
+
+def get_datasource_column_formats(
+    datasource: Optional[Union["BaseDatasource", "Query"]],
+) -> tuple[dict[str, str | None], dict[str, str]]:
+    """Return saved metric formats and verbose labels from a datasource."""
+    if not datasource:
+        return {}, {}
+
+    datasource_data = datasource.data
+    return (
+        datasource_data.get("column_formats") or {},
+        datasource_data.get("verbose_map") or {},
+    )
+
+
+def get_datasource_currency_formats(
+    datasource: Optional[Union["BaseDatasource", "Query"]],
+) -> tuple[dict[str, dict[str, Any]], dict[str, str]]:
+    """Return saved metric currencies and verbose labels from a datasource.
+
+    The frontend derives ``datasource.currencyFormats`` from each metric's
+    ``currency`` property in ``hydrateExplore.ts``. Report processing receives
+    the raw datasource payload, so it performs the same derivation here.
+    """
+    if not datasource:
+        return {}, {}
+
+    datasource_data = datasource.data
+    stored_currency_formats = datasource_data.get("currency_formats")
+    currency_formats: dict[str, dict[str, Any]] = (
+        {
+            metric: currency
+            for metric, currency in stored_currency_formats.items()
+            if isinstance(metric, str) and isinstance(currency, dict)
+        }
+        if isinstance(stored_currency_formats, dict)
+        else {}
+    )
+    currency_formats.update(
+        {
+            metric["metric_name"]: metric["currency"]
+            for metric in datasource_data.get("metrics") or []
+            if isinstance(metric, dict)
+            and isinstance(metric.get("metric_name"), str)
+            and isinstance(metric.get("currency"), dict)
+            and metric["currency"].get("symbol")
+        }
+    )
+    return currency_formats, datasource_data.get("verbose_map") or {}
+
+
+def get_datasource_currency_column(
+    datasource: Optional[Union["BaseDatasource", "Query"]],
+    df: pd.DataFrame,
+) -> Optional[str]:
+    """Return the currency-code column name as represented in ``df``."""
+    if not datasource:
+        return None
+
+    datasource_data = datasource.data
+    currency_column = datasource_data.get("currency_code_column")
+    if not isinstance(currency_column, str):
+        return None
+    if currency_column in df.columns:
+        return currency_column
+
+    verbose_column = (datasource_data.get("verbose_map") or {}).get(
+        currency_column, currency_column
+    )
+    return verbose_column if verbose_column in df.columns else None
+
+
+def currency_context_value(value: Any) -> tuple[str, ...]:
+    """Convert a truthy row currency value to frontend-compatible context."""
+    try:
+        if value is None or pd.isna(value) or not value:
+            return ()
+    except (TypeError, ValueError):
+        return ()
+    return (str(value),)
+
+
+def build_pivot_currency_context(
+    df: pd.DataFrame,
+    currency_column: str,
+    metrics: list[str],
+    pivot_options: dict[str, Any],
+) -> pd.DataFrame:
+    """
+    Pivot contributing currency sets through the same layout as metric values.
+
+    This mirrors the frontend Pivot Table aggregators' ``currencySet``: every
+    output cell, subtotal, and total carries the union of currencies from its
+    contributing records while numeric aggregation remains unchanged.
+    """
+    currency_source = df.copy()
+    row_context = currency_source[currency_column].map(currency_context_value)
+    for metric in metrics:
+        currency_source[metric] = row_context
+
+    currency_pivot_options = {
+        **pivot_options,
+        "aggfunc": CURRENCY_CONTEXT_AGGREGATION,
+        "metric_name_aggfunc": pivot_options["aggfunc"],
+    }
+    return pivot_df(currency_source, **currency_pivot_options)
+
+
+def get_pivot_currency_format(form_data: dict[str, Any]) -> dict[str, Any]:
+    """Return Pivot Table currency config from transformed or stored form data."""
+    currency_format = form_data.get("currencyFormat") or form_data.get(
+        "currency_format"
+    )
+    return currency_format if isinstance(currency_format, dict) else {}
+
+
+def has_auto_currency_format(
+    form_data: dict[str, Any],
+    datasource: Optional[Union["BaseDatasource", "Query"]] = None,
+) -> bool:
+    """Return whether the Pivot Table has a global or per-metric AUTO format."""
+    currency_formats = [
+        get_pivot_currency_format(form_data),
+        *merge_currency_formats(form_data, datasource).values(),
+    ]
+    return any(
+        isinstance(currency, dict) and currency.get("symbol") == AUTO_CURRENCY
+        for currency in currency_formats
+    )
+
+
+def merge_column_formats(
+    form_data: dict[str, Any],
+    datasource: Optional[Union["BaseDatasource", "Query"]],
+) -> dict[str, str | None]:
+    """Merge saved formats with truthy chart overrides using verbose labels."""
+    saved_formats, verbose_map = get_datasource_column_formats(datasource)
+    column_formats = {
+        verbose_map.get(metric, metric): d3_format
+        for metric, d3_format in saved_formats.items()
+    }
+    column_formats.update(
+        {
+            verbose_map.get(metric, metric): d3_format
+            for metric, d3_format in (form_data.get("columnFormats") or {}).items()
+            if d3_format
+        }
+    )
+    return column_formats
+
+
+def merge_currency_formats(
+    form_data: dict[str, Any],
+    datasource: Optional[Union["BaseDatasource", "Query"]],
+) -> dict[str, dict[str, Any]]:
+    """Merge saved metric currencies with chart overrides by verbose label."""
+    saved_formats, verbose_map = get_datasource_currency_formats(datasource)
+    currency_formats = {
+        verbose_map.get(metric, metric): currency
+        for metric, currency in saved_formats.items()
+    }
+    currency_formats.update(
+        {
+            verbose_map.get(metric, metric): currency
+            for metric, currency in (form_data.get("currencyFormats") or {}).items()
+            if isinstance(currency, dict) and currency.get("symbol")
+        }
+    )
+    return currency_formats
+
+
 def pivot_table_v2(
     df: pd.DataFrame,
     form_data: dict[str, Any],
     datasource: Optional[Union["BaseDatasource", "Query"]] = None,
+    apply_number_format: bool = True,
+    detected_currency: Optional[str] = None,
 ) -> pd.DataFrame:
     """
     Pivot table v2.
     """
     verbose_map = datasource.data["verbose_map"] if datasource else None
+    metrics = get_metric_names(form_data["metrics"], verbose_map)
+    pivot_options: dict[str, Any] = {
+        "rows": get_column_names(form_data.get("groupbyRows"), verbose_map),
+        "columns": get_column_names(form_data.get("groupbyColumns"), verbose_map),
+        "metrics": metrics,
+        "aggfunc": form_data.get("aggregateFunction", "Sum"),
+        "transpose_pivot": bool(form_data.get("transposePivot")),
+        "combine_metrics": bool(form_data.get("combineMetric")),
+        "show_rows_total": bool(form_data.get("rowTotals")),
+        "show_columns_total": bool(form_data.get("colTotals")),
+        "apply_metrics_on_rows": form_data.get("metricsLayout") == "ROWS",
+    }
 
-    return pivot_df(
-        df,
-        rows=get_column_names(form_data.get("groupbyRows"), verbose_map),
-        columns=get_column_names(form_data.get("groupbyColumns"), verbose_map),
-        metrics=get_metric_names(form_data["metrics"], verbose_map),
-        aggfunc=form_data.get("aggregateFunction", "Sum"),
-        transpose_pivot=bool(form_data.get("transposePivot")),
-        combine_metrics=bool(form_data.get("combineMetric")),
-        show_rows_total=bool(form_data.get("rowTotals")),
-        show_columns_total=bool(form_data.get("colTotals")),
-        apply_metrics_on_rows=form_data.get("metricsLayout") == "ROWS",
-    )
+    pivoted = pivot_df(df, **pivot_options)
+    if apply_number_format:
+        currency_context = None
+        if (
+            pivot_options["aggfunc"] not in PIVOT_AGGREGATIONS_WITHOUT_CURRENCY_CONTEXT
+            and has_auto_currency_format(form_data, datasource)
+            and (currency_column := get_datasource_currency_column(datasource, df))
+        ):
+            currency_context = build_pivot_currency_context(
+                df,
+                currency_column,
+                metrics,
+                pivot_options,
+            )
+        return apply_pivot_number_formats(
+            pivoted,
+            form_data,
+            detected_currency,
+            datasource,
+            currency_context,
+        )
+    return pivoted
+
+
+def apply_pivot_number_formats(
+    df: pd.DataFrame,
+    form_data: dict[str, Any],
+    detected_currency: Optional[str] = None,
+    datasource: Optional[Union["BaseDatasource", "Query"]] = None,
+    currency_context: Optional[pd.DataFrame] = None,
+) -> pd.DataFrame:
+    """
+    Apply `valueFormat`/`columnFormats` and currency config to pivot values.
+
+    The metric name is the first column level, or the last when `combineMetric`
+    moves it there; in the ROWS metrics layout it is on the index instead.
+    Per-metric overrides fall back to the global value format.
+    """
+    value_format = form_data.get("valueFormat")
+    column_formats = merge_column_formats(form_data, datasource)
+    currency_format = get_pivot_currency_format(form_data)
+    currency_formats = merge_currency_formats(form_data, datasource)
+    metric_level = -1 if form_data.get("combineMetric") else 0
+    metrics_on_rows = form_data.get("metricsLayout") == "ROWS"
+
+    if metrics_on_rows:
+        df = df.T
+        if currency_context is not None:
+            currency_context = currency_context.T
+
+    for column in df.columns:
+        metric = column[metric_level] if isinstance(column, tuple) else column
+        column_currency_context = (
+            currency_context[column]
+            if currency_context is not None and column in currency_context.columns
+            else None
+        )
+        column_number_format = column_formats.get(metric) or value_format
+        column_currency = currency_formats.get(metric) or currency_format
+        if not column_number_format and not column_currency.get("symbol"):
+            # The frontend Pivot Table formatter falls back to SMART_NUMBER when
+            # neither a value format nor a currency is configured
+            # (``getNumberFormatter``'s default key), so unconfigured metric
+            # cells must not be left raw.
+            column_number_format = SMART_NUMBER
+        format_column(
+            df,
+            column,
+            column_number_format,
+            column_currency,
+            detected_currency,
+            column_currency_context,
+        )
+
+    return df.T if metrics_on_rows else df
 
 
 def table(
     df: pd.DataFrame,
     form_data: dict[str, Any],
-    datasource: Optional[  # pylint: disable=unused-argument
-        Union["BaseDatasource", "Query"]
-    ] = None,
+    datasource: Optional[Union["BaseDatasource", "Query"]] = None,
+    apply_number_format: bool = True,
+    detected_currency: Optional[str] = None,
 ) -> pd.DataFrame:
     """
     Table.
     """
-    # apply `d3NumberFormat` to columns, if present
-    column_config = form_data.get("column_config", {})
-    for column, config in column_config.items():
-        if "d3NumberFormat" in config:
-            format_ = "{:" + config["d3NumberFormat"] + "}"
-            try:
-                df[column] = df[column].apply(format_.format)
-            except Exception:  # pylint: disable=broad-except  # noqa: S110
-                # if we can't format the column for any reason, send as is
-                pass
+    if not apply_number_format:
+        return df
+
+    saved_formats, verbose_map = get_datasource_column_formats(datasource)
+    saved_currency_formats, _ = get_datasource_currency_formats(datasource)
+    currency_column = get_datasource_currency_column(datasource, df)
+    row_currency_context = (
+        df[currency_column].map(currency_context_value) if currency_column else None
+    )
+    column_config = form_data.get("column_config") or {}
+
+    def label_of(name: str) -> str:
+        """Return the column label as it appears in ``df`` (verbose-renamed)."""
+        return name if name in df.columns else verbose_map.get(name, name)
+
+    # Index the per-column overrides by the label present in ``df`` so numeric
+    # and metric columns can be looked up while iterating the frame.
+    number_format_by_label = {
+        label_of(name): fmt for name, fmt in saved_formats.items()
+    }
+    currency_by_label = {
+        label_of(name): currency for name, currency in saved_currency_formats.items()
+    }
+    config_by_label = {label_of(name): config for name, config in column_config.items()}
+    metric_labels = {
+        label_of(name) for name in get_metric_names(form_data.get("metrics"))
+    }
+    # Percent metric columns are emitted with a leading ``%`` and are not
+    # verbose-renamed, so match them by that prefixed label.
+    percent_metric_labels = {
+        f"%{name}"
+        for name in get_metric_names(form_data.get("percent_metrics"))
+        if f"%{name}" in df.columns
+    }
+
+    # Mirror the Table plugin's per-column formatter selection in
+    # ``plugin-chart-table/src/transformProps.ts``: percent metrics default to
+    # PERCENT_3_POINT, every (numeric) metric gets a formatter that defaults to
+    # SMART_NUMBER, and other numeric columns are only formatted when an explicit
+    # format or currency is configured. Dimension and non-numeric columns are
+    # left untouched, matching the browser.
+    for column in df.columns:
+        config = config_by_label.get(column) or {}
+        configured_currency = config.get("currencyFormat") or {}
+        number_format = config.get("d3NumberFormat") or number_format_by_label.get(
+            column
+        )
+        currency = (
+            configured_currency
+            if configured_currency.get("symbol")
+            else currency_by_label.get(column) or {}
+        )
+
+        is_number = pd.api.types.is_numeric_dtype(df[column])
+
+        if column in percent_metric_labels:
+            format_column(df, column, number_format or PERCENT_3_POINT, {})
+        elif (column in metric_labels and is_number) or (
+            is_number and (number_format or currency.get("symbol"))
+        ):
+            if not number_format and not currency.get("symbol"):
+                number_format = SMART_NUMBER
+            format_column(
+                df,
+                column,
+                number_format,
+                currency,
+                detected_currency,
+                row_currency_context,
+                fallback_to_detected=False,
+            )
 
     return df
 
@@ -393,7 +794,14 @@ def apply_client_processing(  # noqa: C901
         if datasource:
             df.rename(columns=datasource.data["verbose_map"], inplace=True)
 
-        processed_df = post_processor(df, form_data, datasource)
+        apply_number_format = query["result_format"] == ChartDataResultFormat.JSON
+        processed_df = post_processor(
+            df,
+            form_data,
+            datasource,
+            apply_number_format,
+            query.get("detected_currency"),
+        )
 
         query["colnames"] = list(processed_df.columns)
         query["indexnames"] = list(processed_df.index)

superset/charts/data/api.py

@@ -646,8 +646,6 @@ class ChartDataRestApi(ChartRestApi):
             return self.response_422(message=sanitize_error_message(exc.message))
         except ChartDataQueryFailedError as exc:
             return self.response_400(message=sanitize_error_message(exc.message))
-        except QueryObjectValidationError as exc:
-            return self.response_400(message=sanitize_error_message(exc.message))
 
             # Log is_cached if extra payload callback is provided
         materialized_result = result.materialize()

superset/commands/database/sync_permissions.py

@@ -73,6 +73,7 @@ class SyncPermissionsCommand(BaseCommand):
         self.username = username
         self._old_db_connection_name: str | None = old_db_connection_name
         self._db_connection: Database | None = db_connection
+        self._user_id: int | None = None
 
         self.async_mode: bool = app.config["SYNC_DB_PERMISSIONS_IN_ASYNC_MODE"]
 
@@ -99,11 +100,15 @@ class SyncPermissionsCommand(BaseCommand):
         if not self._db_connection:
             raise DatabaseNotFoundError()
 
-        # Need user info to impersonate for OAuth2 connections
-        if not self.username or not security_manager.get_user_by_username(
-            self.username
+        # Need user info to impersonate for OAuth2 connections. The id is
+        # captured here, at validation/enqueue time, so that an async run of
+        # this command binds to whoever held the username right now, rather
+        # than re-resolving the (mutable) username at execution time.
+        if not self.username or not (
+            user := security_manager.get_user_by_username(self.username)
         ):
             raise UserNotFoundInSessionError()
+        self._user_id = user.id
 
         with self.db_connection.get_sqla_engine() as engine:
             try:
@@ -126,7 +131,7 @@ class SyncPermissionsCommand(BaseCommand):
         self.validate()
         if self.async_mode:
             sync_database_permissions_task.delay(
-                self.db_connection_id, self.username, self.old_db_connection_name
+                self.db_connection_id, self._user_id, self.old_db_connection_name
             )
             return
 
@@ -313,14 +318,14 @@ class SyncPermissionsCommand(BaseCommand):
 
 @celery_app.task(name="sync_database_permissions", soft_time_limit=600)
 def sync_database_permissions_task(
-    database_id: int, username: str, old_db_connection_name: str
+    database_id: int, user_id: int, old_db_connection_name: str
 ) -> None:
     """
     Celery task that triggers the SyncPermissionsCommand in async mode.
     """
     with app.test_request_context():
         try:
-            user = security_manager.get_user_by_username(username)
+            user = security_manager.get_user_by_id(user_id)
             if not user:
                 raise UserNotFoundInSessionError()
             g.user = user
@@ -336,7 +341,7 @@ def sync_database_permissions_task(
 
             SyncPermissionsCommand(
                 database_id,
-                username,
+                user.username,
                 old_db_connection_name=old_db_connection_name,
                 db_connection=db_connection,
             ).sync_database_permissions()

superset/commands/database/validate_sql.py

@@ -21,6 +21,7 @@ from typing import Any, Optional
 from flask import current_app as app
 from flask_babel import gettext as __
 
+from superset import security_manager
 from superset.commands.base import BaseCommand
 from superset.commands.database.exceptions import (
     DatabaseNotFoundError,
@@ -69,6 +70,17 @@ class ValidateSQLCommand(BaseCommand):
         schema = self._properties.get("schema")
         template_params = self._properties.get("template_params") or {}
 
+        # Check access before rendering the Jinja template (mirrors the SQL
+        # Lab execute path).
+        security_manager.raise_for_access(
+            database=self._model,
+            sql=sql,
+            catalog=catalog,
+            schema=schema,
+            template_params=template_params,
+            force_dataset_match=True,
+        )
+
         try:
             # Render Jinja templates to handle template syntax before
             # validation. Note: The ENABLE_TEMPLATE_PROCESSING feature flag is

superset/commands/sql_lab/estimate.py

@@ -23,8 +23,9 @@ from flask import current_app as app
 from flask_babel import gettext as __
 from jinja2.exceptions import TemplateError
 
-from superset import db, is_feature_enabled, security_manager
+from superset import is_feature_enabled, security_manager
 from superset.commands.base import BaseCommand
+from superset.daos.database import DatabaseDAO
 from superset.errors import ErrorLevel, SupersetError, SupersetErrorType
 from superset.exceptions import (
     SupersetDisallowedSQLFunctionException,
@@ -66,8 +67,10 @@ class QueryEstimationCommand(BaseCommand):
         self._catalog = params.get("catalog")
 
     def validate(self) -> None:
-        self._database = db.session.query(Database).get(self._database_id)
-        if not self._database:
+        # Load the database through the DAO so ``DatabaseFilter`` scopes
+        # visibility the same way it does on the SQL Lab execution path.
+        database = DatabaseDAO.find_by_id(self._database_id)
+        if not database:
             raise SupersetErrorException(
                 SupersetError(
                     message=__("The database could not be found"),
@@ -76,7 +79,17 @@ class QueryEstimationCommand(BaseCommand):
                 ),
                 status=404,
             )
-        security_manager.raise_for_access(database=self._database)
+        self._database = database
+        # Pass the SQL so table-level authorization runs, mirroring the SQL
+        # Lab execution path. Runs before Jinja templating in ``run()``.
+        security_manager.raise_for_access(
+            database=self._database,
+            sql=self._sql,
+            catalog=self._catalog,
+            schema=self._schema or None,
+            template_params=self._template_params,
+            force_dataset_match=True,
+        )
 
     def _apply_sql_security(self, sql: str) -> str:
         """Run the disallowed-function/table, DML and RLS controls against the
@@ -150,6 +163,7 @@ class QueryEstimationCommand(BaseCommand):
 
         sql = self._sql
         if self._template_params:
+            # Access is already checked in validate() before any rendering.
             template_processor = get_template_processor(self._database)
             try:
                 sql = template_processor.process_template(sql, **self._template_params)

superset/commands/streaming_export/base.py

@@ -85,6 +85,14 @@ class BaseStreamingCSVExportCommand(BaseCommand):
         """
         Get the SQL query, database, catalog, and schema for execution.
 
+        The returned SQL is expected to already carry any
+        ``is_split=False`` mutation applied upstream (e.g. by
+        ``get_query_str_extended`` for charts, or SQL Lab's stored
+        ``executed_sql``); ``run()`` applies the complementary
+        ``is_split=True`` mutation exactly once before execution, so the
+        mutator fires once total under either ``MUTATE_AFTER_SPLIT``
+        setting.
+
         Returns:
             Tuple of (sql_query, database_object, catalog, schema)
         """
@@ -223,13 +231,27 @@ class BaseStreamingCSVExportCommand(BaseCommand):
             # Merge database to prevent DetachedInstanceError
             merged_database = session.merge(database)
 
+            # `is_split=True` mirrors the non-streaming download path exactly:
+            # Database.get_df() -> _execute_sql_with_mutation_and_logging()
+            # always calls mutate_sql_based_on_config(..., is_split=True) on
+            # SQL Lab's stored select_sql/executed_sql, and the chart query
+            # path applies its own mutation upstream (in
+            # get_query_str_extended, with is_split=False) before landing
+            # here. Since `is_split` and `MUTATE_AFTER_SPLIT` are compared
+            # for equality, `is_split=True` is the complement of that
+            # upstream chart mutation -- together they mutate the SQL
+            # exactly once for either MUTATE_AFTER_SPLIT setting, instead of
+            # double-mutating when it's False and never mutating when it's
+            # True.
+            mutated_sql = merged_database.mutate_sql_based_on_config(sql, is_split=True)
+
             with merged_database.get_sqla_engine(
                 catalog=catalog, schema=schema
             ) as engine:
                 with engine.connect() as connection:
                     result_proxy = connection.execution_options(
                         stream_results=True
-                    ).execute(text(sql))
+                    ).execute(text(mutated_sql))
 
                     columns = list(result_proxy.keys())
 

superset/common/query_object.py

@@ -440,6 +440,22 @@ class QueryObject:  # pylint: disable=too-many-instance-attributes
         cache_dict: dict[str, Any] = dict(self.to_dict())
         cache_dict.update(extra)
 
+        if "extra_cache_keys" in cache_dict:
+            # Order carries no meaning here (an unordered set of opaque
+            # Jinja url_param()-derived values), but hash_from_dict only
+            # sorts dict keys, not list values, so an unsorted list makes
+            # the cache key depend on Python's per-process hash-randomized
+            # set iteration order (see SqlaTable.get_extra_cache_keys).
+            # Normalize once here so every producer of extra_cache_keys is
+            # safe by construction. Sort on (type name, str value) rather
+            # than a bare str() so values that stringify identically but
+            # differ in type (e.g. 1 and "1") still sort deterministically
+            # instead of falling back to input order.
+            cache_dict["extra_cache_keys"] = sorted(
+                cache_dict["extra_cache_keys"],
+                key=lambda value: (type(value).__name__, str(value)),
+            )
+
         # TODO: the below KVs can all be cleaned up and moved to `to_dict()` at some
         #  predetermined point in time when orgs are aware that the previously
         #  cached results will be invalidated.

superset/connectors/sqla/models.py

@@ -1651,7 +1651,18 @@ class SqlaTable(
     def dttm_cols(self) -> list[str]:
         l = [c.column_name for c in self.columns if c.is_dttm]  # noqa: E741
         if self.main_dttm_col and self.main_dttm_col not in l:
-            l.append(self.main_dttm_col)
+            # Only treat ``main_dttm_col`` as a datetime column when the column it
+            # points to is actually temporal. A column whose "Is Temporal" flag was
+            # removed must not keep being reported as a datetime column just because
+            # it is still referenced by ``main_dttm_col`` (#30510). When the column
+            # is not present on the dataset, fall back to the legacy behavior of
+            # trusting ``main_dttm_col``.
+            main_dttm_column: TableColumn | None = next(
+                (c for c in self.columns if c.column_name == self.main_dttm_col),
+                None,
+            )
+            if main_dttm_column is None or main_dttm_column.is_dttm:
+                l.append(self.main_dttm_col)
         return l
 
     @property

superset/db_engine_specs/postgres.py

@@ -641,12 +641,11 @@ class PostgresEngineSpec(BasicParametersMixin, PostgresBaseEngineSpec):
         """
         Return the default schema for a given query.
 
-        This method simply uses the parent method after checking that there are no
-        malicious path setting in the query.
+        This method simply uses the parent method after checking that the query
+        cannot rebind the schema used to resolve unqualified table names.
         """
         script = process_jinja_sql(query.sql, database, template_params).script
-        settings = script.get_settings()
-        if "search_path" in settings:
+        if script.changes_default_schema():
             raise SupersetSecurityException(
                 SupersetError(
                     error_type=SupersetErrorType.QUERY_SECURITY_ACCESS_ERROR,

superset/jinja_context.py

@@ -915,6 +915,18 @@ class BaseTemplateProcessor:
         """
         return self._context.copy()
 
+    def get_template_context(self, **kwargs: Any) -> dict[str, Any]:
+        """
+        Build the validated context used to render a template.
+
+        Split out from ``process_template`` so that validation paths which
+        render a pre-parsed template (``superset.sql.parse.process_jinja_sql``)
+        use exactly the same context as execution, keeping the validated SQL
+        identical to the executed SQL.
+        """
+        kwargs.update(self._context)
+        return validate_template_context(self.engine, kwargs)
+
     def process_template(self, sql: str, **kwargs: Any) -> str:
         """Processes a sql template
 
@@ -984,8 +996,7 @@ class BaseTemplateProcessor:
 
             raise SupersetTemplateException(message) from ex
 
-        kwargs.update(self._context)
-        context = validate_template_context(self.engine, kwargs)
+        context = self.get_template_context(**kwargs)
 
         try:
             return template.render(context)
@@ -1133,27 +1144,21 @@ class HiveTemplateProcessor(PrestoTemplateProcessor):
 class SparkTemplateProcessor(HiveTemplateProcessor):
     engine = "spark"
 
-    def process_template(self, sql: str, **kwargs: Any) -> str:
-        template = self.env.from_string(sql)
-        kwargs.update(self._context)
-
+    def get_template_context(self, **kwargs: Any) -> dict[str, Any]:
+        context = super().get_template_context(**kwargs)
         # Backwards compatibility if migrating from Hive.
-        context = validate_template_context(self.engine, kwargs)
         context["hive"] = context["spark"]
-        return template.render(context)
+        return context
 
 
 class TrinoTemplateProcessor(PrestoTemplateProcessor):
     engine = "trino"
 
-    def process_template(self, sql: str, **kwargs: Any) -> str:
-        template = self.env.from_string(sql)
-        kwargs.update(self._context)
-
+    def get_template_context(self, **kwargs: Any) -> dict[str, Any]:
+        context = super().get_template_context(**kwargs)
         # Backwards compatibility if migrating from Presto.
-        context = validate_template_context(self.engine, kwargs)
         context["presto"] = context["trino"]
-        return template.render(context)
+        return context
 
 
 DEFAULT_PROCESSORS = {

superset/mcp_service/chart/schemas.py

@@ -36,6 +36,7 @@ from pydantic import (
     field_validator,
     model_serializer,
     model_validator,
+    StrictBool,
     ValidationError,
 )
 from typing_extensions import Self
@@ -2202,6 +2203,19 @@ class ListChartsRequest(
 ):
     """Request schema for list_charts with clear, unambiguous types."""
 
+    certified: Annotated[
+        StrictBool | None,
+        Field(
+            default=None,
+            description=(
+                "Filter by governance certification status. Use true to return "
+                "only certified charts (preferred when selecting governed "
+                "assets), false to return only uncertified charts, or omit to "
+                "return both (default)."
+            ),
+        ),
+    ]
+
     deleted_state: Annotated[
         Literal["include", "only"] | None,
         Field(

superset/mcp_service/chart/tool/list_charts.py

@@ -80,7 +80,9 @@ async def list_charts(
     """List charts with filtering and search.
 
     Returns chart metadata including id, name, viz_type, URL, and last
-    modified time.
+    modified time. Set ``request.certified`` to true to return only governed
+    charts; false returns only uncertified charts, while omitting it preserves
+    the unfiltered behavior.
 
     **IMPORTANT**: All parameters must be wrapped in a ``request`` object.
     Do NOT pass ``search``, ``page``, ``page_size``, etc. as top-level
@@ -124,7 +126,7 @@ async def list_charts(
         )
     )
 
-    from superset.charts.filters import ChartDeletedStateFilter
+    from superset.charts.filters import ChartCertifiedFilter, ChartDeletedStateFilter
     from superset.daos.chart import ChartDAO
     from superset.mcp_service.common.schema_discovery import (
         CHART_SORTABLE_COLUMNS,
@@ -179,6 +181,13 @@ async def list_charts(
 
     try:
         with event_logger.log_context(action="mcp.list_charts.query"):
+            custom_filters = None
+            if request.certified is not None:
+                custom_filters = {
+                    "certified": tool.build_bound_filter(
+                        ChartCertifiedFilter, request.certified
+                    )
+                }
             result = tool.run_tool(
                 filters=request.filters,
                 search=request.search,
@@ -190,6 +199,7 @@ async def list_charts(
                 created_by_me=request.created_by_me,
                 edited_by_me=request.edited_by_me,
                 deleted_state=request.deleted_state,
+                custom_filters=custom_filters,
             )
         count = len(result.charts) if hasattr(result, "charts") else 0
         total_pages = getattr(result, "total_pages", None)

superset/mcp_service/dataset/schemas.py

@@ -32,6 +32,7 @@ from pydantic import (
     field_validator,
     model_serializer,
     model_validator,
+    StrictBool,
 )
 
 from superset.daos.base import ColumnOperator, ColumnOperatorEnum
@@ -246,6 +247,19 @@ class ListDatasetsRequest(
     test_list_datasets_with_string_filters.
     """
 
+    certified: Annotated[
+        StrictBool | None,
+        Field(
+            default=None,
+            description=(
+                "Filter by governance certification status. Use true to return "
+                "only certified datasets (preferred when selecting governed "
+                "semantic-layer assets), false to return only uncertified "
+                "datasets, or omit to return both (default)."
+            ),
+        ),
+    ]
+
     @field_validator("filters", mode="before")
     @classmethod
     def parse_filters(cls, v: Any) -> Any:

superset/mcp_service/dataset/tool/list_datasets.py

@@ -94,7 +94,9 @@ async def list_datasets(
     """List datasets with filtering and search.
 
     Returns dataset metadata including table name, schema, and last modified
-    time.
+    time. Set ``request.certified`` to true to return only governed,
+    semantic-layer datasets; false returns only uncertified datasets, while
+    omitting it preserves the unfiltered behavior.
 
     **IMPORTANT**: All parameters must be wrapped in a ``request`` object.
     Do NOT pass ``search``, ``page``, ``page_size``, etc. as top-level
@@ -160,6 +162,7 @@ async def list_datasets(
 
     try:
         from superset.daos.dataset import DatasetDAO
+        from superset.datasets.filters import DatasetCertifiedFilter
         from superset.mcp_service.common.schema_discovery import (
             DATASET_SORTABLE_COLUMNS,
             get_all_column_names,
@@ -191,6 +194,13 @@ async def list_datasets(
         )
 
         with event_logger.log_context(action="mcp.list_datasets.query"):
+            custom_filters = None
+            if request.certified is not None:
+                custom_filters = {
+                    "certified": tool.build_bound_filter(
+                        DatasetCertifiedFilter, request.certified
+                    )
+                }
             result = tool.run_tool(
                 filters=request.filters,
                 search=request.search,
@@ -201,6 +211,7 @@ async def list_datasets(
                 page_size=request.page_size,
                 created_by_me=request.created_by_me,
                 edited_by_me=request.edited_by_me,
+                custom_filters=custom_filters,
             )
 
         await ctx.info(

superset/mcp_service/mcp_core.py

@@ -115,28 +115,30 @@ class BaseCore(ABC):
         self.logger.warning(message)
 
 
-class DeletedStateBoundFilter:
-    """Adapt a FAB deleted-state filter for ``BaseDAO.list`` custom_filters.
+class BoundFilter:
+    """Bind a caller value to a FAB filter used by ``BaseDAO.list``.
 
     ``BaseDAO.list`` invokes custom filters as ``apply(query, None)``, but the
-    ``BaseDeletedStateFilter`` subclasses interpret ``None`` as "live rows
-    only". Binding the value at construction lets the DAO-side invocation
-    reach the FAB filter with the caller's actual ``include``/``only`` choice.
-
-    ``model`` re-exposes the FAB filter's SoftDeleteMixin model class so the
-    caller can scope the session visibility bypass without re-consulting the
-    (Optional) filter-class attribute.
+    request value is already available to MCP callers. Binding it at
+    construction preserves that value for the DAO-side invocation.
     """
 
-    def __init__(self, inner: Any, value: str, model: type) -> None:
+    def __init__(self, inner: Any, value: Any) -> None:
         self._inner = inner
         self._value = value
-        self.model = model
 
     def apply(self, query: Any, value: Any) -> Any:
         return self._inner.apply(query, self._value)
 
 
+class DeletedStateBoundFilter(BoundFilter):
+    """Bound deleted-state filter carrying its visibility-bypass model."""
+
+    def __init__(self, inner: Any, value: str, model: type) -> None:
+        super().__init__(inner, value)
+        self.model = model
+
+
 class ModelListCore(BaseCore, Generic[L]):
     """
     Generic tool for listing model objects with filtering, search, pagination, and
@@ -340,6 +342,11 @@ class ModelListCore(BaseCore, Generic[L]):
         inner = self._deleted_state_filter("id", datamodel)
         return DeletedStateBoundFilter(inner, normalized, model)
 
+    def build_bound_filter(self, filter_class: type, value: Any) -> BoundFilter:
+        """Bind an MCP value to a FAB filter for this core's DAO model."""
+        datamodel = SQLAInterface(self.dao_class.model_cls, db.session)
+        return BoundFilter(filter_class("id", datamodel), value)
+
     def run_tool(
         self,
         filters: Any | None = None,
@@ -352,6 +359,7 @@ class ModelListCore(BaseCore, Generic[L]):
         created_by_me: bool = False,
         edited_by_me: bool = False,
         deleted_state: str | None = None,
+        custom_filters: Dict[str, Any] | None = None,
     ) -> L:
         # Clamp page_size to MAX_PAGE_SIZE as defense-in-depth
         page_size = min(page_size, MAX_PAGE_SIZE)
@@ -401,7 +409,9 @@ class ModelListCore(BaseCore, Generic[L]):
             "search": search,
             "columns_to_load": columns_to_load,
         }
+        dao_custom_filters = dict(custom_filters or {})
         if deleted_state_bound is not None:
+            dao_custom_filters["deleted_state"] = deleted_state_bound
             # The soft-delete ORM listener appends ``deleted_at IS NULL`` at
             # execution time, so the session-scoped bypass must span both
             # executions inside DAO.list (count + fetch). The context manager
@@ -410,11 +420,13 @@ class ModelListCore(BaseCore, Generic[L]):
             # which unhidden rows the caller may actually see.
             with skip_visibility_filter(db.session, deleted_state_bound.model):
                 items, total_count = self._call_dao_list(
-                    custom_filters={"deleted_state": deleted_state_bound},
+                    custom_filters=dao_custom_filters,
                     **dao_kwargs,
                 )
         else:
-            items, total_count = self._call_dao_list(**dao_kwargs)
+            items, total_count = self._call_dao_list(
+                custom_filters=dao_custom_filters or None, **dao_kwargs
+            )
         # Serialize items
         item_objs = []
         for item in items:

superset/security/manager.py

@@ -3971,6 +3971,21 @@ class SupersetSecurityManager(  # pylint: disable=too-many-public-methods
             if query in self.session:
                 self.session.expunge(query)
 
+        # When only ``database`` is provided, enforce database-level access
+        # here so the call is not a no-op.
+        if database and not (table or query):
+            if not self.can_access_database(database):
+                raise SupersetSecurityException(
+                    SupersetError(
+                        error_type=SupersetErrorType.DATABASE_SECURITY_ACCESS_ERROR,
+                        message=_(
+                            "You need access to the following database: %(name)s",
+                            name=database.database_name,
+                        ),
+                        level=ErrorLevel.WARNING,
+                    )
+                )
+
         if database and table or query:
             if query:
                 # Type narrow: only SQL Lab Query objects have .database attribute
@@ -4053,6 +4068,24 @@ class SupersetSecurityManager(  # pylint: disable=too-many-public-methods
                             level=ErrorLevel.ERROR,
                         )
                     )
+                # Statements that rebind how unqualified table names resolve
+                # (``USE``, ``SET SCHEMA``, or a ``search_path`` change) make
+                # the qualification below diverge from what the engine uses at
+                # execution time, so reject them regardless of engine.
+                if force_dataset_match and parse_result.script.changes_default_schema():
+                    raise SupersetSecurityException(
+                        SupersetError(
+                            error_type=SupersetErrorType.QUERY_SECURITY_ACCESS_ERROR,
+                            message=_(
+                                "SQL Lab cannot authorise a script that "
+                                "changes the schema used to resolve "
+                                "unqualified table names (e.g. USE or "
+                                "search_path changes). Qualify tables "
+                                "explicitly instead."
+                            ),
+                            level=ErrorLevel.ERROR,
+                        )
+                    )
                 tables = {
                     table_.qualify(
                         catalog=query.catalog or default_catalog,

superset/sql/parse.py

@@ -275,6 +275,9 @@ class RLSTransformer:
 
         return None
 
+    def __call__(self, node: exp.Table) -> exp.Expression:
+        raise NotImplementedError()
+
 
 class RLSAsPredicateTransformer(RLSTransformer):
     """
@@ -298,17 +301,17 @@ class RLSAsPredicateTransformer(RLSTransformer):
     databases without support for subqueries.
     """
 
-    def __call__(self, node: exp.Expression) -> exp.Expression:
-        if not isinstance(node, exp.Table):
-            return node
-
+    def __call__(self, node: exp.Table) -> exp.Expression:
         predicate = self.get_predicate(node)
         if not predicate:
             return node
 
-        # qualify columns with table name
+        # Qualify with the parsed alias node, not the ``node.alias`` string (which drops
+        # quoting and could inject SQL); use the table when the alias has no name.
+        table_alias = node.args.get("alias")
+        qualifier = (table_alias and table_alias.this) or node.this
         for column in predicate.find_all(exp.Column):
-            column.set("table", node.alias or node.this)
+            column.set("table", qualifier.copy())
 
         if isinstance(node.parent, exp.From):
             select = node.parent.parent
@@ -354,13 +357,12 @@ class RLSAsSubqueryTransformer(RLSTransformer):
     all databases.
     """
 
-    def __call__(self, node: exp.Expression) -> exp.Expression:
-        if not isinstance(node, exp.Table):
-            return node
-
+    def __call__(self, node: exp.Table) -> exp.Expression:
         if predicate := self.get_predicate(node):
-            if node.alias:
-                alias = node.alias
+            if existing_alias := node.args.get("alias"):
+                # Reuse the parsed alias node, not the ``node.alias`` string: that drops
+                # quoting (SQL in an alias re-emits as SQL) and the column-alias list.
+                alias = existing_alias
             else:
                 # Use just the table name (not schema-qualified) so that
                 # column references like ``table.column`` still resolve after
@@ -616,6 +618,18 @@ class BaseSQLStatement(Generic[InternalRepresentation]):
         """
         return False
 
+    def changes_default_schema(self) -> bool:
+        """
+        Check if the statement changes the schema used to resolve unqualified
+        table names.
+
+        Defaults to ``False``; engines whose statements can rebind unqualified
+        schema resolution override this.
+
+        :return: True if the statement rebinds default schema resolution
+        """
+        return False
+
     def get_disallowed_tables(
         self,
         tables: set[str],
@@ -751,26 +765,24 @@ class SQLStatement(BaseSQLStatement[exp.Expression]):
         }
     )
 
-    # PostgreSQL constructs that sqlglot represents as an opaque ``exp.Command``
-    # (no structured AST). Each can mutate server state or wrap a DML body that
-    # would otherwise be detected by node-type matching. Used by
-    # ``is_mutating()``.
-    _POSTGRES_MUTATING_COMMAND_NAMES: frozenset[str] = frozenset(
+    # Constructs that sqlglot represents as an opaque ``exp.Command`` (no
+    # structured AST). Each can mutate server state or wrap a DML body that
+    # would otherwise be detected by node-type matching. The head keywords
+    # are not engine-specific (MySQL ``CALL`` / ``LOAD DATA INFILE`` and
+    # MSSQL ``EXEC`` reach the same ``exp.Command`` fallback as their
+    # PostgreSQL counterparts), so ``is_mutating()`` applies this list for
+    # every dialect: an opaque command with one of these heads is treated as
+    # mutating.
+    _MUTATING_COMMAND_NAMES: frozenset[str] = frozenset(
         {
             "DO",  # PL/pgSQL anonymous block
             "PREPARE",  # PREPARE u AS UPDATE ... ; EXECUTE u
             "EXECUTE",  # body is the prepared DML
+            "EXEC",  # MSSQL spelling of EXECUTE; the procedure body may mutate
             "CALL",  # procedure body may mutate
             "COPY",  # server-side file ingest into a table
             "GRANT",
             "REVOKE",
-            # Only the command-fallback forms (e.g. SET ROLE / SET SESSION
-            # AUTHORIZATION, which change the effective user) reach here as an
-            # exp.Command. Structured `SET search_path = ...` /
-            # `SET statement_timeout = ...` parse as exp.Set and are NOT matched
-            # by this command-name path.
-            "SET",
-            "RESET",  # RESET ROLE / RESET ALL reverts SET; same class as SET
             "REFRESH",  # REFRESH MATERIALIZED VIEW
             "REINDEX",
             "VACUUM",
@@ -783,7 +795,9 @@ class SQLStatement(BaseSQLStatement[exp.Expression]):
             "CREATE",
             "ALTER",
             "DROP",
-            "LOAD",  # LOAD '/path/lib.so' dlopens a shared library on the PG host
+            # MySQL LOAD DATA INFILE ingests server files into a table;
+            # PostgreSQL LOAD '/path/lib.so' dlopens a shared library.
+            "LOAD",
             # NOTE: `SHOW` is intentionally NOT included. It is a read (mutates
             # nothing), so classifying it as mutating would be wrong for every
             # is_mutating()/has_mutation() consumer (the commit decision, the
@@ -794,6 +808,20 @@ class SQLStatement(BaseSQLStatement[exp.Expression]):
         }
     )
 
+    # PostgreSQL-only command-fallback heads. Only the command-fallback
+    # forms (e.g. SET ROLE / SET SESSION AUTHORIZATION, which change the
+    # effective user) reach here as an exp.Command; structured
+    # `SET search_path = ...` / `SET statement_timeout = ...` parse as
+    # exp.Set and are NOT matched by this path. On other dialects the `SET`
+    # fallback covers session variables (e.g. Hive `SET hivevar:x=1`),
+    # which do not mutate data, so these heads stay dialect-gated.
+    _POSTGRES_MUTATING_COMMAND_NAMES: frozenset[str] = frozenset(
+        {
+            "SET",
+            "RESET",  # RESET ROLE / RESET ALL reverts SET; same class as SET
+        }
+    )
+
     # Dialects where `SELECT ... INTO target` is CTAS (creates a table, and so
     # mutates schema). Elsewhere the same syntax assigns into a variable and is
     # a read: Oracle PL/SQL `SELECT ... INTO v` and MySQL `SELECT ... INTO @v`
@@ -926,7 +954,7 @@ class SQLStatement(BaseSQLStatement[exp.Expression]):
         """
         return isinstance(self._parsed, exp.Select)
 
-    def is_mutating(self) -> bool:
+    def is_mutating(self) -> bool:  # noqa: C901
         """
         Check if the statement mutates data (DDL/DML).
 
@@ -949,6 +977,14 @@ class SQLStatement(BaseSQLStatement[exp.Expression]):
             exp.Revoke,
             # COMMENT ON TABLE/COLUMN/etc. writes to system catalog pg_description.
             exp.Comment,
+            # A bare COMMIT persists earlier writes on the same connection, so
+            # treat it as mutating.
+            exp.Commit,
+            # EXEC/EXECUTE invokes a stored procedure whose body is opaque;
+            # some dialects (e.g. MSSQL) parse it as this structured node
+            # rather than an opaque exp.Command, so treat it as mutating here
+            # too.
+            exp.Execute,
         )
 
         if self._parsed.find(*mutating_nodes):
@@ -986,37 +1022,76 @@ class SQLStatement(BaseSQLStatement[exp.Expression]):
         ):
             return True
 
-        # depending on the dialect (Oracle, MS SQL) the `ALTER` is parsed as a
-        # command, not an expression - check at root level
-        if isinstance(self._parsed, exp.Command) and self._parsed.name == "ALTER":
-            return True  # pragma: no cover
+        # Statements that sqlglot cannot model parse as an opaque
+        # `exp.Command`. The `.name` attribute on `exp.Command` preserves
+        # the source-case of the head keyword (so `create extension ...`
+        # would yield `'create'`), which means the lookups must be
+        # case-insensitive. This also covers the dialects (Oracle, MS SQL)
+        # where `ALTER` itself is parsed as a command, not an expression.
+        if isinstance(self._parsed, exp.Command):
+            command_name = self._parsed.name.upper()
 
-        # PostgreSQL constructs that sqlglot represents as an opaque
-        # `exp.Command` rather than a structured AST. Each of these can mutate
-        # state or wrap a DML body that would otherwise be detected. The
-        # `.name` attribute on `exp.Command` preserves the source-case of the
-        # head keyword (so `create extension ...` would yield `'create'`),
-        # which means the set lookup must be case-insensitive.
-        if (
-            self._dialect == Dialects.POSTGRES
-            and isinstance(self._parsed, exp.Command)
-            and self._parsed.name.upper() in self._POSTGRES_MUTATING_COMMAND_NAMES
-        ):
-            return True
+            if command_name in self._MUTATING_COMMAND_NAMES:
+                return True
 
-        # Postgres runs DMLs prefixed by `EXPLAIN ANALYZE`, see
-        # https://www.postgresql.org/docs/current/sql-explain.html
-        if (
-            self._dialect == Dialects.POSTGRES
-            and isinstance(self._parsed, exp.Command)
-            and self._parsed.name == "EXPLAIN"
-            and self._parsed.expression.name.upper().startswith("ANALYZE ")
-        ):
-            analyzed_sql = self._parsed.expression.name[len("ANALYZE ") :]
-            return SQLStatement(
-                statement=analyzed_sql,
-                engine=self.engine,
-            ).is_mutating()
+            if (
+                self._dialect == Dialects.POSTGRES
+                and command_name in self._POSTGRES_MUTATING_COMMAND_NAMES
+            ):
+                return True
+
+            # `EXPLAIN ANALYZE <statement>` executes the statement for real
+            # (PostgreSQL and MySQL both run the body), see
+            # https://www.postgresql.org/docs/current/sql-explain.html
+            # The flag may be spelled `ANALYSE`, be separated by any
+            # whitespace, or appear in a parenthesized option list such as
+            # `EXPLAIN (ANALYZE, BUFFERS) ...`, so the raw tail is
+            # normalized before the inner statement is classified. Anything
+            # that carries the flag but cannot be classified is treated as
+            # mutating.
+            if command_name == "EXPLAIN":
+                tail = (
+                    self._parsed.expression.name.strip()
+                    if self._parsed.expression
+                    else ""
+                )
+
+                # sqlglot preserves the raw tail text, comments included;
+                # strip leading comments so an option list hidden behind
+                # `/* ... */` or `-- ...` is still recognized.
+                while True:
+                    if tail.startswith("/*") and "*/" in tail:
+                        tail = tail.split("*/", 1)[1].lstrip()
+                    elif tail.startswith("--"):
+                        parts = tail.split("\n", 1)
+                        tail = parts[1].lstrip() if len(parts) > 1 else ""
+                    else:
+                        break
+
+                has_analyze = False
+                if tail.startswith("("):
+                    options, _, tail = tail[1:].partition(")")
+                    has_analyze = bool(
+                        re.search(r"\b(ANALYZE|ANALYSE)\b", options, re.IGNORECASE)
+                    )
+                else:
+                    while match := re.match(
+                        r"(ANALYZE|ANALYSE|VERBOSE)\s+", tail, re.IGNORECASE
+                    ):
+                        if match.group(1).upper() != "VERBOSE":
+                            has_analyze = True
+                        tail = tail[match.end() :]
+
+                if has_analyze:
+                    if not (inner_sql := tail.strip()):
+                        return True
+                    try:
+                        return SQLStatement(
+                            statement=inner_sql,
+                            engine=self.engine,
+                        ).is_mutating()
+                    except SupersetParseError:
+                        return True
 
         return False
 
@@ -1188,6 +1263,58 @@ class SQLStatement(BaseSQLStatement[exp.Expression]):
             return bool(tokens) and tokens[0].strip('"').lower() == "search_path"
         return False
 
+    def changes_default_schema(self) -> bool:
+        """
+        Return True if the statement rebinds default schema resolution.
+
+        Covers ``USE`` statements (MySQL-, Doris- and Snowflake-family
+        engines) and ``SET [CURRENT] SCHEMA`` / ``SET CATALOG`` variants, in
+        addition to anything that changes the Postgres ``search_path``.
+        Unqualified table names in later statements on the same cursor then
+        resolve against a different schema.
+        """
+        for use in self._parsed.find_all(exp.Use):
+            kind = use.args.get("kind")
+            # `USE WAREHOUSE ...` selects compute, not a namespace, and does
+            # not affect how table names resolve.
+            if kind and kind.name.upper() == "WAREHOUSE":
+                continue
+            return True
+        # `SET SCHEMA 'x'` / `SET CATALOG 'x'` rebind resolution through a
+        # structured setting rather than a search path.
+        rebinding_settings = {
+            "schema",
+            "current_schema",
+            "current schema",
+            "catalog",
+        }
+        if any(
+            key.strip('"').lower() in rebinding_settings for key in self.get_settings()
+        ):
+            return True
+        # A `set_config()` with a non-literal setting name may set
+        # `search_path` at runtime, so treat it as a schema change; literal
+        # names are handled by `changes_search_path`.
+        for func in self._parsed.find_all(exp.Anonymous):
+            if func.name.lower() == "set_config" and not (
+                func.expressions and isinstance(func.expressions[0], exp.Literal)
+            ):
+                return True
+        # `SET SCHEMA` / `SET CATALOG` forms that fall back to an opaque
+        # exp.Command: match the leading setting name, mirroring
+        # `changes_search_path`.
+        parsed = self._parsed
+        if isinstance(parsed, exp.Command) and parsed.name.upper() == "SET":
+            tokens = str(parsed.expression).replace("=", " ").split()
+            while tokens and tokens[0].upper() in {"SESSION", "LOCAL", "CURRENT"}:
+                tokens.pop(0)
+            if tokens and tokens[0].strip('"').strip("'").lower() in {
+                "schema",
+                "catalog",
+            }:
+                return True
+        return self.changes_search_path()
+
     def get_disallowed_tables(
         self,
         tables: set[str],
@@ -1282,6 +1409,15 @@ class SQLStatement(BaseSQLStatement[exp.Expression]):
         Modify the `LIMIT` or `TOP` value of the SQL statement inplace.
         """
         if method == LimitMethod.FORCE_LIMIT:
+            # `SHOW` statements (`SHOW TABLES`, `SHOW DATABASES`, `SHOW CREATE
+            # TABLE`, etc.) have no meaningful `LIMIT` slot to force. On
+            # MySQL/StarRocks, writing one renders a malformed statement with
+            # two `LIMIT` keywords that the engine rejects outright; on dialects
+            # like Snowflake it would render a valid `SHOW ... LIMIT`, but SHOW
+            # returns bounded metadata, so we skip it uniformly rather than
+            # special-case per dialect. Leave them untouched.
+            if isinstance(self._parsed, exp.Show):
+                return
             self._parsed.args["limit"] = exp.Limit(
                 expression=exp.Literal(this=str(limit), is_string=False)
             )
@@ -1408,7 +1544,30 @@ class SQLStatement(BaseSQLStatement[exp.Expression]):
             raise ValueError(f"Invalid RLS method: {method}")
 
         transformer = transformers[method](catalog, schema, predicates)
-        self._parsed = self._parsed.transform(transformer)
+
+        # Rewrite the real table reads -- the same set ``extract_tables_from_statement``
+        # authorizes -- so the filtered set equals the authorized set. (A CTE reference
+        # sharing a rule's table name is not a read here.)
+        seen: set[int] = set()
+        reads: list[exp.Table] = []
+        for scope in traverse_scope(self._parsed):
+            for source in scope.sources.values():
+                # dedupe by identity: a correlated LATERAL reaches one node twice
+                if (
+                    isinstance(source, exp.Table)
+                    and not is_cte(source, scope)
+                    and id(source) not in seen
+                ):
+                    seen.add(id(source))
+                    reads.append(source)
+
+        # Wrap the deepest reads first: a parenthesised-join head carries its join in
+        # its args, so wrapping an ancestor before its descendant would strand the
+        # descendant read's replacement off the live tree.
+        for node in sorted(reads, key=lambda read: read.depth, reverse=True):
+            replacement = transformer(node)
+            if replacement is not node:
+                node.replace(replacement)
 
 
 class KQLSplitState(enum.Enum):
@@ -1819,12 +1978,16 @@ class SQLScript:
     def has_unparseable_statement(self) -> bool:
         """
         True if any statement in the script cannot be fully modeled as an
-        AST whose table references Superset can enumerate. This covers two
-        cases that must both fail closed under strict scoping:
+        AST whose table references Superset can enumerate. This covers the
+        following cases, which must all fail closed under strict scoping:
 
         * SQLGlot ``exp.Command`` nodes: statements sqlglot recognises but
           cannot fully parse (e.g. dynamic SQL inside a stored-procedure
           call); ``extract_tables_from_statement`` cannot see the tables.
+        * ``exp.Show`` statements with no extractable target (e.g.
+          ``SHOW TABLES FROM some_schema``): the statement reads database
+          metadata, but there is no table reference for the per-table check
+          to enforce against.
         * Non-sqlglot engines (e.g. Kusto KQL): the statement class does
           not produce a sqlglot AST at all and its
           ``_extract_tables_from_statement`` returns an empty set, so the
@@ -1835,6 +1998,11 @@ class SQLScript:
                 return True
             if isinstance(statement._parsed, exp.Command):  # noqa: SLF001
                 return True
+            if (
+                isinstance(statement._parsed, exp.Show)  # noqa: SLF001
+                and not statement.tables
+            ):
+                return True
         return False
 
     def get_settings(self) -> dict[str, str | bool]:
@@ -1868,6 +2036,16 @@ class SQLScript:
         """
         return any(statement.is_destructive() for statement in self.statements)
 
+    def changes_default_schema(self) -> bool:
+        """
+        Check if any statement rebinds default schema resolution.
+
+        :return: True if any statement changes the schema (``USE``,
+            ``SET SCHEMA``) or the Postgres ``search_path`` used to resolve
+            unqualified table names
+        """
+        return any(statement.changes_default_schema() for statement in self.statements)
+
     def optimize(self) -> SQLScript:
         """
         Return optimized script.
@@ -1986,6 +2164,31 @@ def extract_tables_from_statement(
         except (ParseError, SupersetParseError):
             return set()
         sources = pseudo_query.find_all(exp.Table)
+    elif isinstance(statement, exp.Show):
+        # Structured metadata statements (`SHOW CREATE TABLE foo.bar`,
+        # `SHOW COLUMNS FROM foo`, ...) reference their target via dedicated
+        # args rather than query sources, so build the table references
+        # explicitly. Statements with no extractable target (e.g.
+        # `SHOW TABLES FROM some_schema`) yield an empty set and are treated
+        # as unparseable for authorization purposes (see
+        # `SQLScript.has_unparseable_statement`).
+        show_tables = {
+            Table(
+                source.name,
+                source.db if source.db != "" else None,
+                source.catalog if source.catalog != "" else None,
+            )
+            for source in statement.find_all(exp.Table)
+        }
+        if target := statement.args.get("target"):
+            db = statement.args.get("db")
+            show_tables.add(
+                Table(
+                    target.name if isinstance(target, exp.Expression) else str(target),
+                    db.name if isinstance(db, exp.Expression) else db,
+                )
+            )
+        return show_tables
     else:
         sources = [
             source
@@ -2006,41 +2209,22 @@ def extract_tables_from_statement(
 
 def is_cte(source: exp.Table, scope: Scope) -> bool:
     """
-    Is the source a CTE?
+    Does this reference resolve to a CTE rather than to a real table?
 
-    CTEs in the parent scope look like tables (and are represented by
-    exp.Table objects), but should not be considered as such;
-    otherwise a user with access to table `foo` could access any table
-    with a query like this:
-
-        WITH foo AS (SELECT * FROM target_table) SELECT * FROM foo
-
-    A CTE name is always a bare identifier: it can never carry a schema or
-    catalog qualifier. A schema/catalog-qualified reference therefore always
-    resolves to a physical table, even when its final name component happens to
-    match a CTE defined in scope. Such a reference must be reported as a real
-    table so it resolves to the correct object; otherwise
-    ``WITH orders AS (...) SELECT * FROM public.orders`` would treat the
-    qualified ``public.orders`` as the CTE and drop the physical table from the
-    extracted set.
-
-    Note: an unqualified reference is always resolved relative to the caller's
-    own schema/catalog before any downstream use, so treating a bare name that
-    matches a CTE as a CTE stays correct and is intentionally left unchanged
-    here.
+    A CTE reference is also an ``exp.Table``, so it must be excluded from a statement's
+    read tables, or a rule on a table could be evaded by wrapping it in a same-named
+    CTE. Resolve the name through ``Scope.cte_sources`` (not ``Scope.sources``, keyed by
+    ``alias_or_name``, which would hide a real table sharing a CTE's alias); a qualified
+    reference (schema or catalog) is always a table. Where sqlglot registers a name
+    differently than SQL scopes it (letter-case, a ``WITH RECURSIVE`` self/forward
+    reference), this errs toward reporting a table -- a spurious check, not a leak.
     """
     if source.db or source.catalog:
         # Qualified references are always physical tables, never CTEs.
         return False
 
-    parent_sources = scope.parent.sources if scope.parent else {}
-    ctes_in_scope = {
-        name
-        for name, parent_scope in parent_sources.items()
-        if isinstance(parent_scope, Scope) and parent_scope.scope_type == ScopeType.CTE
-    }
-
-    return source.name in ctes_in_scope
+    resolved = scope.cte_sources.get(source.name)
+    return isinstance(resolved, Scope) and resolved.scope_type == ScopeType.CTE
 
 
 T = TypeVar("T", str, None)
@@ -2071,6 +2255,17 @@ def remove_quotes(val: T) -> T:
     return val
 
 
+# Jinja macros that execute statements against the analytical database when
+# rendered; their table references are extracted before rendering, and the
+# macros are stubbed out during a validation-time render.
+PARTITION_MACRO_NAMES = (
+    "first_latest_partition",
+    "latest_partition",
+    "latest_partitions",
+    "latest_sub_partition",
+)
+
+
 def process_jinja_sql(
     sql: str, database: Database, template_params: Optional[dict[str, Any]] = None
 ) -> JinjaSQLResult:
@@ -2091,10 +2286,13 @@ def process_jinja_sql(
     :returns: JinjaSQLResult containing the processed script and table references
     :raises SupersetSecurityException: If SQLGlot is unable to parse the SQL statement
     :raises jinja2.exceptions.TemplateError: If the Jinjafied SQL could not be rendered
+    :raises SupersetParseError: If a partition macro references a table that
+        cannot be determined statically
     """
 
     from superset.jinja_context import (  # pylint: disable=import-outside-toplevel
         get_template_processor,
+        NoOpTemplateProcessor,
     )
 
     processor = get_template_processor(database)
@@ -2102,37 +2300,74 @@ def process_jinja_sql(
 
     tables = set()
 
+    def raise_for_unresolvable_macro() -> Any:
+        raise SupersetParseError(
+            sql,
+            database.db_engine_spec.engine,
+            message=(
+                "Unable to determine the table referenced by a partition "
+                "macro; use a single constant table reference"
+            ),
+        )
+
     for node in ast.find_all(nodes.Call):
-        if isinstance(node.node, nodes.Getattr) and node.node.attr in (
-            "latest_partition",
-            "latest_sub_partition",
+        if (
+            isinstance(node.node, nodes.Getattr)
+            and node.node.attr in PARTITION_MACRO_NAMES
         ):
-            # Try to extract the table referenced in the macro.
+            # Extract the table referenced in the macro. The reference must
+            # be statically evaluable; otherwise raise rather than render.
             try:
+                if len(node.args) != 1:
+                    raise nodes.Impossible()
                 tables.add(
                     Table(
                         *[
                             remove_quotes(part.strip())
                             for part in node.args[0].as_const().split(".")[::-1]
-                            if len(node.args) == 1
                         ]
                     )
                 )
             except nodes.Impossible:
-                pass
+                raise_for_unresolvable_macro()
 
             # Replace the potentially problematic Jinja macro with some benign SQL.
             node.__class__ = nodes.TemplateData
             node.fields = nodes.TemplateData.fields
             node.data = "NULL"
 
-    # re-render template back into a string
-    code = processor.env.compile(ast)
-    template = Template.from_code(processor.env, code, globals=processor.env.globals)
-    rendered_sql = template.render(processor.get_context(), **(template_params or {}))
+    # Render the neutralized template once, using the same context
+    # ``process_template`` builds at execution time, so the validated SQL
+    # matches the executed SQL. A no-op processor runs the raw SQL at
+    # execution time, so validate that raw SQL directly.
+    if isinstance(processor, NoOpTemplateProcessor):
+        rendered_sql = processor.process_template(sql)
+    else:
+        code = processor.env.compile(ast)
+        template = Template.from_code(
+            processor.env,
+            code,
+            globals=processor.env.globals,
+        )
+        # Replace live partition macros with stubs so a call that survives
+        # neutralization (e.g. via a dynamic attribute lookup) does not
+        # execute during this render.
+        context = processor.get_template_context(**(template_params or {}))
+        if (engine := getattr(processor, "engine", None)) and isinstance(
+            context.get(engine), dict
+        ):
+            context[engine] = {
+                key: (
+                    (lambda *args, **kwargs: raise_for_unresolvable_macro())
+                    if key in PARTITION_MACRO_NAMES
+                    else value
+                )
+                for key, value in context[engine].items()
+            }
+        rendered_sql = template.render(context)
 
     parsed_script = SQLScript(
-        processor.process_template(rendered_sql),
+        rendered_sql,
         engine=database.db_engine_spec.engine,
     )
     for parsed_statement in parsed_script.statements:

superset/sqllab/api.py

@@ -258,6 +258,14 @@ class SqlLabRestApi(BaseSupersetApi):
                                 else template_params
                             )
                             if template_params:
+                                # Check access before rendering the Jinja
+                                # template (mirrors the SQL Lab execute path).
+                                security_manager.raise_for_access(
+                                    database=database,
+                                    sql=sql,
+                                    template_params=template_params,
+                                    force_dataset_match=True,
+                                )
                                 template_processor = get_template_processor(
                                     database=database
                                 )

superset/sqllab/sqllab_execution_context.py

@@ -22,9 +22,12 @@ from dataclasses import dataclass
 from typing import Any, cast, TYPE_CHECKING
 
 from flask import g
+from flask_babel import gettext as __
 from sqlalchemy.orm.exc import DetachedInstanceError
 
 from superset import is_feature_enabled
+from superset.errors import ErrorLevel, SupersetError, SupersetErrorType
+from superset.exceptions import SupersetErrorException
 from superset.models.sql_lab import Query
 from superset.sql.parse import CTASMethod
 from superset.utils import core as utils, json
@@ -128,9 +131,45 @@ class SqlJsonExecutionContext:  # pylint: disable=too-many-instance-attributes
         if self.catalog is None:
             self.catalog = database.get_default_catalog()
         if self.select_as_cta:
+            self._validate_ctas_is_allowed(database)
             schema_name = self._get_ctas_target_schema_name(database)
             self.create_table_as_select.target_schema_name = schema_name  # type: ignore
 
+    def _validate_ctas_is_allowed(self, database: Database) -> None:
+        """
+        Enforce the per-database CTAS/CVAS grants server-side.
+
+        The database's ``allow_ctas``/``allow_cvas`` flags are checked at
+        submission, mirroring the ``allow_dml`` gate on the execution path.
+        """
+        ctas = cast(CreateTableAsSelect, self.create_table_as_select)
+        if ctas.ctas_method == CTASMethod.TABLE and not database.allow_ctas:
+            raise SupersetErrorException(
+                SupersetError(
+                    message=__(
+                        "This database does not allow creating tables from "
+                        "queries (CTAS). Please contact your administrator "
+                        "for more assistance."
+                    ),
+                    error_type=SupersetErrorType.QUERY_SECURITY_ACCESS_ERROR,
+                    level=ErrorLevel.ERROR,
+                ),
+                status=403,
+            )
+        if ctas.ctas_method == CTASMethod.VIEW and not database.allow_cvas:
+            raise SupersetErrorException(
+                SupersetError(
+                    message=__(
+                        "This database does not allow creating views from "
+                        "queries (CVAS). Please contact your administrator "
+                        "for more assistance."
+                    ),
+                    error_type=SupersetErrorType.QUERY_SECURITY_ACCESS_ERROR,
+                    level=ErrorLevel.ERROR,
+                ),
+                status=403,
+            )
+
     def _get_ctas_target_schema_name(self, database: Database) -> str | None:
         if database.force_ctas_schema:
             return database.force_ctas_schema

superset/utils/number_format.py

@@ -0,0 +1,607 @@
+# 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.
+"""
+Server-side port of the d3-format based number and currency formatters used by
+the Table and Pivot Table chart plugins.
+
+Report notifications that embed a chart as text build the table in Python and
+have no access to the frontend formatters, so chart number/currency format
+configuration has to be reproduced here to render the same values an end user
+sees in the browser.
+
+Only d3-format specifiers (and the ``SMART_NUMBER`` pseudo-formats) are ported.
+The duration, memory, and length formatters depend on separate frontend
+factories and are explicitly rejected, causing the public wrapper to preserve
+the raw value rather than silently misformat it. The fill/align/zero/width d3
+flags are likewise rejected because report text has no equivalent of the
+frontend's padding behavior. Accounting-parenthesis and space-sign modes are
+supported.
+"""
+
+from __future__ import annotations
+
+import math
+import re
+from decimal import Decimal, ROUND_HALF_UP
+from functools import lru_cache
+from typing import Any, Iterable
+
+from babel.numbers import format_currency, get_currency_symbol
+from flask import current_app
+from flask_babel import get_locale
+
+SMART_NUMBER: str = "SMART_NUMBER"
+SMART_NUMBER_SIGNED: str = "SMART_NUMBER_SIGNED"
+AUTO_CURRENCY: str = "AUTO"
+
+UNSUPPORTED_FRONTEND_PRESETS: frozenset[str] = frozenset(
+    {
+        "DURATION",
+        "DURATION_SUB",
+        "DURATION_COL",
+        "MEMORY_DECIMAL",
+        "MEMORY_BINARY",
+        "MEMORY_TRANSFER_RATE_DECIMAL",
+        "MEMORY_TRANSFER_RATE_BINARY",
+        "LENGTH",
+        "LENGTH_CM_KM",
+        "LENGTH_CM_M",
+    }
+)
+
+DEFAULT_LOCALE: str = "en"
+CURRENCY_SYMBOL_LOCALE: str = "en_US"
+
+# SI prefixes keyed by their power-of-1000 exponent, mirroring d3-format.
+SI_PREFIXES: dict[int, str] = {
+    -8: "y",
+    -7: "z",
+    -6: "a",
+    -5: "f",
+    -4: "p",
+    -3: "n",
+    -2: "µ",
+    -1: "m",
+    0: "",
+    1: "k",
+    2: "M",
+    3: "G",
+    4: "T",
+    5: "P",
+    6: "E",
+    7: "Z",
+    8: "Y",
+}
+
+# d3-format specifier grammar:
+# [[fill]align][sign][symbol][0][width][,][.precision][~][type]
+D3_FORMAT_RE: re.Pattern[str] = re.compile(
+    r"^(?:(.)?([<>=^]))?([+\-( ])?([$#])?(0)?(\d+)?(,)?(?:\.(\d+))?(~)?([a-z%])?$",
+    re.IGNORECASE,
+)
+
+
+def resolve_auto_currency(
+    currency: dict[str, Any],
+    detected_currency: str | None,
+    currency_context: Iterable[Any] | float | None = None,
+    fallback_to_detected: bool = True,
+) -> dict[str, Any]:
+    """
+    Resolve an ``AUTO`` currency to the code detected from the data.
+
+    Mirrors ``currency-format/utils.ts::resolveAutoCurrency`` and the per-cell
+    handling in the Table and Pivot Table plugins. A single valid currency in
+    ``currency_context`` takes precedence over the query-wide detection. Mixed
+    cell currencies deliberately keep ``AUTO`` so the caller renders a neutral
+    number. Empty cell context can use the detected fallback when the plugin's
+    behavior allows it.
+
+    :param currency_context: the currencies contributing to a cell. A dense
+        pivot cell provides an iterable of codes, but a sparse 2D pivot passes a
+        scalar ``NaN`` float for a missing cross-product cell (hence the ``float``
+        arm); a missing/NaN/non-iterable context is treated as empty.
+    :return: a copied config containing the detected code, or the input config
+    """
+    if currency.get("symbol") != AUTO_CURRENCY:
+        return currency
+
+    if currency_context is not None:
+        # A dense pivot cell carries an iterable of currency codes, but a sparse
+        # 2D pivot leaves missing cross-product cells as a scalar missing value
+        # (``np.nan``; pandas never runs the union aggregator for them). Test
+        # positively for an iterable so any non-iterable sentinel (``np.nan``,
+        # ``pd.NA``, ``pd.NaT``) falls to the empty-context path instead of
+        # raising and taking down the whole report.
+        context_values: list[Any] = (
+            list(currency_context) if isinstance(currency_context, Iterable) else []
+        )
+        normalized_currencies = {
+            normalized
+            for value in context_values
+            if (normalized := normalize_currency(value)) is not None
+        }
+        if len(normalized_currencies) > 1:
+            return currency
+        if context_values and (cell_currency := normalize_currency(context_values[0])):
+            return {**currency, "symbol": cell_currency}
+        if not fallback_to_detected:
+            return currency
+
+    if detected_currency := normalize_currency(detected_currency):
+        return {**currency, "symbol": detected_currency}
+    return currency
+
+
+def normalize_currency(value: Any) -> str | None:
+    """
+    Normalize a possible ISO-4217 code for AUTO currency resolution.
+
+    Mirrors ``currency-format/CurrencyFormatter.ts::normalizeCurrency``:
+    non-strings and values other than three ASCII letters are rejected, while
+    valid strings are stripped and upper-cased.
+
+    :return: the normalized three-letter code, or ``None``
+    """
+    if not isinstance(value, str):
+        return None
+    normalized = value.strip().upper()
+    return normalized if re.fullmatch(r"[A-Z]{3}", normalized) else None
+
+
+def format_number_with_config(
+    d3_format: str | None,
+    currency: dict[str, Any] | None,
+    value: Any,
+) -> Any:
+    """
+    Format ``value`` using a d3-format string and optional currency config.
+
+    This is the report-side entry point corresponding to
+    ``currency-format/CurrencyFormatter.ts::format`` and the formatter invoked
+    by the Table and Pivot Table plugins.
+
+    :param d3_format: a d3-format specifier (e.g. ``",.2f"``) or ``SMART_NUMBER``
+    :param currency: ``{"symbol": <ISO 4217>, "symbolPosition": "prefix"|"suffix"}``
+    :param value: the raw value to format
+    :return: the formatted string, or the value unchanged when it is not a
+        number that can be formatted
+    """
+    if value is None:
+        return ""
+    if isinstance(value, bool) or not isinstance(value, (int, float, Decimal)):
+        return value
+    if isinstance(value, Decimal):
+        value = float(value)
+    if math.isnan(value) or math.isinf(value):
+        return ""
+
+    try:
+        if currency and currency.get("symbol"):
+            # the frontend strips the currency symbol from the d3 format and
+            # falls back to SMART_NUMBER when no explicit format is set
+            number_format = (d3_format or SMART_NUMBER).replace("$", "")
+            formatted = format_numeric(number_format, value)
+            if currency["symbol"] == AUTO_CURRENCY:
+                return formatted
+            try:
+                return apply_currency(formatted, currency)
+            except Exception:  # pylint: disable=broad-except  # noqa: BLE001
+                return formatted
+        if not d3_format:
+            return raw_string(value)
+        return format_numeric(d3_format, value)
+    except Exception:  # pylint: disable=broad-except  # noqa: BLE001
+        # never let an unexpected value break a whole report table
+        return raw_string(value)
+
+
+def format_numeric(d3_format: str, value: float) -> str:
+    """
+    Format ``value`` according to a d3 number format.
+
+    Delegates to the port of ``createSmartNumberFormatter.ts`` for the two smart
+    pseudo-formats and to the port of ``d3-format/src/locale.js`` for d3
+    specifiers. Registered frontend-only factories are rejected explicitly.
+
+    :return: a formatted number string
+    """
+    if d3_format in UNSUPPORTED_FRONTEND_PRESETS:
+        raise ValueError(f"Frontend preset {d3_format!r} is not available in reports")
+    if d3_format in (SMART_NUMBER, SMART_NUMBER_SIGNED):
+        return format_smart_number(value, signed=d3_format == SMART_NUMBER_SIGNED)
+    return format_d3(d3_format, value)
+
+
+def format_d3(d3_format: str, value: float) -> str:
+    """
+    Format ``value`` with a d3-format specifier.
+
+    Mirrors ``d3-format/src/locale.js`` and ``formatTypes.js``. Supports the
+    subset of the specifier grammar the Table/Pivot plugins emit:
+    the ``+ - ( space`` sign modes, the ``$`` currency prefix, the ``,`` group
+    separator, ``.precision``, the ``~`` trim flag, and the ``s`` (SI), ``r``
+    (significant), ``d`` (integer), ``f``/``e``/``g``/``%`` numeric types.
+    Returns the formatted string and raises ``ValueError`` for an unparseable
+    specifier. Padding flags are rejected because they cannot be represented by
+    the report table path.
+
+    :return: a d3-compatible formatted string
+    """
+    match = D3_FORMAT_RE.match(d3_format)
+    if not match:
+        raise ValueError(d3_format)
+    if any(match.group(index) for index in (1, 2, 5, 6)):
+        raise ValueError(f"d3 padding is not supported in reports: {d3_format!r}")
+
+    sign_mode = match.group(3) or "-"
+    currency_symbol = match.group(4) == "$"
+    comma = "," if match.group(7) else ""
+    precision = int(match.group(8)) if match.group(8) is not None else None
+    trim = bool(match.group(9))
+    type_ = (match.group(10) or "").lower()
+
+    if type_ == "n":
+        comma = ","
+        type_ = "g"
+
+    formatted = format_d3_magnitude(
+        type_, abs(value), precision, trim, comma, d3_format
+    )
+
+    if currency_symbol:
+        formatted = f"${formatted}"
+    return apply_sign(formatted, value, sign_mode)
+
+
+def format_d3_magnitude(
+    type_: str,
+    magnitude: float,
+    precision: int | None,
+    trim: bool,
+    comma: str,
+    d3_format: str,
+) -> str:
+    """
+    Render the unsigned numeric portion of a parsed d3 specifier.
+
+    Mirrors the formatter dispatch in ``d3-format/src/locale.js`` and
+    ``formatTypes.js``. The result excludes sign and currency decoration.
+    """
+    if type_ == "s":
+        return format_si(
+            magnitude, max(1, precision if precision is not None else 6), trim
+        )
+    if type_ == "r":
+        return format_significant(
+            magnitude,
+            max(1, precision if precision is not None else 6),
+            trim,
+            comma,
+        )
+    if type_ == "":
+        return format_general(
+            magnitude, precision if precision is not None else 12, True, comma
+        )
+    if type_ == "d":
+        formatted = format(int(quantize_half_up(magnitude, 0)), f"{comma}d")
+    elif type_ in ("f", "%"):
+        precision = precision if precision is not None else 6
+        scaled = magnitude * 100 if type_ == "%" else magnitude
+        suffix = "%" if type_ == "%" else ""
+        if scaled >= 1e21:
+            formatted = normalize_exponent(repr(float(scaled))) + suffix
+        else:
+            rounded = quantize_half_up(scaled, precision)
+            formatted = format(rounded, f"{comma}.{precision}f") + suffix
+    elif type_ == "e":
+        formatted = format_exponential(
+            magnitude, precision if precision is not None else 6
+        )
+    elif type_ == "g":
+        formatted = format_general(
+            magnitude, precision if precision is not None else 6, trim, comma
+        )
+    else:
+        raise ValueError(d3_format)
+    return trim_trailing_zeros(formatted) if trim else formatted
+
+
+def apply_sign(formatted: str, value: float, sign_mode: str) -> str:
+    """
+    Decorate a formatted magnitude with the d3 sign mode.
+
+    Negative values get a leading ``-`` (or wrapping parentheses for the ``(``
+    accounting mode); positive values get a ``+`` or a leading space only for the
+    ``+`` and space modes respectively. Mirrors the sign decoration in
+    ``d3-format/src/locale.js``.
+
+    :return: the signed or accounting-decorated string
+    """
+    if value < 0:
+        return f"({formatted})" if sign_mode == "(" else f"-{formatted}"
+    if sign_mode == "+":
+        return f"+{formatted}"
+    if sign_mode == " ":
+        return f" {formatted}"
+    return formatted
+
+
+def format_default(value: float, comma: str) -> str:
+    """
+    Format ``value`` the way d3's default (no-type) specifier does.
+
+    d3 aliases an omitted type to ``.12~g``. This preserves fixed notation from
+    ``1e-6`` through twelve significant integer digits, then uses exponent
+    notation outside that range. Mirrors the omitted-type alias in
+    ``d3-format/src/formatSpecifier.js``.
+
+    :return: the ``.12~g`` representation
+    """
+    return format_general(value, 12, True, comma)
+
+
+def format_general(value: float, precision: int, trim: bool, comma: str = "") -> str:
+    """
+    Format d3's ``g`` type with JavaScript ``toPrecision`` thresholds.
+
+    Mirrors ``d3-format/src/formatTypes.js`` and returns fixed or exponential
+    notation with the requested significant-digit precision.
+    """
+    precision = max(1, precision)
+    rounded = round_to_significant(value, precision)
+    exponent = decimal_exponent(rounded)
+    if value and (exponent < -6 or exponent >= precision):
+        formatted = format_exponential(value, precision - 1)
+    else:
+        formatted = format_significant(value, precision, False, comma)
+    return trim_trailing_zeros(formatted) if trim else formatted
+
+
+def format_exponential(value: float, precision: int) -> str:
+    """
+    Format d3's ``e`` type using binary-float, half-up rounding.
+
+    Mirrors the ``e`` formatter in ``d3-format/src/formatTypes.js`` and returns
+    an exponent without redundant leading zeros.
+    """
+    rounded = round_to_significant(value, precision + 1)
+    exponent = decimal_exponent(rounded)
+    mantissa = rounded / (10**exponent) if rounded else 0.0
+    return f"{mantissa:.{precision}f}e{exponent:+d}"
+
+
+def format_smart_number(value: float, signed: bool = False) -> str:
+    """
+    Format ``value`` the way the frontend ``SMART_NUMBER`` formatter does.
+
+    The notation is chosen by magnitude: SI prefixes (with ``G`` shown as ``B``)
+    for ``abs(value) >= 1000``, two decimals down to ``1``, four decimals down to
+    ``0.001``, a micro (``µ``) suffix down to ``1e-6``, and SI prefixes again
+    below that. When ``signed`` is set, positive values are prefixed with ``+``.
+    Mirrors ``number-format/factories/createSmartNumberFormatter.ts``.
+
+    :return: the adaptive frontend-compatible number string
+    """
+    if value == 0:
+        body = "0"
+    else:
+        absolute = abs(value)
+        if absolute >= 1000:
+            body = format_si(value, 3, trim=True, billions=True)
+        elif absolute >= 1:
+            body = trim_trailing_zeros(format(quantize_half_up(value, 2), ".2f"))
+        elif absolute >= 0.001:
+            body = trim_trailing_zeros(format(quantize_half_up(value, 4), ".4f"))
+        elif absolute > 0.000001:
+            body = format_si(value * 1000000, 3, trim=True) + "µ"
+        else:
+            body = format_si(value, 3, trim=True)
+    prefix = "+" if signed and value > 0 else ""
+    return prefix + body
+
+
+def format_si(value: float, precision: int, trim: bool, billions: bool = False) -> str:
+    """
+    Format ``value`` with an SI prefix to ``precision`` significant digits.
+
+    Rounds to ``precision`` significant figures first, then scales into the
+    nearest power-of-1000 bracket (clamped to the ``y``..``Y`` range) and appends
+    the matching SI symbol. Rounding before the divide matches d3 and keeps
+    ``4725`` at ``4.73k`` (the inexact ``4.725`` mantissa would round to
+    ``4.72k``), and lets a value that rounds up into the next bracket pick the
+    right symbol (``999.5k`` -> ``1M``). With ``billions`` set, the ``G`` (giga)
+    symbol is rendered as ``B``. Mirrors d3's
+    ``formatPrefixAuto.js``/``formatRounded.js`` combination.
+
+    :return: a significant-digit mantissa followed by its SI prefix
+    """
+    if value == 0:
+        return format_significant(0.0, precision, trim)
+
+    rounded = round_to_significant(value, precision)
+    exponent = max(-8, min(8, math.floor(math.log10(abs(rounded))) // 3))
+    mantissa = rounded / (10 ** (exponent * 3))
+
+    symbol = SI_PREFIXES[exponent]
+    if billions and symbol == "G":
+        symbol = "B"
+
+    return format_significant(mantissa, precision, trim) + symbol
+
+
+def format_significant(
+    value: float, precision: int, trim: bool, comma: str = ""
+) -> str:
+    """
+    Format to `precision` significant digits in fixed-point notation.
+
+    Serves both the d3 `r` type and SI mantissas, and avoids the scientific
+    notation Python's `g` would switch to. Mirrors the fixed representation
+    produced by ``d3-format/src/formatRounded.js``.
+
+    :return: a fixed-point significant-digit string
+    """
+    rounded = round_to_significant(value, precision)
+    decimals = decimals_for_significant(rounded, precision)
+    formatted = format(rounded, f"{comma}.{decimals}f")
+    return trim_trailing_zeros(formatted) if trim else formatted
+
+
+def round_to_significant(value: float, precision: int) -> float:
+    """
+    Round ``value`` to ``precision`` significant digits.
+
+    The number of decimal places to keep is derived from the value's order of
+    magnitude (``precision - 1 - floor(log10(abs(value)))``) and the rounding is
+    half away from zero, matching d3-format's ``formatDecimalParts`` path.
+
+    :return: the rounded binary-float value
+    """
+    if value == 0:
+        return 0.0
+    return float(quantize_half_up(value, precision - 1 - decimal_exponent(value)))
+
+
+def quantize_half_up(value: float, decimals: int) -> Decimal:
+    """
+    Round to `decimals` places, half away from zero, matching d3-format.
+
+    Quantizes the binary float value (not its decimal string) so the result
+    matches d3, which rounds the IEEE-754 value: ``2.675`` is ``2.67`` because it
+    is really ``2.67499...``, while an exact ``0.125`` rounds up to ``0.13``.
+    This supplies the rounding semantics of ``d3-format/src/formatTypes.js``.
+
+    :return: a ``Decimal`` rounded at the requested decimal place
+    """
+    return Decimal(value).quantize(Decimal(1).scaleb(-decimals), rounding=ROUND_HALF_UP)
+
+
+def decimals_for_significant(value: float, precision: int) -> int:
+    """
+    Return fixed-point decimal places needed for significant-digit formatting.
+
+    This is the report-side equivalent of the exponent adjustment in
+    ``d3-format/src/formatRounded.js``.
+    """
+    integer_digits = 1 if value == 0 else decimal_exponent(value) + 1
+    return max(0, precision - integer_digits)
+
+
+def decimal_exponent(value: float) -> int:
+    """
+    Return the base-10 exponent without ``log10`` boundary drift.
+
+    Used where d3-format derives an exponent through ``formatDecimalParts``.
+    """
+    return Decimal(repr(value)).adjusted() if value else 0
+
+
+def normalize_exponent(formatted: str) -> str:
+    """
+    Drop exponent leading zeros (``1e+07`` to ``1e+7``), as d3 does.
+
+    :return: the exponent string style emitted by ``d3-format``
+    """
+    return re.sub(r"([eE][+-])0*(\d)", r"\1\2", formatted)
+
+
+def get_currency_locale() -> str:
+    """
+    Return the request locale, or the configured default outside a request.
+
+    Report tasks run with a Flask application context but without a request, so
+    Flask-Babel can return ``None``. The config fallback keeps Celery-rendered
+    reports aligned with the locale supplied to the frontend at bootstrap. The
+    result feeds the locale argument used by ``currency-format/symbolPosition.ts``.
+
+    :return: a Babel locale identifier, always with a safe default
+    """
+    try:
+        if locale := get_locale():
+            return str(locale)
+    except RuntimeError:
+        pass
+
+    try:
+        return str(current_app.config.get("BABEL_DEFAULT_LOCALE") or DEFAULT_LOCALE)
+    except RuntimeError:
+        return DEFAULT_LOCALE
+
+
+@lru_cache(maxsize=None)
+def resolve_symbol_position(code: str, locale: str) -> str:
+    """
+    Derive the symbol position from the locale's convention for the currency.
+
+    Mirrors ``currency-format/symbolPosition.ts::resolveSymbolPosition`` and
+    returns ``"prefix"`` on invalid locale/currency input.
+    """
+    try:
+        sample = format_currency(1, code, locale=locale)
+        first_digit = next(i for i, char in enumerate(sample) if char.isdigit())
+        return "prefix" if first_digit > 0 else "suffix"
+    except Exception:  # pylint: disable=broad-except  # noqa: BLE001
+        return "prefix"
+
+
+def apply_currency(formatted: str, currency: dict[str, Any]) -> str:
+    """
+    Add a localized currency symbol to an already formatted number.
+
+    Mirrors ``currency-format/CurrencyFormatter.ts::format``: percentage signs
+    are removed, explicit positions win, and an unset position is locale-driven.
+
+    :return: the number with a prefix or suffix currency symbol
+    """
+    normalized = formatted.replace("%", "")
+    code = currency["symbol"]
+    symbol = get_currency_symbol(code, locale=CURRENCY_SYMBOL_LOCALE) or code
+    position = currency.get("symbolPosition")
+    if position not in ("prefix", "suffix"):
+        position = resolve_symbol_position(code, get_currency_locale())
+    if position == "prefix":
+        return f"{symbol} {normalized}"
+    return f"{normalized} {symbol}"
+
+
+def trim_trailing_zeros(formatted: str) -> str:
+    """
+    Remove insignificant fractional zeros while preserving suffixes.
+
+    Mirrors ``d3-format/src/formatTrim.js`` for decimal, exponent, and percent
+    strings and returns the compact representation.
+    """
+    suffix = "%" if formatted.endswith("%") else ""
+    body = formatted[: -len(suffix)] if suffix else formatted
+    coefficient, separator, exponent = body.partition("e")
+    if "." in coefficient:
+        coefficient = coefficient.rstrip("0").rstrip(".")
+    exponent_suffix = f"{separator}{exponent}" if separator else ""
+    return coefficient + exponent_suffix + suffix
+
+
+def raw_string(value: float) -> str:
+    """
+    Convert an unformatted number to the frontend-like neutral representation.
+
+    Integral floats lose their Python-only ``.0`` suffix. The result is the
+    safe fallback used by ``CurrencyFormatter.ts`` and invalid format handling.
+    """
+    if isinstance(value, float) and value.is_integer():
+        return str(int(value))
+    return str(value)

superset/utils/rls.py

@@ -17,18 +17,41 @@
 
 from __future__ import annotations
 
+import hashlib
 from typing import Any, TYPE_CHECKING
 
 from sqlalchemy import and_, or_
 
-from superset import db
+from superset import db, security_manager
 from superset.sql.parse import Table
+from superset.utils import json
+from superset.utils.core import get_user_id
 
 if TYPE_CHECKING:
     from superset.models.core import Database
     from superset.sql.parse import BaseSQLStatement
 
 
+def _get_cache_identity() -> str:
+    """
+    Build a stable per-session identity to key the parse-failure sentinel on.
+
+    Logged-in users have a stable numeric id from ``get_user_id()``. Guest
+    users (embedded) don't -- ``get_user_id()`` always returns ``None`` for
+    them -- so different guest tokens with different RLS scopes would
+    otherwise all collapse onto the same "user-None" sentinel and share cache
+    entries. Key those on a hash of the guest token's own RLS rules instead,
+    so distinct guest scopes stay isolated from one another.
+    """
+    if guest_user := security_manager.get_current_guest_user_if_guest():
+        rls_rules = guest_user.guest_token.get("rls_rules", [])
+        digest = hashlib.sha256(
+            json.dumps(rls_rules, sort_keys=True).encode("utf-8")
+        ).hexdigest()
+        return f"guest-{digest}"
+    return str(get_user_id())
+
+
 def apply_rls(
     database: Database,
     catalog: str | None,
@@ -204,6 +227,12 @@ def collect_rls_predicates_for_sql(
             }
         )
     except Exception:
-        # If we can't parse the SQL, return empty list
-        # This ensures RLS application failure doesn't break caching
-        return []
+        # If we can't parse the SQL, we can't tell which (if any) RLS
+        # predicates would apply, so we can't contribute a meaningful cache
+        # key component. Returning an empty list here would make every
+        # user's failure collapse onto the same (missing) contribution,
+        # which is unsafe when different users have different RLS scopes on
+        # the underlying tables. Fall back to a per-user marker instead, so
+        # the cache key still varies by user even though we don't know the
+        # actual predicates.
+        return [f"rls-predicate-parse-failed-for-user-{_get_cache_identity()}"]

tests/integration_tests/celery_tests.py

@@ -70,10 +70,26 @@ def get_query_by_id(id: int):
 
 @pytest.fixture(autouse=True, scope="module")
 def setup_sqllab():
+    # These tests exercise CTAS/CVAS, which the example database must be
+    # granted to allow. Enable the grants for the duration of the module and
+    # restore the originals afterwards.
+    with app.app_context():
+        example_db = get_example_database()
+        original_allow_ctas = example_db.allow_ctas
+        original_allow_cvas = example_db.allow_cvas
+        example_db.allow_ctas = True
+        example_db.allow_cvas = True
+        db.session.commit()
+
     yield
+
     # clean up after all tests are done
     # use a new app context
     with app.app_context():
+        example_db = get_example_database()
+        example_db.allow_ctas = original_allow_ctas
+        example_db.allow_cvas = original_allow_cvas
+        db.session.commit()
         db.session.query(Query).delete()
         db.session.commit()
         for tbl in TMP_TABLES:

tests/integration_tests/databases/api_tests.py

@@ -4608,8 +4608,9 @@ class TestDatabaseApi(SupersetTestCase):
         assert rv.status_code == 202
         response = json.loads(rv.data.decode("utf-8"))
         assert response == {"message": "Async task created to sync permissions"}
+        admin_user = security_manager.find_user(username=ADMIN_USERNAME)
         mock_task.assert_called_once_with(
-            test_database.id, ADMIN_USERNAME, test_database.database_name
+            test_database.id, admin_user.id, test_database.database_name
         )
 
         # Cleanup

tests/integration_tests/sql_lab/api_tests.py

@@ -262,8 +262,11 @@ class TestSqlLabApi(SupersetTestCase):
             return_value=formatter_response
         )
 
-        with mock.patch("superset.commands.sql_lab.estimate.db") as mock_superset_db:
-            mock_superset_db.session.query().get.return_value = db_mock
+        with (
+            mock.patch("superset.commands.sql_lab.estimate.DatabaseDAO") as mock_dao,
+            mock.patch("superset.security_manager.raise_for_access"),
+        ):
+            mock_dao.find_by_id.return_value = db_mock
 
             data = {"database_id": 1, "sql": "SELECT 1"}
             rv = self.client.post(

tests/integration_tests/sql_lab/commands_tests.py

@@ -49,8 +49,8 @@ class TestQueryEstimationCommand(SupersetTestCase):
         data: EstimateQueryCostSchema = schema.dump(params)
         command = estimate.QueryEstimationCommand(data)
 
-        with mock.patch("superset.commands.sql_lab.estimate.db") as mock_superset_db:
-            mock_superset_db.session.query().get.return_value = None
+        with mock.patch("superset.commands.sql_lab.estimate.DatabaseDAO") as mock_dao:
+            mock_dao.find_by_id.return_value = None
             with pytest.raises(SupersetErrorException) as ex_info:
                 command.validate()
             assert (
@@ -81,8 +81,11 @@ class TestQueryEstimationCommand(SupersetTestCase):
         db_mock.db_engine_spec.query_cost_formatter = mock.Mock(return_value=None)
         is_feature_enabled.return_value = False
 
-        with mock.patch("superset.commands.sql_lab.estimate.db") as mock_superset_db:
-            mock_superset_db.session.query().get.return_value = db_mock
+        with (
+            mock.patch("superset.commands.sql_lab.estimate.DatabaseDAO") as mock_dao,
+            mock.patch("superset.security_manager.raise_for_access"),
+        ):
+            mock_dao.find_by_id.return_value = db_mock
             with pytest.raises(SupersetErrorException) as ex_info:
                 command.run()
             assert (
@@ -107,8 +110,11 @@ class TestQueryEstimationCommand(SupersetTestCase):
         db_mock.db_engine_spec.estimate_query_cost = mock.Mock(return_value=100)
         db_mock.db_engine_spec.query_cost_formatter = mock.Mock(return_value=payload)
 
-        with mock.patch("superset.commands.sql_lab.estimate.db") as mock_superset_db:
-            mock_superset_db.session.query().get.return_value = db_mock
+        with (
+            mock.patch("superset.commands.sql_lab.estimate.DatabaseDAO") as mock_dao,
+            mock.patch("superset.security_manager.raise_for_access"),
+        ):
+            mock_dao.find_by_id.return_value = db_mock
             result = command.run()
             assert result == payload
 

tests/integration_tests/sqla_models_tests.py

@@ -835,7 +835,17 @@ def test_none_operand_in_filter(login_as_admin, physical_dataset):
             '{{ user_email }}' as email,
             '{{ current_user_roles()|tojson }}' as roles
             """,
-            {1, "abc", "abc@test.com", '["role1", "role2"]'},
+            # The leading `{% set %}` block isn't valid SQL, so parsing this
+            # virtual dataset's SQL for RLS predicates fails and the cache key
+            # picks up the per-user parse-failure sentinel (no user is logged
+            # in for this test, hence "user-None").
+            {
+                1,
+                "abc",
+                "abc@test.com",
+                '["role1", "role2"]',
+                "rls-predicate-parse-failed-for-user-None",
+            },
             True,
         ),
         (
@@ -845,7 +855,9 @@ def test_none_operand_in_filter(login_as_admin, physical_dataset):
             SELECT
             '{{ user_conditional_id }}' as conditional
             """,
-            {1, "abc@test.com"},
+            # Same parse-failure sentinel as above: the leading `{% set %}`
+            # block breaks SQL parsing for RLS predicate collection.
+            {1, "abc@test.com", "rls-predicate-parse-failed-for-user-None"},
             True,
         ),
         (

tests/unit_tests/charts/test_client_processing.py

@@ -16,13 +16,21 @@
 # under the License.
 
 from io import BytesIO, StringIO
+from unittest.mock import MagicMock
 
 import pandas as pd
 import pytest
 from flask_babel import lazy_gettext as _
 from sqlalchemy.orm.session import Session
 
-from superset.charts.client_processing import apply_client_processing, pivot_df, table
+from superset.charts.client_processing import (
+    apply_client_processing,
+    apply_pivot_number_formats,
+    format_column,
+    pivot_df,
+    pivot_table_v2,
+    table,
+)
 from superset.common.chart_data import ChartDataResultFormat
 from superset.utils import excel
 from superset.utils.core import GenericDataType
@@ -1848,6 +1856,729 @@ def test_table():
     )
 
 
+def test_table_applies_currency_format() -> None:
+    """
+    Table reports honor a column's `currencyFormat`.
+    """
+    df = pd.DataFrame.from_dict({"amount": {0: 1234.5}})
+    form_data = {
+        "viz_type": "table",
+        "column_config": {
+            "amount": {
+                "d3NumberFormat": ",.2f",
+                "currencyFormat": {"symbol": "USD", "symbolPosition": "prefix"},
+            }
+        },
+    }
+    formatted = table(df, form_data)
+    assert formatted["amount"].tolist() == ["$ 1,234.50"]
+
+
+def test_table_applies_si_number_format() -> None:
+    """
+    Table reports honor d3 formats that Python's str.format cannot express.
+    """
+    df = pd.DataFrame.from_dict({"amount": {0: 1234.0}})
+    form_data = {
+        "viz_type": "table",
+        "column_config": {"amount": {"d3NumberFormat": ".3s"}},
+    }
+    formatted = table(df, form_data)
+    assert formatted["amount"].tolist() == ["1.23k"]
+
+
+def test_table_applies_smart_number_default_to_unconfigured_metric() -> None:
+    """
+    A metric with no saved d3 format still renders like Explore. The Table
+    plugin gives every metric column a formatter, and ``getNumberFormatter``
+    defaults to SMART_NUMBER, so the report must not leave the value raw.
+    """
+    df = pd.DataFrame.from_dict({"count": {0: 1234567}})
+    form_data = {"viz_type": "table", "metrics": ["count"], "percent_metrics": []}
+    formatted = table(df, form_data)
+    assert formatted["count"].tolist() == ["1.23M"]
+
+
+def test_table_leaves_unconfigured_numeric_dimension_untouched() -> None:
+    """
+    A numeric non-metric column with no configured format is left raw, matching
+    the browser, which only formats numeric dimensions when a format or currency
+    is set.
+    """
+    df = pd.DataFrame.from_dict({"year": {0: 2024}, "count": {0: 1234567}})
+    form_data = {"viz_type": "table", "metrics": ["count"], "percent_metrics": []}
+    formatted = table(df, form_data)
+    assert formatted["year"].tolist() == [2024]
+    assert formatted["count"].tolist() == ["1.23M"]
+
+
+def test_table_applies_percent_3_point_default_to_percent_metric() -> None:
+    """
+    Percent metric columns default to PERCENT_3_POINT in the Table plugin.
+    """
+    df = pd.DataFrame.from_dict({"%count": {0: 0.1234}})
+    form_data = {"viz_type": "table", "metrics": [], "percent_metrics": ["count"]}
+    formatted = table(df, form_data)
+    assert formatted["%count"].tolist() == ["12.340%"]
+
+
+def test_table_applies_datasource_saved_metric_format_without_chart_override() -> None:
+    df = pd.DataFrame.from_dict({"amount": {0: 1234.5}})
+    datasource = MagicMock()
+    datasource.data = {
+        "column_formats": {"amount": ",.2f"},
+        "verbose_map": {},
+    }
+
+    formatted = table(df, {"viz_type": "table"}, datasource)
+
+    assert formatted["amount"].tolist() == ["1,234.50"]
+
+
+def test_table_chart_format_overrides_datasource_saved_metric_format() -> None:
+    df = pd.DataFrame.from_dict({"amount": {0: 1234.5}})
+    datasource = MagicMock()
+    datasource.data = {
+        "column_formats": {"amount": ",.2f"},
+        "verbose_map": {},
+    }
+    form_data = {
+        "viz_type": "table",
+        "column_config": {"amount": {"d3NumberFormat": ",.1f"}},
+    }
+
+    formatted = table(df, form_data, datasource)
+
+    assert formatted["amount"].tolist() == ["1,234.5"]
+
+
+def test_pivot_table_v2_applies_value_format() -> None:
+    """
+    Pivot table reports honor `valueFormat` and per-metric `columnFormats`.
+    """
+    df = pd.DataFrame(
+        {"region": ["A", "B"], "sales": [1234.5, 6789.0], "qty": [10.0, 20.0]}
+    )
+    form_data = {
+        "viz_type": "pivot_table_v2",
+        "groupbyRows": ["region"],
+        "groupbyColumns": [],
+        "metrics": ["sales", "qty"],
+        "aggregateFunction": "Sum",
+        "metricsLayout": "COLUMNS",
+        "valueFormat": ",.2f",
+        "columnFormats": {"qty": ",d"},
+    }
+    formatted = pivot_table_v2(df, form_data)
+    assert formatted[("sales",)].tolist() == ["1,234.50", "6,789.00"]
+    assert formatted[("qty",)].tolist() == ["10", "20"]
+
+
+def test_pivot_table_v2_applies_smart_number_default_without_value_format() -> None:
+    """
+    Pivot value cells with no ``valueFormat`` and no per-metric format default to
+    SMART_NUMBER, mirroring the frontend's ``getNumberFormatter`` default.
+    """
+    df = pd.DataFrame({"region": ["A", "B"], "sales": [1234567.0, 6789.0]})
+    form_data = {
+        "viz_type": "pivot_table_v2",
+        "groupbyRows": ["region"],
+        "groupbyColumns": [],
+        "metrics": ["sales"],
+        "aggregateFunction": "Sum",
+        "metricsLayout": "COLUMNS",
+    }
+    formatted = pivot_table_v2(df, form_data)
+    assert formatted[("sales",)].tolist() == ["1.23M", "6.79k"]
+
+
+def test_pivot_table_v2_applies_datasource_saved_metric_format_without_override() -> (
+    None
+):
+    df = pd.DataFrame({"region": ["A", "B"], "sales": [1234.5, 6789.0]})
+    datasource = MagicMock()
+    datasource.data = {
+        "column_formats": {"sales": ",d"},
+        "verbose_map": {},
+    }
+    form_data = {
+        "viz_type": "pivot_table_v2",
+        "groupbyRows": ["region"],
+        "groupbyColumns": [],
+        "metrics": ["sales"],
+        "aggregateFunction": "Sum",
+        "valueFormat": ",.2f",
+    }
+
+    formatted = pivot_table_v2(df, form_data, datasource)
+
+    assert formatted[("sales",)].tolist() == ["1,235", "6,789"]
+
+
+def test_pivot_table_v2_chart_format_overrides_datasource_saved_metric_format() -> None:
+    df = pd.DataFrame({"region": ["A"], "sales": [1234.5]})
+    datasource = MagicMock()
+    datasource.data = {
+        "column_formats": {"sales": ",d"},
+        "verbose_map": {},
+    }
+    form_data = {
+        "viz_type": "pivot_table_v2",
+        "groupbyRows": ["region"],
+        "groupbyColumns": [],
+        "metrics": ["sales"],
+        "aggregateFunction": "Sum",
+        "valueFormat": ",.2f",
+        "columnFormats": {"sales": ",.1f"},
+    }
+
+    formatted = pivot_table_v2(df, form_data, datasource)
+
+    assert formatted[("sales",)].tolist() == ["1,234.5"]
+
+
+def test_pivot_table_v2_applies_per_metric_format_when_metrics_combined() -> None:
+    """
+    Per-metric formats apply when `combineMetric` moves the metric to the last
+    column level.
+    """
+    df = pd.DataFrame(
+        {
+            "dept": ["A", "B"],
+            "region": ["x", "x"],
+            "sales": [100.0, 200.0],
+            "qty": [1111.0, 2222.0],
+        }
+    )
+    form_data = {
+        "viz_type": "pivot_table_v2",
+        "groupbyRows": ["dept"],
+        "groupbyColumns": ["region"],
+        "metrics": ["sales", "qty"],
+        "aggregateFunction": "Sum",
+        "metricsLayout": "COLUMNS",
+        "combineMetric": True,
+        "valueFormat": ",.2f",
+        "columnFormats": {"qty": ",d"},
+    }
+    formatted = pivot_table_v2(df, form_data)
+    assert formatted[("x", "qty")].tolist() == ["1,111", "2,222"]
+    assert formatted[("x", "sales")].tolist() == ["100.00", "200.00"]
+
+
+def test_table_auto_currency_uses_detected_currency() -> None:
+    """
+    AUTO currency resolves to the payload's `detected_currency`, or falls back
+    to the plain number when detection found mixed currencies.
+    """
+    form_data = {
+        "viz_type": "table",
+        "column_config": {
+            "amount": {
+                "d3NumberFormat": ",.2f",
+                "currencyFormat": {"symbol": "AUTO", "symbolPosition": "prefix"},
+            }
+        },
+    }
+    df = pd.DataFrame.from_dict({"amount": {0: 1234.5}})
+    formatted = table(df, form_data, detected_currency="USD")
+    assert formatted["amount"].tolist() == ["$ 1,234.50"]
+
+    df = pd.DataFrame.from_dict({"amount": {0: 1234.5}})
+    formatted = table(df, form_data, detected_currency=None)
+    assert formatted["amount"].tolist() == ["1,234.50"]
+
+
+def test_table_auto_currency_uses_per_row_currency_context() -> None:
+    form_data = {
+        "viz_type": "table",
+        "column_config": {
+            "amount": {
+                "d3NumberFormat": ",.2f",
+                "currencyFormat": {"symbol": "AUTO", "symbolPosition": "prefix"},
+            }
+        },
+    }
+    datasource = MagicMock()
+    datasource.data = {
+        "column_formats": {},
+        "verbose_map": {},
+        "currency_code_column": "currency",
+    }
+    df = pd.DataFrame(
+        {
+            "amount": [100.0, 200.0, 300.0, 400.0],
+            "currency": ["USD", " eur ", None, "invalid"],
+        }
+    )
+
+    formatted = table(df, form_data, datasource, detected_currency="GBP")
+
+    assert formatted["amount"].tolist() == [
+        "$ 100.00",
+        "€ 200.00",
+        "300.00",
+        "400.00",
+    ]
+
+
+def test_table_saved_auto_currency_uses_per_row_currency_context() -> None:
+    datasource = MagicMock()
+    datasource.data = {
+        "column_formats": {"amount": ",.2f"},
+        "verbose_map": {},
+        "currency_code_column": "currency",
+        "metrics": [
+            {
+                "metric_name": "amount",
+                "currency": {"symbol": "AUTO", "symbolPosition": "prefix"},
+            }
+        ],
+    }
+    df = pd.DataFrame(
+        {
+            "amount": [100.0, 200.0],
+            "currency": ["USD", "EUR"],
+        }
+    )
+
+    formatted = table(df, {"viz_type": "table"}, datasource)
+
+    assert formatted["amount"].tolist() == ["$ 100.00", "€ 200.00"]
+
+
+def test_pivot_table_v2_auto_currency_uses_detected_currency() -> None:
+    """
+    AUTO currency in pivot tables resolves to the payload's `detected_currency`.
+    """
+    df = pd.DataFrame({"region": ["A", "B"], "sales": [1234.5, 6789.0]})
+    form_data = {
+        "viz_type": "pivot_table_v2",
+        "groupbyRows": ["region"],
+        "groupbyColumns": [],
+        "metrics": ["sales"],
+        "aggregateFunction": "Sum",
+        "valueFormat": ",.2f",
+        "currencyFormat": {"symbol": "AUTO", "symbolPosition": "prefix"},
+    }
+    formatted = pivot_table_v2(df, form_data, detected_currency="EUR")
+    assert formatted[("sales",)].tolist() == ["€ 1,234.50", "€ 6,789.00"]
+
+
+def test_pivot_table_v2_auto_currency_uses_per_cell_currency_context() -> None:
+    df = pd.DataFrame(
+        {
+            "region": ["USD cell", "USD cell", "EUR cell", "Mixed", "Mixed", "Empty"],
+            "sales": [100.0, 50.0, 200.0, 300.0, 400.0, 500.0],
+            "currency": ["USD", " usd ", "EUR", "USD", "EUR", None],
+        }
+    )
+    datasource = MagicMock()
+    datasource.data = {
+        "column_formats": {},
+        "verbose_map": {},
+        "currency_code_column": "currency",
+    }
+    form_data = {
+        "viz_type": "pivot_table_v2",
+        "groupbyRows": ["region"],
+        "groupbyColumns": [],
+        "metrics": ["sales"],
+        "aggregateFunction": "Sum",
+        "valueFormat": ",.2f",
+        "currencyFormat": {"symbol": "AUTO", "symbolPosition": "prefix"},
+    }
+
+    formatted = pivot_table_v2(df, form_data, datasource, detected_currency="GBP")
+
+    assert formatted[("sales",)].to_dict() == {
+        ("EUR cell",): "€ 200.00",
+        ("Empty",): "£ 500.00",
+        ("Mixed",): "700.00",
+        ("USD cell",): "$ 150.00",
+    }
+
+
+def test_pivot_table_v2_auto_currency_reads_stored_form_data_key() -> None:
+    datasource = MagicMock()
+    datasource.data = {
+        "column_formats": {},
+        "verbose_map": {},
+        "currency_code_column": "currency",
+    }
+    df = pd.DataFrame(
+        {
+            "region": ["US", "EU"],
+            "sales": [100.0, 200.0],
+            "currency": ["USD", "EUR"],
+        }
+    )
+    form_data = {
+        "viz_type": "pivot_table_v2",
+        "groupbyRows": ["region"],
+        "groupbyColumns": [],
+        "metrics": ["sales"],
+        "aggregateFunction": "Sum",
+        "valueFormat": ",.2f",
+        "currency_format": {"symbol": "AUTO", "symbolPosition": "prefix"},
+    }
+
+    formatted = pivot_table_v2(df, form_data, datasource)
+
+    assert formatted[("sales",)].to_dict() == {
+        ("EU",): "€ 200.00",
+        ("US",): "$ 100.00",
+    }
+
+
+def test_pivot_table_v2_auto_currency_handles_sparse_2d_pivot() -> None:
+    """
+    A pivot with both rows and columns has empty cross-product cells. Pandas
+    fills those with scalar ``NaN`` rather than an empty currency tuple, which
+    must not crash AUTO currency resolution for the whole report.
+    """
+    df = pd.DataFrame(
+        {
+            "region": ["EU", "EU", "US"],
+            "product": ["a", "b", "a"],
+            "sales": [10.0, 20.0, 30.0],
+            "currency": ["EUR", "EUR", "USD"],
+        }
+    )
+    datasource = MagicMock()
+    datasource.data = {
+        "column_formats": {},
+        "verbose_map": {},
+        "currency_code_column": "currency",
+    }
+    form_data = {
+        "viz_type": "pivot_table_v2",
+        "groupbyRows": ["region"],
+        "groupbyColumns": ["product"],
+        "metrics": ["sales"],
+        "aggregateFunction": "Sum",
+        "valueFormat": ",.2f",
+        "currencyFormat": {"symbol": "AUTO", "symbolPosition": "prefix"},
+    }
+
+    formatted = pivot_table_v2(df, form_data, datasource, detected_currency="GBP")
+
+    assert formatted[("sales", "a")].to_dict() == {
+        ("EU",): "€ 10.00",
+        ("US",): "$ 30.00",
+    }
+    # The missing (US, b) combination stays empty; the present EUR cell formats.
+    assert formatted[("sales", "b")].to_dict() == {
+        ("EU",): "€ 20.00",
+        ("US",): "",
+    }
+
+
+def test_pivot_table_v2_saved_auto_currency_uses_per_cell_context() -> None:
+    datasource = MagicMock()
+    datasource.data = {
+        "column_formats": {"sales": ",.2f"},
+        "verbose_map": {},
+        "currency_code_column": "currency",
+        "metrics": [
+            {
+                "metric_name": "sales",
+                "currency": {"symbol": "AUTO", "symbolPosition": "prefix"},
+            }
+        ],
+    }
+    df = pd.DataFrame(
+        {
+            "region": ["US", "EU"],
+            "sales": [100.0, 200.0],
+            "currency": ["USD", "EUR"],
+        }
+    )
+    form_data = {
+        "viz_type": "pivot_table_v2",
+        "groupbyRows": ["region"],
+        "groupbyColumns": [],
+        "metrics": ["sales"],
+        "aggregateFunction": "Sum",
+        "valueFormat": ",.1f",
+    }
+
+    formatted = pivot_table_v2(df, form_data, datasource)
+
+    assert formatted[("sales",)].to_dict() == {
+        ("EU",): "€ 200.00",
+        ("US",): "$ 100.00",
+    }
+
+
+def test_pivot_table_v2_count_auto_currency_uses_detected_fallback() -> None:
+    datasource = MagicMock()
+    datasource.data = {
+        "column_formats": {},
+        "verbose_map": {},
+        "currency_code_column": "currency",
+    }
+    df = pd.DataFrame(
+        {
+            "region": ["US", "EU"],
+            "sales": [100.0, 200.0],
+            "currency": ["USD", "EUR"],
+        }
+    )
+    form_data = {
+        "viz_type": "pivot_table_v2",
+        "groupbyRows": ["region"],
+        "groupbyColumns": [],
+        "metrics": ["sales"],
+        "aggregateFunction": "Count",
+        "valueFormat": ",d",
+        "currencyFormat": {"symbol": "AUTO", "symbolPosition": "prefix"},
+    }
+
+    formatted = pivot_table_v2(df, form_data, datasource, detected_currency="GBP")
+
+    assert formatted[("sales",)].to_dict() == {
+        ("EU",): "£ 1",
+        ("US",): "£ 1",
+    }
+
+
+def test_pivot_table_v2_auto_currency_tracks_mixed_total_context() -> None:
+    df = pd.DataFrame(
+        {
+            "region": ["US", "EU"],
+            "sales": [100.0, 200.0],
+            "currency": ["USD", "EUR"],
+        }
+    )
+    datasource = MagicMock()
+    datasource.data = {
+        "column_formats": {},
+        "verbose_map": {},
+        "currency_code_column": "currency",
+    }
+    form_data = {
+        "viz_type": "pivot_table_v2",
+        "groupbyRows": ["region"],
+        "groupbyColumns": [],
+        "metrics": ["sales"],
+        "aggregateFunction": "Sum",
+        "valueFormat": ",.2f",
+        "currencyFormat": {"symbol": "AUTO", "symbolPosition": "prefix"},
+        "colTotals": True,
+    }
+
+    formatted = pivot_table_v2(df, form_data, datasource)
+
+    assert formatted[("sales",)].to_dict() == {
+        ("EU",): "€ 200.00",
+        ("US",): "$ 100.00",
+        ("Total (Sum)",): "300.00",
+    }
+
+
+def test_pivot_table_v2_auto_currency_tracks_subtotal_context() -> None:
+    df = pd.DataFrame(
+        {
+            "region": ["Mixed", "Mixed", "USD", "USD"],
+            "quarter": ["Q1", "Q2", "Q1", "Q2"],
+            "sales": [100.0, 200.0, 300.0, 400.0],
+            "currency": ["USD", "EUR", "USD", "USD"],
+        }
+    )
+    datasource = MagicMock()
+    datasource.data = {
+        "column_formats": {},
+        "verbose_map": {},
+        "currency_code_column": "currency",
+    }
+    form_data = {
+        "viz_type": "pivot_table_v2",
+        "groupbyRows": ["region"],
+        "groupbyColumns": ["quarter"],
+        "metrics": ["sales"],
+        "aggregateFunction": "Sum",
+        "valueFormat": ",.2f",
+        "currencyFormat": {"symbol": "AUTO", "symbolPosition": "prefix"},
+        "rowTotals": True,
+    }
+
+    formatted = pivot_table_v2(df, form_data, datasource)
+
+    assert formatted.loc[("Mixed",), ("sales", "Q1")] == "$ 100.00"
+    assert formatted.loc[("Mixed",), ("sales", "Q2")] == "€ 200.00"
+    assert formatted.loc[("Mixed",), ("sales", "Subtotal")] == "300.00"
+    assert formatted.loc[("USD",), ("sales", "Subtotal")] == "$ 700.00"
+    assert formatted.loc[("Mixed",), ("Total (Sum)", "")] == "300.00"
+
+
+def test_apply_client_processing_passes_detected_currency() -> None:
+    """
+    The query payload's `detected_currency` reaches the number formatters.
+    """
+    result = {
+        "queries": [
+            {
+                "result_format": ChartDataResultFormat.JSON,
+                "detected_currency": "USD",
+                "data": [{"amount": 1234.5}],
+            }
+        ]
+    }
+    form_data = {
+        "viz_type": "table",
+        "column_config": {
+            "amount": {
+                "d3NumberFormat": ",.2f",
+                "currencyFormat": {"symbol": "AUTO", "symbolPosition": "prefix"},
+            }
+        },
+    }
+    processed = apply_client_processing(result, form_data)
+    assert processed["queries"][0]["data"] == {"amount": {0: "$ 1,234.50"}}
+
+
+def test_pivot_table_v2_applies_per_metric_format_when_metrics_on_rows() -> None:
+    """
+    Per-metric formats apply when `metricsLayout` is "ROWS" and the metric is
+    on the index instead of the columns.
+    """
+    df = pd.DataFrame(
+        {
+            "dept": ["A", "B"],
+            "region": ["x", "x"],
+            "sales": [100.0, 200.0],
+            "qty": [1111.0, 2222.0],
+        }
+    )
+    form_data = {
+        "viz_type": "pivot_table_v2",
+        "groupbyRows": ["dept"],
+        "groupbyColumns": ["region"],
+        "metrics": ["sales", "qty"],
+        "aggregateFunction": "Sum",
+        "metricsLayout": "ROWS",
+        "valueFormat": ",.2f",
+        "columnFormats": {"qty": ",d"},
+        "currencyFormats": {"sales": {"symbol": "USD", "symbolPosition": "prefix"}},
+    }
+    formatted = pivot_table_v2(df, form_data)
+    assert formatted[("x",)].tolist() == ["$ 100.00", "$ 200.00", "1,111", "2,222"]
+
+
+def test_pivot_table_v2_applies_per_metric_format_when_metrics_on_rows_combined() -> (
+    None
+):
+    """
+    Per-metric formats apply when `metricsLayout` is "ROWS" and `combineMetric`
+    moves the metric to the last index level.
+    """
+    df = pd.DataFrame(
+        {
+            "dept": ["A", "B"],
+            "region": ["x", "x"],
+            "sales": [100.0, 200.0],
+            "qty": [1111.0, 2222.0],
+        }
+    )
+    form_data = {
+        "viz_type": "pivot_table_v2",
+        "groupbyRows": ["dept"],
+        "groupbyColumns": ["region"],
+        "metrics": ["sales", "qty"],
+        "aggregateFunction": "Sum",
+        "metricsLayout": "ROWS",
+        "combineMetric": True,
+        "valueFormat": ",.2f",
+        "columnFormats": {"qty": ",d"},
+    }
+    formatted = pivot_table_v2(df, form_data)
+    assert formatted[("x",)].tolist() == ["100.00", "1,111", "200.00", "2,222"]
+
+
+def test_format_column_applies_d3_and_currency() -> None:
+    df = pd.DataFrame({"amount": [1234.5, 6789.0]})
+    format_column(df, "amount", ",.2f", {})
+    assert df["amount"].tolist() == ["1,234.50", "6,789.00"]
+
+    df = pd.DataFrame({"amount": [1234.5]})
+    format_column(df, "amount", ",.2f", {"symbol": "USD", "symbolPosition": "prefix"})
+    assert df["amount"].tolist() == ["$ 1,234.50"]
+
+
+def test_format_column_is_noop_without_format() -> None:
+    df = pd.DataFrame({"amount": [1234.5]})
+    format_column(df, "amount", None, {})
+    assert df["amount"].tolist() == [1234.5]
+
+
+def test_format_column_preserves_numeric_format_when_currency_is_invalid() -> None:
+    df = pd.DataFrame({"amount": [1234.5]})
+    format_column(
+        df,
+        "amount",
+        ",.2f",
+        {"symbol": {"invalid": True}, "symbolPosition": "prefix"},
+    )
+    assert df["amount"].tolist() == ["1,234.50"]
+
+
+def test_format_column_preserves_raw_value_for_invalid_number_format() -> None:
+    df = pd.DataFrame({"amount": [1234.5]})
+    format_column(df, "amount", "not-a-format", {})
+    assert df["amount"].tolist() == ["1234.5"]
+
+
+def test_apply_pivot_number_formats_resolves_metric_level() -> None:
+    df = pd.DataFrame({("sales",): [1234.5], ("qty",): [10.0]})
+    df.columns = pd.MultiIndex.from_tuples([("sales",), ("qty",)])
+    apply_pivot_number_formats(
+        df, {"valueFormat": ",.2f", "columnFormats": {"qty": ",d"}}
+    )
+    assert df[("sales",)].tolist() == ["1,234.50"]
+    assert df[("qty",)].tolist() == ["10"]
+
+
+def test_apply_pivot_number_formats_metric_at_last_level_when_combined() -> None:
+    df = pd.DataFrame({("x", "sales"): [100.0], ("x", "qty"): [1111.0]})
+    df.columns = pd.MultiIndex.from_tuples([("x", "sales"), ("x", "qty")])
+    apply_pivot_number_formats(
+        df,
+        {"combineMetric": True, "valueFormat": ",.2f", "columnFormats": {"qty": ",d"}},
+    )
+    assert df[("x", "sales")].tolist() == ["100.00"]
+    assert df[("x", "qty")].tolist() == ["1,111"]
+
+
+def test_apply_pivot_number_formats_falls_back_to_global_format() -> None:
+    df = pd.DataFrame({("sales",): [1234.5]})
+    df.columns = pd.MultiIndex.from_tuples([("sales",)])
+    apply_pivot_number_formats(
+        df, {"valueFormat": ",.2f", "columnFormats": {"sales": ""}}
+    )
+    assert df[("sales",)].tolist() == ["1,234.50"]
+
+
+def test_apply_pivot_number_formats_preserves_raw_value_on_format_error() -> None:
+    df = pd.DataFrame({("sales",): [1234.5]})
+    df.columns = pd.MultiIndex.from_tuples([("sales",)])
+    apply_pivot_number_formats(df, {"valueFormat": "not-a-format"})
+    assert df[("sales",)].tolist() == ["1234.5"]
+
+
+def test_apply_pivot_number_formats_auto_currency_without_detection() -> None:
+    df = pd.DataFrame({("sales",): [1234.5]})
+    df.columns = pd.MultiIndex.from_tuples([("sales",)])
+    apply_pivot_number_formats(
+        df,
+        {
+            "valueFormat": ",.2f",
+            "currencyFormat": {"symbol": "AUTO", "symbolPosition": "prefix"},
+        },
+    )
+    assert df[("sales",)].tolist() == ["1,234.50"]
+
+
 def test_apply_client_processing_no_form_invalid_viz_type():
     """
     Test with invalid viz type. It should just return the result
@@ -2635,10 +3366,10 @@ def test_apply_client_processing_verbose_map(session: Session):
         "queries": [
             {
                 "result_format": ChartDataResultFormat.JSON,
-                "data": {"COUNT(*)": {"Total (Sum)": 4725}},
+                "data": {"COUNT(*)": {"Total (Sum)": "4.73k"}},
                 "colnames": [("COUNT(*)",)],
                 "indexnames": [("Total (Sum)",)],
-                "coltypes": [GenericDataType.NUMERIC],
+                "coltypes": [GenericDataType.STRING],
                 "rowcount": 1,
             }
         ]

tests/unit_tests/commands/chart/streaming_export_command_test.py

@@ -39,6 +39,9 @@ def _setup_chart_mocks(
     datasource = mocker.MagicMock()
     datasource.get_query_str.return_value = sql
     datasource.database = mocker.MagicMock()
+    datasource.database.mutate_sql_based_on_config.side_effect = (
+        lambda sql_, **kwargs: sql_
+    )
     datasource.catalog = catalog
     datasource.schema = schema
     query_context.datasource = datasource
@@ -296,3 +299,46 @@ def test_catalog_and_schema_passed_to_engine(mocker: MockerFixture) -> None:
         catalog="my_catalog",
         schema="my_schema",
     )
+
+
+def test_streaming_export_mutation_does_not_double_apply(
+    mocker: MockerFixture,
+) -> None:
+    """
+    Regression test for a bug caught in review of apache/superset#40465's fix:
+    ``datasource.get_query_str()`` (used to build the SQL for chart streaming
+    exports) already runs it through
+    ``Database.mutate_sql_based_on_config(sql)`` -- with ``is_split=False`` --
+    upstream in ``get_query_str_extended``. If
+    ``_execute_query_and_stream`` mutated again with the same default
+    ``is_split=False``, a ``SQL_QUERY_MUTATOR`` would run twice under the
+    default ``MUTATE_AFTER_SPLIT=False`` config, and never run at all under
+    ``MUTATE_AFTER_SPLIT=True`` (since neither call would use
+    ``is_split=True``). Passing ``is_split=True`` here is the complement of
+    the upstream call, so exactly one of the two fires for either setting.
+    """
+    mock_db, query_context, datasource = _setup_chart_mocks(
+        mocker, sql="SELECT * FROM test /* mutated */"
+    )
+
+    mock_result = mocker.MagicMock()
+    mock_result.keys.return_value = ["col1"]
+    mock_result.fetchmany.side_effect = [[("val",)], []]
+
+    mock_connection = mocker.MagicMock()
+    mock_connection.execution_options.return_value.execute.return_value = mock_result
+    mock_connection.__enter__.return_value = mock_connection
+    mock_connection.__exit__.return_value = None
+
+    mock_engine = mocker.MagicMock()
+    mock_engine.connect.return_value = mock_connection
+    datasource.database.get_sqla_engine.return_value.__enter__.return_value = (
+        mock_engine
+    )
+
+    command = StreamingCSVExportCommand(query_context)
+    list(command.run()())
+
+    datasource.database.mutate_sql_based_on_config.assert_called_once_with(
+        "SELECT * FROM test /* mutated */", is_split=True
+    )

tests/unit_tests/commands/databases/sync_permissions_identity_test.py

@@ -0,0 +1,140 @@
+# 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.
+"""
+Traces how ``sync_database_permissions_task`` binds an acting identity.
+
+The Celery task receives the immutable ``id`` of the user who enqueued it
+(captured at enqueue time by ``SyncPermissionsCommand.validate``), not a
+mutable username string. At execution time it resolves that id to a user
+record via ``security_manager.get_user_by_id`` and binds the result to
+``flask.g.user`` for the duration of the sync. Because resolution is by id,
+a username change between enqueue and execution has no effect on which user
+record the task acts as.
+
+That identity is not just used for logging: ``Database._get_sqla_engine``
+reads ``g.user.id`` to look up a per-user OAuth2 access token, and, for
+databases with ``impersonate_user`` enabled, ``Database.get_effective_user``
+reads ``g.user.username`` (via ``get_username()``) to pick the identity the
+outgoing connection impersonates at the external database. These tests pin
+down both halves of that chain: the id-based resolution in the task, and the
+fact that the resolved user is what a privileged, identity-sensitive
+codepath consumes downstream.
+"""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock
+
+from flask import g
+from pytest_mock import MockerFixture
+
+from superset.commands.database.sync_permissions import (
+    sync_database_permissions_task,
+)
+from superset.models.core import Database
+
+
+def test_task_binds_g_user_to_whoever_held_the_id_at_enqueue_time(
+    mocker: MockerFixture,
+) -> None:
+    """
+    The task resolves its acting identity from the user id captured at
+    enqueue time, via ``security_manager.get_user_by_id``. A username change
+    that happens between enqueue and execution has no effect on which user
+    record the task binds ``g.user`` to, because the id -- not the mutable
+    username -- is what crosses the enqueue/execute boundary.
+    """
+    # Whoever enqueued the task saw this identity at enqueue time. Its id is
+    # what's passed to the task.
+    enqueuing_user = MagicMock()
+    enqueuing_user.id = 101
+    enqueuing_user.username = "alice"
+
+    get_user_mock = mocker.patch(
+        "superset.commands.database.sync_permissions.security_manager.get_user_by_id",
+        return_value=enqueuing_user,
+    )
+
+    mock_db_connection = MagicMock()
+    mocker.patch(
+        "superset.commands.database.sync_permissions.DatabaseDAO.find_by_id",
+        return_value=mock_db_connection,
+    )
+
+    observed_g_user: list[MagicMock] = []
+
+    def capture_g_user(self: object) -> None:
+        # Read g.user at the moment the sync logic actually runs, the same
+        # way privileged downstream code (e.g. _get_sqla_engine) would.
+        observed_g_user.append(g.user)
+
+    mocker.patch(
+        "superset.commands.database.sync_permissions.SyncPermissionsCommand"
+        ".sync_database_permissions",
+        autospec=True,
+        side_effect=capture_g_user,
+    )
+
+    # By the time the task executes, "alice" has been renamed (and the
+    # username could even have been reassigned to someone else) -- but the
+    # task was enqueued with id 101, so the rename doesn't affect resolution.
+    enqueuing_user.username = "alice_renamed"
+
+    sync_database_permissions_task(1, 101, "old_db_name")
+
+    # Resolution happened purely off the immutable id...
+    get_user_mock.assert_called_once_with(101)
+    # ...and the sync ran under the same user captured at enqueue time,
+    # regardless of the username change in between.
+    assert observed_g_user == [enqueuing_user]
+    assert observed_g_user[0].id == 101
+
+
+def test_g_user_bound_by_the_task_drives_external_db_impersonation_identity(
+    mocker: MockerFixture,
+) -> None:
+    """
+    ``Database.get_effective_user`` -- consulted by ``_get_sqla_engine`` to
+    decide which identity an outgoing, ``impersonate_user``-enabled
+    connection impersonates at the external database -- reads
+    ``g.user.username``. Whatever user object the task bound to ``g.user``
+    (per the previous test, the user resolved from the id captured at
+    enqueue time) is therefore the identity used to connect to the external
+    database.
+    """
+    database = MagicMock(spec=Database)
+    database.impersonate_user = True
+
+    object_url = MagicMock()
+    object_url.username = "url-embedded-user"
+
+    # ``get_effective_user`` calls ``get_username()``, which reads
+    # ``g.user.username`` using the ``g`` imported into
+    # ``superset.utils.core`` (where ``get_username`` is defined) -- patch
+    # that module's ``g``, matching what the running task actually touches.
+    user_a = MagicMock()
+    user_a.username = "user_a"
+    mocker.patch("superset.utils.core.g", MagicMock(user=user_a))
+    assert Database.get_effective_user(database, object_url) == "user_a"
+
+    # A different user bound to g.user (as would happen if a different id
+    # had been captured at enqueue time) changes the impersonated identity
+    # for the exact same database configuration and target URL.
+    user_b = MagicMock()
+    user_b.username = "user_b"
+    mocker.patch("superset.utils.core.g", MagicMock(user=user_b))
+    assert Database.get_effective_user(database, object_url) == "user_b"

tests/unit_tests/commands/databases/sync_permissions_test.py

@@ -100,7 +100,7 @@ def test_sync_permissions_command_async_mode(
         "superset.commands.database.sync_permissions.DatabaseDAO"
     )
     mock_database_dao.find_by_id.return_value = database_with_catalog
-    mocker.patch(
+    mock_user = mocker.patch(
         "superset.commands.database.sync_permissions.security_manager.get_user_by_username"
     )
     async_task_mock = mocker.patch(
@@ -110,7 +110,7 @@ def test_sync_permissions_command_async_mode(
 
     cmmd = SyncPermissionsCommand(1, "admin")
     cmmd.run()
-    async_task_mock.delay.assert_called_once_with(1, "admin", "my_db")
+    async_task_mock.delay.assert_called_once_with(1, mock_user.return_value.id, "my_db")
 
 
 @with_config({"SYNC_DB_PERMISSIONS_IN_ASYNC_MODE": False})
@@ -219,7 +219,7 @@ def test_sync_permissions_command_async_mode_new_db_name(
     Test ``SyncPermissionsCommand`` in async mode when the
     database name changed.
     """
-    mocker.patch(
+    mock_user = mocker.patch(
         "superset.commands.database.sync_permissions.security_manager.get_user_by_username"
     )
     async_task_mock = mocker.patch(
@@ -233,7 +233,9 @@ def test_sync_permissions_command_async_mode_new_db_name(
     )
     cmmd.run()
 
-    async_task_mock.delay.assert_called_once_with(1, "admin", "Old Name")
+    async_task_mock.delay.assert_called_once_with(
+        1, mock_user.return_value.id, "Old Name"
+    )
 
 
 def test_sync_permissions_command_get_catalogs(database_with_catalog: MagicMock):

tests/unit_tests/commands/databases/update_test.py

@@ -122,11 +122,11 @@ def test_update_sync_perms_in_async_mode(
         "superset.commands.database.sync_permissions.sync_database_permissions_task.delay"
     )
     mocker.patch("superset.commands.database.update.get_username", return_value="admin")
-    mocker.patch("superset.security_manager.get_user_by_username")
+    mock_user = mocker.patch("superset.security_manager.get_user_by_username")
 
     UpdateDatabaseCommand(1, {}).run()
 
-    sync_task.assert_called_once_with(1, "admin", "my_db")
+    sync_task.assert_called_once_with(1, mock_user.return_value.id, "my_db")
 
 
 def test_update_without_catalog(

tests/unit_tests/commands/databases/validate_sql_test.py

@@ -44,6 +44,11 @@ def mock_database(mocker: MockerFixture) -> MagicMock:
         "superset.commands.database.validate_sql.DatabaseDAO"
     )
     DatabaseDAO.find_by_id.return_value = database
+    # Access validation runs before template processing; it has its own
+    # coverage, so keep it a no-op here.
+    mocker.patch(
+        "superset.commands.database.validate_sql.security_manager.raise_for_access"
+    )
     return database
 
 

tests/unit_tests/commands/sql_lab/streaming_export_command_test.py

@@ -20,6 +20,7 @@ from decimal import Decimal
 from unittest.mock import MagicMock, Mock, patch
 
 import pytest
+from flask import g
 from pytest_mock import MockerFixture
 
 from superset.commands.sql_lab.streaming_export_command import (
@@ -28,6 +29,7 @@ from superset.commands.sql_lab.streaming_export_command import (
 from superset.errors import SupersetErrorType
 from superset.exceptions import SupersetErrorException, SupersetSecurityException
 from superset.sqllab.limiting_factor import LimitingFactor
+from superset.utils.core import get_username
 
 
 def _setup_sqllab_mocks(
@@ -61,6 +63,7 @@ def mock_query():
     query.database = MagicMock()
     query.database.db_engine_spec = MagicMock()
     query.database.db_engine_spec.engine = "postgresql"
+    query.database.mutate_sql_based_on_config.side_effect = lambda sql_, **kwargs: sql_
     query.raise_for_access = MagicMock()
     return query
 
@@ -771,3 +774,120 @@ def test_csv_export_config_custom_decimal_for_decimal_type(mocker, mock_query) -
     assert "2;56,78" in csv_data
     assert "12.34" not in csv_data
     assert "56.78" not in csv_data
+
+
+def test_streaming_export_applies_sql_mutator(mocker, mock_query, mock_result_proxy):
+    """
+    Regression for apache/superset#40465: the non-streaming SQL execution
+    path (Database._execute_sql_with_mutation_and_logging, used by get_df())
+    calls Database.mutate_sql_based_on_config on every statement before
+    executing it -- that's the hook a deployment's SQL_QUERY_MUTATOR runs
+    through (e.g. to strip a trailing semicolon some engines' HTTP endpoints
+    reject, per the issue's Trino repro). _execute_query_and_stream sends
+    the raw SQL straight to engine.connect().execute(text(sql)) instead,
+    bypassing it entirely.
+    """
+    mock_query.select_sql = None
+    mock_query.executed_sql = "SELECT * FROM test_table;"
+    mock_query.limiting_factor = LimitingFactor.NOT_LIMITED
+
+    mock_db, mock_session = _setup_sqllab_mocks(mocker, mock_query)
+    mock_query.database.mutate_sql_based_on_config.side_effect = (
+        lambda sql_, is_split=False: sql_.rstrip(";")
+    )
+
+    mock_connection = MagicMock()
+    mock_connection.execution_options.return_value.execute.return_value = (
+        mock_result_proxy
+    )
+    mock_connection.__enter__.return_value = mock_connection
+    mock_connection.__exit__.return_value = None
+
+    mock_engine = MagicMock()
+    mock_engine.connect.return_value = mock_connection
+    mock_query.database.get_sqla_engine.return_value.__enter__.return_value = (
+        mock_engine
+    )
+
+    command = StreamingSqlResultExportCommand("test_client_123", chunk_size=10)
+    command.validate()
+
+    csv_generator_callable = command.run()
+    list(csv_generator_callable())
+
+    executed_sql = str(
+        mock_connection.execution_options.return_value.execute.call_args[0][0]
+    )
+    assert not executed_sql.rstrip().endswith(";"), (
+        "streaming export sent the raw, unmutated SQL to the engine -- "
+        "Database.mutate_sql_based_on_config was never called on it"
+    )
+    # Pin the contract the fix relies on: mutate_sql_based_on_config is
+    # called exactly once, with is_split=True -- the complement of the
+    # is_split=False mutation applied upstream. Without this, the assertion
+    # above still passes with is_split=False (rstrip is idempotent).
+    mock_query.database.mutate_sql_based_on_config.assert_called_once_with(
+        "SELECT * FROM test_table;", is_split=True
+    )
+
+
+def test_streaming_export_preserves_impersonated_user_context(
+    app_context, mocker, mock_query, mock_result_proxy
+):
+    """
+    Checks the other half of apache/superset#40465's theory: that the
+    streaming path drops user impersonation because it acquires its engine
+    "without this context" (unlike other call sites, which the issue says
+    use a get_sqla_engine_with_context(user_name=...) that doesn't actually
+    exist anywhere in this codebase). Independently verified this does NOT
+    reproduce: BaseStreamingCSVExportCommand.run() captures flask.g's full
+    __dict__ (including g.user) before entering the deferred generator, and
+    csv_generator() restores it via preserve_g_context before calling
+    _execute_query_and_stream -- so Database.get_sqla_engine's internal
+    get_effective_user()/get_username() call (which every engine-acquisition
+    call site relies on for impersonation, streaming or not) sees the same
+    acting user it would anywhere else. This pins that behavior rather than
+    the bug the issue reports for it.
+    """
+    g.user = Mock(username="alice")
+    assert get_username() == "alice"
+
+    mock_query.select_sql = None
+    mock_query.executed_sql = "SELECT * FROM test_table"
+    mock_query.limiting_factor = LimitingFactor.NOT_LIMITED
+
+    mock_db, mock_session = _setup_sqllab_mocks(mocker, mock_query)
+
+    mock_connection = MagicMock()
+    mock_connection.execution_options.return_value.execute.return_value = (
+        mock_result_proxy
+    )
+    mock_connection.__enter__.return_value = mock_connection
+    mock_connection.__exit__.return_value = None
+
+    mock_engine = MagicMock()
+    mock_engine.connect.return_value = mock_connection
+
+    seen_usernames = []
+
+    def fake_get_sqla_engine(*args, **kwargs):
+        # get_sqla_engine is where Database._get_sqla_engine would call
+        # get_effective_user() -> get_username() -> g.user.username in the
+        # real (unmocked) implementation; capture what g.user resolves to
+        # at the moment this is invoked, from inside the generator's
+        # restored app/g context.
+        seen_usernames.append(get_username())
+        cm = MagicMock()
+        cm.__enter__.return_value = mock_engine
+        cm.__exit__.return_value = None
+        return cm
+
+    mock_query.database.get_sqla_engine.side_effect = fake_get_sqla_engine
+
+    command = StreamingSqlResultExportCommand("test_client_123", chunk_size=10)
+    command.validate()
+
+    csv_generator_callable = command.run()
+    list(csv_generator_callable())
+
+    assert seen_usernames == ["alice"]

tests/unit_tests/commands/sql_lab/test_estimate.py

@@ -57,13 +57,13 @@ def _security_exception() -> SupersetSecurityException:
 
 
 @patch("superset.commands.sql_lab.estimate.security_manager", new_callable=MagicMock)
-@patch("superset.commands.sql_lab.estimate.db")
+@patch("superset.commands.sql_lab.estimate.DatabaseDAO")
 def test_validate_raises_when_database_not_found(
-    mock_db: MagicMock,
+    mock_dao: MagicMock,
     mock_security_manager: MagicMock,
 ) -> None:
     """404 is raised before the access check when the database does not exist."""
-    mock_db.session.query.return_value.get.return_value = None
+    mock_dao.find_by_id.return_value = None
 
     command = QueryEstimationCommand(_make_params())
     with pytest.raises(SupersetErrorException) as exc_info:
@@ -79,23 +79,21 @@ def test_validate_raises_when_database_not_found(
 
 
 @patch("superset.commands.sql_lab.estimate.security_manager", new_callable=MagicMock)
-@patch("superset.commands.sql_lab.estimate.db")
+@patch("superset.commands.sql_lab.estimate.DatabaseDAO")
 def test_validate_raises_when_database_access_denied(
-    mock_db: MagicMock,
+    mock_dao: MagicMock,
     mock_security_manager: MagicMock,
 ) -> None:
     """SupersetSecurityException propagates when raise_for_access denies access."""
     mock_database = MagicMock()
-    mock_db.session.query.return_value.get.return_value = mock_database
+    mock_dao.find_by_id.return_value = mock_database
     mock_security_manager.raise_for_access.side_effect = _security_exception()
 
     command = QueryEstimationCommand(_make_params())
     with pytest.raises(SupersetSecurityException):
         command.validate()
 
-    mock_security_manager.raise_for_access.assert_called_once_with(
-        database=mock_database
-    )
+    mock_security_manager.raise_for_access.assert_called_once()
 
 
 # ---------------------------------------------------------------------------
@@ -104,22 +102,21 @@ def test_validate_raises_when_database_access_denied(
 
 
 @patch("superset.commands.sql_lab.estimate.security_manager", new_callable=MagicMock)
-@patch("superset.commands.sql_lab.estimate.db")
+@patch("superset.commands.sql_lab.estimate.DatabaseDAO")
 def test_validate_succeeds_for_authorised_user(
-    mock_db: MagicMock,
+    mock_dao: MagicMock,
     mock_security_manager: MagicMock,
 ) -> None:
     """validate() completes without error when access is granted."""
     mock_database = MagicMock()
-    mock_db.session.query.return_value.get.return_value = mock_database
+    mock_dao.find_by_id.return_value = mock_database
     mock_security_manager.raise_for_access.return_value = None
 
     command = QueryEstimationCommand(_make_params())
     command.validate()  # must not raise
 
-    mock_security_manager.raise_for_access.assert_called_once_with(
-        database=mock_database
-    )
+    call_kwargs = mock_security_manager.raise_for_access.call_args.kwargs
+    assert call_kwargs["database"] is mock_database
 
 
 # ---------------------------------------------------------------------------
@@ -128,15 +125,15 @@ def test_validate_succeeds_for_authorised_user(
 
 
 @patch("superset.commands.sql_lab.estimate.security_manager", new_callable=MagicMock)
-@patch("superset.commands.sql_lab.estimate.db")
+@patch("superset.commands.sql_lab.estimate.DatabaseDAO")
 def test_raise_for_access_called_with_correct_database(
-    mock_db: MagicMock,
+    mock_dao: MagicMock,
     mock_security_manager: MagicMock,
 ) -> None:
     """The database object fetched from the session is passed to raise_for_access."""
     mock_database = MagicMock()
     mock_database.id = 42
-    mock_db.session.query.return_value.get.return_value = mock_database
+    mock_dao.find_by_id.return_value = mock_database
     mock_security_manager.raise_for_access.return_value = None
 
     command = QueryEstimationCommand(_make_params(database_id=42))
@@ -146,6 +143,39 @@ def test_raise_for_access_called_with_correct_database(
     assert call_kwargs["database"] is mock_database
 
 
+# ---------------------------------------------------------------------------
+# Regression: the SQL to be estimated must be authorized, not just the handle
+# ---------------------------------------------------------------------------
+
+
+@patch("superset.commands.sql_lab.estimate.security_manager", new_callable=MagicMock)
+@patch("superset.commands.sql_lab.estimate.DatabaseDAO")
+def test_validate_authorizes_the_sql_to_be_estimated(
+    mock_dao: MagicMock,
+    mock_security_manager: MagicMock,
+) -> None:
+    """
+    ``raise_for_access`` must receive the SQL so table-level authorization
+    runs; a bare ``database=`` argument matches no branch and checks nothing.
+    """
+    mock_database = MagicMock()
+    mock_dao.find_by_id.return_value = mock_database
+
+    command = QueryEstimationCommand(
+        _make_params(sql="SELECT * FROM secret_table", schema="main")
+    )
+    command.validate()
+
+    mock_security_manager.raise_for_access.assert_called_once_with(
+        database=mock_database,
+        sql="SELECT * FROM secret_table",
+        catalog=None,
+        schema="main",
+        template_params={},
+        force_dataset_match=True,
+    )
+
+
 # ---------------------------------------------------------------------------
 # SQL security controls applied on the estimate path (parity with executor)
 # ---------------------------------------------------------------------------
@@ -386,9 +416,9 @@ def test_apply_sql_security_propagates_engine_schema_gate(
 
 @patch("superset.commands.sql_lab.estimate.get_template_processor")
 @patch("superset.commands.sql_lab.estimate.security_manager", new_callable=MagicMock)
-@patch("superset.commands.sql_lab.estimate.db")
+@patch("superset.commands.sql_lab.estimate.DatabaseDAO")
 def test_run_wraps_raw_jinja_undefined_error(
-    mock_db: MagicMock,
+    mock_dao: MagicMock,
     mock_security_manager: MagicMock,
     mock_get_template_processor: MagicMock,
 ) -> None:
@@ -401,7 +431,7 @@ def test_run_wraps_raw_jinja_undefined_error(
     from jinja2.exceptions import UndefinedError
 
     mock_database = MagicMock()
-    mock_db.session.query.return_value.get.return_value = mock_database
+    mock_dao.find_by_id.return_value = mock_database
     mock_security_manager.raise_for_access.return_value = None
     mock_get_template_processor.return_value.process_template.side_effect = (
         UndefinedError("'foo' is undefined")

tests/unit_tests/connectors/sqla/models_test.py

@@ -1680,3 +1680,46 @@ def test_has_extra_cache_key_calls_scans_guest_token_rls(
 
     get_guest_rls.return_value = [{"clause": "tenant = 'acme'"}]
     assert table.has_extra_cache_key_calls(query_obj) is False
+
+
+def test_dttm_cols_excludes_column_after_temporal_flag_removed(
+    session: Session,
+) -> None:
+    """
+    Regression for #30510: when a column is mistakenly marked temporal, set as the
+    dataset's default datetime (``main_dttm_col``) and saved, then later has its
+    ``is_dttm`` flag removed, the dataset must stop treating that column as temporal.
+
+    Otherwise ``dttm_cols`` (which feeds time-column selection and the default time
+    filter for every chart built on the dataset) keeps returning a non-temporal
+    column, corrupting the dataset with a time filter that cannot be removed.
+    """
+    Database.metadata.create_all(session.bind)
+    database = Database(database_name="my_db", sqlalchemy_uri="sqlite://")
+
+    # A column the user mistakenly marks as temporal ("Is Temporal") and then picks
+    # as the dataset "Default Datetime" (``main_dttm_col``).
+    column = TableColumn(column_name="not_really_a_date", type="VARCHAR", is_dttm=True)
+    dataset = SqlaTable(
+        database=database,
+        table_name="my_table",
+        columns=[column],
+        main_dttm_col="not_really_a_date",
+    )
+    session.add(dataset)
+    session.commit()
+
+    # While flagged temporal, the column is (expectedly) exposed as a datetime column.
+    assert dataset.dttm_cols == ["not_really_a_date"]
+
+    # The user realizes the mistake and unchecks "Is Temporal", then saves. Persisting
+    # the update clears ``is_dttm`` on the column.
+    column.is_dttm = False
+    session.commit()
+
+    # The column is no longer temporal...
+    assert column.is_temporal is False
+    # ...so it must no longer be reported as a datetime column. On master
+    # ``main_dttm_col`` is never cleared, so ``dttm_cols`` still contains the stale,
+    # non-temporal column and this assertion fails (bug reproduced).
+    assert "not_really_a_date" not in dataset.dttm_cols

tests/unit_tests/db_engine_specs/test_postgres.py

@@ -182,6 +182,27 @@ SELECT * FROM some_table;
     )
 
 
+def test_get_default_schema_for_query_set_config(mocker: MockerFixture) -> None:
+    """
+    A ``set_config('search_path', ...)`` call rebinds unqualified-name
+    resolution on the shared cursor just like ``SET search_path``, so it
+    must be rejected too.
+    """
+    database = mocker.MagicMock()
+    query = mocker.MagicMock()
+    query.schema = "foo"
+    query.sql = (
+        "SELECT set_config('search_path', 'tenant_b', false); SELECT * FROM orders"
+    )
+
+    with pytest.raises(SupersetSecurityException) as excinfo:
+        spec.get_default_schema_for_query(database, query)
+    assert (
+        str(excinfo.value)
+        == "Users are not allowed to set a search path for security reasons."
+    )
+
+
 def test_adjust_engine_params() -> None:
     """
     Test `adjust_engine_params`.

tests/unit_tests/mcp_service/chart/tool/test_list_charts.py

@@ -20,6 +20,7 @@ Tests for the list_charts request schema
 """
 
 import importlib
+from copy import copy
 from unittest.mock import Mock, patch
 
 import pytest
@@ -147,6 +148,12 @@ class TestListChartsRequestSchema:
         assert request.page == 2
         assert request.page_size == 50
 
+    @pytest.mark.parametrize("value", ["true", "false", 0, 1])
+    def test_certified_requires_json_boolean(self, value):
+        """Reject values that Pydantic's non-strict bool would coerce."""
+        with pytest.raises(ValueError, match="valid boolean"):
+            ListChartsRequest(certified=value)
+
     def test_invalid_order_direction(self):
         """Test that invalid order direction raises validation error."""
         with pytest.raises(ValueError, match="Input should be 'asc' or 'desc'"):
@@ -354,3 +361,61 @@ async def test_list_charts_no_arguments(mock_list, mcp_server):
         result = await client.call_tool("list_charts", {})
     data = json.loads(result.content[0].text)
     assert "charts" in data
+
+
+@patch("superset.daos.chart.ChartDAO.list")
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    ("certified", "expected_names"),
+    [
+        (True, ["Certified"]),
+        (False, ["Uncertified"]),
+        (None, ["Certified", "Uncertified"]),
+    ],
+)
+async def test_list_charts_certified_filter(
+    mock_list, mcp_server, mock_chart, certified, expected_names
+):
+    """Certification is opt-in and supports certified, uncertified, and all."""
+    certified_chart = copy(mock_chart)
+    certified_chart.id = 1
+    certified_chart.slice_name = "Certified"
+    certified_chart.certified_by = "Data Governance"
+    certified_chart.deleted_at = None
+    uncertified_chart = mock_chart
+    uncertified_chart.id = 2
+    uncertified_chart.slice_name = "Uncertified"
+    uncertified_chart.deleted_at = None
+    charts = [certified_chart, uncertified_chart]
+
+    def list_side_effect(**kwargs):
+        custom_filter = (kwargs.get("custom_filters") or {}).get("certified")
+        if custom_filter is None:
+            selected = charts
+        else:
+            query = Mock()
+            custom_filter.apply(query, None)
+            predicate = str(query.filter.call_args.args[0])
+            expected_predicate = (
+                "slices.certified_by IS NOT NULL"
+                if certified
+                else "slices.certified_by IS NULL"
+            )
+            assert expected_predicate in predicate
+            selected = [charts[0] if certified else charts[1]]
+        return selected, len(selected)
+
+    mock_list.side_effect = list_side_effect
+    request = ListChartsRequest(certified=certified)
+    async with Client(mcp_server) as client:
+        result = await client.call_tool(
+            "list_charts", {"request": request.model_dump()}
+        )
+
+    data = json.loads(result.content[0].text)
+    actual_names = [chart["slice_name"] for chart in data["charts"]]
+    assert len(actual_names) == len(expected_names)
+    assert all(
+        expected in actual
+        for expected, actual in zip(expected_names, actual_names, strict=False)
+    )

tests/unit_tests/mcp_service/dataset/tool/test_dataset_tools.py

@@ -53,6 +53,13 @@ get_dataset_info_module = importlib.import_module(
 )
 
 
+@pytest.mark.parametrize("value", ["true", "false", 0, 1])
+def test_list_datasets_certified_requires_json_boolean(value):
+    """Reject values that Pydantic's non-strict bool would coerce."""
+    with pytest.raises(ValueError, match="valid boolean"):
+        ListDatasetsRequest(certified=value)
+
+
 def _wrapped(value: str) -> str:
     return f"{LLM_CONTEXT_OPEN_DELIMITER}\n{value}\n{LLM_CONTEXT_CLOSE_DELIMITER}"
 
@@ -183,7 +190,7 @@ def mock_auth():
 
 
 @pytest.fixture(autouse=True)
-def allow_data_model_metadata():
+def allow_data_model_metadata():  # noqa: PT004
     """Keep dataset tests in the normal metadata-allowed path by default."""
     with (
         patch.object(
@@ -314,6 +321,52 @@ async def test_list_datasets_basic(mock_list, mcp_server):
         assert "changed_on_humanized" in data["columns_loaded"]
 
 
+@patch("superset.daos.dataset.DatasetDAO.list")
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    ("certified", "expected_names"),
+    [
+        (True, ["Certified"]),
+        (False, ["Uncertified"]),
+        (None, ["Certified", "Uncertified"]),
+    ],
+)
+async def test_list_datasets_certified_filter(
+    mock_list, mcp_server, certified, expected_names
+):
+    """Certification is opt-in and supports certified, uncertified, and all."""
+    certified_dataset = create_mock_dataset(1, "Certified")
+    certified_dataset.extra = '{"certification": {"certified_by": "Governance"}}'
+    uncertified_dataset = create_mock_dataset(2, "Uncertified")
+    datasets = [certified_dataset, uncertified_dataset]
+
+    def list_side_effect(**kwargs):
+        custom_filter = (kwargs.get("custom_filters") or {}).get("certified")
+        if custom_filter is None:
+            selected = datasets
+        else:
+            query = MagicMock()
+            custom_filter.apply(query, None)
+            predicate = str(query.filter.call_args.args[0])
+            if certified:
+                assert "lower(tables.extra) LIKE lower" in predicate
+            else:
+                assert "tables.extra NOT LIKE" in predicate
+                assert "tables.extra IS NULL" in predicate
+            selected = [datasets[0] if certified else datasets[1]]
+        return selected, len(selected)
+
+    mock_list.side_effect = list_side_effect
+    request = ListDatasetsRequest(certified=certified)
+    async with Client(mcp_server) as client:
+        result = await client.call_tool(
+            "list_datasets", {"request": request.model_dump()}
+        )
+
+    data = json.loads(result.content[0].text)
+    assert [dataset["table_name"] for dataset in data["datasets"]] == expected_names
+
+
 @patch("superset.daos.dataset.DatasetDAO.list")
 @pytest.mark.asyncio
 async def test_list_datasets_custom_uuid_columns(mock_list, mcp_server):

tests/unit_tests/queries/query_object_test.py

@@ -21,6 +21,7 @@ from flask_appbuilder.security.sqla.models import User
 from superset.common.query_object import QueryObject
 from superset.connectors.sqla.models import SqlaTable
 from superset.models.core import Database
+from superset.superset_typing import Metric
 from superset.utils.core import override_user
 
 
@@ -86,6 +87,72 @@ def test_cache_key_changes_for_new_query_object_different_params():
     assert query_object2.cache_key() != cache_key1
 
 
+def test_cache_key_stable_regardless_of_extra_cache_keys_order():
+    """
+    Regression for #34543: the cache key must not depend on the order of
+    ``extra_cache_keys``.
+
+    ``SqlaTable.get_extra_cache_keys`` (superset/connectors/sqla/models.py)
+    returns ``list(set(extra_cache_keys))``. Python's string hashing is
+    randomized per-process (``PYTHONHASHSEED``), so the same set of values
+    can iterate in a different order in the Celery worker process (which
+    writes the query results to cache) than in the web process (which
+    re-derives the cache key to read them back). Because ``hash_from_dict``
+    only sorts dict keys and not list values, two ``extra_cache_keys`` lists
+    with identical Jinja ``url_param()`` values but different order hash to
+    different cache keys, causing async chart-data lookups to 422 with
+    "Error loading data from cache" whenever more than one url_param is
+    referenced (a single-element list has only one possible order, which is
+    why the bug is only visible with multiple parameters).
+    """
+    query_object1 = QueryObject(row_limit=1)
+    query_object2 = QueryObject(row_limit=1)
+    same_values_different_order = ["CAR_IDS=1,2,3", "CHASSIS_IDS=100,200"]
+    cache_key1 = query_object1.cache_key(extra_cache_keys=same_values_different_order)
+    cache_key2 = query_object2.cache_key(
+        extra_cache_keys=list(reversed(same_values_different_order))
+    )
+    assert cache_key1 == cache_key2
+
+
+def test_cache_key_stable_for_mixed_type_extra_cache_keys():
+    """
+    ``extra_cache_keys`` values are typed as ``Hashable``, so a mix of
+    strings and non-strings that stringify identically (e.g. ``1`` and
+    ``"1"``) can appear together. Sorting on a bare ``str()`` value treats
+    those as equal keys, so Python's stable sort would fall back to
+    whatever order they arrived in from ``list(set(...))`` -- which is not
+    deterministic across processes. The sort key must also account for
+    type so ordering doesn't silently regress to that non-determinism.
+    """
+    query_object1 = QueryObject(row_limit=1)
+    query_object2 = QueryObject(row_limit=1)
+    mixed_values = ["CAR_IDS=1,2,3", 1, "1", None]
+    cache_key1 = query_object1.cache_key(extra_cache_keys=mixed_values)
+    cache_key2 = query_object2.cache_key(extra_cache_keys=list(reversed(mixed_values)))
+    assert cache_key1 == cache_key2
+
+
+def test_cache_key_sensitive_to_orderby_order():
+    """
+    Negative control for the ``extra_cache_keys`` fix above: unlike that
+    field, ``orderby`` is order-significant (it determines sort direction
+    of the executed SQL), so the cache key must still change when the
+    order of its entries changes. This guards against a fix that
+    canonicalizes list values generically instead of targeting
+    ``extra_cache_keys`` specifically.
+    """
+    metric_a: Metric = "count"
+    metric_b: Metric = "sum__value"
+    query_object1 = QueryObject(
+        row_limit=1, orderby=[(metric_a, True), (metric_b, False)]
+    )
+    query_object2 = QueryObject(
+        row_limit=1, orderby=[(metric_b, False), (metric_a, True)]
+    )
+    assert query_object1.cache_key() != query_object2.cache_key()
+
+
 def test_cache_key_changes_for_new_query_object_same_params():
     """
     When a new query object is created with the same params,

tests/unit_tests/sql/parse_tests.py

@@ -408,10 +408,29 @@ def test_extract_tables_illdefined() -> None:
 def test_extract_tables_show_tables_from() -> None:
     """
     Test `SHOW TABLES FROM`.
+
+    No individual table target is extractable, so the statement must be
+    flagged as unparseable for authorization purposes instead of passing
+    strict scoping with an empty table set.
     """
     assert (
         extract_tables_from_sql("SHOW TABLES FROM s1 like '%order%'", "mysql") == set()
     )
+    assert SQLScript(
+        "SHOW TABLES FROM s1 like '%order%'", "mysql"
+    ).has_unparseable_statement
+
+
+def test_extract_tables_show_create_table() -> None:
+    """
+    Test `SHOW CREATE TABLE`.
+
+    The target table must enter table-level authorization.
+    """
+    assert extract_tables_from_sql("SHOW CREATE TABLE s1.t1", "mysql") == {
+        Table("t1", "s1")
+    }
+    assert not SQLScript("SHOW CREATE TABLE s1.t1", "mysql").has_unparseable_statement
 
 
 def test_format_show_tables() -> None:
@@ -761,30 +780,123 @@ SELECT c FROM z
 
 
 def test_extract_tables_reusing_aliases() -> None:
+    """Test that the parser follows aliases.
+
+    A non-recursive ``WITH`` item sees only items declared before it, so a forward
+    reference resolves to the table of that name -- a real read that must be extracted.
     """
-    Test that the parser follows aliases.
-    """
+    # `q1` first: the `q2` in its body, and `q2`'s `src`, are both tables.
     assert extract_tables_from_sql(
         """
 with q1 as ( select key from q2 where key = '5'),
 q2 as ( select key from src where key = '5')
 select * from (select key from q1) a
 """
-    ) == {Table("src")}
+    ) == {Table("q2"), Table("src")}
 
-    # weird query with circular dependency
-    assert (
-        extract_tables_from_sql(
-            """
+    # `src` first: its `q2` is a table; `q2`'s `src` and the outer `src` are the CTE.
+    assert extract_tables_from_sql(
+        """
 with src as ( select key from q2 where key = '5'),
 q2 as ( select key from src where key = '5')
 select * from (select key from src) a
 """
+    ) == {Table("q2")}
+
+
+def test_extract_tables_cte_name_shared_with_table() -> None:
+    """Test that a CTE's name does not hide reads of the table it is named after.
+
+    Only a reference resolving to the CTE may be excluded; dropping any other costs it
+    both its row filter and its access check.
+    """
+    # A qualified reference -- in the CTE body or elsewhere -- is the table.
+    assert extract_tables_from_sql(
+        "WITH orders AS (SELECT * FROM public.orders) SELECT * FROM orders"
+    ) == {Table("orders", "public")}
+    assert extract_tables_from_sql(
+        "WITH orders AS (SELECT 1 AS d) "
+        "SELECT * FROM (SELECT * FROM public.orders) AS z"
+    ) == {Table("orders", "public")}
+
+    # A non-recursive CTE cannot see itself, so its own name in its body is the table.
+    assert extract_tables_from_sql(
+        "WITH orders AS (SELECT * FROM orders) SELECT * FROM orders"
+    ) == {Table("orders")}
+
+    # A catalog disqualifies like a schema; `cat..orders` is checked only when pivoted.
+    assert extract_tables_from_sql(
+        "WITH orders AS (SELECT 1 AS amt, 'a' AS mth) "
+        "SELECT * FROM cat..orders PIVOT(SUM(amt) FOR mth IN ('a'))",
+        engine="snowflake",
+    ) == {Table("orders", None, "cat")}
+
+
+def test_extract_tables_cte_reference_not_table() -> None:
+    """Test the counterpart: a reference that resolves to a CTE is not a table.
+
+    A recursive item's reference to itself is the shape a bare-name compare gets wrong.
+    """
+    assert (
+        extract_tables_from_sql(
+            "WITH RECURSIVE t AS ("
+            "SELECT 1 AS n UNION ALL SELECT n + 1 FROM t WHERE n < 5"
+            ") SELECT * FROM t"
         )
         == set()
     )
 
 
+def test_extract_tables_pivoted_cte_reference_is_not_a_table() -> None:
+    """Test that pivoting a CTE reference does not make it a table read.
+
+    Pivoting yields a new relation, so sqlglot keeps the reference as an ``exp.Table``
+    -- the one shape where a CTE reference reaches ``is_cte()`` unqualified.
+    """
+    assert extract_tables_from_sql(
+        "WITH c AS (SELECT a, b FROM other_table) "
+        "SELECT * FROM c PIVOT(SUM(b) FOR a IN ('p'))",
+        engine="snowflake",
+    ) == {Table("other_table")}
+    # Also when the pivot sits inside a derived table.
+    assert extract_tables_from_sql(
+        "WITH c AS (SELECT a, b FROM other_table) "
+        "SELECT * FROM (SELECT * FROM c PIVOT(SUM(b) FOR a IN ('p'))) AS z",
+        engine="snowflake",
+    ) == {Table("other_table")}
+
+
+def test_extract_tables_aliased_cte_does_not_hide_table() -> None:
+    """Test that aliasing a CTE reference does not erase a table of the same name.
+
+    ``Scope.sources`` is keyed by ``alias_or_name`` and would file the table under the
+    CTE's alias; ``cte_sources`` is keyed by CTE name only.
+    """
+    assert extract_tables_from_sql(
+        "WITH c AS (SELECT 1 AS n) SELECT s2.* FROM c AS other_table, other_table AS s2"
+    ) == {Table("other_table")}
+    assert extract_tables_from_sql(
+        "WITH c AS (SELECT 1 AS n) "
+        "SELECT s2.* FROM c AS other_table LEFT JOIN other_table AS s2 ON TRUE"
+    ) == {Table("other_table")}
+
+
+def test_extract_tables_cte_reference_over_reported() -> None:
+    """Test the two shapes that over-report a CTE reference as a table.
+
+    A spurious access check, not a missing one. Pinned so a change either way is meant.
+    """
+    # PostgreSQL resolves `foo` to the CTE; this reports the table.
+    assert extract_tables_from_sql("WITH Foo AS (SELECT 1 AS d) SELECT * FROM foo") == {
+        Table("foo")
+    }
+    # Legal under RECURSIVE: `q2` is the CTE declared below, not a table.
+    assert extract_tables_from_sql(
+        "WITH RECURSIVE q1 AS (SELECT key FROM q2), q2 AS (SELECT 1 AS key) "
+        "SELECT * FROM q1"
+    ) == {Table("q2")}
+
+
 def test_extract_tables_multistatement() -> None:
     """
     Test that the parser works with multiple statements.
@@ -1589,6 +1701,44 @@ def test_is_mutating(sql: str, engine: str, expected: bool) -> None:
     assert SQLStatement(sql, engine).is_mutating() == expected
 
 
+@pytest.mark.parametrize(
+    "sql, engine",
+    [
+        # Opaque `exp.Command` fallbacks must fail closed on every dialect,
+        # not only PostgreSQL.
+        ("CALL evil_proc()", "mysql"),
+        ("LOAD '/tmp/x.so'", "postgres"),
+        ("EXEC dbo.evil_proc", "mssql"),
+        # The EXPLAIN ANALYZE unwrap must handle the parenthesized
+        # option-list, whitespace, alternate-spelling, and leading-comment
+        # forms: PostgreSQL executes the inner DML for all of them.
+        ("EXPLAIN (ANALYZE) UPDATE t SET x = 1", "postgresql"),
+        ("EXPLAIN (ANALYZE, BUFFERS) DELETE FROM t", "postgresql"),
+        ("EXPLAIN ANALYZE\nUPDATE t SET x = 1", "postgresql"),
+        ("EXPLAIN ANALYSE UPDATE t SET x = 1", "postgresql"),
+        ("EXPLAIN /* c */ (ANALYZE) UPDATE t SET x = 1", "postgresql"),
+        # A bare COMMIT persists every prior write on the connection even
+        # when the execution layer skips its own commit call.
+        ("COMMIT", "postgresql"),
+        ("COMMIT", "mysql"),
+        # Further EXPLAIN ANALYZE edge forms: a leading line comment before
+        # the option, a VERBOSE qualifier, an empty option list, and an
+        # inner statement that cannot be parsed all fail closed as mutating.
+        ("EXPLAIN --c\nANALYZE UPDATE t SET x = 1", "postgresql"),
+        ("EXPLAIN ANALYZE VERBOSE UPDATE t SET x = 1", "postgresql"),
+        ("EXPLAIN (ANALYZE)", "postgresql"),
+        ("EXPLAIN ANALYZE )))", "postgresql"),
+    ],
+)
+def test_is_mutating_fails_closed_on_gate_blind_spots(sql: str, engine: str) -> None:
+    """
+    `is_mutating` must fail closed on statements that slip past node-type
+    matching: non-PostgreSQL command fallbacks, normalized `EXPLAIN ANALYZE`
+    variants, and structured `COMMIT`.
+    """
+    assert SQLStatement(sql, engine).is_mutating()
+
+
 @pytest.mark.parametrize(
     "sql, expected",
     [
@@ -2330,6 +2480,52 @@ def test_set_limit_value(
     assert statement.format() == expected
 
 
+@pytest.mark.parametrize(
+    "engine",
+    [
+        # Engines whose sqlglot dialect parses `SHOW` into a real `exp.Show`
+        # node (as opposed to falling back to an opaque `exp.Command`, which
+        # doesn't expose a `limit` arg and so was never affected by this bug).
+        "starrocks",
+        "mysql",
+        "snowflake",
+    ],
+)
+@pytest.mark.parametrize(
+    "sql",
+    [
+        "SHOW TABLES",
+        "SHOW DATABASES",
+        "SHOW CREATE TABLE test.will_test1",
+    ],
+)
+def test_set_limit_value_leaves_show_statements_unchanged(
+    sql: str, engine: str
+) -> None:
+    """
+    Regression for #36939: FORCE_LIMIT must not touch ``SHOW`` statements.
+
+    ``SHOW`` statements have no `LIMIT` clause in sqlglot's expression tree,
+    so forcing one via ``args["limit"]`` doesn't reject cleanly, it produces
+    a malformed statement with two ``LIMIT`` keywords (one from a stray
+    rendering of the bare ``Limit`` expression, one from the forced value).
+    StarRocks (and presumably other engines) reject that outright: "Getting
+    syntax error ... Unexpected input 'LIMIT'". The statement should be
+    left untouched instead, matching how ``SELECT`` statements without a
+    scannable row source aren't force-limited either.
+
+    Covers multiple engines, not just StarRocks: the fix guards on the AST
+    node type (``exp.Show``), not the dialect, so any engine whose sqlglot
+    dialect parses ``SHOW`` into a real ``Show`` node (e.g. MySQL, Snowflake)
+    is equally exposed and must be equally protected.
+    """
+    statement = SQLStatement(sql, engine)
+    original = statement.format()
+    statement.set_limit_value(1000, LimitMethod.FORCE_LIMIT)
+    assert statement.format() == original
+    assert "LIMIT" not in statement.format()
+
+
 @pytest.mark.parametrize(
     "kql, limit, expected",
     [
@@ -2810,6 +3006,112 @@ FROM (
 LIMIT 100
         """.strip(),
         ),
+        (
+            'SELECT * FROM tbl_a AS "x AND 1 = 0 OR 1 = 1"',
+            {Table("tbl_a", "schema1", "catalog1"): "id = 42"},
+            """
+SELECT
+  *
+FROM (
+  SELECT
+    *
+  FROM tbl_a
+  WHERE
+    id = 42
+) AS "x AND 1 = 0 OR 1 = 1"
+            """.strip(),
+        ),
+        (
+            "SELECT c1 FROM tbl_a AS x (c1, c2)",
+            {Table("tbl_a", "schema1", "catalog1"): "id = 42"},
+            """
+SELECT
+  c1
+FROM (
+  SELECT
+    *
+  FROM tbl_a
+  WHERE
+    id = 42
+) AS x(c1, c2)
+            """.strip(),
+        ),
+        # A CTE sharing the rule's table name is not a read of it: only the real read
+        # inside the CTE body is wrapped; the CTE reference keeps its own projection.
+        (
+            "WITH some_table AS (SELECT id FROM some_table) SELECT * FROM some_table",
+            {Table("some_table", "schema1", "catalog1"): "id = 42"},
+            """
+WITH some_table AS (
+  SELECT
+    id
+  FROM (
+    SELECT
+      *
+    FROM some_table
+    WHERE
+      id = 42
+  ) AS "some_table"
+)
+SELECT
+  *
+FROM some_table
+            """.strip(),
+        ),
+        # A correlated ``LATERAL`` reaches the outer read through two scopes: wrapped
+        # once, not twice. The lateral's own read is a distinct node, wrapped in place.
+        (
+            "SELECT * FROM some_table, LATERAL ("
+            "SELECT * FROM other_table WHERE other_table.x = some_table.x) t",
+            {
+                Table("some_table", "schema1", "catalog1"): "id = 42",
+                Table("other_table", "schema1", "catalog1"): "id = 7",
+            },
+            """
+SELECT
+  *
+FROM (
+  SELECT
+    *
+  FROM some_table
+  WHERE
+    id = 42
+) AS "some_table", LATERAL (
+  SELECT
+    *
+  FROM (
+    SELECT
+      *
+    FROM other_table
+    WHERE
+      id = 7
+  ) AS "other_table"
+  WHERE
+    other_table.x = some_table.x
+) AS t
+            """.strip(),
+        ),
+        # A read in a DML statement's subquery is filtered in place, not refused: the
+        # ``UPDATE`` target is not a source, so only the ``SELECT`` read of ``t`` wraps.
+        (
+            "UPDATE dst SET x = 1 WHERE id IN (SELECT id FROM t)",
+            {Table("t", "schema1", "catalog1"): "id = 42"},
+            """
+UPDATE dst SET x = 1
+WHERE
+  id IN (
+    SELECT
+      id
+    FROM (
+      SELECT
+        *
+      FROM t
+      WHERE
+        id = 42
+    ) AS "t"
+  )
+            """.strip(),
+        ),
     ],
 )
 def test_rls_subquery_transformer(
@@ -2830,6 +3132,63 @@ def test_rls_subquery_transformer(
     assert statement.format() == expected
 
 
+@pytest.mark.parametrize(
+    "sql, read_counts",
+    [
+        ("SELECT * FROM t", {"t": 1}),
+        ("SELECT * FROM t JOIN u ON t.id = u.id", {"t": 1, "u": 1}),
+        ("SELECT * FROM t, u", {"t": 1, "u": 1}),
+        ("SELECT * FROM t WHERE id IN (SELECT id FROM u)", {"t": 1, "u": 1}),
+        # A self-join reads the table through two distinct nodes; both are wrapped.
+        ("SELECT * FROM t AS a JOIN t AS b ON a.id = b.id", {"t": 2}),
+        # The CTE body's read of ``t`` and the outer read of ``t`` are both wrapped;
+        # the CTE reference ``c`` is not a read and carries no rule.
+        (
+            "WITH c AS (SELECT id FROM t) SELECT * FROM c JOIN t AS t2 ON c.id = t2.id",
+            {"t": 2},
+        ),
+        ("SELECT * FROM (SELECT * FROM t) AS x", {"t": 1}),
+        # Pins the deepest-first ordering. The parenthesised join head ``t`` carries the
+        # join to ``u`` in its own args, so ``u`` must be wrapped before ``t``; wrapping
+        # ``t`` first would copy ``u`` into ``t``'s subquery and drop ``u``'s filter.
+        # Flipping the sort to ``reverse=False`` makes this case fail.
+        ("SELECT * FROM (t JOIN u ON t.id = u.id)", {"t": 1, "u": 1}),
+        # A correlated ``LATERAL`` reaches the outer read through two scopes; it is
+        # wrapped once, and the lateral's own read is wrapped once.
+        (
+            "SELECT * FROM some_table, LATERAL ("
+            "SELECT * FROM other_table WHERE other_table.x = some_table.x) t",
+            {"some_table": 1, "other_table": 1},
+        ),
+    ],
+)
+def test_rls_subquery_filters_every_authorized_read(
+    sql: str,
+    read_counts: dict[str, int],
+) -> None:
+    """The set the rewrite filters equals the set authorization enforces.
+
+    Each read gets a table-specific sentinel predicate; its count in the output must
+    equal that table's real-read node count, catching a dropped read or a double-wrap.
+    """
+    authorized = {t.table for t in extract_tables_from_statement(parse_one(sql), None)}
+    assert authorized == set(read_counts)
+
+    statement = SQLStatement(sql)
+    statement.apply_rls(
+        "catalog1",
+        "schema1",
+        {
+            Table(table, "schema1", "catalog1"): [parse_one(f"rls_{table} = 1")]
+            for table in read_counts
+        },
+        RLSMethod.AS_SUBQUERY,
+    )
+    output = statement.format()
+    for table, count in read_counts.items():
+        assert output.count(f"rls_{table} = 1") == count
+
+
 def test_rls_invalid_method(mocker: MockerFixture) -> None:
     """
     Test that an invalid RLS method raises an error.
@@ -3154,6 +3513,58 @@ VALUES
   (1, 2)
             """.strip(),
         ),
+        (
+            'SELECT * FROM tbl_a AS "x AND 1 = 0 OR 1 = 1"',
+            {Table("tbl_a", "schema1", "catalog1"): "id = 42"},
+            """
+SELECT
+  *
+FROM tbl_a AS "x AND 1 = 0 OR 1 = 1"
+WHERE
+  "x AND 1 = 0 OR 1 = 1".id = 42
+            """.strip(),
+        ),
+        (
+            'SELECT * FROM tbl_a AS "a.b"',
+            {Table("tbl_a", "schema1", "catalog1"): "id = 42"},
+            """
+SELECT
+  *
+FROM tbl_a AS "a.b"
+WHERE
+  "a.b".id = 42
+            """.strip(),
+        ),
+        # A column-list alias has no name (``this`` is ``None``); qualify with the table
+        # so the predicate does not resolve outward into an enclosing scope.
+        (
+            "SELECT * FROM tbl_a AS (c1, c2)",
+            {Table("tbl_a", "schema1", "catalog1"): "id = 42"},
+            """
+SELECT
+  *
+FROM tbl_a AS _t0(c1, c2)
+WHERE
+  tbl_a.id = 42
+            """.strip(),
+        ),
+        # A table heading a parenthesised join is a read, but its parent is the wrapping
+        # ``Subquery``, not a ``From``/``Join``, so the predicate method leaves it --
+        # fail-closed (the subquery method filters it). Pinned to catch a shape change.
+        (
+            "SELECT * FROM (some_table JOIN other_table "
+            "ON some_table.id = other_table.id)",
+            {Table("some_table", "schema1", "catalog1"): "id = 42"},
+            """
+SELECT
+  *
+FROM (
+  some_table
+    JOIN other_table
+      ON some_table.id = other_table.id
+)
+            """.strip(),
+        ),
     ],
 )
 def test_rls_predicate_transformer(
@@ -3481,6 +3892,7 @@ def test_sqlstatement_format_preserves_multi_arg_distinct(engine: str) -> None:
     assert "CASE WHEN" not in formatted
 
 
+@with_feature_flags(ENABLE_TEMPLATE_PROCESSING=True)
 @pytest.mark.parametrize(
     "engine",
     [
@@ -3509,12 +3921,12 @@ def test_sqlstatement_format_preserves_multi_arg_distinct(engine: str) -> None:
             {Table(table="bar", schema="foo")},
         ),
         (
-            "latest_partition('foo.%s'|format(str('bar')))",
-            set(),
+            "latest_partitions('foo.bar')",
+            {Table(table="bar", schema="foo")},
         ),
         (
-            "latest_partition('foo.{}'.format('bar'))",
-            set(),
+            "first_latest_partition('foo.bar')",
+            {Table(table="bar", schema="foo")},
         ),
     ],
 )
@@ -3533,6 +3945,42 @@ def test_extract_tables_from_jinja_sql(
     )
 
 
+@pytest.mark.parametrize(
+    "engine",
+    [
+        "hive",
+        "presto",
+        "trino",
+    ],
+)
+@pytest.mark.parametrize(
+    "macro",
+    [
+        "latest_partition('foo.%s'|format(str('bar')))",
+        "latest_partition('foo.{}'.format('bar'))",
+        "latest_partitions('foo.{}'.format('bar'))",
+        # A partition macro with the wrong number of arguments cannot be
+        # resolved to a single table, so it must also fail closed.
+        "latest_partition('foo.bar', 'extra')",
+    ],
+)
+def test_extract_tables_from_jinja_sql_fails_closed(
+    mocker: MockerFixture,
+    engine: str,
+    macro: str,
+) -> None:
+    """
+    A partition macro whose table reference cannot be evaluated statically
+    must fail closed, as the macro would otherwise execute against a table
+    that never entered the authorization check.
+    """
+    with pytest.raises(SupersetParseError):
+        process_jinja_sql(
+            sql=f"'{{{{ {engine}.{macro} }}}}'",
+            database=mocker.MagicMock(backend=engine),
+        )
+
+
 @with_feature_flags(ENABLE_TEMPLATE_PROCESSING=False)
 def test_extract_tables_from_jinja_sql_disabled(mocker: MockerFixture) -> None:
     """
@@ -3622,6 +4070,31 @@ def test_process_jinja_sql_template_params_parameter(mocker: MockerFixture) -> N
     assert result.tables == {Table("table_name")}
 
 
+@with_feature_flags(ENABLE_TEMPLATE_PROCESSING=True)
+def test_process_jinja_sql_renders_exactly_once(mocker: MockerFixture) -> None:
+    """
+    The authorization path must validate exactly the SQL that executes.
+
+    A template whose first render emits Jinja comment markers inside SQL
+    comments used to be rendered a second time, which stripped the markers
+    and everything between them from the validated SQL while the executed
+    SQL (rendered once) kept the extra statement text.
+    """
+    database = mocker.MagicMock(backend="postgresql")
+    database.db_engine_spec.engine = "postgresql"
+
+    result = process_jinja_sql(
+        sql=(
+            'SELECT * FROM granted /*{{ "{#" }}*/ '
+            'UNION SELECT * FROM restricted /*{{ "#}" }}*/'
+        ),
+        database=database,
+    )
+
+    assert Table("restricted") in result.tables
+    assert Table("granted") in result.tables
+
+
 @pytest.mark.parametrize(
     "sql, engine, expected",
     [
@@ -4164,6 +4637,60 @@ def test_changes_search_path(sql: str, expected: bool) -> None:
     assert SQLStatement(sql, "postgresql").changes_search_path() == expected
 
 
+@pytest.mark.parametrize(
+    "sql, engine, expected",
+    [
+        # `USE` rebinds the schema for every later statement on the cursor.
+        ("USE tenant_b; SELECT * FROM orders", "mysql", True),
+        ("use `tenant_b`", "mysql", True),
+        ("USE SCHEMA tenant_b", "snowflake", True),
+        # Warehouse selection changes compute, not name resolution.
+        ("USE WAREHOUSE compute_wh", "snowflake", False),
+        # Search-path changes are schema rebinds too.
+        ("SET search_path = tenant_b", "postgresql", True),
+        (
+            "SELECT set_config('search_path', 'tenant_b', false)",
+            "postgresql",
+            True,
+        ),
+        # A `set_config()` with a computed setting name fails closed.
+        (
+            "SELECT set_config('search' || '_path', 'tenant_b', false)",
+            "postgresql",
+            True,
+        ),
+        # `SET SCHEMA` is an alias for a search-path rebind on Postgres and
+        # a schema rebind on DB2-family engines.
+        ("SET SCHEMA 'tenant_b'", "postgresql", True),
+        ("SELECT * FROM orders", "mysql", False),
+        ("SET statement_timeout = 10", "postgresql", False),
+        # A structured `SET current_schema = ...` rebinds resolution through
+        # a setting rather than a search path.
+        ("SET current_schema = foo", "postgresql", True),
+        # `SET CATALOG`/`SET SCHEMA` that fall back to an opaque command are
+        # schema rebinds, including the `CURRENT` spelling; an unrelated `SET`
+        # command (e.g. `SET ROLE`) is not.
+        ("SET CATALOG tenant_b", "postgresql", True),
+        ("SET CURRENT SCHEMA foo", "postgresql", True),
+        ("SET ROLE admin", "postgresql", False),
+        # A `set_config()` whose setting name is a column reference rather than
+        # a literal is treated conservatively as a schema change.
+        ("SELECT set_config(schema_col, 'tenant_b', false)", "postgresql", True),
+        # Engines without a sqlglot AST (e.g. Kusto KQL) do not rebind schema
+        # resolution through these forms.
+        ("print x = 1", "kustokql", False),
+    ],
+)
+def test_changes_default_schema(sql: str, engine: str, expected: bool) -> None:
+    """
+    `changes_default_schema` detects statements that rebind unqualified-name
+    resolution (`USE`, `SET SCHEMA`, search-path changes) so the SQL Lab
+    authorization path can reject the script before qualifying tables with
+    the schema the user selected.
+    """
+    assert SQLScript(sql, engine).changes_default_schema() == expected
+
+
 @pytest.mark.parametrize(
     "sql, denylist, expected",
     [

tests/unit_tests/sql_lab_execution_context.py

@@ -16,8 +16,12 @@
 # under the License.
 # pylint: disable=import-outside-toplevel, invalid-name, unused-argument, too-many-locals
 
+from unittest.mock import MagicMock
+
 import pytest
 
+from superset.errors import SupersetErrorType
+from superset.exceptions import SupersetErrorException
 from superset.sql.parse import CTASMethod
 from superset.sqllab.sqllab_execution_context import (
     CreateTableAsSelect,
@@ -101,3 +105,45 @@ def test_create_table_as_select():
     assert ctas.ctas_method == CTASMethod.TABLE
     assert ctas.target_schema_name == "public"
     assert ctas.target_table_name == "temp_table"
+
+
+def test_set_database_rejects_ctas_when_database_disallows_it(query_params):
+    """
+    ``allow_ctas`` must be enforced server-side at submission: the
+    ``select_as_cta``/``ctas_method`` payload fields are client-supplied.
+    """
+    query_params["select_as_cta"] = True
+    query_params["ctas_method"] = "TABLE"
+    query_params["tmp_table_name"] = "tmp_target"
+    context = SqlJsonExecutionContext(query_params)
+
+    database = MagicMock()
+    database.allow_ctas = False
+
+    with pytest.raises(SupersetErrorException) as exc_info:
+        context.set_database(database)
+
+    assert (
+        exc_info.value.error.error_type == SupersetErrorType.QUERY_SECURITY_ACCESS_ERROR
+    )
+
+
+def test_set_database_rejects_cvas_when_database_disallows_it(query_params):
+    """
+    ``allow_cvas`` must be enforced server-side at submission, mirroring the
+    ``allow_ctas``/VIEW branch of ``_validate_ctas_is_allowed``.
+    """
+    query_params["select_as_cta"] = True
+    query_params["ctas_method"] = "VIEW"
+    query_params["tmp_table_name"] = "tmp_target"
+    context = SqlJsonExecutionContext(query_params)
+
+    database = MagicMock()
+    database.allow_cvas = False
+
+    with pytest.raises(SupersetErrorException) as exc_info:
+        context.set_database(database)
+
+    assert (
+        exc_info.value.error.error_type == SupersetErrorType.QUERY_SECURITY_ACCESS_ERROR
+    )

tests/unit_tests/sqllab/api_test.py

@@ -17,9 +17,14 @@
 from __future__ import annotations
 
 import re
+from typing import Any
 from unittest.mock import MagicMock, patch
 
 from flask import Flask
+from pytest_mock import MockerFixture
+
+from superset.errors import ErrorLevel, SupersetError, SupersetErrorType
+from superset.exceptions import SupersetSecurityException
 
 
 def _disposition_filename(form_filename: str | None) -> str:
@@ -62,3 +67,44 @@ def test_streaming_csv_falls_back_when_filename_empty() -> None:
 
     assert filename.startswith("sqllab_abc123_")
     assert filename.endswith(".csv")
+
+
+def test_format_sql_checks_access_before_rendering(
+    mocker: MockerFixture,
+    client: Any,
+    full_api_access: None,
+) -> None:
+    """
+    Access must be checked before Jinja rendering, as some Jinja macros
+    execute statements against the database upon rendering.
+    """
+    database = mocker.MagicMock()
+    database.db_engine_spec.engine = "presto"
+    mocker.patch(
+        "superset.sqllab.api.DatabaseDAO.find_by_id",
+        return_value=database,
+    )
+    get_template_processor = mocker.patch("superset.sqllab.api.get_template_processor")
+    raise_for_access = mocker.patch(
+        "superset.sqllab.api.security_manager.raise_for_access",
+        side_effect=SupersetSecurityException(
+            SupersetError(
+                error_type=SupersetErrorType.TABLE_SECURITY_ACCESS_ERROR,
+                message="You need access to the following tables: `s.t`",
+                level=ErrorLevel.ERROR,
+            )
+        ),
+    )
+
+    response = client.post(
+        "/api/v1/sqllab/format_sql/",
+        json={
+            "sql": "SELECT '{{ presto.latest_partition('s.t') }}'",
+            "database_id": 1,
+            "template_params": '{"foo": "bar"}',
+        },
+    )
+
+    assert response.status_code == 403
+    raise_for_access.assert_called_once()
+    get_template_processor.assert_not_called()

tests/unit_tests/utils/number_format_test.py

@@ -0,0 +1,562 @@
+# 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.
+from decimal import Decimal
+from typing import Any
+from unittest.mock import MagicMock, patch
+
+import pandas as pd
+import pytest
+from flask import current_app
+
+from superset.utils.number_format import (
+    format_d3,
+    format_default,
+    format_number_with_config,
+    format_numeric,
+    get_currency_locale,
+    resolve_auto_currency,
+    resolve_symbol_position,
+)
+
+# --- Helper behaviour the d3 parity matrix below cannot cover ----------------
+
+
+@pytest.mark.parametrize(
+    "d3_format,value,expected",
+    [
+        # SMART_NUMBER is a Superset formatter (adaptive SI, half-up), not raw d3
+        ("SMART_NUMBER", 4725, "4.73k"),
+        ("SMART_NUMBER", 80679663, "80.7M"),
+        ("SMART_NUMBER", 1234567890, "1.23B"),
+        ("SMART_NUMBER", 0, "0"),
+        (".2~f", 1200.0, "1200"),
+        (None, 42, "42"),
+        ("not-a-real-format!!", 42, "42"),
+    ],
+)
+def test_format_number(d3_format: str | None, value: Any, expected: Any) -> None:
+    assert format_number_with_config(d3_format, None, value) == expected
+
+
+@pytest.mark.parametrize(
+    "currency,value,expected",
+    [
+        ({"symbol": "USD", "symbolPosition": "prefix"}, 1234.5, "$ 1,234.50"),
+        ({"symbol": "EUR", "symbolPosition": "suffix"}, 1234.5, "1,234.50 €"),
+        ({"symbol": "BRL", "symbolPosition": None}, 1234.5, "R$ 1,234.50"),
+        ({"symbol": "ZZZ", "symbolPosition": None}, 1234.5, "ZZZ 1,234.50"),
+    ],
+)
+def test_format_number_with_currency(
+    currency: dict[str, Any], value: float, expected: str
+) -> None:
+    assert format_number_with_config(",.2f", currency, value) == expected
+
+
+def test_currency_defaults_to_smart_number_when_no_d3_format() -> None:
+    assert (
+        format_number_with_config(
+            None, {"symbol": "USD", "symbolPosition": "prefix"}, 1234567
+        )
+        == "$ 1.23M"
+    )
+
+
+def test_auto_currency_formats_without_symbol() -> None:
+    assert (
+        format_number_with_config(
+            ",.2f", {"symbol": "AUTO", "symbolPosition": "prefix"}, 1234.5
+        )
+        == "1,234.50"
+    )
+
+
+def test_resolve_auto_currency_uses_detected_single_currency() -> None:
+    currency = {"symbol": "AUTO", "symbolPosition": "prefix"}
+    assert resolve_auto_currency(currency, "USD") == {
+        "symbol": "USD",
+        "symbolPosition": "prefix",
+    }
+    assert resolve_auto_currency(currency, None) is currency
+    explicit = {"symbol": "EUR", "symbolPosition": "suffix"}
+    assert resolve_auto_currency(explicit, "USD") is explicit
+
+
+def test_resolve_auto_currency_prefers_cell_context_and_detects_mixed() -> None:
+    currency = {"symbol": "AUTO", "symbolPosition": "prefix"}
+
+    assert resolve_auto_currency(
+        currency, "GBP", currency_context=frozenset({" usd "})
+    ) == {"symbol": "USD", "symbolPosition": "prefix"}
+    assert (
+        resolve_auto_currency(
+            currency, "GBP", currency_context=frozenset({"USD", "EUR"})
+        )
+        is currency
+    )
+    assert resolve_auto_currency(currency, "GBP", currency_context=frozenset()) == {
+        "symbol": "GBP",
+        "symbolPosition": "prefix",
+    }
+    assert (
+        resolve_auto_currency(
+            currency,
+            "GBP",
+            currency_context=frozenset(),
+            fallback_to_detected=False,
+        )
+        is currency
+    )
+
+
+@pytest.mark.parametrize("missing", [float("nan"), pd.NA, pd.NaT])
+def test_resolve_auto_currency_coerces_non_iterable_context(missing: Any) -> None:
+    """
+    A sparse 2D pivot leaves missing cells as a scalar missing value rather than
+    an empty tuple. AUTO resolution must treat any non-iterable context (the
+    ``np.nan`` the live path produces, plus other pandas sentinels) as empty
+    instead of raising ``TypeError`` on ``list(context)``. ``missing`` is typed
+    ``Any`` on purpose: it pins the runtime guard without widening the narrow
+    ``Iterable[Any] | float | None`` contract to cover these sentinels.
+    """
+    currency = {"symbol": "AUTO", "symbolPosition": "prefix"}
+
+    # A missing context with fallback enabled behaves like an empty context and
+    # uses the query-wide detected currency.
+    assert resolve_auto_currency(currency, "GBP", currency_context=missing) == {
+        "symbol": "GBP",
+        "symbolPosition": "prefix",
+    }
+
+    # A missing context without fallback keeps AUTO, like the empty-context path.
+    assert (
+        resolve_auto_currency(
+            currency,
+            "GBP",
+            currency_context=missing,
+            fallback_to_detected=False,
+        )
+        is currency
+    )
+
+
+@pytest.mark.parametrize(
+    "value,expected",
+    [
+        (5e-7, "5e-7"),
+        (9.999999e-7, "9.999999e-7"),
+        (1e-6, "0.000001"),
+        (999_999_999_999, "999,999,999,999"),
+        (1e12, "1e+12"),
+        (1e20, "1e+20"),
+        (999_999_999_999_999_900_000, "1e+21"),
+        (1e21, "1e+21"),
+    ],
+)
+def test_default_format_matches_d3_exponent_boundaries(
+    value: float, expected: str
+) -> None:
+    assert format_number_with_config(",", None, value) == expected
+
+
+def test_currency_position_uses_request_locale() -> None:
+    with patch("superset.utils.number_format.get_locale", return_value="fr_FR"):
+        assert (
+            format_number_with_config(
+                ",.2f", {"symbol": "EUR", "symbolPosition": None}, 1234.5
+            )
+            == "1,234.50 €"
+        )
+
+
+def test_currency_position_uses_configured_locale_without_request() -> None:
+    with (
+        patch("superset.utils.number_format.get_locale", return_value=None),
+        patch.dict(current_app.config, {"BABEL_DEFAULT_LOCALE": "fr_FR"}),
+    ):
+        assert (
+            format_number_with_config(
+                ",.2f", {"symbol": "EUR", "symbolPosition": None}, 1234.5
+            )
+            == "1,234.50 €"
+        )
+
+
+@pytest.mark.parametrize(
+    "code,locale,expected",
+    [
+        ("USD", "en_US", "prefix"),
+        ("EUR", "fr_FR", "suffix"),
+        ("ZZZ", "not_a_locale", "prefix"),
+    ],
+)
+def test_resolve_symbol_position(code: str, locale: str, expected: str) -> None:
+    resolve_symbol_position.cache_clear()
+    assert resolve_symbol_position(code, locale) == expected
+
+
+def test_get_currency_locale_handles_missing_babel_and_app_context() -> None:
+    app = MagicMock()
+    with (
+        patch("superset.utils.number_format.get_locale", side_effect=RuntimeError),
+        patch("superset.utils.number_format.current_app", app),
+    ):
+        app.config.get.return_value = "de_DE"
+        assert get_currency_locale() == "de_DE"
+
+        app.config.get.side_effect = RuntimeError
+        assert get_currency_locale() == "en"
+
+
+def test_non_numeric_value_is_returned_as_is() -> None:
+    assert format_number_with_config(",.2f", None, "abc") == "abc"
+    assert format_number_with_config(",.2f", None, None) == ""
+
+
+def test_currency_error_keeps_formatted_number() -> None:
+    assert (
+        format_number_with_config(
+            ",.2f", {"symbol": {"bad": 1}, "symbolPosition": "prefix"}, 1234.5
+        )
+        == "1,234.50"
+    )
+
+
+def test_decimal_values_are_formatted() -> None:
+    assert format_number_with_config(",.2f", None, Decimal("1234.5")) == "1,234.50"
+    assert (
+        format_number_with_config(
+            ",.2f", {"symbol": "USD", "symbolPosition": "prefix"}, Decimal("1234.5")
+        )
+        == "$ 1,234.50"
+    )
+    assert format_number_with_config(",.2f", None, Decimal("NaN")) == ""
+
+
+# --- Parity with the frontend d3-format --------------------------------------
+#
+# EXPECTED is the authoritative output of the frontend's ``d3-format`` (the same
+# library the Table/Pivot charts render with) for every number preset in
+# ``D3_FORMAT_OPTIONS``. Regenerate from ``superset-frontend`` with::
+#
+#     node -e 'const {format}=require("d3-format");
+#     const p=["~g",",d",".1s",".3s",",.1%",".2%",".3%",".4r",
+#              ",.1f",",.2f",",.3f","+,","$,.2f"];
+#     const v=[12345.432,0,4725.0,80679663,1234567890,-1234.5,
+#              0.0123,999.9,1000.0];
+#     const o={}; for(const f of p){o[f]={};
+#       for(const x of v) o[f][x]=format(f)(x);}
+#     console.log(JSON.stringify(o));'
+#
+# One intentional deviation: d3 emits a Unicode minus (U+2212); the Python helper
+# emits an ASCII "-" for email/CSV safety, so the comparison normalizes it.
+
+VALUES: list[float] = [
+    12345.432,
+    0,
+    4725.0,
+    80679663,
+    1234567890,
+    -1234.5,
+    0.0123,
+    999.9,
+    1000.0,
+]
+
+EXPECTED: dict[str, list[str]] = {
+    "~g": [
+        "12345.4",
+        "0",
+        "4725",
+        "8.06797e+7",
+        "1.23457e+9",
+        "−1234.5",
+        "0.0123",
+        "999.9",
+        "1000",
+    ],  # noqa: E501
+    ",d": [
+        "12,345",
+        "0",
+        "4,725",
+        "80,679,663",
+        "1,234,567,890",
+        "−1,235",
+        "0",
+        "1,000",
+        "1,000",
+    ],  # noqa: E501
+    ".1s": ["10k", "0", "5k", "80M", "1G", "−1k", "10m", "1k", "1k"],
+    ".3s": [
+        "12.3k",
+        "0.00",
+        "4.73k",
+        "80.7M",
+        "1.23G",
+        "−1.23k",
+        "12.3m",
+        "1.00k",
+        "1.00k",
+    ],
+    ",.1%": [
+        "1,234,543.2%",
+        "0.0%",
+        "472,500.0%",
+        "8,067,966,300.0%",
+        "123,456,789,000.0%",
+        "−123,450.0%",
+        "1.2%",
+        "99,990.0%",
+        "100,000.0%",
+    ],  # noqa: E501
+    ".2%": [
+        "1234543.20%",
+        "0.00%",
+        "472500.00%",
+        "8067966300.00%",
+        "123456789000.00%",
+        "−123450.00%",
+        "1.23%",
+        "99990.00%",
+        "100000.00%",
+    ],  # noqa: E501
+    ".3%": [
+        "1234543.200%",
+        "0.000%",
+        "472500.000%",
+        "8067966300.000%",
+        "123456789000.000%",
+        "−123450.000%",
+        "1.230%",
+        "99990.000%",
+        "100000.000%",
+    ],  # noqa: E501
+    ".4r": [
+        "12350",
+        "0.000",
+        "4725",
+        "80680000",
+        "1235000000",
+        "−1235",
+        "0.01230",
+        "999.9",
+        "1000",
+    ],  # noqa: E501
+    ",.1f": [
+        "12,345.4",
+        "0.0",
+        "4,725.0",
+        "80,679,663.0",
+        "1,234,567,890.0",
+        "−1,234.5",
+        "0.0",
+        "999.9",
+        "1,000.0",
+    ],  # noqa: E501
+    ",.2f": [
+        "12,345.43",
+        "0.00",
+        "4,725.00",
+        "80,679,663.00",
+        "1,234,567,890.00",
+        "−1,234.50",
+        "0.01",
+        "999.90",
+        "1,000.00",
+    ],  # noqa: E501
+    ",.3f": [
+        "12,345.432",
+        "0.000",
+        "4,725.000",
+        "80,679,663.000",
+        "1,234,567,890.000",
+        "−1,234.500",
+        "0.012",
+        "999.900",
+        "1,000.000",
+    ],  # noqa: E501
+    "+,": [
+        "+12,345.432",
+        "+0",
+        "+4,725",
+        "+80,679,663",
+        "+1,234,567,890",
+        "−1,234.5",
+        "+0.0123",
+        "+999.9",
+        "+1,000",
+    ],  # noqa: E501
+    "$,.2f": [
+        "$12,345.43",
+        "$0.00",
+        "$4,725.00",
+        "$80,679,663.00",
+        "$1,234,567,890.00",
+        "−$1,234.50",
+        "$0.01",
+        "$999.90",
+        "$1,000.00",
+    ],  # noqa: E501
+    "(,.2f": [
+        "12,345.43",
+        "0.00",
+        "4,725.00",
+        "80,679,663.00",
+        "1,234,567,890.00",
+        "(1,234.50)",
+        "0.01",
+        "999.90",
+        "1,000.00",
+    ],  # noqa: E501
+    "($,.2f": [
+        "$12,345.43",
+        "$0.00",
+        "$4,725.00",
+        "$80,679,663.00",
+        "$1,234,567,890.00",
+        "($1,234.50)",
+        "$0.01",
+        "$999.90",
+        "$1,000.00",
+    ],  # noqa: E501
+    " ,.2f": [
+        " 12,345.43",
+        " 0.00",
+        " 4,725.00",
+        " 80,679,663.00",
+        " 1,234,567,890.00",
+        "−1,234.50",
+        " 0.01",
+        " 999.90",
+        " 1,000.00",
+    ],  # noqa: E501
+}
+
+
+@pytest.mark.parametrize("d3_format", list(EXPECTED))
+def test_matches_frontend_d3_format(d3_format: str) -> None:
+    for value, expected in zip(VALUES, EXPECTED[d3_format], strict=True):
+        result = format_number_with_config(d3_format, None, value)
+        assert result == expected.replace("−", "-"), (
+            f"{d3_format!r} of {value}: got {result!r}, expected {expected!r}"
+        )
+
+
+@pytest.mark.parametrize(
+    "d3_format,value,expected",
+    [
+        (",", 4725.0, "4,725"),
+        (",", 1000.0, "1,000"),
+        (",", 12345.432, "12,345.432"),
+        (",", 4725.5, "4,725.5"),
+        (",", 0.00005, "0.00005"),
+        (",", -1234.5, "-1,234.5"),
+        ("+,", 4725.0, "+4,725"),
+        ("+,", 1000.0, "+1,000"),
+        ("+,", -1234.5, "-1,234.5"),
+    ],
+)
+def test_default_format_matches_d3_for_floats(
+    d3_format: str, value: float, expected: str
+) -> None:
+    assert format_number_with_config(d3_format, None, value) == expected
+
+
+@pytest.mark.parametrize(
+    "d3_format,value,expected",
+    [
+        (".2f", 0.125, "0.13"),
+        ("$,.2f", 0.125, "$0.13"),
+        (".0f", 2.5, "3"),
+        (".1f", 0.25, "0.3"),
+        (".0%", 0.125, "13%"),
+        (".3s", 2.675, "2.67"),
+        (".4r", 0.12345, "0.1235"),
+        (".2f", 1.005, "1.00"),
+        (".1f", 0.35, "0.3"),
+    ],
+)
+def test_rounding_matches_d3_binary_half_up(
+    d3_format: str, value: float, expected: str
+) -> None:
+    assert format_number_with_config(d3_format, None, value) == expected
+
+
+@pytest.mark.parametrize(
+    "d3_format,value,expected",
+    [
+        ("~g", 0.00005, "0.00005"),
+        ("~g", 0.000005, "0.000005"),
+        (".0s", 4725, "5k"),
+        (".0e", 2.5, "3e+0"),
+        (".2~e", 1000, "1e+3"),
+        (",.2f", 1e21, "1e+21"),
+        (",.1%", 1e20, "1e+22%"),
+    ],
+)
+def test_additional_d3_parity_cases(
+    d3_format: str, value: float, expected: str
+) -> None:
+    assert format_number_with_config(d3_format, None, value) == expected
+
+
+@pytest.mark.parametrize(
+    "d3_format,value,expected",
+    [
+        ("SMART_NUMBER", 12.345, "12.35"),
+        ("SMART_NUMBER", 0.12345, "0.1235"),
+        ("SMART_NUMBER", 0.00005, "50µ"),
+        ("SMART_NUMBER", 5e-7, "500n"),
+        ("SMART_NUMBER_SIGNED", 12.345, "+12.35"),
+        ("n", 1234.5, "1,234.50"),
+        ("x", 12, "12"),
+    ],
+)
+def test_number_format_branch_coverage(
+    d3_format: str, value: float, expected: str
+) -> None:
+    assert format_number_with_config(d3_format, None, value) == expected
+
+
+def test_default_helper_and_whole_float_fallback() -> None:
+    assert format_default(1000, ",") == "1,000"
+    assert format_number_with_config(None, None, 42.0) == "42"
+
+
+@pytest.mark.parametrize(
+    "preset",
+    [
+        "DURATION",
+        "DURATION_SUB",
+        "DURATION_COL",
+        "MEMORY_DECIMAL",
+        "MEMORY_BINARY",
+        "MEMORY_TRANSFER_RATE_DECIMAL",
+        "MEMORY_TRANSFER_RATE_BINARY",
+    ],
+)
+def test_unported_frontend_presets_are_explicitly_rejected(preset: str) -> None:
+    with pytest.raises(ValueError, match="not available"):
+        format_numeric(preset, 66000)
+    assert format_number_with_config(preset, None, 66000) == "66000"
+
+
+@pytest.mark.parametrize("d3_format", ["08,.2f", "*>12,.2f"])
+def test_unsupported_d3_padding_is_explicitly_rejected(d3_format: str) -> None:
+    with pytest.raises(ValueError, match="padding"):
+        format_d3(d3_format, 1234.5)
+    assert format_number_with_config(d3_format, None, 1234.5) == "1234.5"

tests/unit_tests/utils/rls_test.py

@@ -0,0 +1,225 @@
+# 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.
+"""
+Traces the exact scope of the fallback in
+``superset.utils.rls.collect_rls_predicates_for_sql``: a SQL-parse failure
+there makes that function return a per-user marker instead of the real
+predicates, but this module is wired in as a *cache-key* input only
+(``SqlaTable.get_extra_cache_keys``), not as part of the code path that
+actually attaches RLS predicates to a query's WHERE clause
+(``BaseDatasource.get_sqla_row_level_filters``, consumed directly by
+``get_sqla_query``). These tests pin down that separation: a parse failure
+in the cache-key helper only ever affects the cache key contribution (kept
+distinct per user via the marker), and never "RLS predicates stop being
+applied to the query".
+"""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock, patch
+
+import pytest
+from flask import Flask
+from sqlalchemy.sql.elements import TextClause
+
+from superset.connectors.sqla.models import BaseDatasource
+from superset.utils.rls import collect_rls_predicates_for_sql
+
+
+@pytest.fixture
+def mock_database() -> MagicMock:
+    database = MagicMock()
+    database.db_engine_spec.engine = "sqlite"
+    database.get_default_catalog.return_value = None
+    return database
+
+
+def test_collect_rls_predicates_for_sql_returns_per_user_sentinel_on_parse_failure(
+    mock_database: MagicMock,
+) -> None:
+    """
+    A SQL-parse exception inside ``collect_rls_predicates_for_sql`` is
+    swallowed, and the function returns a marker derived from the current
+    user's id instead of propagating the exception or silently returning an
+    empty list.
+    """
+    with (
+        patch(
+            "superset.sql.parse.SQLScript",
+            side_effect=ValueError("cannot parse"),
+        ),
+        patch("superset.utils.rls.get_user_id", return_value=42),
+    ):
+        result = collect_rls_predicates_for_sql(
+            "SELECT * FROM some_table",
+            mock_database,
+            catalog=None,
+            schema="public",
+        )
+
+    assert result == ["rls-predicate-parse-failed-for-user-42"]
+
+
+def test_parse_failure_produces_different_cache_contributions_for_different_users(
+    mock_database: MagicMock,
+) -> None:
+    """
+    Two virtual datasets whose underlying RLS predicates differ (one has a
+    predicate, the other has none) would normally contribute different
+    strings to the cache key. If SQL parsing fails before predicates are
+    even collected, the actual predicate difference never gets a chance to
+    be collected -- but each user still contributes a marker scoped to their
+    own id, so the two calls don't collapse onto the same cache key
+    contribution.
+    """
+    with (
+        patch(
+            "superset.sql.parse.SQLScript",
+            side_effect=ValueError("cannot parse"),
+        ),
+        patch(
+            "superset.utils.rls.get_predicates_for_table",
+            side_effect=[["tenant_id = 1"], []],
+        ) as mock_get_predicates,
+        patch(
+            "superset.utils.rls.get_user_id",
+            side_effect=[1, 2],
+        ),
+    ):
+        result_user_one = collect_rls_predicates_for_sql(
+            "SELECT * FROM some_table",
+            mock_database,
+            catalog=None,
+            schema="public",
+        )
+        result_user_two = collect_rls_predicates_for_sql(
+            "SELECT * FROM some_table",
+            mock_database,
+            catalog=None,
+            schema="public",
+        )
+
+    # get_predicates_for_table was never reached: the parse exception fires
+    # first, so the per-user predicate difference never had a chance to be
+    # collected in the first place.
+    mock_get_predicates.assert_not_called()
+    assert result_user_one != result_user_two
+    assert result_user_one == ["rls-predicate-parse-failed-for-user-1"]
+    assert result_user_two == ["rls-predicate-parse-failed-for-user-2"]
+
+
+def test_parse_failure_sentinel_distinguishes_guest_tokens_by_rls_scope(
+    mock_database: MagicMock,
+) -> None:
+    """
+    ``get_user_id()`` always returns ``None`` for guest users, so keying the
+    parse-failure sentinel on it alone would collapse every guest token onto
+    the same cache contribution regardless of the RLS rules baked into each
+    token. Guest sessions must instead be distinguished by (a hash of) their
+    own token's ``rls_rules``, so two guests with different row-level scopes
+    never share a cache entry, while two guests with the *same* scope do.
+    """
+
+    def _guest_user(rls_rules: list[dict[str, str]]) -> MagicMock:
+        guest_user = MagicMock()
+        guest_user.guest_token = {"rls_rules": rls_rules}
+        return guest_user
+
+    scope_a = [{"dataset": "1", "clause": "tenant_id = 1"}]
+    scope_b = [{"dataset": "1", "clause": "tenant_id = 2"}]
+
+    with (
+        patch(
+            "superset.sql.parse.SQLScript",
+            side_effect=ValueError("cannot parse"),
+        ),
+        patch(
+            "superset.utils.rls.security_manager.get_current_guest_user_if_guest",
+            side_effect=[
+                _guest_user(scope_a),
+                _guest_user(scope_b),
+                _guest_user(scope_a),
+            ],
+        ),
+    ):
+        result_guest_scope_a = collect_rls_predicates_for_sql(
+            "SELECT * FROM some_table",
+            mock_database,
+            catalog=None,
+            schema="public",
+        )
+        result_guest_scope_b = collect_rls_predicates_for_sql(
+            "SELECT * FROM some_table",
+            mock_database,
+            catalog=None,
+            schema="public",
+        )
+        result_guest_scope_a_again = collect_rls_predicates_for_sql(
+            "SELECT * FROM some_table",
+            mock_database,
+            catalog=None,
+            schema="public",
+        )
+
+    assert result_guest_scope_a[0].startswith(
+        "rls-predicate-parse-failed-for-user-guest-"
+    )
+    assert result_guest_scope_a != result_guest_scope_b
+    assert result_guest_scope_a == result_guest_scope_a_again
+
+
+def test_real_rls_enforcement_does_not_go_through_the_cache_key_helper(
+    app: Flask,
+) -> None:
+    """
+    ``get_sqla_row_level_filters`` -- the method ``get_sqla_query`` actually
+    calls to build a query's WHERE clause -- reaches the RLS rules directly
+    via ``security_manager.get_rls_filters`` and never touches
+    ``collect_rls_predicates_for_sql``. So even in a request where SQL
+    parsing inside the cache-key helper fails, the predicate is still
+    attached to the real, executed query: the failure mode is confined to
+    the cache key, not the query itself.
+    """
+    datasource = MagicMock(spec=BaseDatasource)
+    datasource.get_template_processor.return_value = MagicMock()
+    datasource.get_template_processor.return_value.process_template = lambda x: x
+    datasource.text = lambda x: TextClause(x)
+
+    configured_filter = MagicMock()
+    configured_filter.clause = "tenant_id = 1"
+    configured_filter.group_key = None
+
+    with (
+        patch(
+            "superset.connectors.sqla.models.security_manager.get_rls_filters",
+            return_value=[configured_filter],
+        ),
+        patch(
+            "superset.connectors.sqla.models.is_feature_enabled",
+            return_value=False,
+        ),
+        patch(
+            "superset.utils.rls.collect_rls_predicates_for_sql",
+            side_effect=AssertionError(
+                "get_sqla_row_level_filters must not call the cache-key helper"
+            ),
+        ),
+    ):
+        filters = BaseDatasource.get_sqla_row_level_filters(datasource)
+
+    assert len(filters) == 1
+    assert "tenant_id" in str(filters[0])