fix(ci): diff pre-commit changed files against the live base, cap the env payload

github.event.pull_request.base.sha is frozen at PR creation, so the
changed-files detection on a long-lived PR diffs against the original
branch point and picks up all of master's churn since then. This week
that list crossed the kernel's per-env-var size limit (MAX_ARG_STRLEN),
killing the pre-commit step with "Argument list too long" before bash
could start — observed on #41127, where the frozen base predated the
antd v6 upgrade and the i18n catalog canonicalization.

Diff against HEAD^1 instead: HEAD is the PR merge commit, so its first
parent is the current base-branch tip and the diff is exactly the PR's
own files. Also fall back to --all-files when the computed list exceeds
100KB, so a PR that legitimately touches thousands of files degrades to
a full sweep instead of an unstartable step.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

.github/workflows/pre-commit.yml

@@ -81,7 +81,6 @@ jobs:
         id: changed_files
         env:
           EVENT_NAME: ${{ github.event_name }}
-          BASE_SHA: ${{ github.event.pull_request.base.sha }}
           BEFORE_SHA: ${{ github.event.before }}
         run: |
           set -euo pipefail
@@ -95,7 +94,15 @@ jobs:
           # Resolve the commit to diff against.
           base=""
           if [ "${EVENT_NAME}" = "pull_request" ]; then
-            base="${BASE_SHA}"
+            # HEAD is the PR merge commit, so its first parent is the current
+            # tip of the base branch. github.event.pull_request.base.sha is
+            # NOT that: GitHub freezes it at PR creation, so on a long-lived
+            # PR it points at the original branch point and the diff picks up
+            # all of the base branch's churn since then — thousands of paths,
+            # enough for the CHANGED_FILES env var below to exceed the
+            # kernel's per-variable size limit and kill the step with
+            # "Argument list too long" before bash even starts.
+            base="$(git rev-parse HEAD^1 2>/dev/null || true)"
           elif [ -n "${BEFORE_SHA:-}" ] && \
                [ "${BEFORE_SHA}" != "0000000000000000000000000000000000000000" ]; then
             base="${BEFORE_SHA}"
@@ -114,6 +121,15 @@ jobs:
           # Files present in HEAD that changed since the base (drop deletions).
           files="$(git diff --name-only --diff-filter=ACMRT "${base}...HEAD")"
 
+          # Env vars have a hard per-variable size limit (E2BIG at step
+          # start). A PR that legitimately touches thousands of files is
+          # better served by --all-files anyway.
+          if [ "$(printf '%s' "${files}" | wc -c)" -gt 100000 ]; then
+            echo "::notice::Changed-file list too large to pass via env; falling back to --all-files."
+            echo "mode=all" >> "$GITHUB_OUTPUT"
+            exit 0
+          fi
+
           if [ -z "${files}" ]; then
             echo "mode=none" >> "$GITHUB_OUTPUT"
           else

superset-frontend/src/features/home/Menu.test.tsx

@@ -59,13 +59,18 @@ jest.mock('@apache-superset/core/theme', () => ({
   useTheme: jest.fn(),
 }));
 
-jest.mock('antd', () => ({
-  ...jest.requireActual('antd'),
-  Grid: {
-    ...jest.requireActual('antd').Grid,
-    useBreakpoint: () => ({ md: true }),
-  },
-}));
+const mockUseBreakpoint = jest.fn<{ md?: boolean }, []>(() => ({ md: true }));
+
+jest.mock('antd', () => {
+  const actual = jest.requireActual('antd');
+  return {
+    ...actual,
+    Grid: {
+      ...actual.Grid,
+      useBreakpoint: () => mockUseBreakpoint(),
+    },
+  };
+});
 
 const dropdownItems = [
   {
@@ -302,6 +307,8 @@ beforeEach(() => {
   applicationRootMock.mockReturnValue('');
   // By default useTheme returns the real default theme (brandLogoUrl is falsy)
   useThemeMock.mockReturnValue(CoreTheme.supersetTheme);
+  // By default simulate a desktop viewport (md breakpoint active)
+  mockUseBreakpoint.mockReturnValue({ md: true });
 });
 
 test('should render', async () => {
@@ -1088,3 +1095,66 @@ describe('active tab highlighting (regression #36403)', () => {
     );
   });
 });
+
+test('navbar renders horizontal when breakpoints are not yet measured on a wide viewport (regression for layout flash)', async () => {
+  // Simulate first paint: useBreakpoint returns {} before the viewport is
+  // measured, so the layout falls back to the viewport width (jsdom defaults
+  // to 1024px, above the md threshold) → mode="horizontal".
+  mockUseBreakpoint.mockReturnValue({});
+  useSelectorMock.mockReturnValue({ roles: user.roles });
+  render(<Menu {...mockedProps} />, {
+    useRedux: true,
+    useQueryParams: true,
+    useRouter: true,
+    useTheme: true,
+  });
+  const navbar = await screen.findByTestId('navbar-top');
+  expect(navbar).toHaveClass('ant-menu-horizontal');
+  expect(navbar).not.toHaveClass('ant-menu-inline');
+});
+
+test('navbar renders inline when breakpoints are not yet measured on a narrow viewport', async () => {
+  // Simulate first paint on a mobile-sized window: useBreakpoint returns {}
+  // and the viewport-width fallback (below the md threshold) → mode="inline",
+  // so mobile users don't see a horizontal flash either.
+  const originalInnerWidth = window.innerWidth;
+  Object.defineProperty(window, 'innerWidth', {
+    configurable: true,
+    writable: true,
+    value: 500,
+  });
+  try {
+    mockUseBreakpoint.mockReturnValue({});
+    useSelectorMock.mockReturnValue({ roles: user.roles });
+    render(<Menu {...mockedProps} />, {
+      useRedux: true,
+      useQueryParams: true,
+      useRouter: true,
+      useTheme: true,
+    });
+    const navbar = await screen.findByTestId('navbar-top');
+    expect(navbar).toHaveClass('ant-menu-inline');
+    expect(navbar).not.toHaveClass('ant-menu-horizontal');
+  } finally {
+    Object.defineProperty(window, 'innerWidth', {
+      configurable: true,
+      writable: true,
+      value: originalInnerWidth,
+    });
+  }
+});
+
+test('navbar renders inline on mobile viewport (md: false)', async () => {
+  // Simulate a mobile viewport where the md breakpoint has resolved to false,
+  // which takes precedence over the viewport-width fallback → mode="inline".
+  mockUseBreakpoint.mockReturnValue({ md: false });
+  useSelectorMock.mockReturnValue({ roles: user.roles });
+  render(<Menu {...mockedProps} />, {
+    useRedux: true,
+    useQueryParams: true,
+    useRouter: true,
+    useTheme: true,
+  });
+  const navbar = await screen.findByTestId('navbar-top');
+  expect(navbar).toHaveClass('ant-menu-inline');
+});

superset-frontend/src/features/home/Menu.tsx

@@ -201,6 +201,11 @@ export function Menu({
   const screens = useBreakpoint();
   const uiConfig = useUiConfig();
   const theme = useTheme();
+  // screens.md is undefined on the first render before breakpoints are measured;
+  // fall back to the actual viewport width (using the same threshold as antd's
+  // md media query) so the first paint matches the device layout instead of
+  // flashing to the wrong mode on either desktop or mobile
+  const isMd = screens.md ?? window.innerWidth >= theme.screenMDMin;
 
   enum Paths {
     Explore = '/explore',
@@ -313,7 +318,7 @@ export function Menu({
     return {
       key,
       label,
-      ...(screens.md && {
+      ...(isMd && {
         icon: <Icons.DownOutlined iconSize="xs" />,
         popupOffset: NAVBAR_MENU_POPUP_OFFSET,
       }),
@@ -417,7 +422,7 @@ export function Menu({
             </StyledBrandText>
           )}
           <StyledMainNav
-            mode={screens.md ? 'horizontal' : 'inline'}
+            mode={isMd ? 'horizontal' : 'inline'}
             data-test="navbar-top"
             className="main-nav"
             selectedKeys={activeTabs}
@@ -445,7 +450,7 @@ export function Menu({
         </StyledCol>
         <Col md={8} xs={24}>
           <RightMenu
-            align={screens.md ? 'flex-end' : 'flex-start'}
+            align={isMd ? 'flex-end' : 'flex-start'}
             settings={settings}
             navbarRight={navbarRight}
             isFrontendRoute={isFrontendRoute}

superset/db_engine_specs/presto.py

@@ -499,7 +499,10 @@ class PrestoBaseEngineSpec(BaseEngineSpec, metaclass=ABCMeta):
         if filters:
             l = []  # noqa: E741
             for field, value in filters.items():
-                l.append(f"{field} = '{value}'")
+                # Escape single quotes so a ``'`` in the caller-supplied value
+                # cannot break out of the SQL string literal. See #41869.
+                escaped_value: str = str(value).replace("'", "''")
+                l.append(f"{field} = '{escaped_value}'")
             where_clause = "WHERE " + " AND ".join(l)
 
         # Partition select syntax changed in v0.199, so check here.
@@ -672,9 +675,9 @@ class PrestoBaseEngineSpec(BaseEngineSpec, metaclass=ABCMeta):
             )
 
         part_fields = indexes[0]["column_names"]
-        for k in kwargs.keys():  # pylint: disable=consider-iterating-dictionary
-            if k not in k in part_fields:  # pylint: disable=comparison-with-itself
-                msg = f"Field [{k}] is not part of the portioning key"
+        for k in kwargs:
+            if k not in part_fields:
+                msg: str = f"Field [{k}] is not part of the partitioning key"
                 raise SupersetTemplateException(msg)
         if len(kwargs.keys()) != len(part_fields) - 1:
             # pylint: disable=consider-using-f-string

superset/mcp_service/app.py

@@ -408,6 +408,8 @@ Chart Types You Can CREATE with generate_chart/generate_explore_link:
   (metrics + distribute_across required — distribute_across is the sample
    axis, e.g. a temporal column; dimensions splits into one box per value;
    whisker_type: tukey | min_max | percentile)
+- chart_type="waterfall": Waterfall chart of cumulative increases/decreases
+  (x_axis + metric required; optional single breakdown column, show_total)
 
 Time grain for temporal x-axis (time_grain parameter):
 - PT1H (hourly), P1D (daily), P1W (weekly), P1M (monthly), P1Y (yearly)
@@ -415,9 +417,9 @@ Time grain for temporal x-axis (time_grain parameter):
 Chart Types in Existing Charts (viewable via list_charts/get_chart_info):
 Each chart returned by list_charts / get_chart_info includes a
 chart_type_display_name field with a human-readable name when available.
-This field is populated only for the 9 chart types supported by generate_chart
+This field is populated only for the 10 chart types supported by generate_chart
 (xy, pie, table, pivot_table, big_number, mixed_timeseries, handlebars,
-histogram, box_plot).
+histogram, box_plot, waterfall).
 For all other viz_types (Funnel, Gauge, Heatmap, etc.) it will be null —
 use the raw viz_type field instead when referring to those chart types.
 

superset/mcp_service/chart/chart_utils.py

@@ -46,6 +46,7 @@ from superset.mcp_service.chart.schemas import (
     PivotTableChartConfig,
     SortByConfig,
     TableChartConfig,
+    WaterfallChartConfig,
     XYChartConfig,
 )
 from superset.mcp_service.utils.url_utils import get_superset_base_url
@@ -377,7 +378,7 @@ def map_config_to_form_data(
 ) -> Dict[str, Any]:
     """Map chart config to Superset form_data via the plugin registry.
 
-    The previous if/elif chain across all 7 chart types has been replaced by a
+    The previous per-chart-type if/elif chain has been replaced by a
     single registry lookup. Cross-field constraints (e.g. BigNumber trendline
     temporal check) are now owned by each plugin's post_map_validate() method
     rather than being baked into this dispatcher.
@@ -951,6 +952,41 @@ def map_box_plot_config(config: "BoxPlotChartConfig") -> Dict[str, Any]:
     return form_data
 
 
+def map_waterfall_config(config: WaterfallChartConfig) -> Dict[str, Any]:
+    """Map waterfall config to Superset form_data (viz_type waterfall).
+
+    Matches the frontend Waterfall buildQuery contract: a single ``x_axis``
+    column, an optional single-select ``groupby`` breakdown, and one
+    ``metric``; the query orders by the axis columns ascending, which the
+    frontend derives from these keys.
+    """
+    form_data: Dict[str, Any] = {
+        "viz_type": "waterfall",
+        "x_axis": config.x_axis.name,
+        "groupby": [config.breakdown.name] if config.breakdown else [],
+        "metric": create_metric_object(config.metric),
+        "show_total": config.show_total,
+        "show_legend": config.show_legend,
+        "increase_label": config.increase_label,
+        "decrease_label": config.decrease_label,
+        "total_label": config.total_label,
+        "x_axis_time_format": config.x_axis_time_format,
+        "y_axis_format": config.y_axis_format,
+        "row_limit": config.row_limit,
+    }
+    # Bucket a temporal x_axis: the grain (time_grain_sqla) needs the temporal
+    # column it applies to (granularity_sqla), mirroring the xy path's
+    # configure_temporal_handling and the frontend buildQuery's
+    # `x_axis || granularity_sqla`. Providing time_grain signals temporal
+    # intent; Superset ignores both for a non-temporal column.
+    if config.time_grain:
+        form_data["time_grain_sqla"] = config.time_grain
+        form_data["granularity_sqla"] = config.x_axis.name
+    add_currency_format(form_data, config.currency_format)
+    _add_adhoc_filters(form_data, config.filters)
+    return form_data
+
+
 def map_big_number_config(
     config: BigNumberChartConfig, dataset_id: int | str | None = None
 ) -> Dict[str, Any]:

superset/mcp_service/chart/plugins/__init__.py

@@ -37,6 +37,7 @@ from superset.mcp_service.chart.plugins.mixed_timeseries import (
 from superset.mcp_service.chart.plugins.pie import PieChartPlugin
 from superset.mcp_service.chart.plugins.pivot_table import PivotTableChartPlugin
 from superset.mcp_service.chart.plugins.table import TableChartPlugin
+from superset.mcp_service.chart.plugins.waterfall import WaterfallChartPlugin
 from superset.mcp_service.chart.plugins.xy import XYChartPlugin
 from superset.mcp_service.chart.registry import register
 
@@ -50,6 +51,7 @@ register(HandlebarsChartPlugin())
 register(BigNumberChartPlugin())
 register(HistogramChartPlugin())
 register(BoxPlotChartPlugin())
+register(WaterfallChartPlugin())
 
 __all__ = [
     "BigNumberChartPlugin",
@@ -60,5 +62,6 @@ __all__ = [
     "PieChartPlugin",
     "PivotTableChartPlugin",
     "TableChartPlugin",
+    "WaterfallChartPlugin",
     "XYChartPlugin",
 ]

superset/mcp_service/chart/plugins/waterfall.py

@@ -0,0 +1,186 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""Waterfall chart type plugin."""
+
+from __future__ import annotations
+
+from collections.abc import Mapping
+from typing import Any, ClassVar
+
+from superset.mcp_service.chart.chart_utils import (
+    _summarize_filters,
+    is_column_truly_temporal,
+    map_waterfall_config,
+)
+from superset.mcp_service.chart.plugin import BaseChartPlugin
+from superset.mcp_service.chart.schemas import ColumnRef, WaterfallChartConfig
+from superset.mcp_service.chart.validation.dataset_validator import DatasetValidator
+from superset.mcp_service.common.error_schemas import ChartGenerationError
+
+
+class WaterfallChartPlugin(BaseChartPlugin):
+    """Plugin for waterfall chart type."""
+
+    chart_type = "waterfall"
+    display_name = "Waterfall Chart"
+    native_viz_types: ClassVar[Mapping[str, str]] = {
+        "waterfall": "Waterfall Chart",
+    }
+
+    def pre_validate(
+        self,
+        config: dict[str, Any],
+    ) -> ChartGenerationError | None:
+        missing_fields = []
+        if "x_axis" not in config:
+            missing_fields.append(
+                "'x_axis' (period/category column, one step per value)"
+            )
+        if "metric" not in config:
+            missing_fields.append("'metric' (value whose per-step change is plotted)")
+
+        if missing_fields:
+            return ChartGenerationError(
+                error_type="missing_waterfall_fields",
+                message=(
+                    f"Waterfall missing required fields: {', '.join(missing_fields)}"
+                ),
+                details=(
+                    "Waterfall charts show how sequential increases and "
+                    "decreases across the x_axis values accumulate into a total"
+                ),
+                suggestions=[
+                    "Add 'x_axis': {'name': 'month'}",
+                    "Add 'metric': {'name': 'revenue_delta', 'aggregate': 'SUM'}",
+                    "Example: {'chart_type': 'waterfall', "
+                    "'x_axis': {'name': 'month'}, "
+                    "'metric': {'name': 'revenue_delta', 'aggregate': 'SUM'}}",
+                ],
+                error_code="MISSING_WATERFALL_FIELDS",
+            )
+        return None
+
+    def extract_column_refs(self, config: Any) -> list[ColumnRef]:
+        if not isinstance(config, WaterfallChartConfig):
+            return []
+        refs: list[ColumnRef] = [config.x_axis, config.metric]
+        if config.breakdown:
+            refs.append(config.breakdown)
+        if config.filters:
+            for f in config.filters:
+                refs.append(ColumnRef(name=f.column))
+        return refs
+
+    def to_form_data(
+        self, config: Any, dataset_id: int | str | None = None
+    ) -> dict[str, Any]:
+        return map_waterfall_config(config)
+
+    def generate_name(self, config: Any, dataset_name: str | None = None) -> str:
+        metric_name = config.metric.label or config.metric.name
+        axis = config.x_axis.label or config.x_axis.name
+        what = f"{metric_name} waterfall by {axis}"
+        if config.breakdown:
+            what += f" ({config.breakdown.label or config.breakdown.name})"
+        context = _summarize_filters(config.filters)
+        return self._with_context(what, context)
+
+    def post_map_validate(
+        self,
+        config: Any,
+        form_data: dict[str, Any],
+        dataset_id: int | str | None = None,
+    ) -> ChartGenerationError | None:
+        """Reject time_grain on a non-temporal x_axis.
+
+        The mapper sets time_grain_sqla/granularity_sqla when time_grain is
+        given; applying DATE_TRUNC to a non-temporal x_axis (e.g. a category
+        or a BIGINT year) errors or produces nonsense at query time. Only
+        validated when time_grain is set and dataset context is available.
+        """
+        if not isinstance(config, WaterfallChartConfig):
+            return None
+        if not config.time_grain or dataset_id is None:
+            return None
+        if is_column_truly_temporal(config.x_axis.name or "", dataset_id):
+            return None
+        return ChartGenerationError(
+            error_type="non_temporal_waterfall_grain",
+            message=(
+                f"time_grain='{config.time_grain}' requires a temporal "
+                f"x_axis, but '{config.x_axis.name}' is not a DATE/DATETIME/"
+                "TIMESTAMP column."
+            ),
+            details=(
+                "time_grain buckets a temporal x_axis; for a category or "
+                "numeric x_axis, omit time_grain (each value is already one "
+                "waterfall step)."
+            ),
+            suggestions=[
+                "Remove time_grain for a non-temporal x_axis",
+                "Use get_dataset_info to find a temporal column for x_axis",
+            ],
+            error_code="NON_TEMPORAL_WATERFALL_GRAIN",
+        )
+
+    def resolve_viz_type(self, config: Any) -> str:
+        return "waterfall"
+
+    def normalize_column_refs(self, config: Any, dataset_context: Any) -> Any:
+        config_dict = config.model_dump()
+
+        for key in ("x_axis", "breakdown"):
+            col = config_dict.get(key)
+            if col and not col.get("sql_expression") and not col.get("saved_metric"):
+                col["name"] = DatasetValidator.get_canonical_column_name(
+                    col["name"], dataset_context
+                )
+        if metric := config_dict.get("metric"):
+            if metric.get("sql_expression"):
+                pass
+            elif metric.get("saved_metric"):
+                metric["name"] = DatasetValidator.get_canonical_metric_name(
+                    metric["name"], dataset_context
+                )
+            else:
+                metric["name"] = DatasetValidator.get_canonical_column_name(
+                    metric["name"], dataset_context
+                )
+        DatasetValidator.normalize_filters(config_dict, dataset_context)
+        return WaterfallChartConfig.model_validate(config_dict)
+
+    def schema_error_hint(self) -> ChartGenerationError | None:
+        return ChartGenerationError(
+            error_type="waterfall_validation_error",
+            message="Waterfall configuration validation failed",
+            details=(
+                "The waterfall configuration is missing required fields or "
+                "has invalid structure"
+            ),
+            suggestions=[
+                "Ensure 'x_axis' has 'name' pointing at a period/category "
+                "column (often temporal)",
+                "Ensure 'metric' has 'name' and 'aggregate' (or saved_metric=True)",
+                "'breakdown' (alias: groupby) is a single optional category column",
+                "Example: {'chart_type': 'waterfall', "
+                "'x_axis': {'name': 'month'}, "
+                "'metric': {'name': 'revenue_delta', 'aggregate': 'SUM'}, "
+                "'breakdown': {'name': 'region'}}",
+            ],
+            error_code="WATERFALL_VALIDATION_ERROR",
+        )

superset/mcp_service/chart/schemas.py

@@ -2017,6 +2017,116 @@ class BoxPlotChartConfig(UnknownFieldCheckMixin):
         return self
 
 
+class WaterfallChartConfig(UnknownFieldCheckMixin):
+    """Config for waterfall charts (viz_type ``waterfall``)."""
+
+    model_config = ConfigDict(extra="ignore", populate_by_name=True)
+
+    chart_type: Literal["waterfall"] = "waterfall"
+    x_axis: ColumnRef = Field(
+        ...,
+        description="Category or period column along the x-axis (often "
+        "temporal, e.g. month); each value is one waterfall step",
+    )
+    metric: ColumnRef = Field(
+        ...,
+        description="Metric whose per-step change is plotted (use aggregate "
+        "e.g. SUM for ad-hoc, or saved_metric=True for a saved metric)",
+    )
+    breakdown: ColumnRef | None = Field(
+        None,
+        validation_alias=AliasChoices("breakdown", "groupby"),
+        description="Optional single category column that breaks each "
+        "x-axis period into per-category steps (form_data 'groupby'; the "
+        "frontend Breakdowns control is single-select)",
+    )
+    time_grain: TimeGrain | None = Field(
+        None,
+        description="Time bucket for a temporal x_axis (PT1H, P1D, P1W, "
+        "P1M, P1Y); each bucket becomes one waterfall step. Ignored for a "
+        "non-temporal x_axis.",
+        validation_alias=AliasChoices("time_grain", "time_grain_sqla"),
+    )
+    show_total: bool = Field(
+        True, description="Append a total bar per period (frontend default)"
+    )
+    show_legend: bool = Field(
+        False, description="Show the legend (frontend default: off)"
+    )
+    increase_label: str = Field("Increase", max_length=50)
+    decrease_label: str = Field("Decrease", max_length=50)
+    total_label: str = Field("Total", max_length=50)
+    x_axis_time_format: str = Field(
+        "smart_date",
+        description="Time format for a temporal x-axis (e.g. 'smart_date', '%Y-%m-%d')",
+        max_length=50,
+    )
+    y_axis_format: str = Field("SMART_NUMBER", max_length=50)
+    currency_format: CurrencyFormat | None = Field(
+        None,
+        description="Currency symbol applied to the metric value",
+    )
+    filters: List[FilterConfig] | None = Field(
+        None,
+        description="Structured filters (column/op/value). "
+        "Do NOT use adhoc_filters or raw SQL expressions.",
+    )
+    row_limit: int = Field(
+        10000,
+        description="Max grouped rows (frontend shared default)",
+        ge=1,
+        le=50000,
+    )
+
+    @model_validator(mode="before")
+    @classmethod
+    def unwrap_list_breakdown(cls, data: Any) -> Any:
+        """Accept the native form_data shape where ``groupby`` is a list.
+
+        The frontend Breakdowns control is single-select but stores its value
+        as a one-item list (e.g. ``["region"]``), so a config round-tripped
+        from existing waterfall form_data sends a list. Unwrap a length-1
+        list to the single value; reject longer lists rather than silently
+        dropping breakdowns. Native form_data also names physical columns as
+        bare strings, so a bare string in groupby/breakdown/x_axis is coerced to
+        ``{"name": ...}``.
+        """
+
+        def _coerce(value: Any) -> Any:
+            if isinstance(value, list):
+                if len(value) > 1:
+                    raise ValueError(
+                        "waterfall breakdown is single-select; pass at most one column"
+                    )
+                value = value[0] if value else None
+            if isinstance(value, str):
+                return {"name": value}
+            return value
+
+        if isinstance(data, dict):
+            for key in ("groupby", "breakdown"):
+                if key in data:
+                    data[key] = _coerce(data[key])
+            # x_axis is a bare column name in native form_data too.
+            if isinstance(data.get("x_axis"), str):
+                data["x_axis"] = {"name": data["x_axis"]}
+        return data
+
+    @model_validator(mode="after")
+    def reject_metric_style_dimensions(self) -> "WaterfallChartConfig":
+        """x_axis and breakdown are dimensions, not metrics."""
+        for ref, field_name in ((self.x_axis, "x_axis"), (self.breakdown, "breakdown")):
+            if ref is None:
+                continue
+            _reject_sql_expression_on_dimension(ref, field_name)
+            if ref.saved_metric:
+                raise ValueError(
+                    f"{field_name} cannot use saved_metric=True; "
+                    "saved metrics belong in the 'metric' field"
+                )
+        return self
+
+
 # Discriminated union for runtime validation (not exposed in JSON Schema)
 ChartConfig = Annotated[
     XYChartConfig
@@ -2027,13 +2137,14 @@ ChartConfig = Annotated[
     | HandlebarsChartConfig
     | BigNumberChartConfig
     | HistogramChartConfig
-    | BoxPlotChartConfig,
+    | BoxPlotChartConfig
+    | WaterfallChartConfig,
     Field(
         discriminator="chart_type",
         description=(
             "Chart configuration - specify chart_type as 'xy', 'table', "
             "'pie', 'pivot_table', 'mixed_timeseries', 'handlebars', "
-            "'big_number', 'histogram', or 'box_plot'"
+            "'big_number', 'histogram', 'box_plot', or 'waterfall'"
         ),
     ),
 ]

superset/mcp_service/chart/tool/generate_chart.py

@@ -107,7 +107,7 @@ async def generate_chart(  # noqa: C901
     - Use numeric dataset ID or UUID (NOT schema.table_name format)
     - MUST include chart_type in config (one of: 'xy', 'table', 'pie',
       'pivot_table', 'mixed_timeseries', 'handlebars', 'big_number',
-      'histogram', 'box_plot')
+      'histogram', 'box_plot', 'waterfall')
 
     IMPORTANT: The 'chart_type' field in the config is a DISCRIMINATOR that determines
     which chart configuration schema to use. It MUST be included and MUST match the
@@ -146,6 +146,9 @@ async def generate_chart(  # noqa: C901
       Required fields: metrics, distribute_across (the sample axis, e.g. a
       temporal column); use dimensions to split into one box per value;
       optional whisker_type ('tukey'|'min_max'|'percentile')
+    - Use chart_type='waterfall' for cumulative increase/decrease breakdowns
+      Required fields: x_axis, metric; optional: breakdown (single category
+      column, alias: groupby), show_total
 
     Quick lookup — natural-language ask -> chart_type (+ kind if applicable):
     - "bar chart" / "line chart" / "area chart" / "scatter plot"

superset/mcp_service/chart/tool/get_chart_data.py

@@ -102,6 +102,9 @@ _VIZ_CATEGORY: dict[str, str] = {
     "box_plot": "box_plot",
     "world_map": "map",
     "pivot_table_v2": "table",
+    # Own category: cumulative-flow semantics differ from a plain bar, like
+    # funnel/gauge carry distinct categories.
+    "waterfall": "waterfall",
 }
 
 _MAX_RECOMMENDATIONS = 4

superset/mcp_service/chart/tool/get_chart_preview.py

@@ -543,6 +543,9 @@ class VegaLitePreviewStrategy(PreviewFormatStrategy):
                 "echarts_timeseries_column",
                 "bar",
                 "column",
+                # Waterfall is a bar-mark construction; a bar preview is the
+                # closest faithful approximation.
+                "waterfall",
             ],
             "area": ["echarts_area", "area"],
             "scatter": ["echarts_timeseries_scatter", "scatter"],

superset/mcp_service/chart/tool/get_chart_type_schema.py

@@ -36,6 +36,7 @@ from superset.mcp_service.chart.schemas import (
     PieChartConfig,
     PivotTableChartConfig,
     TableChartConfig,
+    WaterfallChartConfig,
     XYChartConfig,
 )
 
@@ -52,6 +53,7 @@ _CHART_TYPE_ADAPTERS: Dict[str, TypeAdapter[Any]] = {
     "big_number": TypeAdapter(BigNumberChartConfig),
     "histogram": TypeAdapter(HistogramChartConfig),
     "box_plot": TypeAdapter(BoxPlotChartConfig),
+    "waterfall": TypeAdapter(WaterfallChartConfig),
 }
 
 VALID_CHART_TYPES = sorted(_CHART_TYPE_ADAPTERS.keys())
@@ -159,6 +161,20 @@ _CHART_EXAMPLES: Dict[str, list[Dict[str, Any]]] = {
             "percentile_high": 90,
         },
     ],
+    "waterfall": [
+        {
+            "chart_type": "waterfall",
+            "x_axis": {"name": "month"},
+            "metric": {"name": "revenue_delta", "aggregate": "SUM"},
+        },
+        {
+            "chart_type": "waterfall",
+            "x_axis": {"name": "quarter"},
+            "metric": {"name": "profit", "aggregate": "SUM"},
+            "breakdown": {"name": "region"},
+            "show_total": True,
+        },
+    ],
 }
 
 
@@ -223,7 +239,8 @@ def get_chart_type_schema(
     for a chart configuration before calling generate_chart or update_chart.
 
     Valid chart_type values: xy, table, pie, pivot_table,
-    mixed_timeseries, handlebars, big_number, histogram, box_plot.
+    mixed_timeseries, handlebars, big_number, histogram, box_plot,
+    waterfall.
 
     Returns the JSON Schema for the requested chart type, optionally
     with working examples.

superset/mcp_service/chart/validation/dataset_validator.py

@@ -271,8 +271,9 @@ class DatasetValidator:
         """Extract all column references from configuration via the plugin registry.
 
         Previously only handled TableChartConfig and XYChartConfig, causing
-        5 of 7 chart types to silently skip column validation. Now delegates
-        to the plugin for each chart type so all types are covered.
+        most chart types to silently skip column validation. Now delegates
+        to the plugin for each registered chart type; a config whose type has
+        no registered plugin yields no refs (rather than raising).
         """
         # Local import: plugins call DatasetValidator helpers from
         # normalize_column_refs().
@@ -390,7 +391,7 @@ class DatasetValidator:
         so we need to ensure column names match exactly.
 
         Previously only XYChartConfig and TableChartConfig were normalized; now
-        all 7 chart types are handled via the plugin registry.
+        all registered chart types are handled via the plugin registry.
 
         Args:
             config: Chart configuration with column references

superset/models/helpers.py

@@ -268,6 +268,28 @@ def json_to_dict(json_str: str) -> dict[Any, Any]:
     return {}
 
 
+UUID_NATIVE_TYPE_RE: re.Pattern[str] = re.compile(
+    r"\b(uuid|uniqueidentifier)\b", re.IGNORECASE
+)
+
+
+def is_uuid_native_type(native_type: Optional[str]) -> bool:
+    """
+    Return True if a native column type represents a UUID.
+
+    Engines such as PostgreSQL and ClickHouse expose native UUID column types
+    (e.g. ``UUID``, ``Nullable(UUID)``) that map to ``GenericDataType.STRING``
+    yet reject LIKE/ILIKE against the raw column, so these columns need an
+    explicit cast to string before pattern matching. SQL Server's equivalent
+    native type is ``uniqueidentifier``, which is matched too. The match is
+    on whole words so unrelated types that merely contain one of these as a
+    substring (e.g. a hypothetical ``uuidish`` type) aren't misclassified.
+    """
+    return native_type is not None and bool(
+        UUID_NATIVE_TYPE_RE.search(native_type.strip())
+    )
+
+
 def convert_uuids(obj: Any) -> Any:
     """
     Convert UUID objects to str so we can use yaml.safe_dump
@@ -4000,22 +4022,23 @@ class ExploreMixin:  # pylint: disable=too-many-public-methods
                     elif op in {
                         utils.FilterOperator.ILIKE,
                         utils.FilterOperator.LIKE,
+                        utils.FilterOperator.NOT_LIKE,
+                        utils.FilterOperator.NOT_ILIKE,
                     }:
-                        if target_generic_type != GenericDataType.STRING:
+                        # Native UUID columns report GenericDataType.STRING but
+                        # reject LIKE/ILIKE without a cast (see issue #41795)
+                        needs_string_cast_for_like: bool = (
+                            target_generic_type != GenericDataType.STRING
+                            or is_uuid_native_type(col_type)
+                        )
+                        if needs_string_cast_for_like:
                             sqla_col = sa.cast(sqla_col, sa.String)
 
                         if op == utils.FilterOperator.LIKE:
                             target_clause_list.append(sqla_col.like(eq))
-                        else:
+                        elif op == utils.FilterOperator.ILIKE:
                             target_clause_list.append(sqla_col.ilike(eq))
-                    elif op in {
-                        utils.FilterOperator.NOT_LIKE,
-                        utils.FilterOperator.NOT_ILIKE,
-                    }:
-                        if target_generic_type != GenericDataType.STRING:
-                            sqla_col = sa.cast(sqla_col, sa.String)
-
-                        if op == utils.FilterOperator.NOT_LIKE:
+                        elif op == utils.FilterOperator.NOT_LIKE:
                             target_clause_list.append(sqla_col.not_like(eq))
                         else:
                             target_clause_list.append(sqla_col.not_ilike(eq))

tests/unit_tests/db_engine_specs/test_presto.py

@@ -410,3 +410,69 @@ def test_extract_errors_maps_401_to_access_denied() -> None:
     result = PrestoEngineSpec.extract_errors(Exception(msg))
     assert len(result) == 1
     assert result[0].error_type == SupersetErrorType.CONNECTION_ACCESS_DENIED_ERROR
+
+
+def test_latest_sub_partition_rejects_unknown_field(
+    mocker: MockerFixture,
+) -> None:
+    """Regression test for #41869.
+
+    ``PrestoBaseEngineSpec.latest_sub_partition`` previously used a chained
+    comparison (``k not in k in part_fields``) that Python evaluates as
+    ``(k not in k) and (k in part_fields)``. Since ``k not in k`` is always
+    ``False`` for strings, the guard was unreachable and unknown kwarg names
+    were silently accepted, flowing into ``_partition_query`` and enabling
+    SQL injection via the ``latest_sub_partition`` Jinja macro. This test
+    locks in that unknown fields are now rejected before reaching the query
+    builder.
+    """
+    from superset.db_engine_specs.presto import PrestoBaseEngineSpec
+    from superset.exceptions import SupersetTemplateException
+
+    database: mock.MagicMock = mocker.MagicMock()
+    database.get_indexes.return_value = [{"column_names": ["ds", "event_type"]}]
+    table: mock.MagicMock = mocker.MagicMock()
+    with pytest.raises(SupersetTemplateException) as exc_info:
+        PrestoBaseEngineSpec.latest_sub_partition(
+            database,
+            table,
+            unknown_field="anything",
+        )
+
+    assert "unknown_field" in str(exc_info.value)
+    assert "not part of the partitioning key" in str(exc_info.value)
+
+
+def test_partition_query_escapes_single_quote_in_filter_value(
+    mocker: MockerFixture,
+) -> None:
+    """Regression test for #41869.
+
+    ``_partition_query`` previously interpolated filter values directly into
+    the SQL ``WHERE`` clause with an f-string, allowing SQL injection via any
+    caller that let user input reach ``filters``. Values must be escaped
+    (single-quote doubling per SQL standard) so a ``'`` in the value cannot
+    break out of the string literal.
+    """
+    from superset.db_engine_specs.presto import PrestoBaseEngineSpec
+
+    database: mock.MagicMock = mocker.MagicMock()
+    database.get_extra.return_value = {}
+    table: Table = Table("my_table", "my_schema")
+
+    injected: str = "2024-01-01' UNION SELECT secret FROM other_table--"
+    sql: str = PrestoBaseEngineSpec._partition_query(
+        table,
+        indexes=[{"column_names": ["ds", "event_type"]}],
+        database=database,
+        filters={"ds": injected},
+    )
+
+    # The single quote in the value must be doubled so the injection stays
+    # inside the SQL string literal — this is the whole payload wrapped in
+    # ONE literal, escape sequence and all.
+    assert "'2024-01-01'' UNION SELECT secret FROM other_table--'" in sql
+    # The pre-escape form (single quote closing the literal early followed
+    # by injected SQL) must NOT appear anywhere in the output — that would
+    # mean the payload broke out of the literal.
+    assert "'2024-01-01' UNION SELECT" not in sql

tests/unit_tests/mcp_service/chart/test_waterfall_chart.py

@@ -0,0 +1,316 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""Tests for the waterfall chart type plugin.
+
+Schema validation, form_data mapping (matching the frontend Waterfall
+buildQuery/transformProps contract for viz_type ``waterfall``), native
+vocabulary aliases, and registry integration.
+"""
+
+import pytest
+from pydantic import TypeAdapter, ValidationError
+
+from superset.mcp_service.chart.chart_utils import map_waterfall_config
+from superset.mcp_service.chart.schemas import ChartConfig, WaterfallChartConfig
+
+
+class TestWaterfallChartConfigSchema:
+    """WaterfallChartConfig schema validation."""
+
+    def test_basic_waterfall_config(self) -> None:
+        config = WaterfallChartConfig(
+            chart_type="waterfall",
+            x_axis={"name": "month"},
+            metric={"name": "revenue_delta", "aggregate": "SUM"},
+        )
+        assert config.x_axis.name == "month"
+        assert config.breakdown is None
+        assert config.show_total is True  # frontend controlPanel default
+
+    def test_waterfall_missing_x_axis(self) -> None:
+        with pytest.raises(ValidationError):
+            WaterfallChartConfig(
+                chart_type="waterfall",
+                metric={"name": "revenue", "aggregate": "SUM"},
+            )
+
+    def test_waterfall_missing_metric(self) -> None:
+        with pytest.raises(ValidationError):
+            WaterfallChartConfig(chart_type="waterfall", x_axis={"name": "month"})
+
+    def test_waterfall_rejects_extra_fields(self) -> None:
+        with pytest.raises(ValidationError):
+            WaterfallChartConfig(
+                chart_type="waterfall",
+                x_axis={"name": "month"},
+                metric={"name": "revenue", "aggregate": "SUM"},
+                bogus=1,
+            )
+
+    def test_waterfall_breakdown_rejects_saved_metric(self) -> None:
+        """The breakdown is a dimension, not a metric."""
+        with pytest.raises(ValidationError):
+            WaterfallChartConfig(
+                chart_type="waterfall",
+                x_axis={"name": "month"},
+                metric={"name": "revenue", "aggregate": "SUM"},
+                breakdown={"name": "count", "saved_metric": True},
+            )
+
+    def test_waterfall_x_axis_rejects_saved_metric(self) -> None:
+        with pytest.raises(ValidationError):
+            WaterfallChartConfig(
+                chart_type="waterfall",
+                x_axis={"name": "count", "saved_metric": True},
+                metric={"name": "revenue", "aggregate": "SUM"},
+            )
+
+    def test_groupby_alias_for_breakdown(self) -> None:
+        """Superset-native 'groupby' is accepted for the breakdown field."""
+        config = WaterfallChartConfig.model_validate(
+            {
+                "chart_type": "waterfall",
+                "x_axis": {"name": "month"},
+                "metric": {"name": "revenue", "aggregate": "SUM"},
+                "groupby": {"name": "region"},
+            }
+        )
+        assert config.breakdown is not None
+        assert config.breakdown.name == "region"
+
+    def test_chart_config_union_dispatches_waterfall(self) -> None:
+        config = TypeAdapter(ChartConfig).validate_python(
+            {
+                "chart_type": "waterfall",
+                "x_axis": {"name": "month"},
+                "metric": {"name": "revenue", "aggregate": "SUM"},
+            }
+        )
+        assert isinstance(config, WaterfallChartConfig)
+
+
+class TestMapWaterfallConfig:
+    """form_data mapping must match the frontend Waterfall buildQuery."""
+
+    def test_basic_waterfall_form_data(self) -> None:
+        config = WaterfallChartConfig(
+            chart_type="waterfall",
+            x_axis={"name": "month"},
+            metric={"name": "revenue_delta", "aggregate": "SUM"},
+        )
+        form_data = map_waterfall_config(config)
+        assert form_data["viz_type"] == "waterfall"
+        assert form_data["x_axis"] == "month"
+        assert form_data["groupby"] == []
+        assert form_data["metric"]["label"] == "SUM(revenue_delta)"
+        assert form_data["show_total"] is True
+
+    def test_waterfall_form_data_with_breakdown_and_filters(self) -> None:
+        config = WaterfallChartConfig(
+            chart_type="waterfall",
+            x_axis={"name": "month"},
+            metric={"name": "revenue", "aggregate": "SUM"},
+            breakdown={"name": "region"},
+            filters=[{"column": "year", "op": "=", "value": 2026}],
+            show_total=False,
+        )
+        form_data = map_waterfall_config(config)
+        # single breakdown maps to the groupby list (frontend multi: false)
+        assert form_data["groupby"] == ["region"]
+        assert form_data["show_total"] is False
+        assert form_data["adhoc_filters"], "filters must map to adhoc_filters"
+
+
+class TestWaterfallPluginRegistry:
+    """Plugin registration and viz-type resolution."""
+
+    def test_waterfall_plugin_registered(self) -> None:
+        from superset.mcp_service.chart import registry
+
+        plugin = registry.get("waterfall")
+        assert plugin is not None
+        assert plugin.resolve_viz_type(None) == "waterfall"
+
+    def test_display_name_resolves(self) -> None:
+        from superset.mcp_service.chart.registry import display_name_for_viz_type
+
+        assert display_name_for_viz_type("waterfall") == "Waterfall Chart"
+
+    def test_pre_validate_missing_fields(self) -> None:
+        from superset.mcp_service.chart import registry
+
+        plugin = registry.get("waterfall")
+        assert plugin is not None
+        error = plugin.pre_validate({"chart_type": "waterfall"})
+        assert error is not None
+        assert "x_axis" in error.message
+        assert "metric" in error.message
+
+
+class TestWaterfallTemporalAndNativeShape:
+    """time_grain bucketing and native form_data groupby-list handling."""
+
+    def test_time_grain_maps_to_time_grain_sqla(self) -> None:
+        config = WaterfallChartConfig(
+            chart_type="waterfall",
+            x_axis={"name": "order_date"},
+            metric={"name": "revenue", "aggregate": "SUM"},
+            time_grain="P1M",
+        )
+        form_data = map_waterfall_config(config)
+        assert form_data["time_grain_sqla"] == "P1M"
+        # the grain needs a temporal column to apply to
+        assert form_data["granularity_sqla"] == "order_date"
+
+    def test_time_grain_omitted_when_unset(self) -> None:
+        config = WaterfallChartConfig(
+            chart_type="waterfall",
+            x_axis={"name": "category"},
+            metric={"name": "revenue", "aggregate": "SUM"},
+        )
+        assert "time_grain_sqla" not in map_waterfall_config(config)
+
+    def test_time_grain_sqla_alias(self) -> None:
+        config = WaterfallChartConfig.model_validate(
+            {
+                "chart_type": "waterfall",
+                "x_axis": {"name": "order_date"},
+                "metric": {"name": "revenue", "aggregate": "SUM"},
+                "time_grain_sqla": "P1D",
+            }
+        )
+        assert config.time_grain == "P1D"
+
+    def test_groupby_list_unwrapped(self) -> None:
+        """Native form_data stores the single breakdown as a one-item list."""
+        config = WaterfallChartConfig.model_validate(
+            {
+                "chart_type": "waterfall",
+                "x_axis": {"name": "month"},
+                "metric": {"name": "revenue", "aggregate": "SUM"},
+                "groupby": [{"name": "region"}],
+            }
+        )
+        assert config.breakdown is not None
+        assert config.breakdown.name == "region"
+
+    def test_empty_groupby_list_is_no_breakdown(self) -> None:
+        config = WaterfallChartConfig.model_validate(
+            {
+                "chart_type": "waterfall",
+                "x_axis": {"name": "month"},
+                "metric": {"name": "revenue", "aggregate": "SUM"},
+                "groupby": [],
+            }
+        )
+        assert config.breakdown is None
+
+    def test_multi_item_groupby_rejected(self) -> None:
+        with pytest.raises(ValidationError):
+            WaterfallChartConfig.model_validate(
+                {
+                    "chart_type": "waterfall",
+                    "x_axis": {"name": "month"},
+                    "metric": {"name": "revenue", "aggregate": "SUM"},
+                    "groupby": [{"name": "region"}, {"name": "product"}],
+                }
+            )
+
+    def test_groupby_string_list_coerced(self) -> None:
+        """Native form_data names physical columns as bare strings."""
+        config = WaterfallChartConfig.model_validate(
+            {
+                "chart_type": "waterfall",
+                "x_axis": {"name": "month"},
+                "metric": {"name": "revenue", "aggregate": "SUM"},
+                "groupby": ["region"],
+            }
+        )
+        assert config.breakdown is not None
+        assert config.breakdown.name == "region"
+
+    def test_breakdown_scalar_string_coerced(self) -> None:
+        config = WaterfallChartConfig.model_validate(
+            {
+                "chart_type": "waterfall",
+                "x_axis": {"name": "month"},
+                "metric": {"name": "revenue", "aggregate": "SUM"},
+                "breakdown": "region",
+            }
+        )
+        assert config.breakdown is not None
+        assert config.breakdown.name == "region"
+
+
+class TestWaterfallNativeXAxisAndTemporalValidation:
+    """x_axis string coercion, recommendation category, temporal grain check."""
+
+    def test_x_axis_string_coerced(self) -> None:
+        config = WaterfallChartConfig.model_validate(
+            {
+                "chart_type": "waterfall",
+                "x_axis": "order_date",
+                "metric": {"name": "revenue", "aggregate": "SUM"},
+            }
+        )
+        assert config.x_axis.name == "order_date"
+
+    def test_waterfall_in_recommendation_category_map(self) -> None:
+        from superset.mcp_service.chart.tool.get_chart_data import _VIZ_CATEGORY
+
+        assert _VIZ_CATEGORY.get("waterfall") == "waterfall"
+
+    def test_time_grain_on_non_temporal_x_axis_rejected(self) -> None:
+        from unittest.mock import patch
+
+        from superset.mcp_service.chart import registry
+
+        plugin = registry.get("waterfall")
+        assert plugin is not None
+        config = WaterfallChartConfig(
+            chart_type="waterfall",
+            x_axis={"name": "region"},
+            metric={"name": "revenue", "aggregate": "SUM"},
+            time_grain="P1M",
+        )
+        with patch(
+            "superset.mcp_service.chart.plugins.waterfall.is_column_truly_temporal",
+            return_value=False,
+        ):
+            error = plugin.post_map_validate(config, {}, dataset_id=1)
+        assert error is not None
+        assert error.error_type == "non_temporal_waterfall_grain"
+
+    def test_time_grain_on_temporal_x_axis_passes(self) -> None:
+        from unittest.mock import patch
+
+        from superset.mcp_service.chart import registry
+
+        plugin = registry.get("waterfall")
+        assert plugin is not None
+        config = WaterfallChartConfig(
+            chart_type="waterfall",
+            x_axis={"name": "order_date"},
+            metric={"name": "revenue", "aggregate": "SUM"},
+            time_grain="P1M",
+        )
+        with patch(
+            "superset.mcp_service.chart.plugins.waterfall.is_column_truly_temporal",
+            return_value=True,
+        ):
+            assert plugin.post_map_validate(config, {}, dataset_id=1) is None

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

@@ -21,6 +21,7 @@ import pytest
 
 from superset.mcp_service.chart.tool.get_chart_type_schema import (
     _CHART_EXAMPLES,
+    _CHART_TYPE_ADAPTERS,
     _get_chart_type_schema_impl as _call_schema,
     VALID_CHART_TYPES,
 )
@@ -94,9 +95,11 @@ class TestGetChartTypeSchema:
             assert example["chart_type"] == "pie"
 
     def test_valid_chart_types_constant(self) -> None:
-        assert len(VALID_CHART_TYPES) == 9
-        assert "xy" in VALID_CHART_TYPES
-        assert "table" in VALID_CHART_TYPES
+        # Parity with the adapter registry rather than a brittle magic number,
+        # so adding a chart type doesn't fail this test spuriously.
+        assert set(VALID_CHART_TYPES) == set(_CHART_TYPE_ADAPTERS)
+        # A few load-bearing members that must always be present.
+        assert {"xy", "table", "pie", "waterfall"} <= set(VALID_CHART_TYPES)
 
     def test_all_chart_types_have_examples(self) -> None:
         for chart_type in VALID_CHART_TYPES:

tests/unit_tests/models/helpers_test.py

@@ -29,7 +29,8 @@ from pytest_mock import MockerFixture
 from sqlalchemy import create_engine
 from sqlalchemy.orm.session import Session
 from sqlalchemy.pool import StaticPool
-from sqlalchemy.sql.elements import ColumnElement
+from sqlalchemy.sql.elements import Cast, ColumnElement
+from sqlalchemy.sql.visitors import iterate
 
 from superset.superset_typing import AdhocColumn, AdhocMetric, OrderBy
 from superset.utils.core import FilterOperator, GenericDataType
@@ -3720,3 +3721,82 @@ def test_simple_metric_quotes_column_requiring_quoting(database: Database) -> No
     assert f"SUM({column_name})" not in rendered, (
         f"Column requiring quoting was emitted unquoted: {rendered}"
     )
+
+
+@pytest.mark.parametrize(
+    "native_type",
+    [
+        "UUID",
+        "uuid",
+        "Nullable(UUID)",
+        "uniqueidentifier",
+        "LowCardinality(UUID)",
+        "LowCardinality(Nullable(UUID))",
+    ],
+)
+@pytest.mark.parametrize(
+    "op",
+    ["LIKE", "ILIKE", "NOT LIKE", "NOT ILIKE"],
+)
+def test_like_filter_on_uuid_column_casts_to_string(
+    database: Database, native_type: str, op: str
+) -> None:
+    """
+    LIKE-family filters on native UUID columns must cast the column to string.
+
+    UUID columns map to ``GenericDataType.STRING``, so the generic-type guard
+    alone skips the string cast — but engines such as PostgreSQL and ClickHouse
+    reject LIKE/ILIKE against a raw UUID column (issue #41795: table chart
+    server-pagination search fails with e.g. "Illegal type UUID of argument of
+    function ilike"). The native column type must force the cast.
+    """
+    from superset.connectors.sqla.models import SqlaTable, TableColumn
+
+    table = SqlaTable(
+        database=database,
+        schema=None,
+        table_name="t",
+        columns=[TableColumn(column_name="event_id", type=native_type)],
+    )
+
+    result = table.get_sqla_query(
+        columns=["event_id"],
+        metrics=[],
+        extras={},
+        filter=[{"col": "event_id", "op": op, "val": "abc%"}],
+        granularity=None,
+        is_timeseries=False,
+        orderby=[],
+    )
+    whereclause: ColumnElement = result.sqla_query.whereclause
+    assert any(isinstance(node, Cast) for node in iterate(whereclause)), (
+        f"Expected a Cast node in the filter expression: {whereclause}"
+    )
+
+
+def test_like_filter_on_string_column_does_not_cast(database: Database) -> None:
+    """
+    LIKE-family filters on plain string columns must not add a redundant cast.
+    """
+    from superset.connectors.sqla.models import SqlaTable, TableColumn
+
+    table = SqlaTable(
+        database=database,
+        schema=None,
+        table_name="t",
+        columns=[TableColumn(column_name="b", type="TEXT")],
+    )
+
+    result = table.get_sqla_query(
+        columns=["b"],
+        metrics=[],
+        extras={},
+        filter=[{"col": "b", "op": "ILIKE", "val": "abc%"}],
+        granularity=None,
+        is_timeseries=False,
+        orderby=[],
+    )
+    whereclause: ColumnElement = result.sqla_query.whereclause
+    assert not any(isinstance(node, Cast) for node in iterate(whereclause)), (
+        f"Unexpected Cast node in the filter expression: {whereclause}"
+    )