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/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/commands/dataset/update.py

@@ -123,66 +123,13 @@ class UpdateDatasetCommand(UpdateMixin, BaseCommand):
         # we know we have a valid model
         self._model = cast(SqlaTable, self._model)
         database_id = self._properties.pop("database_id", None)
-        new_db_connection = self._get_new_database_connection(database_id, exceptions)
-        db = new_db_connection or self._model.database
-        database_changed = new_db_connection is not None
-
-        # Detect a caller-supplied change to the source binding, inspected
-        # before the catalog normalization below injects derived values.
-        source_changed = database_changed or any(
-            field in self._properties
-            and self._properties[field] != getattr(self._model, field)
-            for field in ("catalog", "schema", "table_name")
-        )
-
-        catalog, schema, table = self._resolve_catalog_schema_table(db, exceptions)
-
-        # Repointing to a different database connection requires access to
-        # that connection, independent of the caller's editorship of this
-        # dataset -- only persist the change once that's confirmed.
-        if new_db_connection:
-            self._apply_database_repoint(new_db_connection, table, exceptions)
-
-        # Validate uniqueness
-        if not DatasetDAO.validate_update_uniqueness(
-            db,
-            table,
-            self._model_id,
-        ):
-            exceptions.append(DatasetExistsValidationError(table))
-
-        # Repointing a physical dataset (or converting a virtual dataset to a
-        # physical one) runs the same data-access check as the create path.
-        # Skip it when the database connection itself changed: that case is
-        # already covered by the repoint check above, against the same
-        # (db, table) pair.
-        sql = self._properties.get("sql", self._model.sql)
-        if (
-            not new_db_connection
-            and not sql
-            and (source_changed or ("sql" in self._properties and self._model.sql))
-        ):
-            self._validate_table_access(db, table, exceptions)
-
-        self._validate_sql_access(db, catalog, schema, exceptions)
-
-    def _get_new_database_connection(
-        self, database_id: int | None, exceptions: list[ValidationError]
-    ) -> Database | None:
-        # we know we have a valid model
-        self._model = cast(SqlaTable, self._model)
-        if database_id and database_id != self._model.database.id:
-            if new_db_connection := DatasetDAO.get_database_by_id(database_id):
-                return new_db_connection
-            exceptions.append(DatabaseNotFoundValidationError())
-        return None
-
-    def _resolve_catalog_schema_table(
-        self, db: Database, exceptions: list[ValidationError]
-    ) -> tuple[str | None, str | None, Table]:
-        # we know we have a valid model
-        self._model = cast(SqlaTable, self._model)
         catalog = self._properties.get("catalog")
+        new_db_connection: Database | None = None
+
+        if database_id and database_id != self._model.database.id:
+            if not (new_db_connection := DatasetDAO.get_database_by_id(database_id)):
+                exceptions.append(DatabaseNotFoundValidationError())
+        db = new_db_connection or self._model.database
         default_catalog = db.get_default_catalog()
 
         # If multi-catalog is disabled, and catalog provided is not
@@ -214,28 +161,29 @@ class UpdateDatasetCommand(UpdateMixin, BaseCommand):
             schema,
             catalog,
         )
-        return catalog, schema, table
 
-    def _apply_database_repoint(
-        self,
-        new_db_connection: Database,
-        table: Table,
-        exceptions: list[ValidationError],
-    ) -> None:
-        try:
-            security_manager.raise_for_access(database=new_db_connection, table=table)
-        except SupersetSecurityException as ex:
-            exceptions.append(DatasetDataAccessIsNotAllowed(ex.error.message))
-        else:
-            self._properties["database"] = new_db_connection
+        # Repointing to a different database connection requires access to
+        # that connection, independent of the caller's editorship of this
+        # dataset -- only persist the change once that's confirmed.
+        if new_db_connection:
+            try:
+                security_manager.raise_for_access(
+                    database=new_db_connection, table=table
+                )
+            except SupersetSecurityException as ex:
+                exceptions.append(DatasetDataAccessIsNotAllowed(ex.error.message))
+            else:
+                self._properties["database"] = new_db_connection
 
-    def _validate_table_access(
-        self, db: Database, table: Table, exceptions: list[ValidationError]
-    ) -> None:
-        try:
-            security_manager.raise_for_access(database=db, table=table)
-        except SupersetSecurityException as ex:
-            exceptions.append(DatasetDataAccessIsNotAllowed(ex.error.message))
+        # Validate uniqueness
+        if not DatasetDAO.validate_update_uniqueness(
+            db,
+            table,
+            self._model_id,
+        ):
+            exceptions.append(DatasetExistsValidationError(table))
+
+        self._validate_sql_access(db, catalog, schema, exceptions)
 
     def _validate_sql_access(
         self,
@@ -278,9 +226,6 @@ class UpdateDatasetCommand(UpdateMixin, BaseCommand):
             self._validate_metrics(metrics, exceptions)
             self._validate_expressions(metrics, "metrics", exceptions)
 
-        if predicate := self._properties.get("fetch_values_predicate"):
-            self._validate_fetch_values_predicate(predicate, exceptions)
-
         if folders := self._properties.get("folders"):
             valid_uuids: set[UUID] = set()
             if metrics:
@@ -389,34 +334,6 @@ class UpdateDatasetCommand(UpdateMixin, BaseCommand):
                     )
                 )
 
-    def _validate_fetch_values_predicate(
-        self,
-        predicate: str,
-        exceptions: list[ValidationError],
-    ) -> None:
-        """
-        Validate ``fetch_values_predicate`` with the same parser-based
-        validator used for stored column and metric expressions.
-        """
-        self._model = cast(SqlaTable, self._model)
-        database = self._properties.get("database") or self._model.database
-        catalog = self._properties.get("catalog", self._model.catalog)
-        schema = self._properties.get("schema", self._model.schema)
-        try:
-            validate_stored_expression(database, catalog, schema, predicate)
-        except (SupersetSecurityException, QueryClauseValidationException) as ex:
-            message = (
-                ex.error.message
-                if isinstance(ex, SupersetSecurityException)
-                else ex.message
-            )
-            exceptions.append(
-                ValidationError(
-                    message,
-                    field_name="fetch_values_predicate",
-                )
-            )
-
     @staticmethod
     def _get_duplicates(data: list[dict[str, Any]], key: str) -> list[str]:
         duplicates = [

superset/commands/report/alert.py

@@ -39,9 +39,7 @@ from superset.commands.report.exceptions import (
     AlertValidatorConfigError,
     ReportScheduleExecutorNotFoundError,
 )
-from superset.exceptions import SupersetSecurityException
 from superset.reports.models import ReportSchedule, ReportScheduleValidatorType
-from superset.sql.parse import SQLScript
 from superset.tasks.utils import get_executor
 from superset.utils import json
 from superset.utils.core import override_user
@@ -183,18 +181,6 @@ class AlertCommand(BaseCommand):
             "execution_id": self._execution_id,
         }
 
-    def _validate_rendered_sql(self, rendered_sql: str) -> None:
-        """
-        Enforce SQL-level constraints on the rendered alert query: a single
-        statement, and no mutations unless the database allows DML.
-        """
-        database = self._report_schedule.database
-        script = SQLScript(rendered_sql, engine=database.backend)
-        if len(script.statements) != 1:
-            raise AlertQueryError(message=_("Alert query must be a single statement"))
-        if script.has_mutation() and not database.allow_dml:
-            raise AlertQueryError(message=_("Alert query must be read-only"))
-
     @logs_context(context_func=_get_alert_metadata_from_object)
     def _execute_query(self) -> pd.DataFrame:
         """
@@ -210,7 +196,6 @@ class AlertCommand(BaseCommand):
 
         try:
             rendered_sql = sql_template.process_template(self._report_schedule.sql)
-            self._validate_rendered_sql(rendered_sql)
             limited_rendered_sql = self._report_schedule.database.apply_limit_to_sql(
                 rendered_sql, ALERT_SQL_LIMIT
             )
@@ -235,18 +220,6 @@ class AlertCommand(BaseCommand):
                 raise ReportScheduleExecutorNotFoundError(username)
 
             with override_user(user):
-                # Run table-level authorization as the executing user against
-                # the rendered SQL.
-                try:
-                    security_manager.raise_for_access(
-                        database=self._report_schedule.database,
-                        sql=rendered_sql,
-                        force_dataset_match=True,
-                    )
-                except SupersetSecurityException as ex:
-                    raise AlertQueryError(
-                        message=_("Alert query failed the authorization check")
-                    ) from ex
                 start = default_timer()
                 df = self._report_schedule.database.get_df(sql=limited_rendered_sql)
                 stop = default_timer()
@@ -263,10 +236,6 @@ class AlertCommand(BaseCommand):
             # A missing executor user is a configuration problem, not a transient
             # query error; surface the typed error rather than masking it.
             raise
-        except AlertQueryError:
-            # Re-raise the typed validation/authorization errors as-is instead
-            # of masking them behind the generic error below.
-            raise
         except Exception as ex:
             logger.warning("An error occurred when running alert query")
             # The exception message here can reveal to much information to malicious

superset/commands/report/base.py

@@ -15,7 +15,6 @@
 # specific language governing permissions and limitations
 # under the License.
 import logging
-import re
 from typing import Any, Optional
 
 from croniter import croniter, CroniterBadDateError
@@ -26,9 +25,6 @@ from marshmallow import ValidationError
 from superset import security_manager
 from superset.commands.base import BaseCommand
 from superset.commands.report.exceptions import (
-    AlertQueryDataAccessValidationError,
-    AlertQueryDMLNotAllowedValidationError,
-    AlertQueryMultipleStatementsValidationError,
     ChartNotFoundValidationError,
     ChartNotSavedValidationError,
     DashboardNotFoundValidationError,
@@ -42,22 +38,16 @@ from superset.commands.report.exceptions import (
 from superset.daos.base import BaseDAO
 from superset.daos.chart import ChartDAO
 from superset.daos.dashboard import DashboardDAO
-from superset.exceptions import SupersetParseError, SupersetSecurityException
-from superset.models.core import Database
+from superset.exceptions import SupersetSecurityException
 from superset.reports.models import (
     ReportCreationMethod,
     ReportScheduleType,
 )
 from superset.reports.types import ReportScheduleExtra
-from superset.sql.parse import SQLScript
 from superset.utils import json
 
 logger = logging.getLogger(__name__)
 
-# Matches balanced Jinja blocks so templated alert SQL can be recognized and
-# its static validation deferred to execution time.
-_JINJA_BLOCK_RE = re.compile(r"\{\{.*?\}\}|\{%.*?%\}|\{#.*?#\}", re.DOTALL)
-
 
 class BaseReportScheduleCommand(BaseCommand):
     _properties: dict[str, Any]
@@ -68,52 +58,6 @@ class BaseReportScheduleCommand(BaseCommand):
     def validate(self) -> None:
         pass
 
-    def validate_alert_query(
-        self,
-        database: Database,
-        sql: str,
-        exceptions: list[ValidationError],
-    ) -> None:
-        """
-        Validate alert SQL at save time: it must parse as a single statement,
-        must not mutate state unless the database allows DML, and the saving
-        user must be authorized for the tables it reads. Templated SQL that
-        only parses after rendering is validated at execution time on the
-        rendered query.
-        """
-        contains_jinja = bool(_JINJA_BLOCK_RE.search(sql))
-        try:
-            script = SQLScript(sql, engine=database.backend)
-        except SupersetParseError as ex:
-            if not contains_jinja:
-                exceptions.append(
-                    ValidationError(
-                        _("Invalid SQL: %(error)s", error=ex.error.message),
-                        field_name="sql",
-                    )
-                )
-            return
-        if len(script.statements) != 1:
-            exceptions.append(AlertQueryMultipleStatementsValidationError())
-            return
-        if script.has_mutation() and not database.allow_dml:
-            exceptions.append(AlertQueryDMLNotAllowedValidationError())
-            return
-        try:
-            security_manager.raise_for_access(
-                database=database, sql=sql, force_dataset_match=True
-            )
-        except SupersetSecurityException as ex:
-            exceptions.append(AlertQueryDataAccessValidationError(ex.error.message))
-        except SupersetParseError as ex:
-            if not contains_jinja:
-                exceptions.append(
-                    ValidationError(
-                        _("Invalid SQL: %(error)s", error=ex.error.message),
-                        field_name="sql",
-                    )
-                )
-
     def _check_object_access(
         self,
         object_id: int,

superset/commands/report/create.py

@@ -129,8 +129,6 @@ class CreateReportScheduleCommand(CreateMixin, BaseReportScheduleCommand):
                 database_id = self._properties["database"]
                 if database := DatabaseDAO.find_by_id(database_id):
                     self._properties["database"] = database
-                    if sql := self._properties.get("sql"):
-                        self.validate_alert_query(database, sql, exceptions)
                 else:
                     exceptions.append(DatabaseNotFoundValidationError())
             except KeyError:

superset/commands/report/exceptions.py

@@ -40,38 +40,6 @@ class DatabaseNotFoundValidationError(ValidationError):
         super().__init__(_("Database does not exist"), field_name="database")
 
 
-class AlertQueryMultipleStatementsValidationError(ValidationError):
-    """
-    Marshmallow validation error for alert SQL containing multiple statements
-    """
-
-    def __init__(self) -> None:
-        super().__init__(
-            _("Alert query must be a single statement"),
-            field_name="sql",
-        )
-
-
-class AlertQueryDMLNotAllowedValidationError(ValidationError):
-    """
-    Marshmallow validation error for alert SQL that mutates state on a
-    database that does not allow DML
-    """
-
-    def __init__(self) -> None:
-        super().__init__(_("Alert query must be read-only"), field_name="sql")
-
-
-class AlertQueryDataAccessValidationError(ValidationError):
-    """
-    Marshmallow validation error for alert SQL referencing tables the user
-    is not authorized to query
-    """
-
-    def __init__(self, message: str) -> None:
-        super().__init__(message, field_name="sql")
-
-
 class ReportScheduleDatabaseNotAllowedValidationError(ValidationError):
     """
     Marshmallow validation error for database reference on a Report type schedule

superset/commands/report/update.py

@@ -149,19 +149,6 @@ class UpdateReportScheduleCommand(UpdateMixin, BaseReportScheduleCommand):
                 exceptions.append(DatabaseNotFoundValidationError())
             self._properties["database"] = database
 
-        # Re-validate the alert SQL whenever the SQL or the target database
-        # changes, using the stored value for whichever half is absent from
-        # the payload.
-        if report_type == ReportScheduleType.ALERT and (
-            "sql" in self._properties or "database" in self._properties
-        ):
-            effective_database = (
-                self._properties.get("database") or self._model.database
-            )
-            effective_sql = self._properties.get("sql", self._model.sql)
-            if effective_database and effective_sql:
-                self.validate_alert_query(effective_database, effective_sql, exceptions)
-
         # validate report frequency
         try:
             self.validate_report_frequency(

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

@@ -81,7 +81,6 @@ from superset.errors import ErrorLevel, SupersetError, SupersetErrorType
 from superset.exceptions import (
     ColumnNotFoundException,
     DatasetInvalidPermissionEvaluationException,
-    QueryClauseValidationException,
     QueryObjectValidationError,
     SupersetParseError,
     SupersetSecurityException,
@@ -104,7 +103,6 @@ from superset.models.helpers import (
     SoftDeleteMixin,
     SQLA_QUERY_KEYS,
     validate_adhoc_subquery,
-    validate_rendered_expression,
     validate_stored_expression_at_query_time,
 )
 from superset.models.slice import Slice
@@ -1217,14 +1215,6 @@ class TableColumn(AuditMixinNullable, ImportExportMixin, CertificationMixin, Mod
                             msg=msg,
                         )
                     ) from ex
-                if expression != self.expression:
-                    # Re-check the rendered expression before embedding it.
-                    expression = validate_rendered_expression(
-                        expression,
-                        self.database,
-                        self.table.catalog if self.table else None,
-                        self.table.schema if self.table else None,
-                    )
             expression = self._validate_stored_expression(expression)
             col = literal_column(expression, type_=type_)
         else:
@@ -1273,14 +1263,6 @@ class TableColumn(AuditMixinNullable, ImportExportMixin, CertificationMixin, Mod
                             msg=msg,
                         )
                     ) from ex
-                if expression != self.expression:
-                    # Re-check the rendered expression before embedding it.
-                    expression = validate_rendered_expression(
-                        expression,
-                        self.database,
-                        self.table.catalog if self.table else None,
-                        self.table.schema if self.table else None,
-                    )
             expression = self._validate_stored_expression(expression)
             col = literal_column(expression, type_=type_)
         else:
@@ -1386,14 +1368,6 @@ class SqlMetric(AuditMixinNullable, ImportExportMixin, CertificationMixin, Model
                         msg=msg,
                     )
                 ) from ex
-            if expression != self.expression:
-                # Re-check the rendered expression before embedding it.
-                expression = validate_rendered_expression(
-                    expression,
-                    self.table.database,
-                    self.table.catalog,
-                    self.table.schema,
-                )
 
         if expression:
             expression = self._validate_stored_expression(expression)
@@ -1806,24 +1780,7 @@ class SqlaTable(
                 fetch_values_predicate
             )
         try:
-            # Re-validate the rendered predicate with the same parser policy
-            # as stored column and metric expressions before embedding it.
-            validate_stored_expression(
-                self.database, self.catalog, self.schema, fetch_values_predicate
-            )
             return self.text(fetch_values_predicate)
-        except (SupersetSecurityException, QueryClauseValidationException) as ex:
-            message = (
-                ex.error.message
-                if isinstance(ex, SupersetSecurityException)
-                else ex.message
-            )
-            raise QueryObjectValidationError(
-                _(
-                    "Fetch values predicate failed SQL validation: %(msg)s",
-                    msg=message,
-                )
-            ) from ex
         except (TemplateError, SupersetSyntaxErrorException) as ex:
             msg = getattr(ex, "message", str(ex))
             raise QueryObjectValidationError(

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/models/helpers.py

@@ -153,7 +153,7 @@ from superset.utils.date_parser import (
     TimeDeltaAmbiguousError,
 )
 from superset.utils.dates import datetime_to_epoch
-from superset.utils.rls import apply_rls, get_predicates_for_table
+from superset.utils.rls import apply_rls
 
 
 class ValidationResultDict(TypedDict):
@@ -329,62 +329,6 @@ def validate_stored_expression_at_query_time(
     return expression
 
 
-def validate_rendered_expression(
-    expression: str,
-    database: Database,
-    catalog: str | None,
-    schema: str | None,
-) -> str:
-    """
-    Apply the stored-expression validation policy to a rendered expression.
-
-    Query-time counterpart to ``validate_stored_expression``: it runs on the
-    already-rendered expression that is embedded via ``literal_column`` and
-    applies the same policy, failing closed on unparseable results.
-
-    :param expression: the rendered expression
-    :returns: the expression to embed, possibly rewritten with RLS predicates
-    :raises QueryObjectValidationError: on multi-statement, set-operation,
-        disallowed sub-query, or sanitization failures -- matching the
-        ``QueryObjectValidationError`` contract callers already expect from
-        ``validate_stored_expression_at_query_time``, rather than letting a
-        raw ``SupersetSecurityException`` escape uncaught.
-    """
-    engine = database.backend
-    wrapped = f"SELECT {expression}"
-
-    try:
-        parsed = SQLStatement(wrapped, engine)
-    except SupersetParseError as ex:
-        raise QueryObjectValidationError(
-            _("Custom SQL fields cannot be parsed as a single SQL statement.")
-        ) from ex
-
-    if parsed.is_set_operation():
-        raise QueryObjectValidationError(
-            _("Custom SQL fields cannot contain set operations.")
-        )
-
-    try:
-        wrapped = validate_adhoc_subquery(
-            wrapped, database, catalog, schema or "", engine
-        )
-    except SupersetSecurityException as ex:
-        raise QueryObjectValidationError(ex.message) from ex
-    try:
-        wrapped = sanitize_clause(wrapped, engine)
-    except QueryClauseValidationException as ex:
-        raise QueryObjectValidationError(ex.message) from ex
-
-    prefix, expression = re.split(
-        r"SELECT\s+",
-        wrapped,
-        maxsplit=1,
-        flags=re.IGNORECASE,
-    )
-    return expression.strip()
-
-
 def json_to_dict(json_str: str) -> dict[Any, Any]:
     if json_str:
         val = re.sub(",[ \t\r\n]+}", "}", json_str)
@@ -3139,40 +3083,9 @@ class ExploreMixin:  # pylint: disable=too-many-public-methods
                 if rls_applied:
                     from_sql = parsed_script.format()
 
-            except Exception as ex:  # pylint: disable=broad-except
-                # RLS injection failures fail closed: only continue when it is
-                # positively confirmed that no RLS predicates apply to the
-                # referenced tables; any other outcome aborts the query.
-                try:
-                    rls_required = any(
-                        get_predicates_for_table(
-                            table.qualify(
-                                catalog=self.catalog,
-                                schema=self.schema or default_schema or "",
-                            ),
-                            self.database,
-                            self.database.get_default_catalog(),
-                            exclude_dataset_id=self_id,
-                        )
-                        for statement in parsed_script.statements
-                        for table in statement.tables
-                    )
-                except Exception:  # pylint: disable=broad-except
-                    rls_required = True
-                if rls_required:
-                    raise QueryObjectValidationError(
-                        _(
-                            "Row-level security could not be applied to the "
-                            "virtual dataset query, so it cannot be run "
-                            "securely: %(msg)s",
-                            msg=str(ex),
-                        )
-                    ) from ex
-                logger.warning(
-                    "RLS application to virtual dataset SQL failed, but no "
-                    "predicates apply to its tables; continuing: %s",
-                    ex,
-                )
+            except Exception as ex:
+                # Log the error but don't fail - RLS application is best-effort
+                logger.warning("Failed to apply RLS to virtual dataset SQL: %s", ex)
 
         cte = self.db_engine_spec.get_cte_query(from_sql)
         from_clause = (
@@ -3869,11 +3782,6 @@ class ExploreMixin:  # pylint: disable=too-many-public-methods
         if expression := tbl_column.expression:
             if template_processor:
                 expression = template_processor.process_template(expression)
-                if expression != tbl_column.expression:
-                    # Re-check the rendered expression before embedding it.
-                    expression = validate_rendered_expression(
-                        expression, self.database, self.catalog, self.schema
-                    )
             expression = self._validate_stored_expression(expression)
             col = literal_column(expression, type_=type_)
         else:

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
@@ -1407,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)
             )
@@ -1533,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):
@@ -2175,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)

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)

tests/integration_tests/reports/alert_tests.py

@@ -104,9 +104,6 @@ def test_execute_query_as_report_executor(
     )
     command = AlertCommand(report_schedule=report_schedule, execution_id=uuid.uuid4())
     override_user_mock = mocker.patch("superset.commands.report.alert.override_user")
-    # override_user is mocked, so no real executor context is set; the alert
-    # authorization check is covered elsewhere, so keep it a no-op here.
-    mocker.patch("superset.commands.report.alert.security_manager.raise_for_access")
     cm = (
         pytest.raises(type(expected_result))
         if isinstance(expected_result, Exception)
@@ -131,7 +128,6 @@ def test_execute_query_mutate_query_enabled(
 
     app.config["MUTATE_ALERT_QUERY"] = True
     mocker.patch("superset.commands.report.alert.override_user")
-    mocker.patch("superset.commands.report.alert.security_manager.raise_for_access")
     mock_df = mocker.MagicMock(spec=pd.DataFrame)
     mock_df.empty = True
     mock_database = get_example_database()
@@ -175,7 +171,6 @@ def test_execute_query_mutate_query_disabled(
 
     app.config["MUTATE_ALERT_QUERY"] = False
     mocker.patch("superset.commands.report.alert.override_user")
-    mocker.patch("superset.commands.report.alert.security_manager.raise_for_access")
     mock_database = mocker.MagicMock()
 
     admin_user = get_user("admin")

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/dataset/update_test.py

@@ -38,7 +38,6 @@ from superset.commands.dataset.update import (
 from superset.datasets.schemas import FolderSchema
 from superset.errors import ErrorLevel, SupersetError, SupersetErrorType
 from superset.exceptions import SupersetSecurityException
-from superset.sql.parse import Table
 from superset.subjects.exceptions import SubjectsNotFoundValidationError
 from tests.unit_tests.conftest import with_feature_flags
 
@@ -279,60 +278,6 @@ def test_update_dataset_database_id_change_allowed_with_access(
     assert update_kwargs["attributes"]["database"] is mock_new_database
 
 
-def test_update_dataset_physical_repoint_requires_table_access(
-    mocker: MockerFixture,
-) -> None:
-    """
-    Repointing a physical dataset at a different table must pass the same
-    ``raise_for_access(database=..., table=...)`` gate the create path
-    enforces; editorship alone must not grant access to the new table.
-    """
-    mock_dataset_dao = mocker.patch("superset.commands.dataset.update.DatasetDAO")
-    mocker.patch(
-        "superset.commands.dataset.update.security_manager.raise_for_editorship",
-    )
-    mocker.patch("superset.commands.utils.security_manager.is_admin", return_value=True)
-
-    mock_database = mocker.MagicMock()
-    mock_database.id = 1
-    mock_database.get_default_catalog.return_value = "catalog"
-    mock_database.allow_multi_catalog = False
-
-    mock_dataset = mocker.MagicMock()
-    mock_dataset.database = mock_database
-    mock_dataset.catalog = "catalog"
-    mock_dataset.schema = "public"
-    mock_dataset.table_name = "allowed_table"
-    mock_dataset.sql = None  # physical dataset
-    mock_dataset.editors = []
-
-    mock_dataset_dao.find_by_id.return_value = mock_dataset
-    mock_dataset_dao.validate_update_uniqueness.return_value = True
-
-    raise_for_access = mocker.patch(
-        "superset.commands.dataset.update.security_manager.raise_for_access",
-        side_effect=SupersetSecurityException(
-            SupersetError(
-                error_type=SupersetErrorType.DATASOURCE_SECURITY_ACCESS_ERROR,
-                message="You don't have access to the table 'restricted_table'",
-                level=ErrorLevel.ERROR,
-            )
-        ),
-    )
-
-    with pytest.raises(DatasetInvalidError) as excinfo:
-        UpdateDatasetCommand(1, {"table_name": "restricted_table"}).run()
-
-    raise_for_access.assert_called_once_with(
-        database=mock_database,
-        table=Table("restricted_table", "public", "catalog"),
-    )
-    assert any(
-        "You don't have access to the table" in str(exc)
-        for exc in excinfo.value._exceptions
-    )
-
-
 @pytest.mark.parametrize(
     ("payload, exception, error_msg"),
     [
@@ -1274,44 +1219,3 @@ def test_validate_folders_metrics_vs_columns_behavior(mocker: MockerFixture) ->
         command2._validate_semantics([])
     except Exception as e:
         pytest.fail(f"Should work with new metric UUIDs when new metrics provided: {e}")
-
-
-def test_update_dataset_rejects_malicious_fetch_values_predicate(
-    mocker: MockerFixture,
-) -> None:
-    """
-    ``fetch_values_predicate`` is wrapped verbatim into a raw WHERE clause at
-    query time, so the command routes it through the stored-expression
-    validator; a UNION-based predicate is rejected at save time.
-    """
-    mock_dataset_dao = mocker.patch("superset.commands.dataset.update.DatasetDAO")
-    mocker.patch(
-        "superset.commands.dataset.update.security_manager.raise_for_editorship",
-    )
-    mocker.patch("superset.commands.utils.security_manager.is_admin", return_value=True)
-    mocker.patch(
-        "superset.commands.utils.security_manager.get_user_by_id", return_value=None
-    )
-    mock_database = mocker.MagicMock()
-    mock_database.id = 1
-    mock_database.backend = "sqlite"
-    mock_database.allow_multi_catalog = False
-    mock_database.get_default_catalog.return_value = "catalog"
-    mock_dataset = mocker.MagicMock()
-    mock_dataset.database = mock_database
-    mock_dataset.catalog = "catalog"
-    mock_dataset.schema = None
-    mock_dataset_dao.find_by_id.return_value = mock_dataset
-    mock_dataset_dao.get_database_by_id.return_value = mock_database
-    mock_dataset_dao.validate_update_uniqueness.return_value = True
-
-    payload = {
-        "fetch_values_predicate": "1=0 UNION SELECT card_number FROM billing.cards"
-    }
-    with pytest.raises(DatasetInvalidError) as excinfo:
-        UpdateDatasetCommand(1, payload).run()
-    assert any(
-        isinstance(exc, ValidationError)
-        and "fetch_values_predicate" in (exc.field_name or "")
-        for exc in excinfo.value._exceptions
-    )

tests/unit_tests/commands/report/alert_test.py

@@ -511,11 +511,8 @@ def test_execute_query_raises_when_executor_user_missing(
     username, rather than swallowing it into an opaque ``AlertQueryError`` (or
     surfacing a NoneType/AttributeError from the downstream auth flow).
     """
-    template_processor_mock = mocker.Mock()
-    template_processor_mock.process_template.return_value = "SELECT value FROM metrics"
     mocker.patch(
         "superset.commands.report.alert.jinja_context.get_template_processor",
-        return_value=template_processor_mock,
     )
     mocker.patch(
         "superset.commands.report.alert.get_executor",
@@ -529,8 +526,6 @@ def test_execute_query_raises_when_executor_user_missing(
     report_schedule_mock = mocker.Mock()
     report_schedule_mock.id = 1
     report_schedule_mock.sql = "SELECT value FROM metrics"
-    report_schedule_mock.database.backend = "sqlite"
-    report_schedule_mock.database.allow_dml = False
 
     command = AlertCommand(
         report_schedule=report_schedule_mock,

tests/unit_tests/commands/report/base_test.py

@@ -24,7 +24,6 @@ from typing import Any, Callable
 from unittest.mock import patch
 
 import pytest
-from pytest_mock import MockerFixture
 
 from superset.commands.report.base import BaseReportScheduleCommand
 from superset.commands.report.exceptions import (
@@ -319,94 +318,3 @@ def test_validate_report_frequency_using_callable() -> None:
         "1,6 * * * *",
         ReportScheduleType.REPORT,
     )
-
-
-def test_validate_alert_query_rejects_multi_statement_sql() -> None:
-    """
-    Alert SQL is validated at save time; multi-statement SQL cannot be
-    persisted for later raw execution by the alert runner.
-    """
-    from unittest.mock import MagicMock
-
-    from marshmallow import ValidationError
-
-    from superset.commands.report.base import BaseReportScheduleCommand
-    from superset.commands.report.exceptions import (
-        AlertQueryMultipleStatementsValidationError,
-    )
-
-    database = MagicMock()
-    database.backend = "sqlite"
-    database.allow_dml = False
-
-    exceptions: list[ValidationError] = []
-    BaseReportScheduleCommand().validate_alert_query(
-        database, "SELECT 1; DROP TABLE ab_user", exceptions
-    )
-
-    assert len(exceptions) == 1
-    assert isinstance(exceptions[0], AlertQueryMultipleStatementsValidationError)
-
-
-def test_validate_alert_query_rejects_dml_when_not_allowed() -> None:
-    """A mutating alert query is rejected unless the database allows DML."""
-    from unittest.mock import MagicMock
-
-    from marshmallow import ValidationError
-
-    from superset.commands.report.base import BaseReportScheduleCommand
-    from superset.commands.report.exceptions import (
-        AlertQueryDMLNotAllowedValidationError,
-    )
-
-    database = MagicMock()
-    database.backend = "sqlite"
-    database.allow_dml = False
-
-    exceptions: list[ValidationError] = []
-    BaseReportScheduleCommand().validate_alert_query(
-        database, "UPDATE ab_user SET active = 1", exceptions
-    )
-
-    assert len(exceptions) == 1
-    assert isinstance(exceptions[0], AlertQueryDMLNotAllowedValidationError)
-
-
-def test_validate_alert_query_rejects_unauthorized_tables(
-    mocker: MockerFixture,
-) -> None:
-    """A single read-only statement referencing tables the user cannot access
-    is rejected via the table-level authorization check."""
-    from unittest.mock import MagicMock
-
-    from marshmallow import ValidationError
-
-    from superset.commands.report.base import BaseReportScheduleCommand
-    from superset.commands.report.exceptions import (
-        AlertQueryDataAccessValidationError,
-    )
-    from superset.errors import ErrorLevel, SupersetError, SupersetErrorType
-    from superset.exceptions import SupersetSecurityException
-
-    mocker.patch(
-        "superset.commands.report.base.security_manager.raise_for_access",
-        side_effect=SupersetSecurityException(
-            SupersetError(
-                error_type=SupersetErrorType.TABLE_SECURITY_ACCESS_ERROR,
-                message="You need access to the following tables: `secret`",
-                level=ErrorLevel.ERROR,
-            )
-        ),
-    )
-
-    database = MagicMock()
-    database.backend = "sqlite"
-    database.allow_dml = False
-
-    exceptions: list[ValidationError] = []
-    BaseReportScheduleCommand().validate_alert_query(
-        database, "SELECT * FROM secret", exceptions
-    )
-
-    assert len(exceptions) == 1
-    assert isinstance(exceptions[0], AlertQueryDataAccessValidationError)

tests/unit_tests/commands/report/update_test.py

@@ -81,12 +81,6 @@ def _setup_mocks(mocker: MockerFixture, model: Mock) -> None:
         UpdateReportScheduleCommand,
         "validate_report_frequency",
     )
-    # Alert-query validation has dedicated coverage in base_test.py; these
-    # tests focus on database-presence handling, so stub it out here.
-    mocker.patch.object(
-        UpdateReportScheduleCommand,
-        "validate_alert_query",
-    )
     mocker.patch(
         "superset.commands.report.update.compute_subjects",
     )

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/connectors/sqla/models_test.py

@@ -39,11 +39,7 @@ from superset.exceptions import (
     SupersetSecurityException,
 )
 from superset.models.core import Database
-from superset.models.helpers import (
-    ExploreMixin,
-    validate_adhoc_subquery,
-    validate_rendered_expression,
-)
+from superset.models.helpers import ExploreMixin, validate_adhoc_subquery
 from superset.sql.parse import Table
 from superset.superset_typing import QueryObjectDict
 from superset.utils import json
@@ -1649,79 +1645,6 @@ def test_get_sqla_col_catches_subquery_beside_unparseable_syntax(
         tc.get_sqla_col()
 
 
-def test_validate_rendered_expression_rejects_multi_statement(
-    mocker: MockerFixture,
-) -> None:
-    database = _database_for_expression(mocker)
-    with pytest.raises(QueryObjectValidationError):
-        validate_rendered_expression("1; DROP TABLE users", database, None, "public")
-
-
-def test_validate_rendered_expression_rejects_set_operation(
-    mocker: MockerFixture,
-) -> None:
-    database = _database_for_expression(mocker)
-    with pytest.raises(QueryObjectValidationError):
-        validate_rendered_expression(
-            "1 UNION SELECT password FROM ab_user", database, None, "public"
-        )
-
-
-def test_validate_rendered_expression_rejects_subquery(
-    mocker: MockerFixture,
-) -> None:
-    """
-    With ``ALLOW_ADHOC_SUBQUERY=False`` (the default), a rendered expression
-    containing a sub-query is rejected by the same ``validate_adhoc_subquery``
-    gate used for stored and adhoc expressions.
-    """
-    database = _database_for_expression(mocker)
-    mocker.patch("superset.models.helpers.is_feature_enabled", return_value=False)
-    with pytest.raises(QueryObjectValidationError):
-        validate_rendered_expression(
-            "(SELECT password FROM ab_user LIMIT 1)", database, None, "public"
-        )
-
-
-def test_validate_rendered_expression_accepts_valid_expression(
-    mocker: MockerFixture,
-) -> None:
-    """A benign rendered expression is returned unchanged (no RLS applied)."""
-    database = _database_for_expression(mocker)
-    mocker.patch("superset.models.helpers.is_feature_enabled", return_value=False)
-    result = validate_rendered_expression("SUM(amount)", database, None, "public")
-    assert result == "SUM(amount)"
-
-
-def test_get_sqla_col_revalidates_rendered_jinja_expression(
-    mocker: MockerFixture,
-) -> None:
-    """
-    A Jinja block that renders into a sub-query must be rejected at query
-    time: save-time validation only sees the block as a placeholder, so the
-    rendered expression is re-validated before it is embedded via
-    ``literal_column``. The failure surfaces as a chart-level
-    ``QueryObjectValidationError``, matching the stored-expression path,
-    rather than a raw ``SupersetSecurityException``.
-    """
-    # A real Database (not a MagicMock) so the ORM relationship assignment on
-    # SqlaTable has a valid instance state; sqlite gives a concrete backend.
-    database = Database(database_name="t", sqlalchemy_uri="sqlite://")
-    mocker.patch("superset.models.helpers.is_feature_enabled", return_value=False)
-    table = SqlaTable(table_name="t", database=database)
-    tbl_column = TableColumn(
-        column_name="c",
-        expression='{{ "(SELECT password FROM ab_user LIMIT 1)" }}',
-        table=table,
-    )
-    template_processor = mocker.MagicMock()
-    template_processor.process_template.return_value = (
-        "(SELECT password FROM ab_user LIMIT 1)"
-    )
-    with pytest.raises(QueryObjectValidationError):
-        tbl_column.get_sqla_col(template_processor=template_processor)
-
-
 def test_has_extra_cache_key_calls_scans_guest_token_rls(
     mocker: MockerFixture,
 ) -> None:

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/models/test_virtual_dataset_format.py

@@ -38,7 +38,6 @@ import pytest
 from flask import Flask
 from sqlalchemy.sql.elements import TextClause
 
-from superset.exceptions import QueryObjectValidationError
 from superset.models.helpers import ExploreMixin
 from superset.sql.parse import RLSMethod, SQLStatement, Table
 
@@ -381,36 +380,3 @@ class TestRLSSubqueryAlias:
 
         assert "is_green" in result
         assert "WHERE" in result  # RLS predicate applied
-
-
-# ---------------------------------------------------------------------------
-# 4. RLS injection failures must fail closed when predicates apply
-# ---------------------------------------------------------------------------
-
-
-class TestVirtualDatasetRLSFailClosed:
-    """
-    When RLS predicates exist for the underlying tables but cannot be
-    injected into the virtual dataset SQL, the query must be aborted
-    instead of running against the unfiltered inner SQL.
-    """
-
-    @patch(
-        "superset.models.helpers.get_predicates_for_table",
-        return_value=["user_id = 42"],
-    )
-    @patch(
-        "superset.models.helpers.apply_rls",
-        side_effect=NotImplementedError("engine cannot apply RLS"),
-    )
-    def test_raises_when_rls_predicates_cannot_be_applied(
-        self,
-        mock_apply_rls: MagicMock,
-        mock_get_predicates: MagicMock,
-        virtual_datasource: MagicMock,
-        app: Flask,
-    ) -> None:
-        _set_virtual_sql(virtual_datasource, "SELECT pen_id FROM public.pens")
-
-        with pytest.raises(QueryObjectValidationError):
-            virtual_datasource.get_from_clause(template_processor=None)

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

@@ -780,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.
@@ -2387,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",
     [
@@ -2867,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(
@@ -2887,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.
@@ -3211,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(

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"