fix(import): catch KeyError for missing uuid/ssh_tunnel in load_configs

A hand-edited or third-party database export YAML that omits the `uuid`
key (or `ssh_tunnel`) hit `config["uuid"]`/`config["ssh_tunnel"]`
indexing in load_configs() before schema.load() ran, raising a raw
KeyError that escaped the enclosing `except ValidationError` and
surfaced as an opaque 500 from the *//import/ endpoints instead of a
clean validation error. Add a sibling `except KeyError` that logs and
appends a ValidationError, routing the failure into the same aggregated
per-file error path as every other validation failure.

pyproject.toml

@@ -50,7 +50,7 @@ dependencies = [
     "flask-cors>=6.0.5, <7.0",
     "croniter>=6.2.4",
     "cron-descriptor",
-    "cryptography>=50.0.0, <51.0.0",
+    "cryptography>=50.0.1, <51.0.0",
     "deprecation>=2.1.0, <2.2.0",
     "flask>=2.2.5, <4.0.0",
     "flask-appbuilder>=5.2.2, <6.0.0",
@@ -96,7 +96,7 @@ dependencies = [
     "pgsanity",
     "Pillow>=12.3.0, <13", # raise floor to match resolved pin; closes SCA false-positive on 11.x-range CVEs already fixed in 12.3.0
     "polyline>=2.0.4, <3.0",
-    "pydantic>=2.8.0",
+    "pydantic>=2.13.5",
     "pyparsing>=3.3.2, <4",
     "python-dateutil",
     "python-dotenv", # optional dependencies for Flask but required for Superset, see https://flask.palletsprojects.com/en/stable/installation/#optional-dependencies
@@ -142,7 +142,7 @@ bigquery = [
     # 1.17.1 is likely the final release: googleapis/python-bigquery-sqlalchemy
     # was archived 2026-05-16. Both 1.17.0 and 1.17.1 support SQLAlchemy 1.4/2.0.
     "sqlalchemy-bigquery>=1.17.2",
-    "google-cloud-bigquery>=3.42.3",
+    "google-cloud-bigquery>=3.44.0",
 ]
 clickhouse = ["clickhouse-connect>=1.7.2, <2.0"]
 # The `cockroachdb` PyPI package (last released 2021) is abandoned and its
@@ -273,7 +273,7 @@ tdengine = [
     "taospy>=2.8.10",
     "taos-ws-py>=0.7.0"
 ]
-teradata = ["teradatasql>=20.0.0.66"]
+teradata = ["teradatasql>=20.0.0.67"]
 thumbnails = [] # deprecated, will be removed in 7.0
 vertica = ["sqlalchemy-vertica-python>= 0.6.3, < 0.7"]
 netezza = ["nzalchemy>= 11.1.2, < 11.2"]
@@ -288,7 +288,7 @@ development = [
     "docker",
     "flask-testing",
     "freezegun",
-    "grpcio>=1.82.1",
+    "grpcio>=1.83.1",
     "openapi-spec-validator",
     "parameterized",
     "pip",

requirements/base.in

@@ -26,7 +26,7 @@ filelock>=3.20.3,<4.0.0
 brotli>=1.2.0,<2.0.0
 numexpr>=2.9.0
 # Security: CVE-2026-34073 (MEDIUM) - Improper Certificate Validation
-cryptography>=50.0.0,<51.0.0
+cryptography>=50.0.1,<51.0.0
 # Security: Snyk - XSS vulnerability in Mako templates
 mako>=1.4.1,<2.0.0
 # Security: CVE-2024-52338 (CRITICAL) - Deserialization of untrusted data in IPC/Parquet readers

requirements/base.txt

@@ -84,7 +84,7 @@ cron-descriptor==1.4.5
     # via apache-superset (pyproject.toml)
 croniter==6.2.4
     # via apache-superset (pyproject.toml)
-cryptography==50.0.0
+cryptography==50.0.1
     # via
     #   -r requirements/base.in
     #   apache-superset (pyproject.toml)
@@ -297,11 +297,11 @@ pyasn1-modules==0.4.2
     # via google-auth
 pycparser==2.22
     # via cffi
-pydantic==2.13.4
+pydantic==2.13.5
     # via
     #   apache-superset (pyproject.toml)
     #   apache-superset-core
-pydantic-core==2.46.4
+pydantic-core==2.46.5
     # via pydantic
 pygeohash==3.2.2
     # via apache-superset (pyproject.toml)

requirements/development.txt

@@ -179,7 +179,7 @@ croniter==6.2.4
     # via
     #   -c requirements/base-constraint.txt
     #   apache-superset
-cryptography==50.0.0
+cryptography==50.0.1
     # via
     #   -c requirements/base-constraint.txt
     #   apache-superset
@@ -360,7 +360,7 @@ google-auth-oauthlib==1.2.1
     # via
     #   pandas-gbq
     #   pydata-google-auth
-google-cloud-bigquery==3.43.0
+google-cloud-bigquery==3.45.0
     # via
     #   apache-superset
     #   pandas-gbq
@@ -384,7 +384,7 @@ greenlet==3.5.5
     #   sqlalchemy
 griffelib==2.0.2
     # via fastmcp-slim
-grpcio==1.83.0
+grpcio==1.83.1
     # via
     #   apache-superset
     #   google-api-core
@@ -731,7 +731,7 @@ pycparser==2.22
     # via
     #   -c requirements/base-constraint.txt
     #   cffi
-pydantic==2.13.4
+pydantic==2.13.5
     # via
     #   -c requirements/base-constraint.txt
     #   apache-superset
@@ -740,7 +740,7 @@ pydantic==2.13.4
     #   mcp
     #   openapi-pydantic
     #   pydantic-settings
-pydantic-core==2.46.4
+pydantic-core==2.46.5
     # via
     #   -c requirements/base-constraint.txt
     #   pydantic

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

@@ -44,9 +44,14 @@ import {
   ValueFormatter,
 } from '@superset-ui/core';
 import { GenericDataType } from '@apache-superset/core/common';
-import { getOriginalSeries } from '@superset-ui/chart-controls';
+import {
+  getOriginalSeries,
+  getTimeOffset,
+  isDerivedSeries,
+} from '@superset-ui/chart-controls';
 import type { EChartsCoreOption } from 'echarts/core';
 import type { SeriesOption } from 'echarts';
+import type { LineStyleOption } from 'echarts/types/src/util/types';
 import {
   DEFAULT_FORM_DATA,
   EchartsMixedTimeseriesChartTransformedProps,
@@ -100,7 +105,11 @@ import {
   transformSeries,
   transformTimeseriesAnnotation,
 } from '../Timeseries/transformers';
-import { TIMEGRAIN_TO_TIMESTAMP, TIMESERIES_CONSTANTS } from '../constants';
+import {
+  TIMEGRAIN_TO_TIMESTAMP,
+  TIMESERIES_CONSTANTS,
+  OpacityEnum,
+} from '../constants';
 import { getDefaultTooltip } from '../utils/tooltip';
 import {
   createSpacedXAxisFormatter,
@@ -453,6 +462,10 @@ export default function transformProps(
 
   const array = ensureIsArray(chartProps.rawFormData?.time_compare);
   const inverted = invert(verboseMap);
+  // Tracks a stable pattern index per time offset so that derived series
+  // sharing the same comparison window (across both queries A and B) get
+  // the same dash pattern, mirroring the regular Timeseries transform.
+  const offsetPatterns: { [key: string]: number } = {};
 
   // The rendered ECharts series names are display names that can diverge from
   // the backend `label_map` keys: the metric display name is prepended when
@@ -467,6 +480,22 @@ export default function transformProps(
   rawSeriesA.forEach(entry => {
     const entryName = String(entry.name || '');
     const seriesName = inverted[entryName] || entryName;
+    const derivedSeries = isDerivedSeries(
+      entry,
+      chartProps.rawFormData,
+      seriesName,
+    );
+    const lineStyle: LineStyleOption = {};
+    if (derivedSeries && timeShiftColor) {
+      const offset = getTimeOffset(entry, array) || seriesName;
+      if (!offsetPatterns[offset]) {
+        offsetPatterns[offset] = Object.keys(offsetPatterns).length + 1;
+      }
+      const patternIndex = offsetPatterns[offset];
+      // use a combination of dash and dot for the line style
+      lineStyle.type = [(patternIndex % 5) + 1, (patternIndex % 3) + 1];
+      lineStyle.opacity = OpacityEnum.DerivedSeries;
+    }
     const colorScaleKey = getOriginalSeries(seriesName, array);
 
     const labelMapValues = rawLabelMap?.[seriesName];
@@ -544,6 +573,7 @@ export default function transformProps(
         timeShiftColor,
         theme,
         labelPosition,
+        lineStyle,
       },
     );
 
@@ -556,6 +586,23 @@ export default function transformProps(
   rawSeriesB.forEach(entry => {
     const entryName = String(entry.name || '');
     const seriesEntry = inverted[entryName] || entryName;
+    const derivedSeries = isDerivedSeries(
+      entry,
+      chartProps.rawFormData,
+      seriesEntry,
+    );
+    const lineStyle: LineStyleOption = {};
+    if (derivedSeries && timeShiftColor) {
+      const offset = getTimeOffset(entry, array) || seriesEntry;
+      if (!offsetPatterns[offset]) {
+        offsetPatterns[offset] = Object.keys(offsetPatterns).length + 1;
+      }
+      const patternIndex = offsetPatterns[offset];
+      // use a combination of dash and dot for the line style
+      lineStyle.type = [(patternIndex % 5) + 1, (patternIndex % 3) + 1];
+      lineStyle.opacity = OpacityEnum.DerivedSeries;
+    }
+
     const colorScaleKey = getOriginalSeries(seriesEntry, array);
 
     const labelMapValuesB = rawLabelMapB?.[seriesEntry];
@@ -634,6 +681,7 @@ export default function transformProps(
         timeShiftColor,
         theme,
         labelPosition: labelPositionB,
+        lineStyle,
       },
     );
 

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

@@ -21,6 +21,7 @@ import {
   AnnotationType,
   AnnotationSourceType,
   AxisType,
+  ComparisonType,
   DataRecord,
   FormulaAnnotationLayer,
   IntervalAnnotationLayer,
@@ -1562,6 +1563,7 @@ test('y-axis title position: non-Left sets nameLocation to end', () => {
   expect(yAxis[1].nameGap).toEqual(30);
   expect(yAxis[1].nameLocation).toEqual('end');
 });
+
 describe('EchartsMixedTimeseries tooltip truncation', () => {
   const longSeriesName = 'prod-us-east-1-service-checkout-latency-p99';
   const marker = '<span style="background-color:#1f77b4;"></span>';
@@ -1763,3 +1765,162 @@ test('hides the ticks on the x axis and both y axes', () => {
   expect(yAxis[0].axisTick.show).toBe(false);
   expect(yAxis[1].axisTick.show).toBe(false);
 });
+
+test('should apply a dashed lineStyle to derived (time comparison) series only', () => {
+  const queryAData = createTestQueryData(
+    [
+      {
+        sum__num: 100,
+        'sum__num__1 week ago': 80,
+        ds: 599616000000,
+      },
+      {
+        sum__num: 150,
+        'sum__num__1 week ago': 120,
+        ds: 599916000000,
+      },
+    ],
+    {
+      label_map: {
+        ds: ['ds'],
+        sum__num: ['sum__num'],
+        'sum__num__1 week ago': ['sum__num__1 week ago'],
+      },
+    },
+  );
+
+  const chartProps = createEchartsTimeseriesTestChartProps<
+    EchartsMixedTimeseriesFormData,
+    EchartsMixedTimeseriesProps
+  >({
+    ...MIXED_TIMESERIES_CHART_PROPS_DEFAULTS,
+    defaultQueriesData: [queryAData, queriesData[1]],
+    formData: {
+      ...formData,
+      metrics: ['sum__num'],
+      groupby: [],
+      time_compare: ['1 week ago'],
+      comparison_type: ComparisonType.Values,
+      timeShiftColor: true,
+    },
+    queriesData: [queryAData, queriesData[1]],
+  });
+
+  const transformed = transformProps(chartProps);
+  const series = (transformed.echartOptions.series as SeriesOption[]) || [];
+
+  const mainSeries = series.find(s => s.name === 'sum__num') as
+    | (SeriesOption & { lineStyle?: { type?: number[] | string } })
+    | undefined;
+  const derivedSeries = series.find(s => s.name === 'sum__num__1 week ago') as
+    | (SeriesOption & { lineStyle?: { type?: number[] | string } })
+    | undefined;
+
+  expect(mainSeries).toBeDefined();
+  expect(derivedSeries).toBeDefined();
+  // The primary (non-derived) series should not receive a dash pattern
+  expect(mainSeries?.lineStyle?.type).toBeUndefined();
+  // The derived (time comparison) series should receive a dash pattern array
+  expect(Array.isArray(derivedSeries?.lineStyle?.type)).toBe(true);
+});
+
+test('should not apply a dashed lineStyle when comparison_type is not Values', () => {
+  const queryAData = createTestQueryData(
+    [
+      {
+        sum__num: 100,
+        'sum__num__1 week ago': 80,
+        ds: 599616000000,
+      },
+      {
+        sum__num: 150,
+        'sum__num__1 week ago': 120,
+        ds: 599916000000,
+      },
+    ],
+    {
+      label_map: {
+        ds: ['ds'],
+        sum__num: ['sum__num'],
+        'sum__num__1 week ago': ['sum__num__1 week ago'],
+      },
+    },
+  );
+
+  const chartProps = createEchartsTimeseriesTestChartProps<
+    EchartsMixedTimeseriesFormData,
+    EchartsMixedTimeseriesProps
+  >({
+    ...MIXED_TIMESERIES_CHART_PROPS_DEFAULTS,
+    defaultQueriesData: [queryAData, queriesData[1]],
+    formData: {
+      ...formData,
+      metrics: ['sum__num'],
+      groupby: [],
+      time_compare: ['1 week ago'],
+      comparison_type: ComparisonType.Difference,
+    },
+    queriesData: [queryAData, queriesData[1]],
+  });
+
+  const transformed = transformProps(chartProps);
+  const series = (transformed.echartOptions.series as SeriesOption[]) || [];
+
+  const derivedSeries = series.find(s => s.name === 'sum__num__1 week ago') as
+    | (SeriesOption & { lineStyle?: { type?: number[] | string } })
+    | undefined;
+
+  expect(derivedSeries).toBeDefined();
+  expect(derivedSeries?.lineStyle?.type).toBeUndefined();
+});
+
+test('should not apply a dashed lineStyle when timeShiftColor is disabled', () => {
+  const queryAData = createTestQueryData(
+    [
+      {
+        sum__num: 100,
+        'sum__num__1 week ago': 80,
+        ds: 599616000000,
+      },
+      {
+        sum__num: 150,
+        'sum__num__1 week ago': 120,
+        ds: 599916000000,
+      },
+    ],
+    {
+      label_map: {
+        ds: ['ds'],
+        sum__num: ['sum__num'],
+        'sum__num__1 week ago': ['sum__num__1 week ago'],
+      },
+    },
+  );
+
+  const chartProps = createEchartsTimeseriesTestChartProps<
+    EchartsMixedTimeseriesFormData,
+    EchartsMixedTimeseriesProps
+  >({
+    ...MIXED_TIMESERIES_CHART_PROPS_DEFAULTS,
+    defaultQueriesData: [queryAData, queriesData[1]],
+    formData: {
+      ...formData,
+      metrics: ['sum__num'],
+      groupby: [],
+      time_compare: ['1 week ago'],
+      comparison_type: ComparisonType.Values,
+      timeShiftColor: false,
+    },
+    queriesData: [queryAData, queriesData[1]],
+  });
+
+  const transformed = transformProps(chartProps);
+  const series = (transformed.echartOptions.series as SeriesOption[]) || [];
+
+  const derivedSeries = series.find(s => s.name === 'sum__num__1 week ago') as
+    | (SeriesOption & { lineStyle?: { type?: number[] | string } })
+    | undefined;
+
+  expect(derivedSeries).toBeDefined();
+  expect(derivedSeries?.lineStyle?.type).toBeUndefined();
+});

superset/commands/importers/v1/utils.py

@@ -339,6 +339,25 @@ def load_configs(
                 exceptions.append(
                     ValidationError({file_name: {"masked_encrypted_extra": [str(exc)]}})
                 )
+            except KeyError as exc:
+                # Some config fields (e.g. `uuid`, `ssh_tunnel`) are read
+                # directly from the imported YAML before schema validation runs;
+                # a config missing one of these keys raises a raw KeyError
+                # instead of failing validation cleanly like every other
+                # per-file error. Convert it into a ValidationError so it flows
+                # into the same aggregated error path.
+                field = str(exc).strip("'\"")
+                logger.error(
+                    "Missing required key %s in config for %s (prefix: %s)",
+                    exc,
+                    file_name,
+                    prefix,
+                )
+                exceptions.append(
+                    ValidationError(
+                        {file_name: {field: ["Missing data for required field."]}}
+                    )
+                )
 
     return configs
 

superset/embedded/view.py

@@ -59,6 +59,7 @@ class EmbeddedView(BaseSupersetView):
             abort(404)
 
         assert embedded is not None
+        dashboard = embedded.dashboard
 
         # validate request referrer in allowed domains
         is_referrer_allowed = not embedded.allowed_domains
@@ -108,14 +109,11 @@ class EmbeddedView(BaseSupersetView):
             },
         }
 
-        # This page renders before any guest token has been presented, and the
-        # Referer / Sec-Fetch-Dest checks above are browser cooperation only --
-        # a non-browser client can forge or omit both. Serve a neutral shell:
-        # no dashboard title or description here; the embedded SPA fetches
-        # dashboard metadata through the guest-token-authenticated API.
         return self.render_template(
             "superset/spa.html",
             entry="embedded",
+            title=dashboard.dashboard_title,
+            dashboard_description=dashboard.description,
             bootstrap_data=json.dumps(
                 bootstrap_data, default=json.pessimistic_json_iso_dttm_ser
             ),

superset/security/manager.py

@@ -74,7 +74,6 @@ from sqlalchemy.orm.mapper import Mapper
 from sqlalchemy.orm.query import Query as SqlaQuery
 from sqlalchemy.sql import exists
 
-from superset.common.chart_data import ChartDataResultType
 from superset.constants import RouteMethod
 from superset.errors import ErrorLevel, SupersetError, SupersetErrorType
 from superset.exceptions import (
@@ -872,67 +871,28 @@ def _native_filter_query_modified(
     return False
 
 
-def _any_row_expanding_result_type(query_context: "QueryContext") -> bool:
-    """
-    Whether any query in the context asks for a result type that has the
-    server return every column on the datasource
-    (see ``_ROW_EXPANDING_RESULT_TYPES``).
-    """
-    return any(
-        _effective_result_type(
-            getattr(query, "result_type", None), query_context.result_type
-        )
-        in _ROW_EXPANDING_RESULT_TYPES
-        for query in query_context.queries
-    )
-
-
-def _drill_by_row_expanding_result_type(query_context: "QueryContext") -> bool:
-    """
-    Whether a chartless Drill By request (``slice_id`` sentinel ``0`` plus a
-    source ``chart_id``) asks for a result type that expands a query to every
-    column on the datasource.
-
-    ``raise_for_access`` grants Drill By only after confirming the requested
-    ``groupby`` dimensions are configured as drillable columns on the source
-    chart's datasource (see ``has_drill_access``); a guest's entitlement there
-    is that specific dimension allowlist, not the whole table. The
-    samples/drill_detail preparers ignore ``groupby`` entirely and return
-    every column, which would bypass that allowlist - unlike Drill to Detail
-    (no ``slice_id``/``chart_id`` at all), which is already meant to expose a
-    full dataset already attached to the dashboard.
-    """
-    form_data = query_context.form_data or {}
-    if not (form_data.get("slice_id") == 0 and form_data.get("chart_id")):
-        return False
-    return _any_row_expanding_result_type(query_context)
-
-
 def _native_filter_request_modified(query_context: "QueryContext") -> bool:
     """
     Validate a chartless data request that targets a native filter.
 
     Only requests identified as native-filter lookups (by the ``NATIVE_FILTER``
-    type marker or a ``native_filter_id``) are constrained here; other chartless
+    type marker or a ``native_filter_id``) are constrained; other chartless
     paths (drill-to-detail, drill-by, samples) carry neither and are validated by
     the datasource-access checks in raise_for_access, so they are not treated as
-    modified here beyond the Drill By result-type guard in
-    ``_drill_by_row_expanding_result_type``.
+    modified here.
 
     A native filter may only read the column(s) it targets on the dashboard it
     belongs to. The request is treated as modified (and therefore rejected for
     guest users) when it cannot be tied to a native filter on the requesting
-    dashboard, when it asks for a result type that expands the query to raw
-    datasource rows (see ``_ROW_EXPANDING_RESULT_TYPES``), or when any
-    value-returning term (column, group-by, series column, metric,
-    series-limit metric, or order-by) references something other than a
-    target column, a simple aggregate over a target column, or the filter's
-    configured sort metric. Free-form SQL terms and saved metrics other than
-    the configured sort metric are rejected. Row-restricting clauses
-    (``filter``/``extras``) are not constrained here: cross-filters
-    legitimately reference other columns and they do not return column
-    values; that blind-inference surface is a separate concern shared with
-    the chart path.
+    dashboard, or when any value-returning term (column, group-by, series
+    column, metric, series-limit metric, or order-by) references something
+    other than a target column, a simple
+    aggregate over a target column, or the filter's configured sort metric.
+    Free-form SQL terms and saved metrics other than the configured sort metric
+    are rejected. Row-restricting clauses (``filter``/``extras``) are not
+    constrained here: cross-filters legitimately reference other columns and
+    they do not return column values; that blind-inference surface is a separate
+    concern shared with the chart path.
     """
     form_data = query_context.form_data or {}
     if not (
@@ -947,13 +907,6 @@ def _native_filter_request_modified(query_context: "QueryContext") -> bool:
     # intentionally deny every value-returning term below.
     allowed_columns, allowed_metrics = targets
 
-    # The samples/drill_detail preparers replace a query's columns with every
-    # column on the datasource - bypassing the target-column allowlist below
-    # entirely - so reject those result types outright; a native filter never
-    # legitimately needs them.
-    if _any_row_expanding_result_type(query_context):
-        return True
-
     return any(
         _native_filter_query_modified(query, allowed_columns, allowed_metrics)
         for query in query_context.queries
@@ -1662,146 +1615,6 @@ def _columns_metrics_modified(
     return False
 
 
-def _annotation_layer_identity(layer: Any) -> Optional[tuple[str, str]]:
-    """
-    Identity of an annotation layer for tamper comparison: the source type and
-    the underlying source it reads (a native annotation-layer id or a chart
-    id). Cosmetic keys (``name``, styling, overrides) are not part of the
-    identity. Returns ``None`` for a malformed (non-dict) layer.
-    """
-    if not isinstance(layer, dict):
-        return None
-    return (
-        freeze_value(layer.get("sourceType")),
-        freeze_value(layer.get("value")),
-    )
-
-
-def _annotation_layers_modified(
-    query_context: "QueryContext",
-    form_data: dict[str, Any],
-    stored_chart: "Slice",
-    stored_query_context: Optional[dict[str, Any]],
-) -> bool:
-    """
-    Whether the request references annotation layers the stored chart does
-    not already carry.
-
-    ``annotation_layers`` is accepted on any query object, and native layers
-    resolve every annotation of each referenced layer id with no further
-    access check, so a guest injecting a layer the chart was not saved with
-    would read data that was never shared with them. Replaying the chart's
-    own stored layers is not tampering.
-    """
-    requested: set[Optional[tuple[str, str]]] = {
-        _annotation_layer_identity(layer)
-        for layer in form_data.get("annotation_layers") or []
-    }
-    requested.update(
-        _annotation_layer_identity(layer)
-        for query in query_context.queries
-        for layer in getattr(query, "annotation_layers", None) or []
-    )
-    if not requested:
-        return False
-    # A malformed (non-dict) layer is nothing the frontend produces from a
-    # stored chart; treat it as tampering rather than crashing on it later.
-    if None in requested:
-        return True
-
-    stored: set[Optional[tuple[str, str]]] = {
-        _annotation_layer_identity(layer)
-        for layer in stored_chart.params_dict.get("annotation_layers") or []
-    }
-    if stored_query_context:
-        for query in stored_query_context.get("queries") or []:
-            stored.update(
-                _annotation_layer_identity(layer)
-                for layer in query.get("annotation_layers") or []
-            )
-    return not requested.issubset(stored)
-
-
-#: Result types that make the server rewrite the query to return raw rows of
-#: every datasource column (``_prepare_samples_query`` and
-#: ``_prepare_drill_detail_query`` in ``superset.common.query_actions``).
-_ROW_EXPANDING_RESULT_TYPES = {
-    ChartDataResultType.SAMPLES.value,
-    ChartDataResultType.DRILL_DETAIL.value,
-}
-
-
-def _result_type_value(result_type: Any) -> str:
-    """Normalize a result type (enum member or raw string) to its value."""
-    return str(getattr(result_type, "value", result_type)).lower()
-
-
-def _effective_result_type(
-    query_result_type: Any, default_result_type: Any
-) -> Optional[str]:
-    """
-    The result type a query actually runs with: its own ``result_type`` if
-    set, else the query context's top-level default.
-
-    Mirrors ``query_obj.result_type or query_context.result_type``
-    (``QueryContextProcessor.get_payload``), so this reads the same value the
-    server uses to pick the samples/drill_detail preparer for that query.
-    """
-    if query_result_type:
-        return _result_type_value(query_result_type)
-    if default_result_type:
-        return _result_type_value(default_result_type)
-    return None
-
-
-def _result_type_modified(
-    query_context: "QueryContext",
-    stored_query_context: Optional[dict[str, Any]],
-) -> bool:
-    """
-    Whether the request asks for a result type that expands one of its
-    queries to raw datasource rows beyond what the stored chart runs at that
-    same query position.
-
-    The ``samples`` and ``drill_detail`` preparers replace a query's columns
-    with every column on the datasource - and drop its metrics - *after*
-    ``raise_for_access`` has run, so the subset comparisons on columns and
-    metrics in ``query_context_modified`` still pass while the response
-    contains the full underlying table. A guest's entitlement is only what
-    each query on the stored chart itself renders, so each requested query's
-    effective result type is compared against its own corresponding stored
-    query's effective result type by position - never against result types
-    used by other queries in the same query context - matching how
-    ``query_obj.result_type or query_context.result_type`` is resolved
-    per-query at runtime.
-    """
-    stored_queries: list[dict[str, Any]] = []
-    stored_default_result_type: Any = None
-    if stored_query_context:
-        stored_default_result_type = stored_query_context.get("result_type")
-        stored_queries = [
-            stored_query
-            for stored_query in stored_query_context.get("queries") or []
-            if isinstance(stored_query, dict)
-        ]
-
-    for index, query in enumerate(query_context.queries):
-        requested = _effective_result_type(query.result_type, query_context.result_type)
-        if requested not in _ROW_EXPANDING_RESULT_TYPES:
-            continue
-        stored = (
-            _effective_result_type(
-                stored_queries[index].get("result_type"), stored_default_result_type
-            )
-            if index < len(stored_queries)
-            else None
-        )
-        if requested != stored:
-            return True
-
-    return False
-
-
 def query_context_modified(query_context: "QueryContext") -> bool:
     """
     Check if a query context has been modified.
@@ -1815,10 +1628,7 @@ def query_context_modified(query_context: "QueryContext") -> bool:
     # Native-filter data requests have no associated chart (no slice_id). Rather
     # than accepting any payload, constrain them to the column(s) the dashboard's
     # native filter is allowed to target; other chartless paths keep prior
-    # behavior (see _native_filter_request_modified), except Drill By is still
-    # rejected when it asks for a row-expanding result type, since that would
-    # bypass the drillable-column allowlist raise_for_access checked for it
-    # (see _drill_by_row_expanding_result_type).
+    # behavior (see _native_filter_request_modified).
     #
     # SQL extras (extras.where/having) are NOT validated on chartless paths:
     # without a stored chart there is nothing to validate against, and
@@ -1827,9 +1637,7 @@ def query_context_modified(query_context: "QueryContext") -> bool:
     # are still protected by datasource-access checks in raise_for_access.
     # The _sql_filters_modified check below covers chart payloads only.
     if stored_chart is None:
-        return _native_filter_request_modified(
-            query_context
-        ) or _drill_by_row_expanding_result_type(query_context)
+        return _native_filter_request_modified(query_context)
 
     if form_data is None:
         return False
@@ -1872,68 +1680,57 @@ def query_context_modified(query_context: "QueryContext") -> bool:
     # Use ``is not None`` so an empty-but-present stored context reads as present.
     stored_context_state = "present" if stored_query_context is not None else "missing"
 
-    # Each comparator guards one facet of the payload against the stored chart;
-    # the first one that objects rejects the request, with its reason logged
-    # server-side (no payload values) so a 403 is diagnosable.
-    #
-    # - result type: reject types that would have the server expand the query
-    #   to raw datasource rows regardless of the stored chart's columns/metrics.
-    # - columns/metrics/group-by: must be a subset of the stored chart. Order-by
-    #   is handled separately, since a strict subset check there would reject a
-    #   guest legitimately sorting an embedded chart by one of its own columns.
-    # - order-by: may sort only by columns/metrics already on the stored chart;
-    #   new expressions (e.g. ``random()``) are still rejected.
-    # - SQL predicates (extras.where/having, SQL adhoc filters): must match what
-    #   was saved on the chart; injected custom SQL is rejected.
-    # - annotation layers: native layers resolve every annotation of each
-    #   referenced layer with no further access check on this path, so a layer
-    #   the chart was not saved with reads data never shared with the guest.
-    comparators: list[tuple[Callable[[], bool], str]] = [
-        (
-            lambda: _result_type_modified(query_context, stored_query_context),
-            "result type expands the chart to raw datasource rows",
-        ),
-        (
-            lambda: _columns_metrics_modified(
-                query_context, form_data, stored_chart, stored_query_context
-            ),
-            "columns/metrics/group-by not a subset of the stored chart",
-        ),
-        (
-            lambda: _series_limit_metric_modified(
-                query_context, form_data, stored_chart, stored_query_context
-            ),
-            "series-limit metric not on the stored chart",
-        ),
-        (
-            lambda: _orderby_modified(
-                query_context, stored_chart, stored_query_context
-            ),
-            "order-by references a term not on the stored chart",
-        ),
-        (
-            lambda: _sql_filters_modified(
-                query_context, form_data, stored_chart, stored_query_context
-            ),
-            "SQL filter/extras not on the stored chart",
-        ),
-        (
-            lambda: _annotation_layers_modified(
-                query_context, form_data, stored_chart, stored_query_context
-            ),
-            "annotation layer not on the stored chart",
-        ),
-    ]
-    for is_modified, reason in comparators:
-        if is_modified():
-            logger.warning(
-                "Guest chart payload rejected for slice %s: %s "
-                "(stored query_context %s)",
-                stored_chart.id,
-                reason,
-                stored_context_state,
-            )
-            return True
+    # compare columns and metrics in form_data with stored values. Order-by is
+    # handled separately: a strict subset check there would reject a guest
+    # legitimately sorting an embedded chart by one of its existing columns.
+    if _columns_metrics_modified(
+        query_context, form_data, stored_chart, stored_query_context
+    ):
+        logger.warning(
+            "Guest chart payload rejected for slice %s: columns/metrics/group-by "
+            "not a subset of the stored chart (stored query_context %s)",
+            stored_chart.id,
+            stored_context_state,
+        )
+        return True
+
+    if _series_limit_metric_modified(
+        query_context,
+        form_data,
+        stored_chart,
+        stored_query_context,
+    ):
+        logger.warning(
+            "Guest chart payload rejected for slice %s: series-limit metric not "
+            "on the stored chart (stored query_context %s)",
+            stored_chart.id,
+            stored_context_state,
+        )
+        return True
+
+    # Order-by may sort only by columns/metrics already present in the stored
+    # chart; new expressions (e.g. ``random()``) are still rejected.
+    if _orderby_modified(query_context, stored_chart, stored_query_context):
+        logger.warning(
+            "Guest chart payload rejected for slice %s: order-by references a "
+            "term not on the stored chart (stored query_context %s)",
+            stored_chart.id,
+            stored_context_state,
+        )
+        return True
+
+    # SQL predicates (extras.where/having, SQL adhoc filters) must match
+    # what was saved on the chart; injected custom SQL is rejected.
+    if _sql_filters_modified(
+        query_context, form_data, stored_chart, stored_query_context
+    ):
+        logger.warning(
+            "Guest chart payload rejected for slice %s: SQL filter/extras "
+            "not on the stored chart (stored query_context %s)",
+            stored_chart.id,
+            stored_context_state,
+        )
+        return True
 
     return False
 

superset/sql/parse.py

@@ -1453,10 +1453,33 @@ class SQLStatement(BaseSQLStatement[exp.Expression]):
                 found.add(entry)
         return found
 
+    def _has_limit_by(self) -> bool:
+        """
+        Check if the statement has a ClickHouse `LIMIT ... BY` clause.
+
+        `LIMIT n BY <cols>` keeps `n` rows *per group*, so it is a de-duplication
+        clause rather than a row cap. sqlglot models the `BY` columns as the
+        `expressions` of the root `Limit` node, or of the root `Offset` node for
+        the `LIMIT n OFFSET m BY x` and `LIMIT m, n BY x` spellings.
+
+        :return: True if the statement's limit or offset carries `BY` columns.
+        """
+        for arg in ("limit", "offset"):
+            node = self._parsed.args.get(arg)
+            if isinstance(node, exp.Expression) and node.expressions:
+                return True
+
+        return False
+
     def get_limit_value(self) -> int | None:
         """
         Parse a SQL query and return the `LIMIT` or `TOP` value, if present.
         """
+        # `LIMIT 2 BY id` bounds each group, not the result set, so reporting 2
+        # here would make `_set_query_limit()` clamp the whole query to 2 rows.
+        if self._has_limit_by():
+            return None
+
         if limit_node := self._parsed.args.get("limit"):
             literal = limit_node.args.get("expression") or getattr(
                 limit_node, "this", None
@@ -1494,18 +1517,40 @@ class SQLStatement(BaseSQLStatement[exp.Expression]):
         if not isinstance(self._parsed, exp.Query):
             return
 
-        if method == LimitMethod.FORCE_LIMIT:
+        # A ClickHouse `LIMIT ... BY` occupies the very `limit`/`offset` slot that
+        # `FORCE_LIMIT` overwrites, so forcing a row cap in place would drop the
+        # `BY` grouping and silently change what the query returns. The cap can't
+        # be appended alongside it either -- sqlglot rejects ClickHouse's native
+        # `LIMIT n BY x LIMIT m` with "Found multiple 'LIMIT' clauses" -- so it
+        # goes on a wrapping query instead, exactly as `WRAP_SQL` does.
+        if method == LimitMethod.FORCE_LIMIT and not self._has_limit_by():
             self._parsed.args["limit"] = exp.Limit(
                 expression=exp.Literal(this=str(limit), is_string=False)
             )
-        elif method == LimitMethod.WRAP_SQL:
-            self._parsed = exp.Select(
+        elif method in {LimitMethod.FORCE_LIMIT, LimitMethod.WRAP_SQL}:
+            inner = self._parsed.copy()
+            wrapper = exp.Select(
                 expressions=[exp.Star()],
                 limit=exp.Limit(
                     expression=exp.Literal(this=str(limit), is_string=False)
                 ),
-                from_=exp.From(this=exp.Subquery(this=self._parsed.copy())),
+                from_=exp.From(this=exp.Subquery(this=inner)),
             )
+
+            # `FORMAT` and `SETTINGS` configure the query rather than produce
+            # rows, and only mean what they say at the top level: ClickHouse
+            # rejects `FORMAT` inside a subquery outright, and a nested
+            # `SETTINGS` binds to that subquery alone, so top-level-only settings
+            # such as `extremes` would quietly stop applying. Moving them onto
+            # the wrapper keeps their original whole-query scope. Row-producing
+            # modifiers stay in the subquery, where ClickHouse keeps honoring
+            # them: a wrapped `WITH TOTALS` query still emits its totals block,
+            # and `WITH ROLLUP`/`WITH CUBE` still emit their extra rows.
+            for modifier in ("format", "settings"):
+                if value := inner.args.pop(modifier, None):
+                    wrapper.set(modifier, value)
+
+            self._parsed = wrapper
         else:  # method == LimitMethod.FETCH_MANY
             pass
 

tests/integration_tests/embedded/test_view.py

@@ -138,26 +138,3 @@ def test_get_embedded_dashboard_allows_iframe_sec_fetch_dest(
     uri = f"embedded/{embedded.uuid}"
     response = client.get(uri, headers={"Sec-Fetch-Dest": "iframe"})
     assert response.status_code == 200
-
-
-@pytest.mark.usefixtures("load_birth_names_dashboard_with_slices")
-@mock.patch.dict(
-    "superset.extensions.feature_flag_manager._feature_flags",
-    EMBEDDED_SUPERSET=True,
-)
-def test_get_embedded_dashboard_is_neutral_shell(client: FlaskClient[Any]):  # noqa: F811
-    """The pre-token page must not disclose dashboard metadata.
-
-    The Referer / Sec-Fetch-Dest checks are browser cooperation only -- a
-    non-browser client can forge them -- so anything rendered here is
-    effectively public. Title and description belong behind the
-    guest-token-authenticated API.
-    """
-    dash = db.session.query(Dashboard).filter_by(slug="births").first()
-    dash.description = "internal-only dashboard description"
-    embedded = EmbeddedDashboardDAO.upsert(dash, [])
-    db.session.flush()
-    response = client.get(f"embedded/{embedded.uuid}")
-    assert response.status_code == 200
-    assert dash.dashboard_title.encode() not in response.data
-    assert b"internal-only dashboard description" not in response.data

tests/unit_tests/commands/importers/v1/utils_test.py

@@ -18,7 +18,7 @@
 
 import gzip
 import io
-from unittest.mock import patch
+from unittest.mock import MagicMock, patch
 
 import pandas as pd
 import pytest
@@ -168,12 +168,14 @@ class TestLoadYaml:
 
 class TestLoadConfigs:
     """
-    load_configs() merges caller-supplied ``encrypted_extra_secrets`` into the
-    ``masked_encrypted_extra`` field of each config, which comes straight from
-    the imported YAML (before schema validation). A malformed value there used
-    to raise a raw simplejson.JSONDecodeError that escaped uncaught (opaque
-    500); it must instead be collected as a ValidationError like every other
-    per-file failure.
+    Per-file failures inside load_configs() must be collected as
+    ValidationErrors rather than propagating as raw exceptions (opaque 500s):
+
+    - A malformed ``masked_encrypted_extra`` (into which caller-supplied
+      ``encrypted_extra_secrets`` are merged before schema validation) used to
+      raise a raw simplejson.JSONDecodeError.
+    - A config missing its ``uuid`` used to raise a raw KeyError when the
+      password/ssh-tunnel validation looked up ``config["uuid"]``.
     """
 
     @staticmethod
@@ -186,6 +188,17 @@ class TestLoadConfigs:
 
         return TrivialSchema()
 
+    def _database_schemas(self) -> dict[str, object]:
+        from marshmallow import fields, Schema
+
+        class DatabaseSchema(Schema):
+            uuid = fields.UUID(required=True)
+            database_name = fields.String(required=True)
+            sqlalchemy_uri = fields.String(required=True)
+            password = fields.String(required=False, allow_none=True)
+
+        return {"databases/": DatabaseSchema()}
+
     @patch("superset.commands.importers.v1.utils.db")
     def test_invalid_json_in_masked_encrypted_extra_is_collected(
         self, mock_db: object
@@ -267,6 +280,75 @@ class TestLoadConfigs:
         merged = json.loads(configs[file_name]["masked_encrypted_extra"])
         assert merged == {"foo": "actual_secret"}
 
+    @patch("superset.commands.importers.v1.utils.db")
+    def test_missing_uuid_appends_validation_error(self, mock_db: MagicMock) -> None:
+        """A databases config missing `uuid` must not raise a raw KeyError;
+        it should be excluded from the returned configs and a ValidationError
+        appended to the exceptions list instead."""
+        from marshmallow.exceptions import ValidationError
+
+        from superset.commands.importers.v1.utils import load_configs
+
+        mock_db.session.query.return_value.all.return_value = []
+
+        # No `uuid` and no `password`, so the code reaches
+        # `config["uuid"] in db_passwords` and would raise KeyError pre-fix.
+        contents = {
+            "databases/bad.yaml": (
+                "database_name: bad\nsqlalchemy_uri: postgres://localhost\n"
+            ),
+        }
+        exceptions: list[ValidationError] = []
+
+        configs = load_configs(
+            contents,
+            self._database_schemas(),
+            {},
+            exceptions,
+            {},
+            {},
+            {},
+            {},
+        )
+
+        assert "databases/bad.yaml" not in configs
+        assert len(exceptions) == 1
+        assert isinstance(exceptions[0], ValidationError)
+        assert "databases/bad.yaml" in exceptions[0].messages
+
+    @patch("superset.commands.importers.v1.utils.db")
+    def test_uuid_present_loads_successfully(self, mock_db: MagicMock) -> None:
+        """Control: a well-formed databases config loads with no exceptions."""
+        from marshmallow.exceptions import ValidationError
+
+        from superset.commands.importers.v1.utils import load_configs
+
+        mock_db.session.query.return_value.all.return_value = []
+
+        contents = {
+            "databases/good.yaml": (
+                "uuid: 6ff1d5b3-4b0f-4c6a-9d2f-9c8b7a6e5d4c\n"
+                "database_name: good\n"
+                "sqlalchemy_uri: postgres://localhost\n"
+                "password: secret\n"
+            ),
+        }
+        exceptions: list[ValidationError] = []
+
+        configs = load_configs(
+            contents,
+            self._database_schemas(),
+            {},
+            exceptions,
+            {},
+            {},
+            {},
+            {},
+        )
+
+        assert "databases/good.yaml" in configs
+        assert exceptions == []
+
 
 class TestLoadConfigsNonMappingYaml:
     """A syntactically valid YAML document whose top-level value is a

tests/unit_tests/security/manager_test.py

@@ -29,7 +29,6 @@ from flask_appbuilder.const import AUTH_DB, AUTH_REMOTE_USER
 from flask_appbuilder.security.sqla.models import Role, User
 from pytest_mock import MockerFixture
 
-from superset.common.chart_data import ChartDataResultType
 from superset.common.query_object import QueryObject
 from superset.connectors.sqla.models import Database, SqlaTable
 from superset.exceptions import SupersetSecurityException
@@ -1406,156 +1405,6 @@ def test_query_context_modified_malformed_stored_query_context(
     assert query_context_modified(query_context)
 
 
-def test_query_context_modified_injected_annotation_layer(
-    mocker: MockerFixture,
-    stored_metrics: list[AdhocMetric],
-) -> None:
-    """
-    A guest must not be able to inject annotation layers the stored chart
-    does not carry: native annotation layers resolve all their annotations
-    with no further access check. Replaying the chart's own stored layers is
-    not tampering.
-    """
-    layer = {
-        "annotationType": "INTERVAL",
-        "sourceType": "NATIVE",
-        "value": 1,
-        "name": "Incidents",
-    }
-
-    # replaying the stored layer is allowed
-    query_context = mocker.MagicMock()
-    query_context.slice_.id = 42
-    query_context.slice_.query_context = None
-    query_context.slice_.params_dict = {
-        "metrics": stored_metrics,
-        "annotation_layers": [layer],
-    }
-    query_context.form_data = {
-        "slice_id": 42,
-        "metrics": stored_metrics,
-        "annotation_layers": [layer],
-    }
-    query_context.queries = [
-        QueryObject(metrics=stored_metrics, annotation_layers=[layer])  # type: ignore
-    ]
-    assert not query_context_modified(query_context)
-
-    # injecting a layer the chart was not saved with is tampering
-    injected = {**layer, "value": 2}
-    query_context.slice_.params_dict = {
-        "metrics": stored_metrics,
-    }
-    query_context.form_data = {
-        "slice_id": 42,
-        "metrics": stored_metrics,
-        "annotation_layers": [injected],
-    }
-    query_context.queries = [
-        QueryObject(metrics=stored_metrics, annotation_layers=[injected])  # type: ignore
-    ]
-    assert query_context_modified(query_context)
-
-
-def test_query_context_modified_result_type_expansion(
-    mocker: MockerFixture,
-    stored_metrics: list[AdhocMetric],
-) -> None:
-    """
-    Requesting the ``samples``/``drill_detail`` result types is tampering:
-    the server-side preparers expand those queries to every datasource
-    column after the subset checks on columns/metrics have run.
-    """
-    query_context = mocker.MagicMock()
-    query_context.slice_.id = 42
-    query_context.slice_.query_context = None
-    query_context.slice_.params_dict = {
-        "metrics": stored_metrics,
-    }
-    query_context.form_data = {
-        "slice_id": 42,
-        "metrics": stored_metrics,
-    }
-    query_context.queries = [QueryObject(metrics=stored_metrics)]  # type: ignore
-
-    # Top-level result type rewritten to samples.
-    query_context.result_type = ChartDataResultType.SAMPLES
-    assert query_context_modified(query_context)
-
-    # Per-query result type rewritten to drill_detail.
-    query_context.result_type = ChartDataResultType.FULL
-    query_context.queries[0].result_type = ChartDataResultType.DRILL_DETAIL
-    assert query_context_modified(query_context)
-
-    # The chart's own result type is not tampering.
-    query_context.queries[0].result_type = None
-    assert not query_context_modified(query_context)
-
-
-def test_query_context_modified_result_type_per_query_position(
-    mocker: MockerFixture,
-    stored_metrics: list[AdhocMetric],
-) -> None:
-    """
-    A chart's queries are validated by position: only the query stored with
-    ``samples``/``drill_detail`` may request it, and swapping which query
-    index carries that result type is tampering even though the type itself
-    is used somewhere on the stored chart.
-    """
-    query_context = mocker.MagicMock()
-    query_context.slice_.id = 42
-    query_context.slice_.params_dict = {"metrics": stored_metrics}
-    query_context.slice_.query_context = json.dumps(
-        {
-            "result_type": "full",
-            "queries": [
-                {"metrics": stored_metrics, "result_type": "full"},
-                {"metrics": stored_metrics, "result_type": "samples"},
-            ],
-        }
-    )
-    query_context.form_data = {"slice_id": 42, "metrics": stored_metrics}
-    query_context.result_type = ChartDataResultType.FULL
-
-    # Replaying each query's own stored result type, in the stored order, is
-    # not tampering.
-    query_context.queries = [
-        QueryObject(
-            metrics=stored_metrics,  # type: ignore
-            result_type=ChartDataResultType.FULL,
-        ),
-        QueryObject(
-            metrics=stored_metrics,  # type: ignore
-            result_type=ChartDataResultType.SAMPLES,
-        ),
-    ]
-    assert not query_context_modified(query_context)
-
-    # Requesting `samples` for the query stored at index 0 - a position the
-    # chart never renders with raw datasource rows - is tampering, even
-    # though `samples` is the stored result type of a different query.
-    query_context.queries = [
-        QueryObject(
-            metrics=stored_metrics,  # type: ignore
-            result_type=ChartDataResultType.SAMPLES,
-        ),
-        QueryObject(
-            metrics=stored_metrics,  # type: ignore
-            result_type=ChartDataResultType.FULL,
-        ),
-    ]
-    assert query_context_modified(query_context)
-
-    # Neither the query nor the query context names a result type: nothing
-    # resolves to a row-expanding type, so this is not tampering.
-    query_context.result_type = None
-    query_context.queries = [
-        QueryObject(metrics=stored_metrics),  # type: ignore
-        QueryObject(metrics=stored_metrics),  # type: ignore
-    ]
-    assert not query_context_modified(query_context)
-
-
 def test_query_context_modified_singular_metric_param(
     mocker: MockerFixture,
 ) -> None:
@@ -1920,73 +1769,6 @@ def test_query_context_modified_chartless_non_native_filter_allowed(
     assert not query_context_modified(qc)
 
 
-def test_query_context_modified_drill_by_row_expanding_result_type_blocked(
-    mocker: MockerFixture,
-) -> None:
-    """
-    A Drill By request (``slice_id`` 0 sentinel + source ``chart_id``) is
-    rejected when it asks for a row-expanding result type: that would bypass
-    the drillable-dimension allowlist raise_for_access already checked for it
-    and return every column instead.
-    """
-    query = SimpleNamespace(
-        columns=[], metrics=[], groupby=["region"], result_type="samples"
-    )
-    qc = mocker.MagicMock()
-    qc.slice_ = None
-    qc.form_data = {
-        "dashboardId": 10,
-        "slice_id": 0,
-        "chart_id": 5,
-        "groupby": ["region"],
-    }
-    qc.queries = [query]
-    assert query_context_modified(qc)
-
-    query.result_type = "drill_detail"
-    assert query_context_modified(qc)
-
-
-def test_query_context_modified_drill_to_detail_row_expanding_result_type_allowed(
-    mocker: MockerFixture,
-) -> None:
-    """
-    Drill to Detail (no ``slice_id``/``chart_id`` at all) is unaffected: it is
-    already meant to expose every column of a dataset attached to the
-    dashboard, and is validated by raise_for_access rather than here.
-    """
-    query = SimpleNamespace(
-        columns=[], metrics=[], groupby=[], result_type="drill_detail"
-    )
-    qc = mocker.MagicMock()
-    qc.slice_ = None
-    qc.form_data = {"dashboardId": 10}
-    qc.queries = [query]
-    assert not query_context_modified(qc)
-
-
-def test_query_context_modified_native_filter_row_expanding_result_type_blocked(
-    mocker: MockerFixture,
-) -> None:
-    """
-    A native-filter request limited to its target column is still rejected
-    when it asks for a row-expanding result type (``samples``/``drill_detail``):
-    those preparers replace the column list with every datasource column,
-    which would bypass the target-column allowlist entirely.
-    """
-    query = SimpleNamespace(
-        columns=["region"], metrics=[], groupby=[], result_type="samples"
-    )
-    qc = _native_filter_ctx(mocker, [query])
-    assert query_context_modified(qc)
-
-    query = SimpleNamespace(
-        columns=["region"], metrics=[], groupby=[], result_type="drill_detail"
-    )
-    qc = _native_filter_ctx(mocker, [query])
-    assert query_context_modified(qc)
-
-
 def test_query_context_modified_native_filter_without_type_marker_blocked(
     mocker: MockerFixture,
 ) -> None:

tests/unit_tests/sql/parse_tests.py

@@ -2502,6 +2502,21 @@ LATERAL generate_series(1, value) AS i;
         ),
         # not really valid SQL, but let's roll with it
         ("SELECT * FROM my_table LIMIT invalid", "postgresql", None),
+        # A ClickHouse `LIMIT ... BY` caps rows per group, not overall, so it is
+        # not a row limit. sqlglot hangs the `BY` columns off the `Limit` node,
+        # or off the `Offset` node for the `OFFSET` / `m, n` spellings.
+        ("SELECT * FROM t ORDER BY id, val LIMIT 2 BY id", "clickhouse", None),
+        ("SELECT * FROM t ORDER BY id, val LIMIT 2 BY id, val", "clickhouse", None),
+        (
+            "SELECT * FROM t ORDER BY id, val LIMIT 2 OFFSET 1 BY id",
+            "clickhouse",
+            None,
+        ),
+        ("SELECT * FROM t ORDER BY id, val LIMIT 1, 2 BY id", "clickhouse", None),
+        # ... while a plain ClickHouse limit, with or without an offset, is.
+        ("SELECT * FROM t ORDER BY c LIMIT 555", "clickhouse", 555),
+        ("SELECT * FROM t LIMIT 5 OFFSET 3", "clickhouse", 5),
+        ("SELECT * FROM t LIMIT 3, 5", "clickhouse", 5),
     ],
 )
 def test_get_limit_value(sql: str, engine: str, expected: str) -> None:
@@ -2717,6 +2732,158 @@ LIMIT 1000
             LimitMethod.FETCH_MANY,
             "SELECT\n  *\nFROM birth_names\nLIMIT 555",
         ),
+        # A ClickHouse `LIMIT ... BY` shares the `limit`/`offset` slot with the
+        # row limit, so `FORCE_LIMIT` wraps instead of overwriting it.
+        (
+            "SELECT * FROM limit_by ORDER BY id, val LIMIT 2 BY id",
+            "clickhouse",
+            1001,
+            LimitMethod.FORCE_LIMIT,
+            """
+SELECT
+  *
+FROM (
+  SELECT
+    *
+  FROM limit_by
+  ORDER BY
+    id,
+    val
+  LIMIT 2 BY id
+)
+LIMIT 1001
+            """.strip(),
+        ),
+        (
+            "SELECT * FROM limit_by ORDER BY id, val LIMIT 2 BY id, val",
+            "clickhouse",
+            1001,
+            LimitMethod.FORCE_LIMIT,
+            """
+SELECT
+  *
+FROM (
+  SELECT
+    *
+  FROM limit_by
+  ORDER BY
+    id,
+    val
+  LIMIT 2 BY id, val
+)
+LIMIT 1001
+            """.strip(),
+        ),
+        # For `LIMIT n OFFSET m BY x` sqlglot hangs the `BY` columns off the
+        # `Offset` node instead, so the `limit` arg alone doesn't reveal them.
+        (
+            "SELECT * FROM limit_by ORDER BY id, val LIMIT 2 OFFSET 1 BY id",
+            "clickhouse",
+            1001,
+            LimitMethod.FORCE_LIMIT,
+            """
+SELECT
+  *
+FROM (
+  SELECT
+    *
+  FROM limit_by
+  ORDER BY
+    id,
+    val
+  LIMIT 2
+  OFFSET 1 BY id
+)
+LIMIT 1001
+            """.strip(),
+        ),
+        (
+            "SELECT * FROM limit_by ORDER BY id, val LIMIT 1, 2 BY id",
+            "clickhouse",
+            1001,
+            LimitMethod.FORCE_LIMIT,
+            """
+SELECT
+  *
+FROM (
+  SELECT
+    *
+  FROM limit_by
+  ORDER BY
+    id,
+    val
+  LIMIT 2
+  OFFSET 1 BY id
+)
+LIMIT 1001
+            """.strip(),
+        ),
+        # `WITH TOTALS` rides into the subquery untouched: ClickHouse keeps
+        # emitting the totals block for a wrapped query, so the cap really is
+        # the only thing the rewrite adds.
+        (
+            "SELECT id, count() AS c FROM limit_by "
+            "GROUP BY id WITH TOTALS ORDER BY id LIMIT 2 BY id",
+            "clickhouse",
+            1001,
+            LimitMethod.FORCE_LIMIT,
+            """
+SELECT
+  *
+FROM (
+  SELECT
+    id,
+    count() AS c
+  FROM limit_by
+  GROUP BY
+    id
+  WITH TOTALS
+  ORDER BY
+    id
+  LIMIT 2 BY id
+)
+LIMIT 1001
+            """.strip(),
+        ),
+        # `SETTINGS` and `FORMAT` do not survive a demotion into the subquery,
+        # so they move up onto the wrapper instead.
+        (
+            "SELECT * FROM limit_by ORDER BY id LIMIT 2 BY id "
+            "SETTINGS extremes = 1 FORMAT JSONCompact",
+            "clickhouse",
+            1001,
+            LimitMethod.FORCE_LIMIT,
+            """
+SELECT
+  *
+FROM (
+  SELECT
+    *
+  FROM limit_by
+  ORDER BY
+    id
+  LIMIT 2 BY id
+)
+LIMIT 1001
+SETTINGS extremes = 1
+FORMAT JSONCompact
+            """.strip(),
+        ),
+        # A ClickHouse limit without a `BY` still takes the in-place path.
+        (
+            "SELECT * FROM t ORDER BY c LIMIT 555",
+            "clickhouse",
+            1001,
+            LimitMethod.FORCE_LIMIT,
+            "SELECT\n  *\nFROM t\nORDER BY\n  c\nLIMIT 1001",
+        ),
+        (
+            "SELECT * FROM t LIMIT 5 OFFSET 3",
+            "clickhouse",
+            1001,
+            LimitMethod.FORCE_LIMIT,
+            "SELECT\n  *\nFROM t\nLIMIT 1001\nOFFSET 3",
+        ),
     ],
 )
 def test_set_limit_value(
@@ -2731,6 +2898,88 @@ def test_set_limit_value(
     assert statement.format() == expected
 
 
+@pytest.mark.parametrize("engine", ["clickhouse", "clickhousedb"])
+@pytest.mark.parametrize(
+    "sql",
+    [
+        "SELECT * FROM limit_by ORDER BY id, val LIMIT 2 BY id",
+        "SELECT * FROM limit_by ORDER BY id, val LIMIT 2 BY id, val",
+        "SELECT * FROM limit_by ORDER BY id, val LIMIT 2 OFFSET 1 BY id",
+        "SELECT * FROM limit_by ORDER BY id, val LIMIT 1, 2 BY id",
+    ],
+)
+def test_set_limit_value_preserves_clickhouse_limit_by(sql: str, engine: str) -> None:
+    """
+    A row limit must not cannibalize a ClickHouse ``LIMIT ... BY``.
+
+    ``LIMIT 2 BY id`` keeps 2 rows *per id*; ``FORCE_LIMIT`` used to build a
+    fresh ``Limit`` node over ``args["limit"]``, dropping the ``BY`` columns and
+    turning the query into a flat ``LIMIT 1001`` -- a different result set, with
+    no error to hint at it. ``get_limit_value()`` reported the per-group 2 as a
+    row cap on top of that, so ``_set_query_limit()`` clamped the query to 2 rows.
+
+    The cap can't simply be appended next to the ``BY`` either: sqlglot cannot
+    parse ClickHouse's own ``LIMIT n BY x LIMIT m`` ("Found multiple 'LIMIT'
+    clauses"), so the result would not survive a reparse. Wrapping the query is
+    what keeps both the grouping and the cap.
+    """
+    statement = SQLStatement(sql, engine)
+    assert statement.get_limit_value() is None
+
+    statement.set_limit_value(1001, LimitMethod.FORCE_LIMIT)
+    limited = statement.format()
+
+    assert "BY id" in limited
+    assert limited.endswith("LIMIT 1001")
+    # The rewrite has to be valid ClickHouse, not just valid-looking.
+    assert SQLStatement(limited, engine).format() == limited
+
+
+def test_set_limit_value_keeps_clickhouse_top_level_modifiers() -> None:
+    """
+    The wrap must not demote clauses that only work at the top level.
+
+    ClickHouse rejects `FORMAT` inside a subquery outright, and a `SETTINGS`
+    attached to a subquery binds to that subquery alone -- top-level-only
+    settings such as ``extremes`` would silently stop applying. Both therefore
+    move onto the wrapper, which is where the original query had them.
+
+    The row-producing modifiers are left alone, because ClickHouse honors them
+    inside a `FROM` subquery: a wrapped `WITH TOTALS` query still emits its
+    totals block, and `WITH ROLLUP`/`WITH CUBE` still emit their extra rows.
+    Hoisting those would change the result rather than preserve it.
+    """
+    statement = SQLStatement(
+        "SELECT id, count() AS c FROM limit_by "
+        "GROUP BY id WITH TOTALS ORDER BY id LIMIT 2 BY id "
+        "SETTINGS extremes = 1 FORMAT JSONCompact",
+        "clickhouse",
+    )
+    statement.set_limit_value(1001, LimitMethod.FORCE_LIMIT)
+    limited = statement.format()
+
+    assert limited.endswith("LIMIT 1001\nSETTINGS extremes = 1\nFORMAT JSONCompact")
+    # `WITH TOTALS` stays with the aggregation it belongs to.
+    assert "WITH TOTALS\n" in limited.split("LIMIT 2 BY id")[0]
+    assert SQLStatement(limited, "clickhouse").format() == limited
+
+
+@pytest.mark.parametrize(
+    "engine", ["clickhouse", "clickhousedb", "postgresql", "mysql"]
+)
+def test_set_limit_value_without_limit_by_stays_in_place(engine: str) -> None:
+    """
+    Queries with no ``LIMIT ... BY`` keep the cheaper in-place rewrite.
+
+    The wrap is reserved for the ``LIMIT ... BY`` case; everything else -- every
+    non-ClickHouse dialect, and ClickHouse's own plain ``LIMIT`` -- must still
+    have its limit replaced without gaining a subquery.
+    """
+    statement = SQLStatement("SELECT * FROM t ORDER BY c LIMIT 555", engine)
+    statement.set_limit_value(1001, LimitMethod.FORCE_LIMIT)
+    assert statement.format() == "SELECT\n  *\nFROM t\nORDER BY\n  c\nLIMIT 1001"
+
+
 @pytest.mark.parametrize(
     "method",
     [LimitMethod.FORCE_LIMIT, LimitMethod.WRAP_SQL],