fix(dataset): let OAuth2RedirectError reach caller through virtual-table path

get_columns_description wrapped every exception (including OAuth2RedirectError) raised during query execution into SupersetGenericDBErrorException, so a virtual dataset's OAuth2 challenge never reached the create.py guard meant to preserve it. Also restore @safe's opaque-500 behavior for unexpected exceptions on DatasetRestApi.post, since dropping @safe to let OAuth2RedirectError through otherwise let raw exception text leak via Flask's global error handler. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Merge branch 'master' into enxdev/fix/sql-lab-save-dataset
2026-08-28 19:11:16 +00:00 · 2026-08-28 18:28:34 +02:00 · 2026-08-28 14:53:30 +02:00 · 2026-08-28 14:52:12 +02:00 · 2026-08-28 14:42:12 +02:00 · 2026-08-28 11:28:21 +02:00
61 changed files with 737 additions and 1068 deletions
@@ -10,7 +10,6 @@
 .stylelintignore
 .flake8
 .nvmrc
-.npmrc
 .rat-excludes
 .swcrc
 .*log
@@ -493,8 +493,8 @@ Frontend assets (TypeScript, JavaScript, CSS, and images) must be compiled in or

 First, be sure you are using the following versions of Node.js and npm:

- `Node.js`: Version 24 (see `superset-frontend/.nvmrc` for the exact version)
- `npm`: Version 11
+- `Node.js`: Version 22 (LTS)
+- `npm`: Version 10

 We recommend using [nvm](https://github.com/nvm-sh/nvm) to manage your node environment:

@@ -507,8 +507,8 @@ export NVM_DIR="$HOME/.nvm"
 [ -s "$NVM_DIR/bash_completion" ] && \. "$NVM_DIR/bash_completion"  # This loads nvm bash_completion

 cd superset-frontend
-nvm install
-nvm use
+nvm install --lts
+nvm use --lts
 ```

 Or if you use the default macOS starting with Catalina shell `zsh`, try:
@@ -21,7 +21,6 @@ from abc import ABC, abstractmethod
 from typing import Any, Generic, TypeVar

 from pydantic import BaseModel
-
 from superset_core.semantic_layers.view import SemanticView

 ConfigT = TypeVar("ConfigT", bound=BaseModel)
@@ -1 +0,0 @@
-../superset-frontend/.npmrc
@@ -1 +0,0 @@
-min-release-age=3
@@ -392,7 +392,7 @@
    "@luma.gl/shadertools": "~9.2.5",
    "@luma.gl/webgl": "~9.2.5",
    "core-js": "^3.38.1",
-    "dompurify": "^3.4.13",
+    "dompurify": "^3.4.11",
    "esbuild": "^0.28.1",
    "eslint-plugin-import": {
      "eslint": "$eslint"
@@ -38,21 +38,11 @@ function formatMemory(
        : ['B', 'kB', 'MB', 'GB', 'TB', 'PB', 'EB', 'ZB', 'YB', 'RB', 'QB'];
      const base = binary ? 1024 : 1000;

-      let i = Math.max(
-        0,
-        Math.min(
-          suffixes.length - 1,
-          Math.floor(Math.log(absValue) / Math.log(base)),
-        ),
+      const i = Math.min(
+        suffixes.length - 1,
+        Math.floor(Math.log(absValue) / Math.log(base)),
      );
-      let scaled = parseFloat((absValue / Math.pow(base, i)).toFixed(decimals));
-
-      if (scaled >= base && i < suffixes.length - 1) {
-        i += 1;
-        scaled = parseFloat((absValue / Math.pow(base, i)).toFixed(decimals));
-      }
-
-      formatted = `${sign}${scaled}${suffixes[i]}`;
+      formatted = `${sign}${parseFloat((absValue / Math.pow(base, i)).toFixed(decimals))}${suffixes[i]}`;
    }

    if (transfer) {
@@ -60,31 +60,6 @@ test('formats float bytes in human readable format with default options', () =>
  expect(formatter(1200.666)).toBe('1.2kB');
 });

-test('formats values below one byte without dropping the unit', () => {
-  const formatter = createMemoryFormatter();
-  expect(formatter(0.5)).toBe('0.5B');
-  expect(formatter(0.004)).toBe('0B');
-  expect(formatter(-0.25)).toBe('-0.25B');
-
-  const binaryFormatter = createMemoryFormatter({ binary: true });
-  expect(binaryFormatter(0.5)).toBe('0.5B');
-});
-
-test('rolls over to the next unit when rounding reaches the base', () => {
-  const formatter = createMemoryFormatter();
-  expect(formatter(999999)).toBe('1MB');
-  expect(formatter(999995)).toBe('1MB');
-  expect(formatter(999994)).toBe('999.99kB');
-  expect(formatter(-999999)).toBe('-1MB');
-
-  const binaryFormatter = createMemoryFormatter({ binary: true });
-  expect(binaryFormatter(1024 * 1024 - 1)).toBe('1MiB');
-
-  // the largest unit has nothing to roll over into
-  const largest = createMemoryFormatter();
-  expect(largest(Math.pow(1000, 11))).toBe('1000QB');
-});
-
 test('formats bytes in human readable format with additional binary option', () => {
  const formatter = createMemoryFormatter({ binary: true });
  expect(formatter(0)).toBe('0B');
@@ -38,8 +38,6 @@ import { EchartsTimeseriesSeriesType } from '../Timeseries/types';
 import {
  legendSection,
  minorTicks,
-  axisTicks,
-  gridlines,
  richTooltipSection,
  truncateXAxis,
  xAxisBounds,
@@ -393,8 +391,6 @@ const config: ControlPanelConfig = {
        ...createCustomizeSection(t('Query B'), 'B'),
        ['zoomable'],
        [minorTicks],
-        [axisTicks],
-        [gridlines],
        ...legendSection,
        [<ControlSubSectionHeader>{t('X Axis')}</ControlSubSectionHeader>],
        ['x_axis_time_format'],
@@ -184,8 +184,6 @@ export default function transformProps(
    opacityB,
    minorSplitLine,
    minorTicks,
-    gridlines,
-    axisTicks,
    seriesType,
    seriesTypeB,
    showLegend,
@@ -790,8 +788,6 @@ export default function transformProps(
        }),
      },
      minorTick: { show: minorTicks },
-      axisTick: { show: axisTicks ? 'auto' : false },
-      ...(gridlines ? {} : { splitLine: { show: false } }),
      minInterval:
        xAxisType === AxisType.Time && resolvedTimeGrain && !forceMaxInterval
          ? (TIMEGRAIN_TO_TIMESTAMP[
@@ -822,8 +818,6 @@ export default function transformProps(
        min: yAxisMin,
        max: yAxisMax,
        minorTick: { show: minorTicks },
-        axisTick: { show: axisTicks ? 'auto' : false },
-        splitLine: { show: gridlines },
        minorSplitLine: { show: minorSplitLine },
        axisLabel: {
          formatter: getYAxisFormatter(
@@ -846,7 +840,6 @@ export default function transformProps(
        min: minSecondary,
        max: maxSecondary,
        minorTick: { show: minorTicks },
-        axisTick: { show: axisTicks ? 'auto' : false },
        splitLine: { show: false },
        minorSplitLine: { show: minorSplitLine },
        axisLabel: {
@@ -48,8 +48,6 @@ export type EchartsMixedTimeseriesFormData = QueryFormData & {
  // shared properties
  minorSplitLine: boolean;
  minorTicks: boolean;
-  gridlines: boolean;
-  axisTicks: boolean;
  logAxis: boolean;
  logAxisSecondary: boolean;
  yAxisFormat?: string;
@@ -115,8 +113,6 @@ export const DEFAULT_FORM_DATA: EchartsMixedTimeseriesFormData = {
  ...DEFAULT_LEGEND_FORM_DATA,
  annotationLayers: [],
  minorSplitLine: TIMESERIES_DEFAULTS.minorSplitLine,
-  gridlines: TIMESERIES_DEFAULTS.gridlines,
-  axisTicks: TIMESERIES_DEFAULTS.axisTicks,
  truncateYAxis: TIMESERIES_DEFAULTS.truncateYAxis,
  truncateYAxisSecondary: TIMESERIES_DEFAULTS.truncateYAxis,
  logAxis: TIMESERIES_DEFAULTS.logAxis,
@@ -44,8 +44,6 @@ import {
  truncateXAxis,
  xAxisBounds,
  minorTicks,
-  axisTicks,
-  gridlines,
  forceMaxInterval,
 } from '../../controls';
 import { AreaChartStackControlOptions } from '../../constants';
@@ -176,8 +174,6 @@ const config: ControlPanelConfig = {
          },
        ],
        [minorTicks],
-        [axisTicks],
-        [gridlines],
        ['zoomable'],
        ...legendSection,
        [<ControlSubSectionHeader>{t('X Axis')}</ControlSubSectionHeader>],
@@ -133,8 +133,6 @@ const defaultFormData: EchartsTimeseriesFormData & {
  metrics: [],
  minorSplitLine: false,
  minorTicks: false,
-  gridlines: true,
-  axisTicks: true,
  opacity: 1,
  orderDesc: false,
  rowLimit: 0,
@@ -40,8 +40,6 @@ import {
 import {
  legendSection,
  minorTicks,
-  axisTicks,
-  gridlines,
  richTooltipSection,
  seriesOrderSection,
  showValueSectionWithoutStream,
@@ -390,8 +388,6 @@ const config: ControlPanelConfig = {
          },
        ],
        [minorTicks],
-        [axisTicks],
-        [gridlines],
        ['zoomable'],
        ...legendSection,
        [<ControlSubSectionHeader>{t('X Axis')}</ControlSubSectionHeader>],
@@ -37,8 +37,6 @@ import {
 import {
  legendSection,
  minorTicks,
-  axisTicks,
-  gridlines,
  richTooltipSection,
  seriesOrderSection,
  showValueSection,
@@ -158,8 +156,6 @@ const config: ControlPanelConfig = {
        ],
        ['zoomable'],
        [minorTicks],
-        [axisTicks],
-        [gridlines],
        ...legendSection,
        [<ControlSubSectionHeader>{t('X Axis')}</ControlSubSectionHeader>],
        [
@@ -42,8 +42,6 @@ import {
 import {
  legendSection,
  minorTicks,
-  axisTicks,
-  gridlines,
  richTooltipSection,
  seriesOrderSection,
  showValueSection,
@@ -482,8 +480,6 @@ const config: ControlPanelConfig = {
        ],
        ['zoomable'],
        [minorTicks],
-        [axisTicks],
-        [gridlines],
        ...legendSection,
        [<ControlSubSectionHeader>{t('X Axis')}</ControlSubSectionHeader>],
        ...createAxisControl('x'),
@@ -37,8 +37,6 @@ import {
 import {
  legendSection,
  minorTicks,
-  axisTicks,
-  gridlines,
  richTooltipSection,
  seriesOrderSection,
  showValueSectionWithoutStack,
@@ -107,8 +105,6 @@ const config: ControlPanelConfig = {
        ],
        ['zoomable'],
        [minorTicks],
-        [axisTicks],
-        [gridlines],
        ...legendSection,
        [<ControlSubSectionHeader>{t('X Axis')}</ControlSubSectionHeader>],
        [
@@ -35,8 +35,6 @@ import { DEFAULT_FORM_DATA, TIME_SERIES_DESCRIPTION_TEXT } from '../constants';
 import {
  legendSection,
  minorTicks,
-  axisTicks,
-  gridlines,
  richTooltipSection,
  seriesOrderSection,
  showValueSection,
@@ -159,8 +157,6 @@ const config: ControlPanelConfig = {
        ],
        ['zoomable'],
        [minorTicks],
-        [axisTicks],
-        [gridlines],
        ...legendSection,
        [<ControlSubSectionHeader>{t('X Axis')}</ControlSubSectionHeader>],
        [
@@ -67,8 +67,6 @@ export const DEFAULT_FORM_DATA: EchartsTimeseriesFormData = {
  maxMarkerSize: 30,
  minMarkerSize: 5,
  minorSplitLine: false,
-  gridlines: true,
-  axisTicks: true,
  opacity: 0.2,
  orderDesc: true,
  rowLimit: 10000,
@@ -283,8 +283,6 @@ export default function transformProps(
    metrics,
    minorSplitLine,
    minorTicks,
-    gridlines,
-    axisTicks,
    onlyTotal,
    opacity,
    orientation,
@@ -1282,8 +1280,6 @@ export default function transformProps(
      }),
    },
    minorTick: { show: minorTicks },
-    axisTick: { show: axisTicks ? 'auto' : false },
-    ...(gridlines ? {} : { splitLine: { show: false } }),
    minInterval:
      xAxisType === AxisType.Time && resolvedTimeGrain && !forceMaxInterval
        ? (TIMEGRAIN_TO_TIMESTAMP[
@@ -1328,7 +1324,7 @@ export default function transformProps(
    max: yAxisMax,
    minorTick: { show: isSmallChart ? false : minorTicks },
    minorSplitLine: { show: isSmallChart ? false : minorSplitLine },
-    splitLine: { show: isSmallChart ? false : gridlines },
+    splitLine: { show: !isSmallChart },
    axisLabel: {
      show: !isMicroChart,
      showMinLabel: !isMicroChart,
@@ -1342,7 +1338,7 @@ export default function transformProps(
        yAxisFormat,
      ),
    },
-    axisTick: { show: isSmallChart ? false : axisTicks },
+    axisTick: { show: !isSmallChart },
    scale: truncateYAxis,
    name: isSmallChart ? undefined : yAxisTitle,
    nameGap: convertInteger(yAxisTitleMargin),
@@ -73,8 +73,6 @@ export type EchartsTimeseriesFormData = QueryFormData & {
  metrics: QueryFormMetric[];
  minorSplitLine: boolean;
  minorTicks: boolean;
-  gridlines: boolean;
-  axisTicks: boolean;
  opacity: number;
  orderDesc: boolean;
  rowLimit: number;
@@ -495,28 +495,6 @@ export const minorTicks: ControlSetItem = {
  },
 };

-export const axisTicks: ControlSetItem = {
-  name: 'axisTicks',
-  config: {
-    type: 'CheckboxControl',
-    label: t('Axis ticks'),
-    default: true,
-    renderTrigger: true,
-    description: t('Show the main ticks on axes.'),
-  },
-};
-
-export const gridlines: ControlSetItem = {
-  name: 'gridlines',
-  config: {
-    type: 'CheckboxControl',
-    label: t('Gridlines'),
-    default: true,
-    renderTrigger: true,
-    description: t('Draw split lines for the main value axis ticks.'),
-  },
-};
-
 export const forceCategorical: ControlSetItem = {
  name: 'forceCategorical',
  config: {
@@ -116,8 +116,6 @@ const formData: EchartsMixedTimeseriesFormData = {
  markerSizeB: 0,
  minorSplitLine: false,
  minorTicks: false,
-  gridlines: true,
-  axisTicks: true,
  opacity: 0,
  opacityB: 0,
  orderDesc: false,
@@ -1511,56 +1509,3 @@ describe('EchartsMixedTimeseries tooltip truncation', () => {
    expect(html).not.toContain(longSeriesName);
  });
 });
-
-function transformWithChrome(
-  overrides: Partial<EchartsMixedTimeseriesFormData>,
-) {
-  const chartProps = createEchartsTimeseriesTestChartProps<
-    EchartsMixedTimeseriesFormData,
-    EchartsMixedTimeseriesProps
-  >({
-    ...MIXED_TIMESERIES_CHART_PROPS_DEFAULTS,
-    defaultQueriesData: queriesData,
-    formData: { ...formData, ...overrides },
-    queriesData,
-  });
-  const { echartOptions } = transformProps(chartProps);
-  return {
-    xAxis: echartOptions.xAxis as any,
-    yAxis: echartOptions.yAxis as any[],
-  };
-}
-
-test('draws gridlines and axis ticks when both are enabled', () => {
-  const { xAxis, yAxis } = transformWithChrome({});
-
-  expect(yAxis[0].splitLine.show).toBe(true);
-  // Both axes keep ECharts' own default, which the Mixed chart never overrode.
-  expect(yAxis[0].axisTick.show).toBe('auto');
-  expect(xAxis.axisTick.show).toBe('auto');
-});
-
-test('hides the gridlines on the primary axis', () => {
-  const { xAxis, yAxis } = transformWithChrome({ gridlines: false });
-
-  expect(yAxis[0].splitLine.show).toBe(false);
-  // The secondary axis never draws gridlines, so the two grids cannot double up.
-  expect(yAxis[1].splitLine.show).toBe(false);
-  expect(xAxis.splitLine.show).toBe(false);
-});
-
-test('never turns the secondary axis gridlines on', () => {
-  const { xAxis, yAxis } = transformWithChrome({ gridlines: true });
-
-  expect(yAxis[0].splitLine.show).toBe(true);
-  expect(yAxis[1].splitLine.show).toBe(false);
-  expect(xAxis.splitLine).toBeUndefined();
-});
-
-test('hides the ticks on the x axis and both y axes', () => {
-  const { xAxis, yAxis } = transformWithChrome({ axisTicks: false });
-
-  expect(xAxis.axisTick.show).toBe(false);
-  expect(yAxis[0].axisTick.show).toBe(false);
-  expect(yAxis[1].axisTick.show).toBe(false);
-});
@@ -2766,76 +2766,3 @@ describe('tooltip for metrics whose labels end in forecast suffixes', () => {
    expect(html).toContain('>ci<');
  });
 });
-
-test('shows gridlines and axis ticks by default', () => {
-  const { echartOptions } = transformProps(createTestChartProps({}));
-  const xAxis = echartOptions.xAxis as any;
-  const yAxis = echartOptions.yAxis as any;
-
-  expect(yAxis.splitLine.show).toBe(true);
-  expect(yAxis.axisTick.show).toBe(true);
-  // Left to ECharts, which draws no ticks on a banded category axis. Forcing
-  // true would add ticks the chart does not have today.
-  expect(xAxis.axisTick.show).toBe('auto');
-});
-
-test('hides gridlines without touching the minor split lines', () => {
-  const { echartOptions } = transformProps(
-    createTestChartProps({ formData: { gridlines: false } }),
-  );
-  const yAxis = echartOptions.yAxis as any;
-
-  expect(yAxis.splitLine.show).toBe(false);
-  expect(yAxis.minorSplitLine.show).toBe(DEFAULT_FORM_DATA.minorSplitLine);
-  expect(yAxis.axisTick.show).toBe(true);
-});
-
-test('leaves the category axis split lines alone until gridlines are turned off', () => {
-  const shown = transformProps(createTestChartProps({}));
-  // Writing show:true here would draw gridlines on axis types that default to
-  // none, so the key is only ever added to hide them.
-  expect((shown.echartOptions.xAxis as any).splitLine).toBeUndefined();
-
-  const hidden = transformProps(
-    createTestChartProps({ formData: { gridlines: false } }),
-  );
-  expect((hidden.echartOptions.xAxis as any).splitLine.show).toBe(false);
-});
-
-test('hides the ticks on both axes', () => {
-  const { echartOptions } = transformProps(
-    createTestChartProps({ formData: { axisTicks: false } }),
-  );
-
-  expect((echartOptions.yAxis as any).axisTick.show).toBe(false);
-  expect((echartOptions.xAxis as any).axisTick.show).toBe(false);
-  expect((echartOptions.yAxis as any).splitLine.show).toBe(true);
-});
-
-test('keeps gridlines and ticks off on a compact chart even when both are enabled', () => {
-  const { echartOptions } = transformProps(
-    createTestChartProps({
-      height: TIMESERIES_CONSTANTS.compactChartHeight - 1,
-      formData: { gridlines: true, axisTicks: true },
-    }),
-  );
-  const yAxis = echartOptions.yAxis as any;
-
-  expect(yAxis.splitLine.show).toBe(false);
-  expect(yAxis.axisTick.show).toBe(false);
-});
-
-test('applies gridlines to the value axis after a horizontal orientation swaps it', () => {
-  const { echartOptions } = transformProps(
-    createTestChartProps({
-      formData: {
-        orientation: OrientationType.Horizontal,
-        gridlines: false,
-      },
-    }),
-  );
-
-  // The transform swaps the axes for a horizontal chart, so the value axis —
-  // and the gridlines belonging to it — end up on xAxis.
-  expect((echartOptions.xAxis as any).splitLine.show).toBe(false);
-});
@@ -43,6 +43,23 @@ describe('SaveDatasetActionButton', () => {
    expect(saveDatasetBtn).toBeVisible();
  });

+  test('disables only the dataset button when canSaveDataset is false', () => {
+    const onSaveAsExplore = jest.fn();
+    render(
+      <SaveDatasetActionButton
+        setShowSave={() => true}
+        onSaveAsExplore={onSaveAsExplore}
+        canSaveDataset={false}
+      />,
+    );
+
+    // Saving the query needs no results.
+    expect(screen.getByRole('button', { name: 'Save' })).toBeEnabled();
+    expect(
+      screen.getByRole('button', { name: /save dataset/i }),
+    ).toBeDisabled();
+  });
+
  test('disables the save dataset button when the query did not run successfully', async () => {
    render(
      <SaveDatasetActionButton
@@ -19,12 +19,14 @@
 import { act, type ComponentProps } from 'react';
 import {
  cleanup,
+  createStore,
  fireEvent,
  render,
  screen,
  userEvent,
  waitFor,
 } from 'spec/helpers/testing-library';
+import reducerIndex from 'spec/helpers/reducerIndex';
 import fetchMock from 'fetch-mock';
 import { SaveDatasetModal } from 'src/SqlLab/components/SaveDatasetModal';
 import { createDatasource } from 'src/SqlLab/actions/sqlLab';
@@ -63,6 +65,12 @@ beforeEach(() => {
  cleanup();
 });

+afterEach(() => {
+  // In-body restores are skipped when an assertion throws, leaking a
+  // configured spy into later tests.
+  jest.restoreAllMocks();
+});
+
 // Mock createDatasource to return a thunk that resolves with the dataset's
 // new id. The test's mock store includes redux-thunk middleware (from RTK's
 // getDefaultMiddleware), so dispatch(createDatasource(...)) properly unwraps
@@ -518,6 +526,39 @@ describe('SaveDatasetModal', () => {
    });
  });

+  test('surfaces the error and keeps the modal open when saving fails', async () => {
+    // The chart-payload step's toast was built but never dispatched, so a
+    // failure there was silent.
+    const postFormData = jest.spyOn(
+      require('src/explore/exploreUtils/formData'),
+      'postFormData',
+    );
+    postFormData.mockRejectedValue(new Error('Boom'));
+    const onHide = jest.fn();
+    const store = createStore({ user }, reducerIndex);
+
+    render(<SaveDatasetModal {...mockedProps} onHide={onHide} />, { store });
+
+    fireEvent.change(screen.getByDisplayValue(/unimportant/i), {
+      target: { value: 'my dataset' },
+    });
+    userEvent.click(screen.getByRole('button', { name: /save/i }));
+
+    // `createStore` builds its reducer map at runtime, so state isn't typed.
+    const toasts = () =>
+      (
+        store.getState() as unknown as {
+          messageToasts: { toastType: string }[];
+        }
+      ).messageToasts;
+
+    await waitFor(() => {
+      expect(toasts()).toHaveLength(1);
+    });
+    expect(toasts()[0].toastType).toBe('DANGER_TOAST');
+    expect(onHide).not.toHaveBeenCalled();
+  });
+
  test('clearDatasetCache is imported and available', () => {
    const { clearDatasetCache } = require('src/utils/cachedSupersetGet');

@@ -61,6 +61,9 @@ import type Subject from 'src/types/Subject';
 import { openInNewTab, redirect } from 'src/utils/navigationUtils';
 import { mapSubjectValuesToIds } from 'src/features/subjects/SubjectPicker';

+// Derived so it can't drift from what `getClientErrorObject` accepts.
+type SaveErrorSource = Parameters<typeof getClientErrorObject>[0];
+
 interface QueryDatabase {
  id?: number;
 }
@@ -391,9 +394,18 @@ export const SaveDatasetModal = ({
        setDatasetName(getDefaultDatasetName());
        onHide();
      })
-      .catch(() => {
+      .catch((error?: SaveErrorSource) => {
        setLoading(false);
-        addDangerToast(t('An error occurred saving dataset'));
+        // `createDatasource` already toasted the server's message and rejects
+        // with nothing; only the chart-payload step needs its own.
+        if (!error) {
+          return;
+        }
+        getClientErrorObject(error).then(e =>
+          dispatch(
+            addDangerToast(e.error || t('An error occurred saving dataset')),
+          ),
+        );
      });
  };

@@ -27,6 +27,8 @@ import {
 import SaveQuery from 'src/SqlLab/components/SaveQuery';
 import { initialState, databases } from 'src/SqlLab/fixtures';

+const RESULT_COLUMNS = [{ column_name: 'col', type: 'STRING' }];
+
 const mockedProps = {
  queryEditorId: '123',
  animation: false,
@@ -35,7 +37,6 @@ const mockedProps = {
  onSave: () => {},
  saveQueryWarning: null,
  columns: [],
-  canSaveDataset: true,
 };

 const mockState = {
@@ -60,8 +61,31 @@ const splitSaveBtnProps = {
    ...mockedProps.database,
    allows_virtual_table_explore: true,
  },
+  columns: RESULT_COLUMNS,
 };

+const EDITOR_SQL = 'SELECT * FROM t';
+
+const stateWithLatestQuery = ({
+  id,
+  state,
+  sql = EDITOR_SQL,
+}: {
+  id: string;
+  state: string;
+  sql?: string;
+}) => ({
+  ...mockState,
+  sqlLab: {
+    ...mockState.sqlLab,
+    queryEditors: mockState.sqlLab.queryEditors.map(qe => ({
+      ...qe,
+      latestQueryId: id,
+    })),
+    queries: { [id]: { id, state, sql } },
+  },
+});
+
 const middlewares = [thunk];
 const mockStore = configureStore(middlewares);

@@ -97,6 +121,71 @@ describe('SavedQuery', () => {
    expect(saveBtn).toBeVisible();
  });

+  test('blocks "Save dataset" until the query has run successfully', () => {
+    // Without a successful run the save can only fail server-side.
+    render(<SaveQuery {...splitSaveBtnProps} />, {
+      useRedux: true,
+      store: mockStore(stateWithLatestQuery({ id: 'qid-1', state: 'failed' })),
+    });
+
+    expect(
+      screen.getByRole('button', { name: /save dataset/i }),
+    ).toBeDisabled();
+    // Saving the query itself is unaffected.
+    expect(screen.getByRole('button', { name: 'Save' })).toBeEnabled();
+  });
+
+  test('blocks "Save dataset" when no query has been run at all', () => {
+    render(<SaveQuery {...splitSaveBtnProps} />, {
+      useRedux: true,
+      store: mockStore(mockState),
+    });
+
+    expect(
+      screen.getByRole('button', { name: /save dataset/i }),
+    ).toBeDisabled();
+  });
+
+  test('blocks "Save dataset" when the SQL changed after a successful run', () => {
+    // The run succeeded, but not for what is in the editor now -- and it is
+    // the editor's SQL that gets saved.
+    render(<SaveQuery {...splitSaveBtnProps} />, {
+      useRedux: true,
+      store: mockStore(
+        stateWithLatestQuery({
+          id: 'qid-1',
+          state: 'success',
+          sql: 'SELECT 1 AS ran_earlier',
+        }),
+      ),
+    });
+
+    expect(
+      screen.getByRole('button', { name: /save dataset/i }),
+    ).toBeDisabled();
+  });
+
+  test('blocks "Save dataset" when the successful query returned no columns', () => {
+    // e.g. a DDL/DML statement -- there is nothing to introspect into a dataset.
+    render(<SaveQuery {...splitSaveBtnProps} columns={[]} />, {
+      useRedux: true,
+      store: mockStore(stateWithLatestQuery({ id: 'qid-1', state: 'success' })),
+    });
+
+    expect(
+      screen.getByRole('button', { name: /save dataset/i }),
+    ).toBeDisabled();
+  });
+
+  test('enables "Save dataset" once the query has succeeded', () => {
+    render(<SaveQuery {...splitSaveBtnProps} />, {
+      useRedux: true,
+      store: mockStore(stateWithLatestQuery({ id: 'qid-1', state: 'success' })),
+    });
+
+    expect(screen.getByRole('button', { name: /save dataset/i })).toBeEnabled();
+  });
+
  test('renders a save query modal when user clicks save button', () => {
    render(<SaveQuery {...mockedProps} />, {
      useRedux: true,
@@ -234,7 +323,7 @@ describe('SavedQuery', () => {
  test('renders a save dataset modal when user clicks "save dataset" menu item', async () => {
    render(<SaveQuery {...splitSaveBtnProps} />, {
      useRedux: true,
-      store: mockStore(mockState),
+      store: mockStore(stateWithLatestQuery({ id: 'qid-1', state: 'success' })),
    });

    const saveDatasetMenuItem = await screen.findByLabelText(/save dataset/i);
@@ -248,7 +337,7 @@ describe('SavedQuery', () => {
  test('renders the save dataset modal UI', async () => {
    render(<SaveQuery {...splitSaveBtnProps} />, {
      useRedux: true,
-      store: mockStore(mockState),
+      store: mockStore(stateWithLatestQuery({ id: 'qid-1', state: 'success' })),
    });
    const saveDatasetMenuItem = await screen.findByLabelText(/save dataset/i);
    userEvent.click(saveDatasetMenuItem);
@@ -17,6 +17,8 @@
 * under the License.
 */
 import { useState, useEffect, useMemo, ChangeEvent } from 'react';
+import { useSelector } from 'react-redux';
+import { Query, QueryState } from '@superset-ui/core';
 import type { DatabaseObject } from 'src/features/databases/types';
 import { t } from '@apache-superset/core/translation';
 import { styled } from '@apache-superset/core/theme';
@@ -37,7 +39,7 @@ import {
 } from 'src/SqlLab/components/SaveDatasetModal';
 import { getDatasourceAsSaveableDataset } from 'src/utils/datasourceUtils';
 import useQueryEditor from 'src/SqlLab/hooks/useQueryEditor';
-import { QueryEditor } from 'src/SqlLab/types';
+import { QueryEditor, SqlLabRootState } from 'src/SqlLab/types';
 import useLogAction from 'src/logger/useLogAction';
 import {
  LOG_ACTIONS_SQLLAB_CREATE_CHART,
@@ -52,7 +54,6 @@ interface SaveQueryProps {
  onUpdate: (arg0: QueryPayload, id: string) => void;
  saveQueryWarning: string | null;
  database: Partial<DatabaseObject> | undefined;
-  canSaveDataset: boolean;
 }

 export type QueryPayload = {
@@ -82,7 +83,6 @@ const SaveQuery = ({
  saveQueryWarning,
  database,
  columns,
-  canSaveDataset,
 }: SaveQueryProps) => {
  const queryEditor = useQueryEditor(queryEditorId, [
    'autorun',
@@ -113,6 +113,17 @@ const SaveQuery = ({
  const [label, setLabel] = useState<string>(defaultLabel);
  const [showSave, setShowSave] = useState<boolean>(false);
  const [showSaveDatasetModal, setShowSaveDatasetModal] = useState(false);
+  // Saving a dataset runs the SQL to introspect columns, so it needs a
+  // successful run of the SQL being saved that produced at least one column
+  // -- editing after a run invalidates it, and running a selection only
+  // validates that selection.
+  const latestQuery = useSelector<SqlLabRootState, Query | undefined>(
+    ({ sqlLab }) => sqlLab.queries[queryEditor.latestQueryId || ''],
+  );
+  const canSaveDataset =
+    latestQuery?.state === QueryState.Success &&
+    latestQuery.sql === queryEditor.sql &&
+    columns.length > 0;
  const isSaved = !!query.remoteId;
  const isLabelEmpty = label.trim().length === 0;
  const canExploreDatabase = !!database?.allows_virtual_table_explore;
@@ -362,6 +362,7 @@ describe('SqlEditor', () => {
  test('enables the save dataset button when the latest query succeeded', async () => {
    const { findByLabelText } = setupWithLatestQuery({
      state: QueryState.Success,
+      sql: mockedProps.queryEditor.sql,
    });
    expect(await findByLabelText('Save dataset')).toBeEnabled();
  });
@@ -868,7 +868,6 @@ const SqlEditor: FC<Props> = ({
          }
          saveQueryWarning={saveQueryWarning}
          database={database}
-          canSaveDataset={successful && resultColumns.length > 0}
        />
        <ShareSqlLabQuery queryEditorId={queryEditor.id} />
      </>
@@ -312,7 +312,7 @@ export function handleComponentDrop(dropResult: DropResult) {
      source &&
      !(
        // ensure it has moved
-        destination.id === source.id && destination.index === source.index
+        (destination.id === source.id && destination.index === source.index)
      )
    ) {
      dispatch(moveComponent(dropResult));
@@ -126,7 +126,7 @@ function fillNativeFilters(
      !(
        // Treat all-null arrays (range filters use [null, null] as their
        // canonical cleared value) and empty arrays as "no value".
-        Array.isArray(loadedValue) && loadedValue.every(v => v === null)
+        (Array.isArray(loadedValue) && loadedValue.every(v => v === null))
      );
    const loadedHasExtraFormData =
      !!loaded?.extraFormData && Object.keys(loaded.extraFormData).length > 0;
@@ -29,9 +29,7 @@ import { ControlFormItemComponents } from './ControlForm';
 * Column formatting configs.
 */
 export type ColumnConfig = {
-  [
-    key in SharedColumnConfigProp
-  ]?: (typeof SHARED_COLUMN_CONFIG_PROPS)[key]['value'];
+  [key in SharedColumnConfigProp]?: (typeof SHARED_COLUMN_CONFIG_PROPS)[key]['value'];
 } & Record<string, StrictJsonValue>;

 /**
@@ -951,167 +951,3 @@ test('filters the subject select by column verbose_name as well as column_name',
  expect(within(dropdown).getByText('total_count')).toBeInTheDocument();
  expect(within(dropdown).queryByText('Full Name')).not.toBeInTheDocument();
 });
-
-const COLUMN_VALUES_ENDPOINT =
-  'glob:*/api/v1/datasource/*/column/value/values/*';
-
-let columnValues: { result: unknown[]; limit: number } = {
-  result: [],
-  limit: 10000,
-};
-fetchMock.get(COLUMN_VALUES_ENDPOINT, () => columnValues);
-
-const setupWithFilterValues = (result: unknown[], limit = 10000) => {
-  columnValues = { result, limit };
-  const onChange = jest.fn();
-  const validHandler = jest.fn();
-  const spy = jest.spyOn(redux, 'useSelector');
-  spy.mockReturnValue({});
-  const props = {
-    adhocFilter: new AdhocFilter({
-      expressionType: ExpressionTypes.Simple,
-      subject: 'value',
-      operatorId: Operators.In,
-      operator: OPERATOR_ENUM_TO_OPERATOR_TYPE[Operators.In].operation,
-      comparator: [],
-      clause: Clauses.Where,
-    }),
-    onChange,
-    options,
-    datasource: {
-      ...TestDataset,
-      columns: [{ column_name: 'value', type: 'VARCHAR', id: 3 }],
-      filter_select: true,
-    },
-    partitionColumn: 'test',
-    validHandler,
-  };
-  render(
-    <AdhocFilterEditPopoverSimpleTabContent {...(props as unknown as Props)} />,
-  );
-  return props;
-};
-
-const openComparator = async () => {
-  const comparator = screen.getByRole('combobox', {
-    name: 'Comparator option',
-  });
-  userEvent.click(comparator);
-  return comparator;
-};
-
-test('loads comparator values from the server', async () => {
-  setupWithFilterValues(['alpha', 'beta']);
-  await openComparator();
-  expect(await screen.findByTitle('alpha')).toBeInTheDocument();
-});
-
-test('sends the typed text to the server rather than filtering the loaded page', async () => {
-  // The loaded page is bounded, so matching client-side cannot reach a value
-  // beyond the row limit. The search has to reach the database.
-  setupWithFilterValues(['alpha']);
-  const comparator = await openComparator();
-  userEvent.type(comparator, 'gamma');
-
-  await waitFor(
-    () => {
-      const searched = fetchMock.callHistory
-        .calls(COLUMN_VALUES_ENDPOINT)
-        .map(call => String(call.url));
-      expect(searched.some(url => url.includes('q=gamma'))).toBe(true);
-    },
-    { timeout: 3000 },
-  );
-});
-
-test('lets a value the server did not return still be selected', async () => {
-  // Even with server-side search a match can fall outside the page; typing the
-  // exact value has to remain a way through.
-  setupWithFilterValues([]);
-  const comparator = await openComparator();
-  userEvent.type(comparator, 'not-in-the-page');
-  expect(await screen.findByTitle('not-in-the-page')).toBeInTheDocument();
-});
-
-test('does not query for values when the dataset disables them', async () => {
-  fetchMock.clearHistory();
-  setup({
-    adhocFilter: new AdhocFilter({
-      expressionType: ExpressionTypes.Simple,
-      subject: 'value',
-      operatorId: Operators.In,
-      operator: OPERATOR_ENUM_TO_OPERATOR_TYPE[Operators.In].operation,
-      comparator: [],
-      clause: Clauses.Where,
-    }),
-  });
-  await openComparator();
-  expect(fetchMock.callHistory.calls(COLUMN_VALUES_ENDPOINT)).toHaveLength(0);
-});
-
-test('stores the picked value, not the option object', async () => {
-  // AsyncSelect is labelInValue: taking its argument at face value puts
-  // {label, value} into the comparator, and the engine then fails to render it
-  // as a literal.
-  const props = setupWithFilterValues(['Michael']);
-  await openComparator();
-  userEvent.click(await screen.findByTitle('Michael'));
-
-  await waitFor(() => expect(props.onChange).toHaveBeenCalled());
-  const [filter] = props.onChange.mock.calls.at(-1);
-  expect(filter.comparator).toEqual(['Michael']);
-});
-
-test('can remove a value that was saved earlier', async () => {
-  // Reopening the popover restores the comparator from the saved filter, and
-  // the value is not in the freshly loaded page. Removing it has to still work.
-  columnValues = { result: [], limit: 10000 };
-  const onChange = jest.fn();
-  const validHandler = jest.fn();
-  jest.spyOn(redux, 'useSelector').mockReturnValue({});
-  render(
-    <AdhocFilterEditPopoverSimpleTabContent
-      {...({
-        adhocFilter: new AdhocFilter({
-          expressionType: ExpressionTypes.Simple,
-          subject: 'value',
-          operatorId: Operators.In,
-          operator: OPERATOR_ENUM_TO_OPERATOR_TYPE[Operators.In].operation,
-          comparator: ['Michael'],
-          clause: Clauses.Where,
-        }),
-        onChange,
-        options,
-        datasource: {
-          ...TestDataset,
-          columns: [{ column_name: 'value', type: 'VARCHAR', id: 3 }],
-          filter_select: true,
-        },
-        partitionColumn: 'test',
-        validHandler,
-      } as unknown as Props)}
-    />,
-  );
-
-  // Remove it the way a user does: the tag's own close control.
-  userEvent.click(await screen.findByLabelText('close'));
-
-  await waitFor(() => expect(onChange).toHaveBeenCalled());
-  const [filter] = onChange.mock.calls.at(-1);
-  expect(filter.comparator).toEqual([]);
-});
-
-test('says the list is partial when the server capped it', async () => {
-  setupWithFilterValues(['alpha', 'beta'], 2);
-  await openComparator();
-  expect(
-    await screen.findByText(/Only the first 2 values are listed/),
-  ).toBeInTheDocument();
-});
-
-test('does not say the list is partial when it is complete', async () => {
-  setupWithFilterValues(['alpha', 'beta'], 10000);
-  await openComparator();
-  expect(await screen.findByTitle('alpha')).toBeInTheDocument();
-  expect(screen.queryByText(/Only the first/)).not.toBeInTheDocument();
-});
@@ -16,25 +16,13 @@
 * specific language governing permissions and limitations
 * under the License.
 */
-import {
-  FC,
-  ChangeEvent,
-  useCallback,
-  useEffect,
-  useMemo,
-  useState,
-  useRef,
-} from 'react';
+import { FC, ChangeEvent, useEffect, useState, useRef } from 'react';

 import {
-  AsyncSelect,
  Input,
  InputRef,
  Select,
  Tooltip,
-  type AsyncSelectRef,
-  type LabeledValue,
-  type SelectOptionsTypePage,
  type SelectValue,
 } from '@superset-ui/core/components';
 import { t } from '@apache-superset/core/translation';
@@ -69,7 +57,7 @@ import { useDatePickerInAdhocFilter } from '../utils';
 import { useDefaultTimeFilter } from '../../DateFilterControl/utils';
 import { Clauses, ExpressionTypes } from '../types';

-const SelectWithLabel = styled(AsyncSelect)<{ labelText: string }>`
+const SelectWithLabel = styled(Select)<{ labelText: string }>`
  .ant-select-content::after {
    content: ${({ labelText }) => labelText || '\\A0'};
    display: inline-block;
@@ -79,30 +67,6 @@ const SelectWithLabel = styled(AsyncSelect)<{ labelText: string }>`
  }
 `;

-// The server answers with one bounded page, not an offset window: paging would
-// need a stable ORDER BY, and ordering a high-cardinality column is the full
-// scan this search exists to avoid. A page size no response can reach keeps
-// AsyncSelect from asking for a second page.
-const COMPARATOR_PAGE_SIZE = 1_000_000;
-
-const toLabeledValue = (value: unknown): LabeledValue => ({
-  value: value as LabeledValue['value'],
-  label: optionLabel(value as null | number | boolean | string),
-});
-
-// The reverse of toLabeledValue: what AsyncSelect emits is labelled, and the
-// comparator has to be the raw value or the engine cannot render it as a
-// literal.
-const unwrapComparator = (value: unknown): unknown => {
-  if (Array.isArray(value)) {
-    return value.map(unwrapComparator);
-  }
-  if (value !== null && typeof value === 'object' && 'value' in value) {
-    return (value as LabeledValue).value;
-  }
-  return value;
-};
-
 export interface SimpleExpressionType {
  expressionType: keyof typeof ExpressionTypes;
  column: ColumnMeta;
@@ -383,9 +347,11 @@ const AdhocFilterEditPopoverSimpleTabContent: FC<Props> = props => {
  } = useSimpleTabFilterProps(props);
  const [comparator, setComparator] = useState(props.adhocFilter.comparator);
  const comparatorInputRef = useRef<InputRef | null>(null);
-  const comparatorSelectRef = useRef<AsyncSelectRef>(null);
-  const [loadedOptionCount, setLoadedOptionCount] = useState(0);
-  const [optionsTruncated, setOptionsTruncated] = useState(false);
+  const [suggestions, setSuggestions] = useState<
+    Record<'label' | 'value', any>[]
+  >([]);
+  const [loadingComparatorSuggestions, setLoadingComparatorSuggestions] =
+    useState<boolean>(false);
  const [hasFocusedComparator, setHasFocusedComparator] =
    useState<boolean>(false);

@@ -421,8 +387,18 @@ const AdhocFilterEditPopoverSimpleTabContent: FC<Props> = props => {
    />
  );

-  const createSuggestionsPlaceholder = () =>
-    loadedOptionCount ? t('%s option(s)', loadedOptionCount) : '';
+  const getOptionsRemaining = () => {
+    // if select is multi/value is array, we show the options not selected
+    const valuesFromSuggestionsLength = Array.isArray(comparator)
+      ? comparator.filter(v => suggestions.includes(v)).length
+      : 0;
+    return suggestions ? suggestions.length - valuesFromSuggestionsLength : 0;
+  };
+  const createSuggestionsPlaceholder = () => {
+    const optionsRemaining = getOptionsRemaining();
+    const placeholder = t('%s option(s)', optionsRemaining);
+    return optionsRemaining ? placeholder : '';
+  };

  const handleSubjectChange = (subject: string) => {
    setComparator(undefined);
@@ -479,63 +455,21 @@ const AdhocFilterEditPopoverSimpleTabContent: FC<Props> = props => {
    operatorId !== undefined &&
    DISABLE_INPUT_OPERATORS.includes(operatorId as Operators);

-  const canSuggestComparatorValues = Boolean(
-    subjectString &&
-    props.datasource?.filter_select &&
-    props.adhocFilter.clause !== Clauses.Having,
-  );
-
  const hasComparatorOptions =
    (operatorId && MULTI_OPERATORS.has(operatorId as Operators)) ||
-    canSuggestComparatorValues;
-
-  // AsyncSelect is labelInValue, so the value it is given has to be labelled
-  // too. Handed a bare value it still renders, but `handleOnDeselect` then
-  // compares `element.value` against entries that have no `.value`, matches
-  // nothing, and the tag cannot be removed.
-  //
-  // Memoised because AsyncSelect resets its internal selection whenever the
-  // identity of `value` changes. A fresh array every render would wipe out
-  // each pick as soon as it was made.
-  const comparatorSelectValue = useMemo(
-    () =>
-      Array.isArray(comparator)
-        ? comparator.map(toLabeledValue)
-        : isDefined(comparator) && comparator !== ''
-          ? toLabeledValue(comparator)
-          : undefined,
-    [comparator],
-  );
-
-  const handleComparatorChange = useCallback(
-    (value: unknown) => {
-      onComparatorChange(unwrapComparator(value) as string);
-    },
-    // eslint-disable-next-line react-hooks/exhaustive-deps
-    [props.adhocFilter, props.onChange],
-  );
+    suggestions.length > 0;

  const comparatorSelectProps = {
    allowClear: true,
    allowNewOptions: true,
    ariaLabel: t('Comparator option'),
-    pageSize: COMPARATOR_PAGE_SIZE,
-    // A capped list reads as the whole set unless it says otherwise, so an
-    // absent value looks like a value that does not exist. Only shown when the
-    // list is actually cut short.
-    helperText: optionsTruncated
-      ? t(
-          'Only the first %s values are listed. Type to search all of them, ' +
-            'or enter a value that is not listed.',
-          loadedOptionCount,
-        )
-      : undefined,
    mode:
      operatorId && MULTI_OPERATORS.has(operatorId as Operators)
        ? ('multiple' as const)
        : ('single' as const),
-    value: comparatorSelectValue as SelectValue,
-    onChange: handleComparatorChange,
+    loading: loadingComparatorSuggestions,
+    value: comparator as SelectValue,
+    onChange: onComparatorChange,
    notFoundContent: t('Type a value here'),
    placeholder: createSuggestionsPlaceholder(),
  };
@@ -561,89 +495,76 @@ const AdhocFilterEditPopoverSimpleTabContent: FC<Props> = props => {
    onChange: onDatePickerChange,
  });

-  // Element-level array operators (Contains any / Contains all) search inside
-  // the array, so suggest individual elements; whole-array operators (=, In, …)
-  // keep the default distinct-array suggestions.
-  const arrayElements =
-    props.adhocFilter.operatorId === Operators.ContainsAny ||
-    props.adhocFilter.operatorId === Operators.ContainsAll;
-
-  // AsyncSelect throws away every loaded option when the identity of its
-  // `options` callback changes, so this depends on plain values rather than on
-  // `props.datasource`, whose identity the parent does not guarantee.
-  const datasourceType = props.datasource?.type;
-  const datasourceId = props.datasource?.id;
-
-  const loadComparatorOptions = useCallback(
-    async (search: string): Promise<SelectOptionsTypePage> => {
-      const col = subjectString;
-      if (!col || !canSuggestComparatorValues) {
-        return { data: [], totalCount: 0 };
-      }
-
-      const params = new URLSearchParams();
-      if (arrayElements) {
-        params.set('array_elements', 'true');
-      }
-      if (search) {
-        params.set('q', search);
-      }
-      const query = params.toString();
-
-      try {
-        const { json } = await SupersetClient.get({
-          endpoint:
-            `/api/v1/datasource/${datasourceType}/${datasourceId}` +
-            `/column/${encodeURIComponent(col)}/values/${query ? `?${query}` : ''}`,
-        });
-        const data = json.result.map((suggestion: unknown) => {
-          // Complex column values arrive as JS arrays or objects: whole arrays
-          // for MULTI_VALUE columns (e.g. [5, 6, 7]) and Map/Tuple objects for
-          // nested-container columns (e.g. {"a": ["x","y"]}). A raw
-          // array/object is neither a valid single-select value (antd collapses
-          // an array to its first element) nor renderable as a React child (an
-          // object throws). Render it as its literal string, which is also
-          // exactly what the backend's parse_array_literal expects for the
-          // whole-array operators.
-          if (suggestion !== null && typeof suggestion === 'object') {
-            const literal = JSON.stringify(suggestion);
-            return { value: literal, label: literal };
-          }
-          return {
-            value: suggestion as null | number | boolean | string,
-            label: optionLabel(suggestion as null | number | boolean | string),
-          };
-        });
-
-        setLoadedOptionCount(data.length);
-        setOptionsTruncated(isDefined(json.limit) && data.length >= json.limit);
-
-        // The count has to exceed what was returned. AsyncSelect treats
-        // `loaded >= totalCount` as "that is every value", sets allValuesLoaded
-        // and from then on serves searches by filtering the loaded page
-        // client-side -- which is the behaviour this whole change exists to
-        // replace. Pagination is held off by COMPARATOR_PAGE_SIZE instead.
-        return { data, totalCount: data.length + 1 };
-      } catch {
-        setLoadedOptionCount(0);
-        setOptionsTruncated(false);
-        return { data: [], totalCount: 0 };
-      }
-    },
-    [
-      subjectString,
-      canSuggestComparatorValues,
-      datasourceType,
-      datasourceId,
-      arrayElements,
-    ],
-  );
-
-  // Options are cached per search term inside AsyncSelect; a different column
-  // or a switch to element-level suggestions invalidates all of them.
  useEffect(() => {
-    comparatorSelectRef.current?.clearCache();
-  }, [subjectString, arrayElements]);
+    const refreshComparatorSuggestions = () => {
+      const { datasource } = props;
+      const col = props.adhocFilter.subject;
+      const having = props.adhocFilter.clause === Clauses.Having;
+
+      if (col && datasource && datasource.filter_select && !having) {
+        const controller = new AbortController();
+        const { signal } = controller;
+        if (loadingComparatorSuggestions) {
+          controller.abort();
+        }
+        // Element-level array operators (Contains any / Contains all) search
+        // inside the array, so suggest individual elements; whole-array
+        // operators (=, In, …) keep the default distinct-array suggestions.
+        const { operatorId } = props.adhocFilter;
+        const arrayElements =
+          operatorId === Operators.ContainsAny ||
+          operatorId === Operators.ContainsAll;
+        setLoadingComparatorSuggestions(true);
+        SupersetClient.get({
+          signal,
+          endpoint: `/api/v1/datasource/${datasource.type}/${datasource.id}/column/${col}/values/${
+            arrayElements ? '?array_elements=true' : ''
+          }`,
+        })
+          .then(({ json }) => {
+            setSuggestions(
+              json.result.map((suggestion: unknown) => {
+                // Complex column values arrive as JS arrays or objects: whole
+                // arrays for MULTI_VALUE columns (e.g. [5, 6, 7]) and Map/Tuple
+                // objects for nested-container columns (e.g. {"a": ["x","y"]}).
+                // A raw array/object is neither a valid single-select value
+                // (antd collapses an array to its first element) nor renderable
+                // as a React child (an object throws). Render it as its literal
+                // string, which is also exactly what the backend's
+                // parse_array_literal expects for the whole-array operators.
+                if (suggestion !== null && typeof suggestion === 'object') {
+                  const literal = JSON.stringify(suggestion);
+                  return { value: literal, label: literal };
+                }
+                return {
+                  value: suggestion as null | number | boolean | string,
+                  label: optionLabel(
+                    suggestion as null | number | boolean | string,
+                  ),
+                };
+              }),
+            );
+            setLoadingComparatorSuggestions(false);
+          })
+          .catch(() => {
+            setSuggestions([]);
+            setLoadingComparatorSuggestions(false);
+          });
+      }
+    };
+
+    if (!datePicker) {
+      refreshComparatorSuggestions();
+    }
+    // loadingComparatorSuggestions intentionally omitted - set inside effect, would cause infinite loop
+    // eslint-disable-next-line react-hooks/exhaustive-deps
+  }, [
+    props.adhocFilter.subject,
+    props.adhocFilter.clause,
+    props.adhocFilter.operatorId,
+    props.datasource,
+    datePicker,
+  ]);

  useEffect(() => {
    if (isFeatureEnabled(FeatureFlag.EnableAdvancedDataTypes)) {
@@ -749,12 +670,11 @@ const AdhocFilterEditPopoverSimpleTabContent: FC<Props> = props => {
            }
          >
            <SelectWithLabel
-              ref={comparatorSelectRef}
              css={css`
                margin-top: ${theme.marginXS}px;
              `}
              labelText={labelText}
-              options={loadComparatorOptions}
+              options={suggestions}
              {...comparatorSelectProps}
            />
          </Tooltip>
@@ -879,41 +879,10 @@ describe('SelectFilterPlugin', () => {
    expect(await screen.findByTitle('brand-new')).toBeInTheDocument();
  });

-  test('says the list is capped when it hits the row limit', async () => {
-    // 3 rows of data against a limit of 3: the user is looking at a page, not
-    // at every value the column has.
-    getWrapper({ rowLimit: 3 });
-    userEvent.click(screen.getAllByRole('combobox')[0]);
-    expect(
-      await screen.findByText(/Only the first 3 values are listed/),
-    ).toBeInTheDocument();
-  });
-
-  test('offers the ways out that the filter actually supports', async () => {
-    getWrapper({ rowLimit: 3, creatable: true, searchAllOptions: true });
-    userEvent.click(screen.getAllByRole('combobox')[0]);
-    expect(
-      await screen.findByText(/Type to search all of them/),
-    ).toBeInTheDocument();
-    expect(
-      screen.getByText(/You can enter a value that is not listed/),
-    ).toBeInTheDocument();
-  });
-
-  test('says nothing when the whole column fits under the limit', async () => {
-    getWrapper();
-    userEvent.click(screen.getAllByRole('combobox')[0]);
-    expect(await screen.findByRole('combobox')).toBeInTheDocument();
-    expect(screen.queryByText(/Only the first/)).not.toBeInTheDocument();
-  });
-
-  test('shows create option when searchAllOptions is true', async () => {
-    // Server-side search returns a bounded page, so a value that exists in the
-    // data can still be missing from the dropdown. Suppressing the create
-    // option there leaves the user with no way to apply it at all.
+  test('does not show create option when searchAllOptions is true', () => {
    getWrapper({ creatable: true, searchAllOptions: true });
    userEvent.type(screen.getByRole('combobox'), 'brand-new');
-    expect(await screen.findByTitle('brand-new')).toBeInTheDocument();
+    expect(screen.queryByTitle('brand-new')).not.toBeInTheDocument();
  });
 });

@@ -271,10 +271,7 @@ export default function PluginFilterSelect(props: PluginFilterSelectProps) {
            type: 'ownState',
            ownState: {
              coltypeMap: initialColtypeMap,
-              // The dropdown offers `stripSurroundingQuotes(search)` as the
-              // creatable option, so the server has to be asked for the same
-              // string or the two disagree about what was searched for.
-              search: stripSurroundingQuotes(search).trim(),
+              search,
            },
          });
        }
@@ -284,10 +281,8 @@ export default function PluginFilterSelect(props: PluginFilterSelectProps) {

  const handleBlur = useCallback(() => {
    unsetFocusedFilter();
-    if (search) {
-      onSearch('');
-    }
-  }, [onSearch, search, unsetFocusedFilter]);
+    onSearch('');
+  }, [onSearch, unsetFocusedFilter]);

  const handleChange = useCallback(
    (value?: SelectValue | number | string) => {
@@ -309,25 +304,6 @@ export default function PluginFilterSelect(props: PluginFilterSelectProps) {
      ? t('No data')
      : tn('%s option', '%s options', data.length, data.length);

-  // A capped list reads as the whole set, so a value sitting past the row
-  // limit looks like a value that does not exist. Each sentence is only added
-  // when it is actually true of this filter's configuration.
-  const rowLimit = Number(formData.rowLimit) || 0;
-  const helperText = useMemo(() => {
-    if (!rowLimit || data.length < rowLimit) {
-      return undefined;
-    }
-    return [
-      t('Only the first %s values are listed.', data.length),
-      searchAllOptions ? t('Type to search all of them.') : undefined,
-      creatable !== false
-        ? t('You can enter a value that is not listed.')
-        : undefined,
-    ]
-      .filter(Boolean)
-      .join(' ');
-  }, [creatable, data.length, rowLimit, searchAllOptions]);
-
  const formItemExtra = useMemo(() => {
    if (filterState.validateMessage) {
      return (
@@ -360,6 +336,7 @@ export default function PluginFilterSelect(props: PluginFilterSelectProps) {
    const unquotedSearch = stripSurroundingQuotes(search);
    if (
      unquotedSearch &&
+      !searchAllOptions &&
      creatable !== false &&
      !hasOption(unquotedSearch, uniqueOptions, true)
    ) {
@@ -369,7 +346,7 @@ export default function PluginFilterSelect(props: PluginFilterSelectProps) {
      ];
    }
    return uniqueOptions;
-  }, [search, uniqueOptions, creatable]);
+  }, [search, uniqueOptions, creatable, searchAllOptions]);

  const sortComparator = useCallback(
    (a: LabeledValue, b: LabeledValue) => {
@@ -640,7 +617,7 @@ export default function PluginFilterSelect(props: PluginFilterSelectProps) {
              name={formData.nativeFilterId}
              allowClear
              autoClearSearchValue
-              allowNewOptions={creatable !== false}
+              allowNewOptions={!searchAllOptions && creatable !== false}
              allowNewOptionsOnPaste={multiSelect && searchAllOptions}
              allowSelectAll={!searchAllOptions}
              value={multiSelect ? filterState.value || [] : filterState.value}
@@ -649,7 +626,6 @@ export default function PluginFilterSelect(props: PluginFilterSelectProps) {
              showSearch={showSearch}
              mode={multiSelect ? 'multiple' : 'single'}
              placeholder={placeholderText}
-              helperText={helperText}
              onClear={() => onSearch('')}
              onSearch={onSearch}
              onBlur={handleBlur}
@@ -117,38 +117,6 @@ describe('Select buildQuery', () => {
    ]);
  });

-  test('should not sort by the searched column', () => {
-    // Ordering by a high-cardinality column makes the engine sort every match
-    // before applying the row limit; the dropdown re-sorts the page anyway.
-    const queryContext = buildQuery(
-      { ...formData, sortAscending: true },
-      {
-        ownState: {
-          search: 'abc',
-          coltypeMap: { my_col: GenericDataType.String },
-        },
-      },
-    );
-    const [query] = queryContext.queries;
-    expect(query.orderby).toEqual([]);
-  });
-
-  test('should keep the sort metric while searching', () => {
-    // A sort metric decides which rows come back, so dropping it would change
-    // the result set rather than just its order.
-    const queryContext = buildQuery(
-      { ...formData, sortMetric: 'my_metric', sortAscending: false },
-      {
-        ownState: {
-          search: 'abc',
-          coltypeMap: { my_col: GenericDataType.String },
-        },
-      },
-    );
-    const [query] = queryContext.queries;
-    expect(query.orderby).toEqual([['my_metric', false]]);
-  });
-
  test('should add text search parameter for numeric to query filter', () => {
    const queryContext = buildQuery(formData, {
      ownState: {
@@ -54,13 +54,6 @@ const buildQuery: BuildQuery<PluginFilterSelectQueryFormData> = (
    }

    const sortColumns = sortMetric ? [sortMetric] : columns;
-    // Sorting by the searched column makes the engine scan and sort every
-    // match before applying the row limit, which is the dominant cost of
-    // search-as-you-type on a high-cardinality column. The dropdown re-sorts
-    // the returned page client-side, so the server sort buys nothing here. A
-    // sort metric is different: it selects *which* rows come back, so it has
-    // to stay.
-    const skipOrderBy = !!search && !sortMetric;
    const query: QueryObject[] = [
      {
        ...baseQueryObject,
@@ -68,7 +61,7 @@ const buildQuery: BuildQuery<PluginFilterSelectQueryFormData> = (
        metrics: sortMetric ? [sortMetric] : [],
        filters: filters.concat(extraFilters),
        orderby:
-          !skipOrderBy && (sortMetric || sortAscending !== undefined)
+          sortMetric || sortAscending !== undefined
            ? sortColumns.map(column => [column, !!sortAscending])
            : [],
      },
@@ -1 +0,0 @@
-../superset-frontend/.npmrc
@@ -160,7 +160,7 @@ def migrate_by_id(ids: tuple[int, ...], is_downgrade: bool = False) -> None:
    """
    Migrate a subset of charts by IDs.

-    :param ids: Tuple of chart IDs to migrate
+    :param id: Tuple of chart IDs to migrate
    :param is_downgrade: Whether to downgrade the charts. Default is upgrade.
    """
    slices = db.session.query(Slice).filter(Slice.id.in_(ids))
@@ -33,7 +33,18 @@ from superset.commands.dataset.exceptions import (
 )
 from superset.commands.utils import populate_subjects
 from superset.daos.dataset import DatasetDAO
-from superset.exceptions import SupersetParseError, SupersetSecurityException
+from superset.db_engine_specs.exceptions import (
+    SupersetDBAPIConnectionError,
+    SupersetDBAPIDatabaseError,
+    SupersetDBAPIOperationalError,
+)
+from superset.exceptions import (
+    OAuth2RedirectError,
+    SupersetException,
+    SupersetParseError,
+    SupersetSecurityException,
+    SupersetTimeoutException,
+)
 from superset.extensions import security_manager
 from superset.sql.parse import Table
 from superset.utils.decorators import on_error, transaction
@@ -50,7 +61,38 @@ class CreateDatasetCommand(CreateMixin, BaseCommand):
        self.validate()

        dataset = DatasetDAO.create(attributes=self._properties)
-        dataset.fetch_metadata()
+        try:
+            dataset.fetch_metadata()
+        except OAuth2RedirectError:
+            # Must reach the caller unchanged to start the OAuth2 dance.
+            raise
+        except (
+            SupersetTimeoutException,
+            SupersetDBAPIConnectionError,
+            SupersetDBAPIOperationalError,
+            SupersetDBAPIDatabaseError,
+        ):
+            # Infra-level failures (unreachable database, query timeout), not
+            # bad user input: let them propagate with their own status
+            # instead of being coerced into a 422 "invalid table" error.
+            raise
+        except SupersetException as ex:
+            # Not a SQLAlchemyError, so ``on_error`` re-raises it untouched and
+            # it escapes to FAB's ``@safe`` as an opaque 500 "Fatal error".
+            # Deliberately covers the 403 ``SupersetSecurityException`` raised
+            # for mutation/multi-statement SQL too: ``validate()`` already
+            # reports that class of rejection as a 422 on ``sql`` via
+            # ``DatasetDataAccessIsNotAllowed``.
+            raise DatasetInvalidError(
+                exceptions=[
+                    ValidationError(
+                        # ``lazy_gettext`` messages aren't ``str``, so
+                        # marshmallow won't wrap them into a list on its own.
+                        [str(ex.message)],
+                        field_name="sql" if self._properties.get("sql") else "table",
+                    )
+                ]
+            ) from ex
        return dataset

    def validate(self) -> None:  # noqa: C901
@@ -964,7 +964,6 @@ class AnnotationDatasource(BaseDatasource):
        limit: int = 10000,
        denormalize_column: bool = False,
        array_elements: bool = False,
-        search: str | None = None,
    ) -> list[Any]:
        raise NotImplementedError()

@@ -34,6 +34,7 @@ from superset import db
 from superset.constants import LRU_CACHE_MAX_SIZE
 from superset.errors import ErrorLevel, SupersetError, SupersetErrorType
 from superset.exceptions import (
+    SupersetErrorException,
    SupersetGenericDBErrorException,
    SupersetParseError,
    SupersetSecurityException,
@@ -125,12 +126,14 @@ def get_virtual_table_metadata(dataset: SqlaTable) -> list[ResultSetColumnType]:
        # rest (sandbox violations, malformed template syntax, encoding
        # errors) indicate a real problem with the template that must
        # surface. See #38012.
+        # str(ex) stringifies the raw SupersetError list (enum reprs and all).
+        error_message = "; ".join(err.message for err in ex.errors)
        if isinstance(ex.__cause__, UndefinedError):
            raise SupersetVirtualTableParseException(
-                message=_("Template processing error: %(error)s", error=str(ex)),
+                message=_("Template processing error: %(error)s", error=error_message),
            ) from ex
        raise SupersetGenericDBErrorException(
-            message=_("Template processing error: %(error)s", error=str(ex)),
+            message=_("Template processing error: %(error)s", error=error_message),
        ) from ex
    try:
        parsed_script = SQLScript(sql, engine=db_engine_spec.engine)
@@ -209,6 +212,11 @@ def get_columns_description(
                    result, cursor.description, db_engine_spec
                )
            return result_set.columns
+    except SupersetErrorException:
+        # Preserve exceptions that already carry a specific SupersetError
+        # (e.g. OAuth2RedirectError) so callers can act on them instead of
+        # seeing an opaque generic DB error.
+        raise
    except Exception as ex:
        raise SupersetGenericDBErrorException(message=str(ex)) from ex

@@ -87,6 +87,7 @@ from superset.datasets.schemas import (
    openapi_spec_methods_override,
 )
 from superset.exceptions import (
+    OAuth2RedirectError,
    SupersetSyntaxErrorException,
    SupersetTemplateException,
 )
@@ -278,18 +279,6 @@ class DatasetRestApi(SoftDeleteApiMixin, BaseSupersetModelRestApi):
        "database.backend",
        "database.allow_multi_catalog",
        "columns.advanced_data_type",
-        # Certification/warning metadata is stored serialized in the ``extra``
-        # column and surfaced through model properties. Exposing them keeps this
-        # payload consistent with the datasource serialization used by Explore,
-        # so clients hydrating from this endpoint don't lose the badges.
-        "columns.certification_details",
-        "columns.certified_by",
-        "columns.is_certified",
-        "columns.warning_markdown",
-        "metrics.certification_details",
-        "metrics.certified_by",
-        "metrics.is_certified",
-        "metrics.warning_markdown",
        "is_managed_externally",
        "uid",
        "uuid",
@@ -452,7 +441,6 @@ class DatasetRestApi(SoftDeleteApiMixin, BaseSupersetModelRestApi):

    @expose("/", methods=("POST",))
    @protect()
-    @safe
    @statsd_metrics
    @event_logger.log_this_with_context(
        action=lambda self, *args, **kwargs: f"{self.__class__.__name__}.post",
@@ -507,6 +495,12 @@ class DatasetRestApi(SoftDeleteApiMixin, BaseSupersetModelRestApi):
                data=new_model.data,
                uuid=new_model.uuid,
            )
+        except OAuth2RedirectError:
+            # Must reach the client unchanged to start the OAuth2 dance;
+            # ``@safe`` isn't used on this endpoint since it would otherwise
+            # swallow this into an opaque 500 that drops the ``url``/``tab_id``
+            # extras the frontend needs.
+            raise
        except DatasetSoftDeletedTwinExistsError as ex:
            return self.response_422(message=str(ex))
        except DatasetInvalidError as ex:
@@ -519,6 +513,14 @@ class DatasetRestApi(SoftDeleteApiMixin, BaseSupersetModelRestApi):
                exc_info=True,
            )
            return self.response_422(message=str(ex))
+        except Exception:  # pylint: disable=broad-except
+            # ``@safe`` isn't used on this endpoint (it would swallow the
+            # ``OAuth2RedirectError`` re-raised above into an opaque 500), so
+            # replicate its behavior here for any other unexpected exception:
+            # log the full error server-side, but don't echo internal details
+            # (ORM/driver error text, connection info) back to the caller.
+            logger.exception("Unexpected error in DatasetRestApi.post")
+            return self.response_500(message="Fatal error")

    @expose("/<pk>", methods=("PUT",))
    @protect()
@@ -41,9 +41,6 @@ from superset.views.base_api import BaseSupersetApi, statsd_metrics

 logger = logging.getLogger(__name__)

-# Cache lifetime for search-filtered column values, in seconds.
-SEARCH_CACHE_TIMEOUT = 60
-

 class DatasourceRestApi(BaseSupersetApi):
    allow_browser_login = True
@@ -90,14 +87,6 @@ class DatasourceRestApi(BaseSupersetApi):
              type: string
            name: column_name
            description: The name of the column to get values for
-          - in: query
-            schema:
-              type: string
-            name: q
-            description: >-
-              Optional case-insensitive substring; only values containing it are
-              returned. Lets the client search the full column rather than the
-              truncated first page.
          responses:
            200:
              description: A List of distinct values for the column
@@ -147,10 +136,6 @@ class DatasourceRestApi(BaseSupersetApi):
        # Element-level operators (Contains any / Contains all) request the
        # distinct array *elements* rather than distinct whole arrays.
        array_elements = parse_boolean_string(request.args.get("array_elements"))
-        # Server-side search. Without it the client can only match against the
-        # bounded first page, so a value beyond ``FILTER_SELECT_ROW_LIMIT`` is
-        # unfindable on a high-cardinality column.
-        search = (request.args.get("q") or "").strip() or None

        # Cache distinct column-value results so a dashboard with many filters
        # backed by the same (often heavy) virtual dataset doesn't re-execute
@@ -184,7 +169,6 @@ class DatasourceRestApi(BaseSupersetApi):
                        "limit": row_limit,
                        "denorm": denormalize_column,
                        "elements": array_elements,
-                        "q": search,
                        "rls": security_manager.get_rls_cache_key(datasource),
                        "changed_on": str(getattr(datasource, "changed_on", "")),
                    },
@@ -200,7 +184,7 @@ class DatasourceRestApi(BaseSupersetApi):
            logger.debug(
                "column-values cache HIT: uid=%s col=%s", datasource.uid, column_name
            )
-            response = self.response(200, result=cached, limit=row_limit)
+            response = self.response(200, result=cached)
            response.headers["X-Cache-Status"] = "HIT"
            return response

@@ -210,7 +194,6 @@ class DatasourceRestApi(BaseSupersetApi):
                limit=row_limit,
                denormalize_column=denormalize_column,
                array_elements=array_elements,
-                search=search,
            )
        except KeyError:
            return self.response(
@@ -242,15 +225,11 @@ class DatasourceRestApi(BaseSupersetApi):
        timeout = datasource.cache_timeout or app.config.get(
            "CACHE_DEFAULT_TIMEOUT", 300
        )
-        if search:
-            # Every distinct search term is its own key, so a few users typing
-            # would otherwise pin one entry per keystroke for the full timeout.
-            timeout = min(timeout, SEARCH_CACHE_TIMEOUT)
        cache_manager.data_cache.set(cache_key, payload, timeout=timeout)
        logger.debug(
            "column-values cache MISS: uid=%s col=%s", datasource.uid, column_name
        )
-        response = self.response(200, result=payload, limit=row_limit)
+        response = self.response(200, result=payload)
        response.headers["X-Cache-Status"] = "MISS"
        return response

@@ -220,6 +220,8 @@ async def create_virtual_dataset(  # noqa: C901
            error=f"Failed to update dataset metadata (creation rolled back): {exc}",
        )
    except SupersetGenericDBErrorException as exc:
+        # Defensive backstop for direct raises (see
+        # test_create_virtual_dataset_sql_error_is_actionable).
        logger.warning("Virtual dataset SQL validation failed", exc_info=True)
        await ctx.warning(f"Virtual dataset SQL failed validation: {exc}")
        return CreateVirtualDatasetResponse(
@@ -45,9 +45,9 @@ def redefine(
    Redefine the foreign key constraint to include the ON DELETE and ON UPDATE
    constructs for cascading purposes.

-    :param foreign_key: The foreign key constraint
-    :param on_delete: If set, emit ON DELETE <value> when issuing DDL operations
-    :param on_update: If set, emit ON UPDATE <value> when issuing DDL operations
+    :params foreign_key: The foreign key constraint
+    :param ondelete: If set, emit ON DELETE <value> when issuing DDL operations
+    :param onupdate: If set, emit ON UPDATE <value> when issuing DDL operations
    """

    bind = op.get_bind()
@@ -198,41 +198,6 @@ def get_effective_hours_offset(
 R_SUFFIX = "__right_suffix"


-# Escape character for LIKE patterns built from user-supplied search text.
-# Deliberately not a backslash: dialects that escape backslashes when rendering
-# string literals would emit a two-character ESCAPE clause, which is a syntax
-# error on engines that honour standard-conforming strings.
-LIKE_ESCAPE_CHAR = "!"
-
-
-def escape_like_pattern(value: str) -> str:
-    """
-    Neutralize LIKE wildcards in user-supplied search text.
-
-    Without this a user typing ``%`` or ``_`` would match every row, which is
-    both wrong and, on a large table, a scan the search was meant to avoid.
-    """
-    return (
-        value.replace(LIKE_ESCAPE_CHAR, LIKE_ESCAPE_CHAR * 2)
-        .replace("%", f"{LIKE_ESCAPE_CHAR}%")
-        .replace("_", f"{LIKE_ESCAPE_CHAR}_")
-    )
-
-
-def build_like_predicate(
-    expr: ColumnElement[Any],
-    search: str,
-) -> ColumnElement[Any]:
-    """
-    Build a case-insensitive containment predicate for ``expr``.
-
-    ``lower(expr) LIKE lower('%term%')`` is used rather than ``ILIKE`` because
-    the latter is not portable across engines.
-    """
-    pattern = f"%{escape_like_pattern(search)}%".lower()
-    return sa.func.lower(expr).like(pattern, escape=LIKE_ESCAPE_CHAR)
-
-
 def _normalize_mssql_virtual_dataset_sql(
    sql: str, parsed_script: SQLScript, engine: str
 ) -> str:
@@ -1646,28 +1611,16 @@ class ExtraJSONMixin:
        return value


-_EXTRA_DICT_CACHE_UNSET = object()
-
-
 class CertificationMixin:
    """Mixin to add extra certification fields"""

    extra = sa.Column(sa.Text, default="{}")

    def get_extra_dict(self) -> dict[str, Any]:
-        # Cache the parsed ``extra`` payload on the instance, keyed by the raw
-        # string it was parsed from, so callers reading multiple
-        # certification/warning properties off the same object don't each
-        # trigger their own ``json.loads``. The cache is transient (not a
-        # mapped column) and self-invalidates whenever ``extra`` changes.
-        cache_raw = getattr(self, "_extra_dict_cache_raw", _EXTRA_DICT_CACHE_UNSET)
-        if cache_raw is _EXTRA_DICT_CACHE_UNSET or cache_raw != self.extra:
-            try:
-                self._extra_dict_cache = json.loads(self.extra)
-            except (TypeError, json.JSONDecodeError):
-                self._extra_dict_cache = {}
-            self._extra_dict_cache_raw = self.extra
-        return self._extra_dict_cache
+        try:
+            return json.loads(self.extra)
+        except (TypeError, json.JSONDecodeError):
+            return {}

    @property
    def is_certified(self) -> bool:
@@ -4057,7 +4010,6 @@ class ExploreMixin:  # pylint: disable=too-many-public-methods
        limit: int = 10000,
        denormalize_column: bool = False,
        array_elements: bool = False,
-        search: str | None = None,
    ) -> list[Any]:
        # denormalize column name before querying for values
        # unless disabled in the dataset configuration
@@ -4095,9 +4047,6 @@ class ExploreMixin:  # pylint: disable=too-many-public-methods
            .select_from(tbl)
            .distinct()
        )
-        if search:
-            qry = qry.where(build_like_predicate(value_expr, search))
-
        if limit:
            qry = qry.limit(limit)

@@ -404,7 +404,7 @@ class SemanticView(AuditMixinNullable, Model):
                for dimension in dimensions
            },
        }
-        column_formats: dict[str, str | None] = {
+        column_formats = {
            metric.name: metric.d3format for metric in metrics if metric.d3format
        }

@@ -1539,25 +1539,7 @@ class SQLStatement(BaseSQLStatement[exp.Expression]):
        :return: The parsed predicate.
        """
        _check_script_length(predicate, self.engine)
-        try:
-            return sqlglot.parse_one(predicate, dialect=self._dialect)
-        except sqlglot.errors.ParseError as ex:
-            kwargs = (
-                {
-                    "highlight": ex.errors[0]["highlight"],
-                    "line": ex.errors[0]["line"],
-                    "column": ex.errors[0]["col"],
-                }
-                if ex.errors
-                else {}
-            )
-            raise SupersetParseError(predicate, self.engine, **kwargs) from ex
-        except sqlglot.errors.SqlglotError as ex:
-            raise SupersetParseError(
-                predicate,
-                self.engine,
-                message="Unable to parse predicate",
-            ) from ex
+        return sqlglot.parse_one(predicate, dialect=self._dialect)

    def apply_rls(
        self,
@@ -333,7 +333,7 @@ class BaseScreenshot:
        Computes the thumbnail and caches the result

        :param user: If no user is given will use the current context
-        :param cache_key: The cache key to store the thumbnail payload under
+        :param cache: The cache to keep the thumbnail payload
        :param window_size: The window size from which will process the thumb
        :param thumb_size: The final thumbnail size
        :param force: Will force the computation even if it's already cached
@@ -20,7 +20,6 @@ import copy
 import unittest
 from datetime import timedelta
 from io import BytesIO
-from typing import Any
 from unittest.mock import ANY, patch
 from zipfile import is_zipfile, ZipFile

@@ -73,27 +72,6 @@ from tests.integration_tests.fixtures.importexport import (
    dataset_ui_export,
 )

-# Fields the dataset ``show`` payload exposes but the ``PUT`` schema doesn't
-# accept: audit timestamps plus attributes derived from the model (type
-# affinity and the certification/warning metadata stored in ``extra``).
-DATASET_READ_ONLY_ITEM_FIELDS = (
-    "changed_on",
-    "created_on",
-    "type_generic",
-    "certification_details",
-    "certified_by",
-    "is_certified",
-    "warning_markdown",
-)
-
-
-def strip_read_only_fields(items: list[dict[str, Any]]) -> list[dict[str, Any]]:
-    """Drop read-only fields so a ``show`` payload can be fed back to ``PUT``."""
-    for item in items:
-        for field in DATASET_READ_ONLY_ITEM_FIELDS:
-            item.pop(field, None)
-    return items
-

 class TestDatasetApi(SupersetTestCase):
    fixture_tables_names = ("ab_permission", "ab_permission_view", "ab_view_menu")
@@ -1338,10 +1316,17 @@ class TestDatasetApi(SupersetTestCase):
        rv = self.get_assert_metric(uri, "get")
        data = json.loads(rv.data.decode("utf-8"))

-        strip_read_only_fields(data["result"]["columns"])
+        for column in data["result"]["columns"]:
+            column.pop("changed_on", None)
+            column.pop("created_on", None)
+            column.pop("type_generic", None)
        data["result"]["columns"].append(new_column_data)

-        strip_read_only_fields(data["result"]["metrics"])
+        for metric in data["result"]["metrics"]:
+            metric.pop("changed_on", None)
+            metric.pop("created_on", None)
+            metric.pop("type_generic", None)
+
        data["result"]["metrics"].append(new_metric_data)

        with freeze_time() as frozen:
@@ -1419,7 +1404,11 @@ class TestDatasetApi(SupersetTestCase):
        rv = self.get_assert_metric(uri, "get")
        data = json.loads(rv.data.decode("utf-8"))

-        strip_read_only_fields(data["result"]["columns"])
+        for column in data["result"]["columns"]:
+            column.pop("changed_on", None)
+            column.pop("created_on", None)
+            column.pop("type_generic", None)
+
        data["result"]["columns"].append(new_column_data)
        rv = self.client.put(uri, json={"columns": data["result"]["columns"]})

@@ -1454,7 +1443,10 @@ class TestDatasetApi(SupersetTestCase):
        # Get current cols and alter one
        rv = self.get_assert_metric(uri, "get")
        resp_columns = json.loads(rv.data.decode("utf-8"))["result"]["columns"]
-        strip_read_only_fields(resp_columns)
+        for column in resp_columns:
+            column.pop("changed_on", None)
+            column.pop("created_on", None)
+            column.pop("type_generic", None)

        resp_columns[0]["groupby"] = False
        resp_columns[0]["filterable"] = False
@@ -155,7 +155,6 @@ class TestDatasourceApi(SupersetTestCase):
            limit=10000,
            denormalize_column=False,
            array_elements=False,
-            search=None,
        )

    @pytest.mark.usefixtures("app_context", "virtual_dataset")
@@ -171,79 +170,6 @@ class TestDatasourceApi(SupersetTestCase):
        )
        assert values_for_column_mock.call_args.kwargs["array_elements"] is True

-    @pytest.mark.usefixtures("app_context", "virtual_dataset")
-    def test_get_column_values_search_filters_server_side(self):
-        """``?q=`` narrows the values in the database rather than client-side,
-        which is what makes a value beyond the row limit reachable at all."""
-        self.login(ADMIN_USERNAME)
-        table = self.get_virtual_dataset()
-        rv = self.client.get(
-            f"api/v1/datasource/table/{table.id}/column/col2/values/?q=b"
-        )
-        assert rv.status_code == 200
-        assert json.loads(rv.data.decode("utf-8"))["result"] == ["b"]
-
-    @pytest.mark.usefixtures("app_context", "virtual_dataset")
-    def test_get_column_values_search_is_case_insensitive(self):
-        self.login(ADMIN_USERNAME)
-        table = self.get_virtual_dataset()
-        rv = self.client.get(
-            f"api/v1/datasource/table/{table.id}/column/col2/values/?q=B"
-        )
-        assert rv.status_code == 200
-        assert json.loads(rv.data.decode("utf-8"))["result"] == ["b"]
-
-    @pytest.mark.usefixtures("app_context", "virtual_dataset")
-    def test_get_column_values_search_escapes_wildcards(self):
-        """A literal ``%`` must not be treated as "match everything"."""
-        self.login(ADMIN_USERNAME)
-        table = self.get_virtual_dataset()
-        rv = self.client.get(
-            f"api/v1/datasource/table/{table.id}/column/col2/values/?q=%25"
-        )
-        assert rv.status_code == 200
-        assert json.loads(rv.data.decode("utf-8"))["result"] == []
-
-    @pytest.mark.usefixtures("app_context", "virtual_dataset")
-    @patch("superset.models.helpers.ExploreMixin.values_for_column")
-    def test_get_column_values_blank_search_is_ignored(self, values_for_column_mock):
-        """Whitespace is not a search term; it must not narrow the list."""
-        values_for_column_mock.return_value = []
-        self.login(ADMIN_USERNAME)
-        table = self.get_virtual_dataset()
-        self.client.get(
-            f"api/v1/datasource/table/{table.id}/column/col2/values/?q=%20%20"
-        )
-        assert values_for_column_mock.call_args.kwargs["search"] is None
-
-    @pytest.mark.usefixtures("app_context", "virtual_dataset")
-    def test_get_column_values_returns_applied_limit(self):
-        """The client needs the limit to tell a short list from a truncated
-        one, so it can say the list is partial instead of implying it is whole."""
-        self.login(ADMIN_USERNAME)
-        table = self.get_virtual_dataset()
-        rv = self.client.get(f"api/v1/datasource/table/{table.id}/column/col2/values/")
-        assert rv.status_code == 200
-        assert json.loads(rv.data.decode("utf-8"))["limit"] == 10000
-
-    @pytest.mark.usefixtures("app_context", "virtual_dataset")
-    @patch("superset.models.helpers.ExploreMixin.values_for_column")
-    def test_get_column_values_cache_isolated_per_search(self, values_for_column_mock):
-        """Search terms must partition the cache; sharing one entry would serve
-        the results of somebody else's search."""
-        cache_manager.data_cache.clear()
-        values_for_column_mock.return_value = ["x"]
-        self.login(ADMIN_USERNAME)
-        table = self.get_virtual_dataset()
-        url = f"api/v1/datasource/table/{table.id}/column/col2/values/"
-
-        self.client.get(url)
-        self.client.get(f"{url}?q=a")
-        self.client.get(f"{url}?q=b")
-        self.client.get(f"{url}?q=a")
-
-        assert values_for_column_mock.call_count == 3
-
    @pytest.mark.usefixtures("app_context", "virtual_dataset")
    @patch("superset.db_engine_specs.base.BaseEngineSpec.denormalize_name")
    def test_get_column_values_not_denormalize_column(self, denormalize_name_mock):
@@ -265,7 +191,6 @@ class TestDatasourceApi(SupersetTestCase):
            limit=10000,
            denormalize_column=True,
            array_elements=False,
-            search=None,
        )

    @pytest.mark.usefixtures("app_context", "virtual_dataset")
@@ -18,11 +18,18 @@ from unittest.mock import Mock, patch

 import pytest
 from marshmallow import ValidationError
+from pytest_mock import MockerFixture

 from superset.commands.dataset.create import CreateDatasetCommand
 from superset.commands.dataset.exceptions import DatasetInvalidError
+from superset.db_engine_specs.exceptions import SupersetDBAPIConnectionError
 from superset.errors import ErrorLevel, SupersetError, SupersetErrorType
-from superset.exceptions import SupersetParseError
+from superset.exceptions import (
+    OAuth2RedirectError,
+    SupersetGenericDBErrorException,
+    SupersetParseError,
+    SupersetTimeoutException,
+)
 from superset.models.core import Database


@@ -250,3 +257,161 @@ def test_create_dataset_generic_exists_error_when_no_twin() -> None:
                            )
                            with pytest.raises(DatasetInvalidError):
                                command.validate()
+
+
+def test_create_dataset_metadata_fetch_error_is_structured(
+    mocker: MockerFixture,
+) -> None:
+    """A metadata-fetch failure must surface the engine's own message.
+
+    ``run()`` executes the SQL to introspect columns; the resulting
+    ``SupersetGenericDBErrorException`` used to escape as a 500 "Fatal error".
+    """
+    mocker.patch.object(CreateDatasetCommand, "validate")
+    dataset = Mock()
+    dataset.fetch_metadata.side_effect = SupersetGenericDBErrorException(
+        message="Invalid SQL: Unable to parse: SELECT ...",
+    )
+    mocker.patch(
+        "superset.commands.dataset.create.DatasetDAO.create",
+        return_value=dataset,
+    )
+
+    command = CreateDatasetCommand(
+        {
+            "database": 1,
+            "table_name": "dataset wrong",
+            "sql": "SELECT ...",
+        }
+    )
+
+    with pytest.raises(DatasetInvalidError) as exc_info:
+        command.run()
+
+    validation_errors = exc_info.value._exceptions
+    assert len(validation_errors) == 1
+    assert validation_errors[0].field_name == "sql"
+    assert "Invalid SQL: Unable to parse: SELECT ..." in str(
+        validation_errors[0].messages[0]
+    )
+
+
+def test_create_dataset_metadata_fetch_error_physical_table(
+    mocker: MockerFixture,
+) -> None:
+    """The same conversion applies to physical datasets, keyed on ``table``."""
+    mocker.patch.object(CreateDatasetCommand, "validate")
+    dataset = Mock()
+    dataset.fetch_metadata.side_effect = SupersetGenericDBErrorException(
+        message="(psycopg2.OperationalError) could not connect to server",
+    )
+    mocker.patch(
+        "superset.commands.dataset.create.DatasetDAO.create",
+        return_value=dataset,
+    )
+
+    command = CreateDatasetCommand({"database": 1, "table_name": "physical_table"})
+
+    with pytest.raises(DatasetInvalidError) as exc_info:
+        command.run()
+
+    validation_errors = exc_info.value._exceptions
+    assert validation_errors[0].field_name == "table"
+    assert "could not connect to server" in str(validation_errors[0].messages[0])
+
+
+def test_create_dataset_oauth2_redirect_propagates_unchanged(
+    mocker: MockerFixture,
+) -> None:
+    """OAuth2 redirects must not be flattened into a DatasetInvalidError."""
+    mocker.patch.object(CreateDatasetCommand, "validate")
+    dataset = Mock()
+    oauth2_error = OAuth2RedirectError(
+        url="https://example.org/oauth2/authorize",
+        tab_id="tab-123",
+        redirect_uri="https://superset.example.org/oauth2/redirect",
+    )
+    dataset.fetch_metadata.side_effect = oauth2_error
+    mocker.patch(
+        "superset.commands.dataset.create.DatasetDAO.create",
+        return_value=dataset,
+    )
+
+    command = CreateDatasetCommand(
+        {"database": 1, "table_name": "good_dataset", "sql": "SELECT 1 AS a"}
+    )
+
+    with pytest.raises(OAuth2RedirectError) as exc_info:
+        command.run()
+
+    assert exc_info.value is oauth2_error
+    assert exc_info.value.error.extra["url"] == "https://example.org/oauth2/authorize"
+    assert exc_info.value.error.extra["tab_id"] == "tab-123"
+
+
+def test_create_dataset_timeout_propagates_unchanged(
+    mocker: MockerFixture,
+) -> None:
+    """A query timeout is an infra failure, not bad user input: it must not
+    be flattened into a 422 DatasetInvalidError on ``table``/``sql``."""
+    mocker.patch.object(CreateDatasetCommand, "validate")
+    dataset = Mock()
+    timeout_error = SupersetTimeoutException(
+        error_type=SupersetErrorType.CONNECTION_DATABASE_TIMEOUT,
+        message="Connection timed out",
+        level=ErrorLevel.ERROR,
+    )
+    dataset.fetch_metadata.side_effect = timeout_error
+    mocker.patch(
+        "superset.commands.dataset.create.DatasetDAO.create",
+        return_value=dataset,
+    )
+
+    command = CreateDatasetCommand({"database": 1, "table_name": "physical_table"})
+
+    with pytest.raises(SupersetTimeoutException) as exc_info:
+        command.run()
+
+    assert exc_info.value is timeout_error
+
+
+def test_create_dataset_connection_error_propagates_unchanged(
+    mocker: MockerFixture,
+) -> None:
+    """An unreachable database must not be reported as an invalid table name."""
+    mocker.patch.object(CreateDatasetCommand, "validate")
+    dataset = Mock()
+    connection_error = SupersetDBAPIConnectionError(
+        "could not connect to server: Connection refused"
+    )
+    dataset.fetch_metadata.side_effect = connection_error
+    mocker.patch(
+        "superset.commands.dataset.create.DatasetDAO.create",
+        return_value=dataset,
+    )
+
+    command = CreateDatasetCommand({"database": 1, "table_name": "physical_table"})
+
+    with pytest.raises(SupersetDBAPIConnectionError) as exc_info:
+        command.run()
+
+    assert exc_info.value is connection_error
+
+
+def test_create_dataset_run_succeeds_when_metadata_fetch_works(
+    mocker: MockerFixture,
+) -> None:
+    """Control: the happy path still returns the created dataset."""
+    mocker.patch.object(CreateDatasetCommand, "validate")
+    dataset = Mock()
+    mocker.patch(
+        "superset.commands.dataset.create.DatasetDAO.create",
+        return_value=dataset,
+    )
+
+    command = CreateDatasetCommand(
+        {"database": 1, "table_name": "good_dataset", "sql": "SELECT 1 AS a"}
+    )
+
+    assert command.run() is dataset
+    dataset.fetch_metadata.assert_called_once()
@@ -191,6 +191,39 @@ def test_get_virtual_table_metadata_template_security_error_is_not_softened():
    assert "Template processing error" in str(exc_info.value.message)


+def test_get_virtual_table_metadata_template_error_message_is_clean():
+    """The message must be the SupersetError's own text, not str(ex)."""
+    mock_dataset = Mock(spec=SqlaTable)
+    mock_database = Mock(spec=Database)
+    mock_dataset.database = mock_database
+    mock_dataset.sql = "SELECT 1 {% if %}"
+
+    ex = SupersetSyntaxErrorException(
+        [
+            SupersetError(
+                message="Malformed template, expected 'endif'",
+                error_type=SupersetErrorType.GENERIC_COMMAND_ERROR,
+                level=ErrorLevel.ERROR,
+            )
+        ]
+    )
+    ex.__cause__ = SecurityError("unrelated cause, not UndefinedError")
+    mock_template_processor = Mock()
+    mock_template_processor.process_template.side_effect = ex
+    mock_dataset.get_template_processor.return_value = mock_template_processor
+    mock_dataset.template_params_dict = {}
+
+    with pytest.raises(SupersetGenericDBErrorException) as exc_info:
+        get_virtual_table_metadata(mock_dataset)
+
+    message = str(exc_info.value.message)
+    assert message == (
+        "Template processing error: Malformed template, expected 'endif'"
+    )
+    assert "SupersetError(" not in message
+    assert "error_type=<" not in message
+
+
 def test_get_virtual_table_metadata_multiple_statements_not_allowed():
    """Test that multiple SQL statements raise security error."""
    mock_dataset = Mock(spec=SqlaTable)
@@ -26,7 +26,7 @@ from superset.connectors.sqla.utils import (
    get_columns_description,
    get_virtual_table_metadata,
 )
-from superset.exceptions import SupersetSecurityException
+from superset.exceptions import OAuth2RedirectError, SupersetSecurityException
 from superset.models.core import Database


@@ -102,6 +102,28 @@ def test_returns_column_descriptions(mocker: MockerFixture) -> None:
    ]


+def test_get_columns_description_propagates_oauth2_redirect(
+    mocker: MockerFixture,
+) -> None:
+    """
+    ``get_columns_description`` wraps every exception raised while executing
+    the metadata query into ``SupersetGenericDBErrorException`` -- but an
+    ``OAuth2RedirectError`` raised by the driver (e.g. a database requiring
+    per-user OAuth2 tokens) must reach the caller unchanged so the frontend
+    can start the OAuth2 dance, instead of being flattened into an opaque
+    generic DB error.
+    """
+    database = mocker.MagicMock()
+    cursor = mocker.MagicMock()
+    oauth2_error = OAuth2RedirectError("https://example.org/oauth2", "tab-id", "uri")
+
+    database.get_raw_connection.return_value.__enter__.return_value.cursor.return_value = cursor  # noqa: E501
+    database.db_engine_spec.execute.side_effect = oauth2_error
+
+    with pytest.raises(OAuth2RedirectError):
+        get_columns_description(database, "catalog", "schema", "SELECT * FROM table")
+
+
 def _create_zero_row_database(tmp_path: Path) -> tuple[Database, str]:
    """
    Create a real SQLite-backed ``Database`` and a query that matches zero rows.
@@ -21,7 +21,6 @@ from unittest.mock import MagicMock, patch
 from sqlalchemy.orm.session import Session

 from superset import db
-from superset.utils import json


 def test_put_invalid_dataset(
@@ -217,61 +216,134 @@ def test_handle_filters_args_returns_request_scoped_filters(
    assert fresh_filters.get_joined_filters.call_count == 2


-def test_get_dataset_exposes_certification_metadata(
+def test_post_dataset_with_invalid_sql_returns_actionable_422(
    session: Session,
    client: Any,
    full_api_access: None,
 ) -> None:
-    """
-    Dataset API: Test that the show payload exposes the certification and
-    warning metadata for both columns and metrics.
+    """Saving a dataset over unrunnable SQL must explain what is wrong.

-    Regression test for #43279: Explore hydrates its datasource from this
-    endpoint after a dataset save or swap. Without these fields the certified
-    and warning badges disappeared until the page was reloaded, because the
-    Explore bootstrap payload serializes them but this endpoint did not.
+    With blanket database access ``validate()`` never parses the SQL, so
+    ``run()``'s column introspection is the first thing to reject it. That
+    used to surface as a bare 500 ``{"message": "Fatal error"}``.
    """
-    from superset.connectors.sqla.models import SqlaTable, SqlMetric, TableColumn
+    from superset.connectors.sqla.models import SqlaTable
    from superset.models.core import Database

    SqlaTable.metadata.create_all(db.session.get_bind())

-    extra = json.dumps(
-        {
-            "certification": {
-                "certified_by": "Data Platform",
-                "details": "Reviewed quarterly",
-            },
-            "warning_markdown": "This is a **warning**",
-        }
-    )
-    database = Database(
-        database_name="my_db",
-        sqlalchemy_uri="sqlite://",
-    )
-    dataset = SqlaTable(
-        table_name="test_certification_table",
-        database=database,
-        columns=[
-            TableColumn(column_name="ds", type="TIMESTAMP", extra=extra),
-            TableColumn(
-                column_name="calculated",
-                type="INTEGER",
-                expression="1 + 1",
-                extra=extra,
-            ),
-        ],
-        metrics=[SqlMetric(metric_name="cnt", expression="COUNT(*)", extra=extra)],
-    )
-    db.session.add(dataset)
+    database = Database(database_name="invalid_sql_db", sqlalchemy_uri="sqlite://")
+    db.session.add(database)
    db.session.flush()

-    response = client.get(f"/api/v1/dataset/{dataset.id}")
+    response = client.post(
+        "/api/v1/dataset/",
+        json={
+            "database": database.id,
+            "schema": "main",
+            "table_name": "dataset wrong",
+            "sql": "SELECT ...",
+        },
+    )

-    assert response.status_code == 200
-    result = response.json["result"]
-    for item in [*result["columns"], *result["metrics"]]:
-        assert item["is_certified"] is True
-        assert item["certified_by"] == "Data Platform"
-        assert item["certification_details"] == "Reviewed quarterly"
-        assert item["warning_markdown"] == "This is a **warning**"
+    assert response.status_code == 422
+    message = response.json["message"]
+    assert "Fatal error" not in str(message)
+    # Not the parser's exact wording -- that would break on a sqlglot bump.
+    assert message["sql"][0].startswith("Invalid SQL")
+
+    # The failed create must not leave a half-built dataset behind.
+    assert (
+        db.session.query(SqlaTable).filter_by(table_name="dataset wrong").one_or_none()
+        is None
+    )
+
+
+def test_post_dataset_oauth2_redirect_propagates_unchanged(
+    session: Session,
+    client: Any,
+    full_api_access: None,
+) -> None:
+    """OAuth2RedirectError must reach the client with its ``url``/``tab_id``
+    extras intact so the frontend can start the OAuth2 dance.
+
+    ``DatasetRestApi.post`` doesn't use flask-appbuilder's ``@safe``
+    decorator for this reason: ``@safe`` catches any uncaught exception and
+    flattens it into an opaque 500, which would strip those extras.
+    """
+    from superset.connectors.sqla.models import SqlaTable
+    from superset.exceptions import OAuth2RedirectError
+    from superset.models.core import Database
+
+    SqlaTable.metadata.create_all(db.session.get_bind())
+
+    database = Database(database_name="oauth2_db", sqlalchemy_uri="sqlite://")
+    db.session.add(database)
+    db.session.flush()
+
+    with patch(
+        "superset.datasets.api.CreateDatasetCommand.run",
+        side_effect=OAuth2RedirectError(
+            "http://example.org/auth", "tab-1", "/redirect"
+        ),
+    ):
+        response = client.post(
+            "/api/v1/dataset/",
+            json={
+                "database": database.id,
+                "schema": "main",
+                "table_name": "oauth2_table",
+            },
+        )
+
+    assert response.status_code == 403
+    error = response.json["errors"][0]
+    assert error["error_type"] == "OAUTH2_REDIRECT"
+    assert error["extra"] == {
+        "url": "http://example.org/auth",
+        "tab_id": "tab-1",
+        "redirect_uri": "/redirect",
+    }
+
+
+def test_post_dataset_unexpected_error_returns_sanitized_500(
+    session: Session,
+    client: Any,
+    full_api_access: None,
+) -> None:
+    """An unexpected, non-``SupersetException`` failure must be flattened
+    into an opaque 500 -- the same contract ``@safe`` used to provide --
+    instead of leaking raw exception text (e.g. driver/connection details)
+    through Flask's catch-all error handler.
+
+    ``DatasetRestApi.post`` doesn't use ``@safe`` so that ``OAuth2RedirectError``
+    can reach the client unchanged (see
+    ``test_post_dataset_oauth2_redirect_propagates_unchanged``); it must
+    replicate ``@safe``'s opaque-500 behavior itself for everything else.
+    """
+    from superset.connectors.sqla.models import SqlaTable
+    from superset.models.core import Database
+
+    SqlaTable.metadata.create_all(db.session.get_bind())
+
+    database = Database(database_name="unexpected_db", sqlalchemy_uri="sqlite://")
+    db.session.add(database)
+    db.session.flush()
+
+    secret = "postgresql://admin:s3cr3t@internal-db.example.com/prod"  # noqa: S105
+    with patch(
+        "superset.datasets.api.CreateDatasetCommand.run",
+        side_effect=RuntimeError(secret),
+    ):
+        response = client.post(
+            "/api/v1/dataset/",
+            json={
+                "database": database.id,
+                "schema": "main",
+                "table_name": "unexpected_table",
+            },
+        )
+
+    assert response.status_code == 500
+    assert response.json == {"message": "Fatal error"}
+    assert secret not in response.get_data(as_text=True)
@@ -105,86 +105,6 @@ def test_values_for_column(database: Database) -> None:
        assert table.values_for_column("a") == [1, None]


-@pytest.mark.parametrize(
-    "raw,expected",
-    [
-        ("plain", "plain"),
-        ("50%", "50!%"),
-        ("a_b", "a!_b"),
-        ("wow!", "wow!!"),
-        ("!%_", "!!!%!_"),
-    ],
-)
-def test_escape_like_pattern(raw: str, expected: str) -> None:
-    """Wildcards typed by a user are data, not pattern syntax."""
-    from superset.models.helpers import escape_like_pattern
-
-    assert escape_like_pattern(raw) == expected
-
-
-def test_build_like_predicate_is_case_insensitive_and_escaped() -> None:
-    import sqlalchemy as sa
-
-    from superset.models.helpers import build_like_predicate
-
-    compiled = str(
-        build_like_predicate(sa.column("c"), "50%").compile(
-            dialect=sa.dialects.registry.load("postgresql")(),
-            compile_kwargs={"literal_binds": True},
-        )
-    ).replace("%%", "%")
-
-    assert compiled == "lower(c) LIKE '%50!%%' ESCAPE '!'"
-
-
-def test_values_for_column_search(database: Database) -> None:
-    """``search`` narrows the distinct-value query in the database."""
-    import pandas as pd
-
-    from superset.connectors.sqla.models import SqlaTable, TableColumn
-
-    table = SqlaTable(
-        database=database,
-        schema=None,
-        table_name="t",
-        columns=[TableColumn(column_name="a")],
-    )
-
-    with patch(
-        "pandas.read_sql_query",
-        return_value=pd.DataFrame({"column_values": ["Alice"]}),
-    ) as read_sql_query:
-        assert table.values_for_column("a", search="ali") == ["Alice"]
-
-    sql = str(read_sql_query.call_args.kwargs["sql"])
-    assert "LIKE" in sql
-    assert "'%ali%'" in sql
-
-
-def test_values_for_column_without_search_has_no_predicate(
-    database: Database,
-) -> None:
-    """The unsearched list must stay a plain bounded DISTINCT scan."""
-    import pandas as pd
-
-    from superset.connectors.sqla.models import SqlaTable, TableColumn
-
-    table = SqlaTable(
-        database=database,
-        schema=None,
-        table_name="t",
-        columns=[TableColumn(column_name="a")],
-    )
-
-    with patch(
-        "pandas.read_sql_query",
-        return_value=pd.DataFrame({"column_values": ["Alice"]}),
-    ) as read_sql_query:
-        table.values_for_column("a")
-
-    assert "LIKE" not in str(read_sql_query.call_args.kwargs["sql"])
-
-
 def test_values_for_column_passes_catalog_and_schema(
    mocker: MockerFixture,
    session: Session,
@@ -5578,20 +5578,6 @@ def test_parse_predicate_length_check() -> None:
        stmt.parse_predicate("x" * 101)


-def test_parse_predicate_invalid_sql_raises_superset_parse_error() -> None:
-    """
-    A syntactically invalid RLS predicate raises ``SupersetParseError``.
-
-    ``parse_predicate`` is reachable via ``apply_rls`` for any RLS clause
-    configured on a queried table; an invalid clause must surface as the
-    typed 422 parse error rather than leaking a raw ``sqlglot`` exception.
-    """
-    stmt = SQLStatement("SELECT 1", "postgresql")
-    with pytest.raises(SupersetParseError) as excinfo:
-        stmt.parse_predicate("a >")
-    assert excinfo.value.status == 422
-
-
@pytest.mark.usefixtures("_small_parse_cap")
 def test_transpile_to_dialect_length_check() -> None:
    """
Author	SHA1	Message	Date
Enzo MartellucciandClaude Sonnet 5	390be50f37	fix(dataset): let OAuth2RedirectError reach caller through virtual-table path get_columns_description wrapped every exception (including OAuth2RedirectError) raised during query execution into SupersetGenericDBErrorException, so a virtual dataset's OAuth2 challenge never reached the create.py guard meant to preserve it. Also restore @safe's opaque-500 behavior for unexpected exceptions on DatasetRestApi.post, since dropping @safe to let OAuth2RedirectError through otherwise let raw exception text leak via Flask's global error handler. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>	2026-08-28 18:28:34 +02:00
Enzo Martellucci	7aa6fcaf1e	Merge branch 'master' into enxdev/fix/sql-lab-save-dataset	2026-08-28 14:53:30 +02:00
Enzo MartellucciandClaude Sonnet 5	8dbdfb4fc6	fix(dataset): don't report infra failures as invalid table/sql on create Query timeouts and unreachable-database errors were being caught by the broad SupersetException handler in CreateDatasetCommand and coerced into a 422 DatasetInvalidError, indistinguishable from a bad table name. Let SupersetTimeoutException and SupersetDBAPIConnectionError/OperationalError/ DatabaseError propagate so they reach the client with their own status via the existing SupersetException error handler. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>	2026-08-28 14:52:12 +02:00
Enzo Martellucci	2b1df8d462	Merge branch 'master' into enxdev/fix/sql-lab-save-dataset	2026-08-28 14:42:12 +02:00
Enzo Martellucci	feee3dea2f	Merge branch 'master' into enxdev/fix/sql-lab-save-dataset # Conflicts: # superset-frontend/src/SqlLab/components/SqlEditor/SqlEditor.test.tsx	2026-08-28 11:28:21 +02:00
Enzo MartellucciandClaude Sonnet 5	192ddf9a6d	fix(dataset): let OAuth2RedirectError survive DatasetRestApi.post msyavuz's review noted that CreateDatasetCommand's OAuth2RedirectError re-raise doesn't survive the API layer: DatasetRestApi.post is wrapped in flask-appbuilder's @safe, which catches any uncaught exception and flattens it into an opaque 500, dropping the url/tab_id extras the frontend needs to start the OAuth2 dance. Mirror the pattern already used by DatabaseRestApi (no @safe + explicit except OAuth2RedirectError: raise) so the app-level SupersetErrorException handler can build the correct 403 response instead. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>	2026-08-28 11:19:18 +02:00
Enzo MartellucciandClaude Sonnet 5	53b88da3b9	fix(dataset): preserve OAuth2 redirects and clean up error messages in CreateDatasetCommand Addresses review feedback on #43334: - OAuth2RedirectError is a SupersetException too, so it was being flattened into a generic DatasetInvalidError, dropping the url/tab_id/redirect_uri the frontend needs to start the OAuth2 dance. Re-raise it unchanged, matching the pattern already used elsewhere (e.g. commands/database/create.py). - get_virtual_table_metadata() built its "Template processing error" message from str() of a SupersetSyntaxErrorException, which stringifies the raw SupersetError list (enum reprs included) instead of the actual error text. Use the errors' own .message instead. - Documented why the SupersetGenericDBErrorException backstop in create_virtual_dataset.py is kept despite being unreachable via CreateDatasetCommand's current fetch_metadata() path. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>	2026-08-25 12:53:59 +02:00
Enzo Martellucci	3f4fdf5f07	Merge branch 'master' into enxdev/fix/sql-lab-save-dataset	2026-08-24 17:29:09 +02:00
Enzo MartellucciandClaude Sonnet 5	5a6c1b977b	refactor(sqllab): drop the redundant canSaveDataset prop from SaveQuery SaveQuery already receives the query's result columns and computes its own SQL-staleness check, so master's canSaveDataset prop (threaded through SqlEditor -> SaveQuery -> SaveDatasetActionButton) duplicated the same "did the query succeed" condition. Fold it into the single check SaveQuery already owns instead of ANDing two independently computed booleans together. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>	2026-08-21 14:17:12 +02:00
Enzo Martellucci	e8577368d3	Merge branch 'master' into enxdev/fix/sql-lab-save-dataset # Conflicts: # superset-frontend/src/SqlLab/components/SaveDatasetActionButton/SaveDatasetActionButton.test.tsx # superset-frontend/src/SqlLab/components/SaveDatasetActionButton/index.tsx # superset-frontend/src/SqlLab/components/SaveQuery/index.tsx	2026-08-21 14:15:55 +02:00
Enzo Martellucci	540f8cb2d0	fix(sqllab): invalidate the dataset-save gate when the editor SQL changes	2026-08-19 16:52:40 +02:00
Enzo Martellucci	aae997e546	fix(sql_lab): return 400 not 500 when raise_for_access hits malformed Jinja	2026-08-19 16:02:09 +02:00